Repository: dragonflyoss/Dragonfly2
Pull request #1600 (2022-08-26 15:47:40)

feat: refactor peertask option
<!--- Provide a general summary of your changes in the Title above -->

## Description

<!--- Describe your changes in detail -->

## Related Issue

<!--- This project only accepts pull requests related to open issues -->
<!--- If suggesting a new feature or change, please discuss it in an issue first -->
<!--- If fixing a bug, there should be an issue describing it with steps to reproduce -->
<!--- Please link to the issue here: -->

## Motivation and Context

<!--- Why is this change required? What problem does it solve? -->

## Screenshots (if appropriate)

## Types of changes

<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)
- [ ] Documentation Update (if none of the other choices apply)

## Checklist

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [ ] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1600](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (42ce8b7) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/c1c2d5e4f0a1a6882cdbea8bfe5c0cd9645f4e96?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (c1c2d5e) will **increase** coverage by `0.03%`.
> The diff coverage is `56.09%`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600/graphs/tree.svg?width=650&height=150&src=pr&token=PUICJYSAT7&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main    #1600      +/-   ##
==========================================
+ Coverage   42.39%   42.42%   +0.03%     
==========================================
  Files         120      120              
  Lines       11353    11339      -14     
==========================================
- Hits         4813     4811       -2     
+ Misses       6029     6018      -11     
+ Partials      511      510       -1     
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `42.42% <56.09%> (+0.03%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [client/config/peerhost.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2NvbmZpZy9wZWVyaG9zdC5nbw==) | `38.25% <ø> (ø)` | |
| [client/config/peerhost\_linux.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2NvbmZpZy9wZWVyaG9zdF9saW51eC5nbw==) | `0.00% <0.00%> (ø)` | |
| [client/daemon/peer/piece\_manager.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BpZWNlX21hbmFnZXIuZ28=) | `52.71% <0.00%> (ø)` | |
| [client/daemon/peer/peertask\_manager.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX21hbmFnZXIuZ28=) | `26.06% <9.52%> (+1.76%)` | :arrow_up: |
| [client/daemon/peer/peertask\_stream.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3N0cmVhbS5nbw==) | `59.47% <33.33%> (ø)` | |
| [...ent/daemon/peer/peertask\_piecetask\_synchronizer.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19zeW5jaHJvbml6ZXIuZ28=) | `48.36% <71.42%> (+0.84%)` | :arrow_up: |
| [client/daemon/peer/peertask\_conductor.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2NvbmR1Y3Rvci5nbw==) | `55.93% <73.33%> (-0.22%)` | :arrow_down: |
| [client/daemon/peer/peertask\_reuse.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3JldXNlLmdv) | `68.44% <83.33%> (ø)` | |
| [client/daemon/peer/peertask\_file.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1600/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2ZpbGUuZ28=) | `57.25% <100.00%> (ø)` | |


Patch:
diff --git a/client/config/peerhost.go b/client/config/peerhost.go
index 0b4b970e..1c217634 100644
--- a/client/config/peerhost.go
+++ b/client/config/peerhost.go
@@ -28,7 +28,6 @@ import (
 	"os"
 	"path/filepath"
 	"regexp"
-	"strings"
 	"time"
 
 	"gopkg.in/yaml.v3"
@@ -40,6 +39,7 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/dfnet"
 	netip "d7y.io/dragonfly/v2/pkg/net/ip"
+	"d7y.io/dragonfly/v2/pkg/serialize"
 	"d7y.io/dragonfly/v2/pkg/unit"
 )
 
@@ -174,9 +174,9 @@ func ConvertPattern(p string, defaultPattern commonv1.Pattern) commonv1.Pattern
 }
 
 type GlobalSecurityOption struct {
-	AutoIssueCert bool       `mapstructure:"autoIssueCert" yaml:"autoIssueCert"`
-	CACert        PEMContent `mapstructure:"caCert" yaml:"caCert"`
-	TLSVerify     bool       `mapstructure:"tlsVerify" yaml:"tlsVerify"`
+	AutoIssueCert bool                 `mapstructure:"autoIssueCert" yaml:"autoIssueCert"`
+	CACert        serialize.PEMContent `mapstructure:"caCert" yaml:"caCert"`
+	TLSVerify     bool                 `mapstructure:"tlsVerify" yaml:"tlsVerify"`
 }
 
 type SchedulerOption struct {
@@ -527,12 +527,12 @@ type UnixListenOption struct {
 
 type SecurityOption struct {
 	// Insecure indicate enable tls or not
-	Insecure  bool        `mapstructure:"insecure" yaml:"insecure"`
-	CACert    PEMContent  `mapstructure:"caCert" yaml:"caCert"`
-	Cert      PEMContent  `mapstructure:"cert" yaml:"cert"`
-	Key       PEMContent  `mapstructure:"key" yaml:"key"`
-	TLSVerify bool        `mapstructure:"tlsVerify" yaml:"tlsVerify"`
-	TLSConfig *tls.Config `mapstructure:"tlsConfig" yaml:"tlsConfig"`
+	Insecure  bool                 `mapstructure:"insecure" yaml:"insecure"`
+	CACert    serialize.PEMContent `mapstructure:"caCert" yaml:"caCert"`
+	Cert      serialize.PEMContent `mapstructure:"cert" yaml:"cert"`
+	Key       serialize.PEMContent `mapstructure:"key" yaml:"key"`
+	TLSVerify bool                 `mapstructure:"tlsVerify" yaml:"tlsVerify"`
+	TLSConfig *tls.Config          `mapstructure:"tlsConfig" yaml:"tlsConfig"`
 }
 
 type StorageOption struct {
@@ -562,58 +562,10 @@ type ReloadOption struct {
 	Interval util.Duration `mapstructure:"interval" yaml:"interval"`
 }
 
-// PEMContent supports load PEM format from file or just inline PEM format content
-type PEMContent string
-
-func (p *PEMContent) UnmarshalJSON(b []byte) error {
-	var s string
-	err := json.Unmarshal(b, &s)
-	if err != nil {
-		return err
-	}
-
-	return p.loadPEM(s)
-}
-
-func (p *PEMContent) UnmarshalYAML(node *yaml.Node) error {
-	var s string
-	switch node.Kind {
-	case yaml.ScalarNode:
-		if err := node.Decode(&s); err != nil {
-			return err
-		}
-	default:
-		return errors.New("invalid pem content")
-	}
-
-	return p.loadPEM(s)
-}
-
-func (p *PEMContent) loadPEM(content string) error {
-	if content == "" {
-		*p = PEMContent("")
-		return nil
-	}
-	// inline PEM, just return
-	if strings.HasPrefix(strings.TrimSpace(content), "-----BEGIN ") {
-		val := strings.TrimSpace(content)
-		*p = PEMContent(val)
-		return nil
-	}
-
-	file, err := os.ReadFile(content)
-	if err != nil {
-		return err
-	}
-	val := strings.TrimSpace(string(file))
-	*p = PEMContent(val)
-	return nil
-}
-
 type tlsConfigFiles struct {
-	Cert   PEMContent `yaml:"cert" json:"cert"`
-	Key    PEMContent `yaml:"key" json:"key"`
-	CACert PEMContent `yaml:"caCert" json:"caCert"`
+	Cert   serialize.PEMContent `yaml:"cert" json:"cert"`
+	Key    serialize.PEMContent `yaml:"key" json:"key"`
+	CACert serialize.PEMContent `yaml:"caCert" json:"caCert"`
 }
 
 type TLSConfig struct {
diff --git a/client/daemon/daemon.go b/client/daemon/daemon.go
index 327e177d..1c7f7cc7 100644
--- a/client/daemon/daemon.go
+++ b/client/daemon/daemon.go
@@ -218,9 +218,25 @@ func New(opt *config.DaemonOption, d dfpath.Dfpath) (Daemon, error) {
 			return nil, err
 		}
 	}
-	peerTaskManager, err := peer.NewPeerTaskManager(host, pieceManager, storageManager, sched, opt.Scheduler,
-		opt.Download.PerPeerRateLimit.Limit, opt.Storage.Multiplex, opt.Download.Prefetch, opt.Download.CalculateDigest,
-		opt.Download.GetPiecesMaxRetry, opt.Download.WatchdogTimeout, credentials, opt.Download.GRPCDialTimeout)
+
+	peerTaskManagerOption := &peer.TaskManagerOption{
+		TaskOption: peer.TaskOption{
+			PeerHost:        host,
+			SchedulerOption: opt.Scheduler,
+			PieceManager:    pieceManager,
+			StorageManager:  storageManager,
+			WatchdogTimeout: opt.Download.WatchdogTimeout,
+			CalculateDigest: opt.Download.CalculateDigest,
+			GRPCCredentials: credentials,
+			GRPCDialTimeout: opt.Download.GRPCDialTimeout,
+		},
+		SchedulerClient:   sched,
+		PerPeerRateLimit:  opt.Download.PerPeerRateLimit.Limit,
+		Multiplex:         opt.Storage.Multiplex,
+		Prefetch:          opt.Download.Prefetch,
+		GetPiecesMaxRetry: opt.Download.GetPiecesMaxRetry,
+	}
+	peerTaskManager, err := peer.NewPeerTaskManager(peerTaskManagerOption)
 	if err != nil {
 		return nil, err
 	}
diff --git a/client/daemon/peer/peertask_conductor.go b/client/daemon/peer/peertask_conductor.go
index b02f715c..d236d64d 100644
--- a/client/daemon/peer/peertask_conductor.go
+++ b/client/daemon/peer/peertask_conductor.go
@@ -31,6 +31,7 @@ import (
 	"go.uber.org/atomic"
 	"golang.org/x/time/rate"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/status"
 
 	commonv1 "d7y.io/api/pkg/apis/common/v1"
@@ -67,8 +68,9 @@ var _ Task = (*peerTaskConductor)(nil)
 
 // peerTaskConductor will fetch all pieces from other peers and send pieces info to broker
 type peerTaskConductor struct {
+	TaskOption
 	*logger.SugaredLoggerOnWith
-	ptm *peerTaskManager
+
 	// ctx is with span info for tracing
 	// we use successCh and failCh mark task success or fail
 	ctx context.Context
@@ -77,8 +79,6 @@ type peerTaskConductor struct {
 	// when back source, cancel all piece download action
 	pieceDownloadCancel context.CancelFunc
 
-	// host info about current host
-	host *schedulerv1.PeerHost
 	// request is the original PeerTaskRequest
 	request *schedulerv1.PeerTaskRequest
 
@@ -86,15 +86,10 @@ type peerTaskConductor struct {
 	needBackSource *atomic.Bool
 	seed           bool
 
-	// pieceManager will be used for downloading piece
-	pieceManager    PieceManager
-	storageManager  storage.Manager
 	peerTaskManager *peerTaskManager
 
 	storage storage.TaskStorageDriver
 
-	// schedule options
-	schedulerOption config.SchedulerOption
 	schedulerClient schedulerclient.Client
 
 	// peer task meta info
@@ -171,6 +166,21 @@ type peerTaskConductor struct {
 	sourceErrorStatus *status.Status
 }
 
+type TaskOption struct {
+	// PeerHost info about current PeerHost
+	PeerHost *schedulerv1.PeerHost
+	// PieceManager will be used for downloading piece
+	PieceManager   PieceManager
+	StorageManager storage.Manager
+	// schedule options
+	SchedulerOption config.SchedulerOption
+	CalculateDigest bool
+	GRPCCredentials credentials.TransportCredentials
+	GRPCDialTimeout time.Duration
+	// WatchdogTimeout > 0 indicates to start watch dog for every single peer task
+	WatchdogTimeout time.Duration
+}
+
 func (ptm *peerTaskManager) newPeerTaskConductor(
 	ctx context.Context,
 	request *schedulerv1.PeerTaskRequest,
@@ -181,8 +191,8 @@ func (ptm *peerTaskManager) newPeerTaskConductor(
 	// use a new context with span info
 	ctx = trace.ContextWithSpan(context.Background(), trace.SpanFromContext(ctx))
 	ctx, span := tracer.Start(ctx, config.SpanPeerTask, trace.WithSpanKind(trace.SpanKindClient))
-	span.SetAttributes(config.AttributePeerHost.String(ptm.host.Id))
-	span.SetAttributes(semconv.NetHostIPKey.String(ptm.host.Ip))
+	span.SetAttributes(config.AttributePeerHost.String(ptm.PeerHost.Id))
+	span.SetAttributes(semconv.NetHostIPKey.String(ptm.PeerHost.Ip))
 	span.SetAttributes(config.AttributePeerID.String(request.PeerId))
 	span.SetAttributes(semconv.HTTPURLKey.String(request.Url))
 
@@ -210,15 +220,12 @@ func (ptm *peerTaskManager) newPeerTaskConductor(
 	span.SetAttributes(config.AttributeTaskID.String(taskID))
 
 	ptc := &peerTaskConductor{
-		ptm:                 ptm,
+		TaskOption:          ptm.TaskOption,
+		peerTaskManager:     ptm,
+		request:             request,
 		startTime:           time.Now(),
 		ctx:                 ctx,
 		broker:              newPieceBroker(),
-		host:                ptm.host,
-		request:             request,
-		pieceManager:        ptm.pieceManager,
-		storageManager:      ptm.storageManager,
-		peerTaskManager:     ptm,
 		peerPacketReady:     make(chan bool, 1),
 		peerID:              request.PeerId,
 		taskID:              taskID,
@@ -235,19 +242,17 @@ func (ptm *peerTaskManager) newPeerTaskConductor(
 		contentLength:       atomic.NewInt64(-1),
 		totalPiece:          atomic.NewInt32(-1),
 		digest:              atomic.NewString(""),
-		schedulerOption:     ptm.schedulerOption,
 		limiter:             rate.NewLimiter(limit, int(limit)),
 		completedLength:     atomic.NewInt64(0),
 		usedTraffic:         atomic.NewUint64(0),
 		SugaredLoggerOnWith: log,
 		seed:                seed,
-
-		parent: parent,
-		rg:     rg,
+		parent:              parent,
+		rg:                  rg,
 	}
 
 	ptc.pieceTaskPoller = &pieceTaskPoller{
-		getPiecesMaxRetry: ptm.getPiecesMaxRetry,
+		getPiecesMaxRetry: ptm.GetPiecesMaxRetry,
 		peerTaskConductor: ptc,
 	}
 
@@ -261,7 +266,7 @@ func (pt *peerTaskConductor) register() error {
 	pt.Debugf("request overview, pid: %s, url: %s, filter: %s, tag: %s, range: %s, digest: %s, header: %#v",
 		pt.request.PeerId, pt.request.Url, pt.request.UrlMeta.Filter, pt.request.UrlMeta.Tag, pt.request.UrlMeta.Range, pt.request.UrlMeta.Digest, pt.request.UrlMeta.Header)
 	// trace register
-	regCtx, cancel := context.WithTimeout(pt.ctx, pt.peerTaskManager.schedulerOption.ScheduleTimeout.Duration)
+	regCtx, cancel := context.WithTimeout(pt.ctx, pt.SchedulerOption.ScheduleTimeout.Duration)
 	defer cancel()
 	regCtx, regSpan := tracer.Start(regCtx, config.SpanRegisterTask)
 
@@ -273,7 +278,7 @@ func (pt *peerTaskConductor) register() error {
 	)
 
 	pt.Infof("step 1: peer %s start to register", pt.request.PeerId)
-	pt.schedulerClient = pt.peerTaskManager.schedulerClient
+	pt.schedulerClient = pt.peerTaskManager.SchedulerClient
 
 	result, err := pt.schedulerClient.RegisterPeerTask(regCtx, pt.request)
 	regSpan.RecordError(err)
@@ -281,10 +286,10 @@ func (pt *peerTaskConductor) register() error {
 
 	if err != nil {
 		if err == context.DeadlineExceeded {
-			pt.Errorf("scheduler did not response in %s", pt.peerTaskManager.schedulerOption.ScheduleTimeout.Duration)
+			pt.Errorf("scheduler did not response in %s", pt.SchedulerOption.ScheduleTimeout.Duration)
 		}
 		pt.Errorf("step 1: peer %s register failed: %s", pt.request.PeerId, err)
-		if pt.peerTaskManager.schedulerOption.DisableAutoBackSource {
+		if pt.SchedulerOption.DisableAutoBackSource {
 			// when peer register failed, some actions need to do with peerPacketStream
 			pt.peerPacketStream = &dummyPeerPacketStream{}
 			pt.Errorf("register peer task failed: %s, peer id: %s, auto back source disabled", err, pt.request.PeerId)
@@ -494,8 +499,8 @@ func (pt *peerTaskConductor) markBackSource() {
 		MainPeer:      nil,
 		CandidatePeers: []*schedulerv1.PeerPacket_DestPeer{
 			{
-				Ip:      pt.host.Ip,
-				RpcPort: pt.host.RpcPort,
+				Ip:      pt.PeerHost.Ip,
+				RpcPort: pt.PeerHost.RpcPort,
 				PeerId:  pt.peerID,
 			},
 		},
@@ -519,7 +524,7 @@ func (pt *peerTaskConductor) backSource() {
 
 	ctx, span := tracer.Start(pt.ctx, config.SpanBackSource)
 	pt.SetContentLength(-1)
-	err := pt.pieceManager.DownloadSource(ctx, pt, pt.request, pt.rg)
+	err := pt.PieceManager.DownloadSource(ctx, pt, pt.request, pt.rg)
 	if err != nil {
 		pt.Errorf("download from source error: %s", err)
 		span.SetAttributes(config.AttributePeerTaskSuccess.Bool(false))
@@ -581,7 +586,7 @@ func (pt *peerTaskConductor) storeTinyPeerTask() {
 	pt.SetTotalPieces(1)
 	ctx := pt.ctx
 	var err error
-	storageDriver, err := pt.peerTaskManager.storageManager.RegisterTask(ctx,
+	storageDriver, err := pt.StorageManager.RegisterTask(ctx,
 		&storage.RegisterTaskRequest{
 			PeerTaskMetadata: storage.PeerTaskMetadata{
 				PeerID: pt.tinyData.PeerID,
@@ -865,7 +870,7 @@ func (pt *peerTaskConductor) pullSinglePiece() {
 		DstAddr: pt.singlePiece.DstAddr,
 	}
 
-	if result, err := pt.pieceManager.DownloadPiece(ctx, request); err == nil {
+	if result, err := pt.PieceManager.DownloadPiece(ctx, request); err == nil {
 		pt.reportSuccessResult(request, result)
 		pt.PublishPieceInfo(request.piece.PieceNum, request.piece.RangeSize)
 
@@ -1045,8 +1050,8 @@ func (pt *peerTaskConductor) waitFirstPeerPacket() (done bool, backSource bool)
 		pt.span.AddEvent("back source due to scheduler says need back source")
 		pt.backSource()
 		return false, true
-	case <-time.After(pt.schedulerOption.ScheduleTimeout.Duration):
-		if pt.schedulerOption.DisableAutoBackSource {
+	case <-time.After(pt.SchedulerOption.ScheduleTimeout.Duration):
+		if pt.SchedulerOption.DisableAutoBackSource {
 			pt.cancel(commonv1.Code_ClientScheduleTimeout, reasonBackSourceDisabled)
 			err := fmt.Errorf("%s, auto back source disabled", pt.failedReason)
 			pt.span.RecordError(err)
@@ -1212,8 +1217,8 @@ func (pt *peerTaskConductor) downloadPiece(workerID int32, request *DownloadPiec
 		"dest peer id: %s, piece num: %d, range start: %d, range size: %d",
 		workerID, request.DstPid, request.piece.PieceNum, request.piece.RangeStart, request.piece.RangeSize)
 	// download piece
-	// result is always not nil, pieceManager will report begin and end time
-	result, err := pt.pieceManager.DownloadPiece(ctx, request)
+	// result is always not nil, PieceManager will report begin and end time
+	result, err := pt.PieceManager.DownloadPiece(ctx, request)
 	if err != nil {
 		pt.ReportPieceResult(request, result, err)
 		span.SetAttributes(config.AttributePieceSuccess.Bool(false))
@@ -1384,7 +1389,7 @@ func (pt *peerTaskConductor) reportSuccessResult(request *DownloadPieceRequest,
 			EndTime:       uint64(result.FinishTime),
 			Success:       true,
 			Code:          commonv1.Code_Success,
-			HostLoad:      nil, // TODO(jim): update host load
+			HostLoad:      nil, // TODO(jim): update PeerHost load
 			FinishedCount: pt.readyPieces.Settled(),
 			// TODO range_start, range_size, piece_md5, piece_offset, piece_style
 		})
@@ -1422,7 +1427,7 @@ func (pt *peerTaskConductor) reportFailResult(request *DownloadPieceRequest, res
 func (pt *peerTaskConductor) initStorage(desiredLocation string) (err error) {
 	// prepare storage
 	if pt.parent == nil {
-		pt.storage, err = pt.storageManager.RegisterTask(pt.ctx,
+		pt.storage, err = pt.StorageManager.RegisterTask(pt.ctx,
 			&storage.RegisterTaskRequest{
 				PeerTaskMetadata: storage.PeerTaskMetadata{
 					PeerID: pt.GetPeerID(),
@@ -1434,7 +1439,7 @@ func (pt *peerTaskConductor) initStorage(desiredLocation string) (err error) {
 				PieceMd5Sign:    pt.GetPieceMd5Sign(),
 			})
 	} else {
-		pt.storage, err = pt.storageManager.RegisterSubTask(pt.ctx,
+		pt.storage, err = pt.StorageManager.RegisterSubTask(pt.ctx,
 			&storage.RegisterSubTaskRequest{
 				Parent: storage.PeerTaskMetadata{
 					PeerID: pt.parent.GetPeerID(),
@@ -1551,9 +1556,9 @@ func (pt *peerTaskConductor) done() {
 		&schedulerv1.PeerResult{
 			TaskId:          pt.GetTaskID(),
 			PeerId:          pt.GetPeerID(),
-			SrcIp:           pt.host.Ip,
-			SecurityDomain:  pt.peerTaskManager.host.SecurityDomain,
-			Idc:             pt.peerTaskManager.host.Idc,
+			SrcIp:           pt.PeerHost.Ip,
+			SecurityDomain:  pt.PeerHost.SecurityDomain,
+			Idc:             pt.PeerHost.Idc,
 			Url:             pt.request.Url,
 			ContentLength:   pt.GetContentLength(),
 			Traffic:         pt.GetTraffic(),
@@ -1623,9 +1628,9 @@ func (pt *peerTaskConductor) fail() {
 	peerResult := &schedulerv1.PeerResult{
 		TaskId:          pt.GetTaskID(),
 		PeerId:          pt.GetPeerID(),
-		SrcIp:           pt.peerTaskManager.host.Ip,
-		SecurityDomain:  pt.peerTaskManager.host.SecurityDomain,
-		Idc:             pt.peerTaskManager.host.Idc,
+		SrcIp:           pt.PeerHost.Ip,
+		SecurityDomain:  pt.PeerHost.SecurityDomain,
+		Idc:             pt.PeerHost.Idc,
 		Url:             pt.request.Url,
 		ContentLength:   pt.GetContentLength(),
 		Traffic:         pt.GetTraffic(),
@@ -1668,7 +1673,7 @@ func (pt *peerTaskConductor) Validate() error {
 		return err
 	}
 
-	if !pt.peerTaskManager.calculateDigest {
+	if !pt.CalculateDigest {
 		return nil
 	}
 	err = pt.GetStorage().ValidateDigest(
diff --git a/client/daemon/peer/peertask_file.go b/client/daemon/peer/peertask_file.go
index 111a0d03..c5a22a15 100644
--- a/client/daemon/peer/peertask_file.go
+++ b/client/daemon/peer/peertask_file.go
@@ -166,7 +166,7 @@ func (f *fileTask) syncProgress() {
 }
 
 func (f *fileTask) storeToOutput() {
-	err := f.peerTaskConductor.storageManager.Store(
+	err := f.peerTaskConductor.StorageManager.Store(
 		f.ctx,
 		&storage.StoreRequest{
 			CommonTaskRequest: storage.CommonTaskRequest{
diff --git a/client/daemon/peer/peertask_manager.go b/client/daemon/peer/peertask_manager.go
index 09469fe6..aff2bba7 100644
--- a/client/daemon/peer/peertask_manager.go
+++ b/client/daemon/peer/peertask_manager.go
@@ -24,19 +24,16 @@ import (
 	"fmt"
 	"io"
 	"sync"
-	"time"
 
 	"github.com/go-http-utils/headers"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/time/rate"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/status"
 
 	commonv1 "d7y.io/api/pkg/apis/common/v1"
 	schedulerv1 "d7y.io/api/pkg/apis/scheduler/v1"
 
-	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/client/daemon/metrics"
 	"d7y.io/dragonfly/v2/client/daemon/storage"
 	"d7y.io/dragonfly/v2/client/util"
@@ -121,63 +118,27 @@ func init() {
 }
 
 type peerTaskManager struct {
-	host            *schedulerv1.PeerHost
-	schedulerClient schedulerclient.Client
-	schedulerOption config.SchedulerOption
-	pieceManager    PieceManager
-	storageManager  storage.Manager
-
+	TaskManagerOption
 	conductorLock    sync.Locker
 	runningPeerTasks sync.Map
-
-	perPeerRateLimit rate.Limit
-
-	// enableMultiplex indicates to reuse the data of completed peer tasks
-	enableMultiplex bool
-	// enablePrefetch indicates to prefetch the whole files of ranged requests
-	enablePrefetch bool
-	// watchdogTimeout > 0 indicates to start watch dog for every single peer task
-	watchdogTimeout time.Duration
-
-	calculateDigest bool
-
-	getPiecesMaxRetry int
-
-	grpcCredentials credentials.TransportCredentials
-	grpcDialTimeout time.Duration
 }
 
-func NewPeerTaskManager(
-	host *schedulerv1.PeerHost,
-	pieceManager PieceManager,
-	storageManager storage.Manager,
-	schedulerClient schedulerclient.Client,
-	schedulerOption config.SchedulerOption,
-	perPeerRateLimit rate.Limit,
-	multiplex bool,
-	prefetch bool,
-	calculateDigest bool,
-	getPiecesMaxRetry int,
-	watchdog time.Duration,
-	grpcCredentials credentials.TransportCredentials,
-	grpcDialTimeout time.Duration) (TaskManager, error) {
+type TaskManagerOption struct {
+	TaskOption
+	SchedulerClient  schedulerclient.Client
+	PerPeerRateLimit rate.Limit
+	// Multiplex indicates to reuse the data of completed peer tasks
+	Multiplex bool
+	// Prefetch indicates to prefetch the whole files of ranged requests
+	Prefetch          bool
+	GetPiecesMaxRetry int
+}
 
+func NewPeerTaskManager(opt *TaskManagerOption) (TaskManager, error) {
 	ptm := &peerTaskManager{
-		host:              host,
+		TaskManagerOption: *opt,
 		runningPeerTasks:  sync.Map{},
 		conductorLock:     &sync.Mutex{},
-		pieceManager:      pieceManager,
-		storageManager:    storageManager,
-		schedulerClient:   schedulerClient,
-		schedulerOption:   schedulerOption,
-		perPeerRateLimit:  perPeerRateLimit,
-		enableMultiplex:   multiplex,
-		enablePrefetch:    prefetch,
-		watchdogTimeout:   watchdog,
-		calculateDigest:   calculateDigest,
-		getPiecesMaxRetry: getPiecesMaxRetry,
-		grpcCredentials:   grpcCredentials,
-		grpcDialTimeout:   grpcDialTimeout,
 	}
 	return ptm, nil
 }
@@ -254,14 +215,14 @@ func (ptm *peerTaskManager) getOrCreatePeerTaskConductor(
 }
 
 func (ptm *peerTaskManager) enabledPrefetch(rg *util.Range) bool {
-	return ptm.enablePrefetch && rg != nil
+	return ptm.Prefetch && rg != nil
 }
 
 func (ptm *peerTaskManager) prefetchParentTask(request *schedulerv1.PeerTaskRequest, desiredLocation string) *peerTaskConductor {
 	req := &schedulerv1.PeerTaskRequest{
 		Url:         request.Url,
 		PeerId:      request.PeerId,
-		PeerHost:    ptm.host,
+		PeerHost:    ptm.PeerHost,
 		HostLoad:    request.HostLoad,
 		IsMigrating: request.IsMigrating,
 		Pattern:     request.Pattern,
@@ -282,8 +243,8 @@ func (ptm *peerTaskManager) prefetchParentTask(request *schedulerv1.PeerTaskRequ
 	req.PeerId = idgen.PeerID(req.PeerHost.Ip)
 
 	var limit = rate.Inf
-	if ptm.perPeerRateLimit > 0 {
-		limit = ptm.perPeerRateLimit
+	if ptm.PerPeerRateLimit > 0 {
+		limit = ptm.PerPeerRateLimit
 	}
 
 	logger.Infof("prefetch peer task %s/%s", taskID, req.PeerId)
@@ -300,10 +261,10 @@ func (ptm *peerTaskManager) prefetchParentTask(request *schedulerv1.PeerTaskRequ
 }
 
 func (ptm *peerTaskManager) StartFileTask(ctx context.Context, req *FileTaskRequest) (chan *FileTaskProgress, *TinyData, error) {
-	if req.KeepOriginalOffset && !ptm.enablePrefetch {
+	if req.KeepOriginalOffset && !ptm.Prefetch {
 		return nil, nil, fmt.Errorf("please enable prefetch when use original offset feature")
 	}
-	if ptm.enableMultiplex {
+	if ptm.Multiplex {
 		progress, ok := ptm.tryReuseFilePeerTask(ctx, req)
 		if ok {
 			metrics.PeerTaskCacheHitCount.Add(1)
@@ -312,8 +273,8 @@ func (ptm *peerTaskManager) StartFileTask(ctx context.Context, req *FileTaskRequ
 	}
 	// TODO ensure scheduler is ok first
 	var limit = rate.Inf
-	if ptm.perPeerRateLimit > 0 {
-		limit = ptm.perPeerRateLimit
+	if ptm.PerPeerRateLimit > 0 {
+		limit = ptm.PerPeerRateLimit
 	}
 	if req.Limit > 0 {
 		limit = rate.Limit(req.Limit)
@@ -323,7 +284,7 @@ func (ptm *peerTaskManager) StartFileTask(ctx context.Context, req *FileTaskRequ
 		return nil, nil, err
 	}
 
-	// FIXME when failed due to schedulerClient error, relocate schedulerClient and retry
+	// FIXME when failed due to SchedulerClient error, relocate SchedulerClient and retry
 	progress, err := pt.Start(ctx)
 	return progress, nil, err
 }
@@ -333,13 +294,13 @@ func (ptm *peerTaskManager) StartStreamTask(ctx context.Context, req *StreamTask
 		Url:         req.URL,
 		UrlMeta:     req.URLMeta,
 		PeerId:      req.PeerID,
-		PeerHost:    ptm.host,
+		PeerHost:    ptm.PeerHost,
 		HostLoad:    nil,
 		IsMigrating: false,
 		Pattern:     req.Pattern,
 	}
 
-	if ptm.enableMultiplex {
+	if ptm.Multiplex {
 		r, attr, ok := ptm.tryReuseStreamPeerTask(ctx, req)
 		if ok {
 			metrics.PeerTaskCacheHitCount.Add(1)
@@ -352,7 +313,7 @@ func (ptm *peerTaskManager) StartStreamTask(ctx context.Context, req *StreamTask
 		return nil, nil, err
 	}
 
-	// FIXME when failed due to schedulerClient error, relocate schedulerClient and retry
+	// FIXME when failed due to SchedulerClient error, relocate SchedulerClient and retry
 	readCloser, attribute, err := pt.Start(ctx)
 	return readCloser, attribute, err
 }
@@ -365,8 +326,8 @@ func (ptm *peerTaskManager) StartSeedTask(ctx context.Context, req *SeedTaskRequ
 	}
 
 	var limit = rate.Inf
-	if ptm.perPeerRateLimit > 0 {
-		limit = ptm.perPeerRateLimit
+	if ptm.PerPeerRateLimit > 0 {
+		limit = ptm.PerPeerRateLimit
 	}
 	if req.Limit > 0 {
 		limit = rate.Limit(req.Limit)
@@ -427,26 +388,26 @@ func (ptm *peerTaskManager) StatTask(ctx context.Context, taskID string) (*sched
 		TaskId: taskID,
 	}
 
-	return ptm.schedulerClient.StatTask(ctx, req)
+	return ptm.SchedulerClient.StatTask(ctx, req)
 }
 
 func (ptm *peerTaskManager) GetPieceManager() PieceManager {
-	return ptm.pieceManager
+	return ptm.PieceManager
 }
 
 func (ptm *peerTaskManager) AnnouncePeerTask(ctx context.Context, meta storage.PeerTaskMetadata, url string, taskType commonv1.TaskType, urlMeta *commonv1.UrlMeta) error {
-	// Check if the given task is completed in local storageManager.
-	if ptm.storageManager.FindCompletedTask(meta.TaskID) == nil {
+	// Check if the given task is completed in local StorageManager.
+	if ptm.StorageManager.FindCompletedTask(meta.TaskID) == nil {
 		return errors.New("task not found in local storage")
 	}
 
 	// Prepare AnnounceTaskRequest.
-	totalPieces, err := ptm.storageManager.GetTotalPieces(ctx, &meta)
+	totalPieces, err := ptm.StorageManager.GetTotalPieces(ctx, &meta)
 	if err != nil {
 		return err
 	}
 
-	piecePacket, err := ptm.storageManager.GetPieces(ctx, &commonv1.PieceTaskRequest{
+	piecePacket, err := ptm.StorageManager.GetPieces(ctx, &commonv1.PieceTaskRequest{
 		TaskId:   meta.TaskID,
 		DstPid:   meta.PeerID,
 		StartNum: 0,
@@ -455,15 +416,15 @@ func (ptm *peerTaskManager) AnnouncePeerTask(ctx context.Context, meta storage.P
 	if err != nil {
 		return err
 	}
-	piecePacket.DstAddr = fmt.Sprintf("%s:%d", ptm.host.Ip, ptm.host.DownPort)
+	piecePacket.DstAddr = fmt.Sprintf("%s:%d", ptm.PeerHost.Ip, ptm.PeerHost.DownPort)
 
 	// Announce peer task to scheduler
-	if err := ptm.schedulerClient.AnnounceTask(ctx, &schedulerv1.AnnounceTaskRequest{
+	if err := ptm.SchedulerClient.AnnounceTask(ctx, &schedulerv1.AnnounceTaskRequest{
 		TaskId:      meta.TaskID,
 		TaskType:    taskType,
 		Url:         url,
 		UrlMeta:     urlMeta,
-		PeerHost:    ptm.host,
+		PeerHost:    ptm.PeerHost,
 		PiecePacket: piecePacket,
 	}); err != nil {
 		return err
diff --git a/client/daemon/peer/peertask_piecetask_synchronizer.go b/client/daemon/peer/peertask_piecetask_synchronizer.go
index 934deac9..69e59b98 100644
--- a/client/daemon/peer/peertask_piecetask_synchronizer.go
+++ b/client/daemon/peer/peertask_piecetask_synchronizer.go
@@ -170,13 +170,13 @@ func (s *pieceTaskSyncManager) newPieceTaskSynchronizer(
 	}
 
 	var credentialOpt grpc.DialOption
-	if s.peerTaskConductor.peerTaskManager.grpcCredentials != nil {
-		credentialOpt = grpc.WithTransportCredentials(s.peerTaskConductor.peerTaskManager.grpcCredentials)
+	if s.peerTaskConductor.GRPCCredentials != nil {
+		credentialOpt = grpc.WithTransportCredentials(s.peerTaskConductor.GRPCCredentials)
 	} else {
 		credentialOpt = grpc.WithTransportCredentials(insecure.NewCredentials())
 	}
 
-	dialCtx, cancel := context.WithTimeout(ctx, s.peerTaskConductor.peerTaskManager.grpcDialTimeout)
+	dialCtx, cancel := context.WithTimeout(ctx, s.peerTaskConductor.GRPCDialTimeout)
 	client, err := dfdaemonclient.GetClient(dialCtx, netAddr.String(), credentialOpt)
 	cancel()
 
@@ -226,7 +226,7 @@ func (s *pieceTaskSyncManager) newMultiPieceTaskSynchronizer(
 	desiredPiece int32) (legacyPeers []*schedulerv1.PeerPacket_DestPeer) {
 	s.Lock()
 	defer func() {
-		if s.peerTaskConductor.ptm.watchdogTimeout > 0 {
+		if s.peerTaskConductor.WatchdogTimeout > 0 {
 			s.resetWatchdog(destPeers[0])
 		}
 		s.Unlock()
@@ -275,7 +275,7 @@ func (s *pieceTaskSyncManager) resetWatchdog(mainPeer *schedulerv1.PeerPacket_De
 	}
 	s.watchdog.mainPeer.Store(mainPeer)
 	s.peerTaskConductor.Infof("start new watchdog")
-	go s.watchdog.watch(s.peerTaskConductor.ptm.watchdogTimeout)
+	go s.watchdog.watch(s.peerTaskConductor.WatchdogTimeout)
 }
 
 func compositePieceResult(peerTaskConductor *peerTaskConductor, destPeer *schedulerv1.PeerPacket_DestPeer, code commonv1.Code) *schedulerv1.PieceResult {
diff --git a/client/daemon/peer/peertask_reuse.go b/client/daemon/peer/peertask_reuse.go
index 0ad1ba58..729c53c7 100644
--- a/client/daemon/peer/peertask_reuse.go
+++ b/client/daemon/peer/peertask_reuse.go
@@ -58,9 +58,9 @@ func (ptm *peerTaskManager) tryReuseFilePeerTask(ctx context.Context,
 	)
 
 	if ptm.enabledPrefetch(request.Range) {
-		reuse = ptm.storageManager.FindCompletedSubTask(taskID)
+		reuse = ptm.StorageManager.FindCompletedSubTask(taskID)
 	} else {
-		reuse = ptm.storageManager.FindCompletedTask(taskID)
+		reuse = ptm.StorageManager.FindCompletedTask(taskID)
 	}
 
 	if reuse == nil {
@@ -70,7 +70,7 @@ func (ptm *peerTaskManager) tryReuseFilePeerTask(ctx context.Context,
 		// for ranged request, check the parent task
 		reuseRange = request.Range
 		taskID = idgen.ParentTaskID(request.Url, request.UrlMeta)
-		reuse = ptm.storageManager.FindPartialCompletedTask(taskID, reuseRange)
+		reuse = ptm.StorageManager.FindPartialCompletedTask(taskID, reuseRange)
 		if reuse == nil {
 			return nil, false
 		}
@@ -97,8 +97,8 @@ func (ptm *peerTaskManager) tryReuseFilePeerTask(ctx context.Context,
 	}
 
 	_, span := tracer.Start(ctx, config.SpanReusePeerTask, trace.WithSpanKind(trace.SpanKindClient))
-	span.SetAttributes(config.AttributePeerHost.String(ptm.host.Id))
-	span.SetAttributes(semconv.NetHostIPKey.String(ptm.host.Ip))
+	span.SetAttributes(config.AttributePeerHost.String(ptm.PeerHost.Id))
+	span.SetAttributes(semconv.NetHostIPKey.String(ptm.PeerHost.Ip))
 	span.SetAttributes(config.AttributeTaskID.String(taskID))
 	span.SetAttributes(config.AttributePeerID.String(request.PeerId))
 	span.SetAttributes(config.AttributeReusePeerID.String(reuse.PeerID))
@@ -124,7 +124,7 @@ func (ptm *peerTaskManager) tryReuseFilePeerTask(ctx context.Context,
 			TotalPieces:    reuse.TotalPieces,
 			OriginalOffset: request.KeepOriginalOffset,
 		}
-		err = ptm.storageManager.Store(ctx, storeRequest)
+		err = ptm.StorageManager.Store(ctx, storeRequest)
 	} else {
 		err = ptm.storePartialFile(ctx, request, log, reuse, reuseRange)
 	}
@@ -169,7 +169,7 @@ func (ptm *peerTaskManager) storePartialFile(ctx context.Context, request *FileT
 		log.Errorf("open dest file error when reuse peer task: %s", err)
 		return err
 	}
-	rc, err := ptm.storageManager.ReadAllPieces(ctx,
+	rc, err := ptm.StorageManager.ReadAllPieces(ctx,
 		&storage.ReadAllPiecesRequest{PeerTaskMetadata: reuse.PeerTaskMetadata, Range: rg})
 	if err != nil {
 		log.Errorf("read pieces error when reuse peer task: %s", err)
@@ -199,9 +199,9 @@ func (ptm *peerTaskManager) tryReuseStreamPeerTask(ctx context.Context,
 	)
 
 	if ptm.enabledPrefetch(request.Range) {
-		reuse = ptm.storageManager.FindCompletedSubTask(taskID)
+		reuse = ptm.StorageManager.FindCompletedSubTask(taskID)
 	} else {
-		reuse = ptm.storageManager.FindCompletedTask(taskID)
+		reuse = ptm.StorageManager.FindCompletedTask(taskID)
 	}
 
 	if reuse == nil {
@@ -211,7 +211,7 @@ func (ptm *peerTaskManager) tryReuseStreamPeerTask(ctx context.Context,
 		// for ranged request, check the parent task
 		reuseRange = request.Range
 		taskID = idgen.ParentTaskID(request.URL, request.URLMeta)
-		reuse = ptm.storageManager.FindPartialCompletedTask(taskID, reuseRange)
+		reuse = ptm.StorageManager.FindPartialCompletedTask(taskID, reuseRange)
 		if reuse == nil {
 			return nil, nil, false
 		}
@@ -237,8 +237,8 @@ func (ptm *peerTaskManager) tryReuseStreamPeerTask(ctx context.Context,
 	}
 
 	ctx, span := tracer.Start(ctx, config.SpanStreamTask, trace.WithSpanKind(trace.SpanKindClient))
-	span.SetAttributes(config.AttributePeerHost.String(ptm.host.Id))
-	span.SetAttributes(semconv.NetHostIPKey.String(ptm.host.Ip))
+	span.SetAttributes(config.AttributePeerHost.String(ptm.PeerHost.Id))
+	span.SetAttributes(semconv.NetHostIPKey.String(ptm.PeerHost.Ip))
 	span.SetAttributes(config.AttributeTaskID.String(taskID))
 	span.SetAttributes(config.AttributePeerID.String(request.PeerID))
 	span.SetAttributes(config.AttributeReusePeerID.String(reuse.PeerID))
@@ -248,7 +248,7 @@ func (ptm *peerTaskManager) tryReuseStreamPeerTask(ctx context.Context,
 	}
 	defer span.End()
 
-	rc, err := ptm.storageManager.ReadAllPieces(ctx,
+	rc, err := ptm.StorageManager.ReadAllPieces(ctx,
 		&storage.ReadAllPiecesRequest{PeerTaskMetadata: reuse.PeerTaskMetadata, Range: reuseRange})
 	if err != nil {
 		log.Errorf("read pieces error when reuse peer task: %s", err)
@@ -257,7 +257,7 @@ func (ptm *peerTaskManager) tryReuseStreamPeerTask(ctx context.Context,
 		return nil, nil, false
 	}
 
-	exa, err := ptm.storageManager.GetExtendAttribute(ctx, &reuse.PeerTaskMetadata)
+	exa, err := ptm.StorageManager.GetExtendAttribute(ctx, &reuse.PeerTaskMetadata)
 	if err != nil {
 		log.Errorf("get extend attribute error when reuse peer task: %s", err)
 		span.SetAttributes(config.AttributePeerTaskSuccess.Bool(false))
@@ -305,9 +305,9 @@ func (ptm *peerTaskManager) tryReuseSeedPeerTask(ctx context.Context,
 	)
 
 	if ptm.enabledPrefetch(request.Range) {
-		reuse = ptm.storageManager.FindCompletedSubTask(taskID)
+		reuse = ptm.StorageManager.FindCompletedSubTask(taskID)
 	} else {
-		reuse = ptm.storageManager.FindCompletedTask(taskID)
+		reuse = ptm.StorageManager.FindCompletedTask(taskID)
 	}
 
 	if reuse == nil {
@@ -320,7 +320,7 @@ func (ptm *peerTaskManager) tryReuseSeedPeerTask(ctx context.Context,
 		// for ranged request, check the parent task
 		//reuseRange = request.Range
 		//taskID = idgen.ParentTaskID(request.Url, request.UrlMeta)
-		//reuse = ptm.storageManager.FindPartialCompletedTask(taskID, reuseRange)
+		//reuse = ptm.StorageManager.FindPartialCompletedTask(taskID, reuseRange)
 		//if reuse == nil {
 		//	return nil, false
 		//}
@@ -337,8 +337,8 @@ func (ptm *peerTaskManager) tryReuseSeedPeerTask(ctx context.Context,
 	}
 
 	ctx, span := tracer.Start(ctx, config.SpanReusePeerTask, trace.WithSpanKind(trace.SpanKindClient))
-	span.SetAttributes(config.AttributePeerHost.String(ptm.host.Id))
-	span.SetAttributes(semconv.NetHostIPKey.String(ptm.host.Ip))
+	span.SetAttributes(config.AttributePeerHost.String(ptm.PeerHost.Id))
+	span.SetAttributes(semconv.NetHostIPKey.String(ptm.PeerHost.Ip))
 	span.SetAttributes(config.AttributeTaskID.String(taskID))
 	span.SetAttributes(config.AttributePeerID.String(request.PeerId))
 	span.SetAttributes(config.AttributeReusePeerID.String(reuse.PeerID))
diff --git a/client/daemon/peer/peertask_stream.go b/client/daemon/peer/peertask_stream.go
index 2c779cd1..713724c0 100644
--- a/client/daemon/peer/peertask_stream.go
+++ b/client/daemon/peer/peertask_stream.go
@@ -71,8 +71,8 @@ func (ptm *peerTaskManager) newStreamTask(
 	rg *util.Range) (*streamTask, error) {
 	metrics.StreamTaskCount.Add(1)
 	var limit = rate.Inf
-	if ptm.perPeerRateLimit > 0 {
-		limit = ptm.perPeerRateLimit
+	if ptm.PerPeerRateLimit > 0 {
+		limit = ptm.PerPeerRateLimit
 	}
 
 	// prefetch parent request
@@ -131,7 +131,7 @@ func (s *streamTask) Start(ctx context.Context) (io.ReadCloser, map[string]strin
 				attr[k] = v
 			}
 		}
-		rc, err := s.peerTaskConductor.peerTaskManager.storageManager.ReadAllPieces(
+		rc, err := s.peerTaskConductor.StorageManager.ReadAllPieces(
 			ctx,
 			&storage.ReadAllPiecesRequest{
 				PeerTaskMetadata: storage.PeerTaskMetadata{
diff --git a/cmd/dependency/dependency.go b/cmd/dependency/dependency.go
index 9488ee02..a1538cc3 100644
--- a/cmd/dependency/dependency.go
+++ b/cmd/dependency/dependency.go
@@ -50,6 +50,7 @@ import (
 	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/net/fqdn"
 	"d7y.io/dragonfly/v2/pkg/net/ip"
+	"d7y.io/dragonfly/v2/pkg/serialize"
 	"d7y.io/dragonfly/v2/pkg/unit"
 	"d7y.io/dragonfly/v2/version"
 )
@@ -238,7 +239,7 @@ func initDecoderConfig(dc *mapstructure.DecoderConfig) {
 			reflect.TypeOf(util.Duration{}),
 			reflect.TypeOf(&config.ProxyOption{}),
 			reflect.TypeOf(config.TCPListenPortRange{}),
-			reflect.TypeOf(config.PEMContent("")),
+			reflect.TypeOf(serialize.PEMContent("")),
 			reflect.TypeOf(config.URL{}),
 			reflect.TypeOf(config.CertPool{}),
 			reflect.TypeOf(config.Regexp{}):
diff --git a/manager/config/config.go b/manager/config/config.go
index 448f850d..4ab2155a 100644
--- a/manager/config/config.go
+++ b/manager/config/config.go
@@ -25,6 +25,7 @@ import (
 
 	"d7y.io/dragonfly/v2/cmd/dependency/base"
 	"d7y.io/dragonfly/v2/pkg/objectstorage"
+	"d7y.io/dragonfly/v2/pkg/serialize"
 )
 
 type Config struct {
@@ -250,10 +251,10 @@ type SecurityConfig struct {
 	Enable bool `yaml:"enable" mapstructure:"enable"`
 
 	// CACert is file path PEM-encoded certificate
-	CACert string `mapstructure:"caCert" yaml:"caCert"`
+	CACert serialize.PEMContent `mapstructure:"caCert" yaml:"caCert"`
 
 	// CAKey is file path of PEM-encoded private key.
-	CAKey string `mapstructure:"caKey" yaml:"caKey"`
+	CAKey serialize.PEMContent `mapstructure:"caKey" yaml:"caKey"`
 }
 
 // New config instance.
diff --git a/manager/manager.go b/manager/manager.go
index b9d1c3d9..4c166313 100644
--- a/manager/manager.go
+++ b/manager/manager.go
@@ -157,7 +157,7 @@ func New(cfg *config.Config, d dfpath.Dfpath) (*Server, error) {
 	// Initialize global certificate.
 	var options []rpcserver.Option
 	if cfg.Security.Enable {
-		cert, err := tls.LoadX509KeyPair(cfg.Security.CACert, cfg.Security.CAKey)
+		cert, err := tls.X509KeyPair([]byte(cfg.Security.CACert), []byte(cfg.Security.CAKey))
 		if err != nil {
 			return nil, err
 		}
diff --git a/pkg/serialize/pemcontent.go b/pkg/serialize/pemcontent.go
new file mode 100644
index 00000000..6c90e15e
--- /dev/null
+++ b/pkg/serialize/pemcontent.go
@@ -0,0 +1,74 @@
+/*
+ *     Copyright 2022 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package serialize
+
+import (
+	"encoding/json"
+	"errors"
+	"os"
+	"strings"
+
+	"gopkg.in/yaml.v3"
+)
+
+// PEMContent supports load PEM format from file or just inline PEM format content
+type PEMContent string
+
+func (p *PEMContent) UnmarshalJSON(b []byte) error {
+	var s string
+	err := json.Unmarshal(b, &s)
+	if err != nil {
+		return err
+	}
+
+	return p.loadPEM(s)
+}
+
+func (p *PEMContent) UnmarshalYAML(node *yaml.Node) error {
+	var s string
+	switch node.Kind {
+	case yaml.ScalarNode:
+		if err := node.Decode(&s); err != nil {
+			return err
+		}
+	default:
+		return errors.New("invalid pem content")
+	}
+
+	return p.loadPEM(s)
+}
+
+func (p *PEMContent) loadPEM(content string) error {
+	if content == "" {
+		*p = PEMContent("")
+		return nil
+	}
+	// inline PEM, just return
+	if strings.HasPrefix(strings.TrimSpace(content), "-----BEGIN ") {
+		val := strings.TrimSpace(content)
+		*p = PEMContent(val)
+		return nil
+	}
+
+	file, err := os.ReadFile(content)
+	if err != nil {
+		return err
+	}
+	val := strings.TrimSpace(string(file))
+	*p = PEMContent(val)
+	return nil
+}


Test patch:
diff --git a/client/config/peerhost_test.go b/client/config/peerhost_test.go
index b77ef506..7c657e1b 100644
--- a/client/config/peerhost_test.go
+++ b/client/config/peerhost_test.go
@@ -31,6 +31,7 @@ import (
 	"d7y.io/dragonfly/v2/cmd/dependency/base"
 	"d7y.io/dragonfly/v2/manager/model"
 	"d7y.io/dragonfly/v2/pkg/dfnet"
+	"d7y.io/dragonfly/v2/pkg/serialize"
 	"d7y.io/dragonfly/v2/pkg/unit"
 )
 
@@ -225,9 +226,9 @@ func TestPeerHostOption_Load(t *testing.T) {
 	_cert, _ := os.ReadFile("./testdata/certs/sca.crt")
 	_key, _ := os.ReadFile("./testdata/certs/sca.key")
 
-	caCert := PEMContent(strings.TrimSpace(string(_caCert)))
-	cert := PEMContent(strings.TrimSpace(string(_cert)))
-	key := PEMContent(strings.TrimSpace(string(_key)))
+	caCert := serialize.PEMContent(strings.TrimSpace(string(_caCert)))
+	cert := serialize.PEMContent(strings.TrimSpace(string(_cert)))
+	key := serialize.PEMContent(strings.TrimSpace(string(_key)))
 
 	peerHostOption := &DaemonOption{
 		Options: base.Options{
diff --git a/client/daemon/peer/peertask_manager_test.go b/client/daemon/peer/peertask_manager_test.go
index fb64e31c..e7ab19f9 100644
--- a/client/daemon/peer/peertask_manager_test.go
+++ b/client/daemon/peer/peertask_manager_test.go
@@ -296,26 +296,30 @@ func setupMockManager(ctrl *gomock.Controller, ts *testSpec, opt componentsOptio
 		scheduleTimeout = util.Duration{Duration: ts.scheduleTimeout}
 	}
 	ptm := &peerTaskManager{
-		calculateDigest: true,
-		host: &schedulerv1.PeerHost{
-			Ip: "127.0.0.1",
-		},
 		conductorLock:    &sync.Mutex{},
 		runningPeerTasks: sync.Map{},
-		pieceManager: &pieceManager{
-			calculateDigest: true,
-			pieceDownloader: opt.pieceDownloader,
-			computePieceSize: func(contentLength int64) uint32 {
-				return opt.pieceSize
+		TaskManagerOption: TaskManagerOption{
+			SchedulerClient: schedulerClient,
+			TaskOption: TaskOption{
+				CalculateDigest: true,
+				PeerHost: &schedulerv1.PeerHost{
+					Ip: "127.0.0.1",
+				},
+				PieceManager: &pieceManager{
+					calculateDigest: true,
+					pieceDownloader: opt.pieceDownloader,
+					computePieceSize: func(contentLength int64) uint32 {
+						return opt.pieceSize
+					},
+				},
+				StorageManager: storageManager,
+				SchedulerOption: config.SchedulerOption{
+					ScheduleTimeout: scheduleTimeout,
+				},
+				GRPCDialTimeout: time.Second,
+				GRPCCredentials: insecure.NewCredentials(),
 			},
 		},
-		storageManager:  storageManager,
-		schedulerClient: schedulerClient,
-		schedulerOption: config.SchedulerOption{
-			ScheduleTimeout: scheduleTimeout,
-		},
-		grpcDialTimeout: time.Second,
-		grpcCredentials: insecure.NewCredentials(),
 	}
 	return &mockManager{
 		testSpec:        ts,
diff --git a/client/daemon/peer/peertask_reuse_test.go b/client/daemon/peer/peertask_reuse_test.go
index a57357e9..4eb52603 100644
--- a/client/daemon/peer/peertask_reuse_test.go
+++ b/client/daemon/peer/peertask_reuse_test.go
@@ -320,11 +320,15 @@ func TestReuseFilePeerTask(t *testing.T) {
 			sm := mocks.NewMockManager(ctrl)
 			tc.storageManager(sm)
 			ptm := &peerTaskManager{
-				host:            &schedulerv1.PeerHost{},
-				enablePrefetch:  tc.enablePrefetch,
-				storageManager:  sm,
-				grpcDialTimeout: time.Second,
-				grpcCredentials: insecure.NewCredentials(),
+				TaskManagerOption: TaskManagerOption{
+					TaskOption: TaskOption{
+						PeerHost:        &schedulerv1.PeerHost{},
+						StorageManager:  sm,
+						GRPCDialTimeout: time.Second,
+						GRPCCredentials: insecure.NewCredentials(),
+					},
+					Prefetch: tc.enablePrefetch,
+				},
 			}
 			tc.verify(ptm.tryReuseFilePeerTask(context.Background(), tc.request))
 		})
@@ -699,11 +703,15 @@ func TestReuseStreamPeerTask(t *testing.T) {
 			sm := mocks.NewMockManager(ctrl)
 			tc.storageManager(sm)
 			ptm := &peerTaskManager{
-				host:            &schedulerv1.PeerHost{},
-				enablePrefetch:  tc.enablePrefetch,
-				storageManager:  sm,
-				grpcDialTimeout: time.Second,
-				grpcCredentials: insecure.NewCredentials(),
+				TaskManagerOption: TaskManagerOption{
+					Prefetch: tc.enablePrefetch,
+					TaskOption: TaskOption{
+						PeerHost:        &schedulerv1.PeerHost{},
+						StorageManager:  sm,
+						GRPCDialTimeout: time.Second,
+						GRPCCredentials: insecure.NewCredentials(),
+					},
+				},
 			}
 			tc.verify(ptm.tryReuseStreamPeerTask(context.Background(), tc.request))
 		})
diff --git a/client/daemon/peer/peertask_stream_backsource_partial_test.go b/client/daemon/peer/peertask_stream_backsource_partial_test.go
index a6aff1d6..ee10c8d8 100644
--- a/client/daemon/peer/peertask_stream_backsource_partial_test.go
+++ b/client/daemon/peer/peertask_stream_backsource_partial_test.go
@@ -251,20 +251,24 @@ func TestStreamPeerTask_BackSource_Partial_WithContentLength(t *testing.T) {
 		},
 	}
 	ptm := &peerTaskManager{
-		calculateDigest: true,
-		host: &schedulerv1.PeerHost{
-			Ip: "127.0.0.1",
-		},
 		conductorLock:    &sync.Mutex{},
 		runningPeerTasks: sync.Map{},
-		pieceManager:     pm,
-		storageManager:   storageManager,
-		schedulerClient:  schedulerClient,
-		schedulerOption: config.SchedulerOption{
-			ScheduleTimeout: util.Duration{Duration: 10 * time.Minute},
+		TaskManagerOption: TaskManagerOption{
+			SchedulerClient: schedulerClient,
+			TaskOption: TaskOption{
+				CalculateDigest: true,
+				PeerHost: &schedulerv1.PeerHost{
+					Ip: "127.0.0.1",
+				},
+				PieceManager:   pm,
+				StorageManager: storageManager,
+				SchedulerOption: config.SchedulerOption{
+					ScheduleTimeout: util.Duration{Duration: 10 * time.Minute},
+				},
+				GRPCDialTimeout: time.Second,
+				GRPCCredentials: insecure.NewCredentials(),
+			},
 		},
-		grpcDialTimeout: time.Second,
-		grpcCredentials: insecure.NewCredentials(),
 	}
 	req := &schedulerv1.PeerTaskRequest{
 		Url: url,