Repository: dragonflyoss/Dragonfly2
Pull request #1159 (2022-03-14 08:28:20)

Fix empty client output path
Signed-off-by: sunwp <244372610@qq.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description
If the client output parameter is not specified, the client cannot automatically obtain the target file path
<!--- 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! -->

- [ ] Code compiles correctly.
- [ ] My code follows the code style of this project.
- [ ] 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.
- [ ] All new and existing tests passed.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1159?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1159](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1159?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (f61c468) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/9155d1b895999c29c1a069f1cd665d29655d0767?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (9155d1b) will **increase** coverage by `0.02%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1159/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/1159?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main    #1159      +/-   ##
==========================================
+ Coverage   46.00%   46.02%   +0.02%     
==========================================
  Files         130      130              
  Lines       10045    10042       -3     
==========================================
+ Hits         4621     4622       +1     
+ Misses       4900     4896       -4     
  Partials      524      524              
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `46.02% <ø> (+0.02%)` | :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/1159?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [client/config/dfget.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1159/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2NvbmZpZy9kZmdldC5nbw==) | `21.62% <ø> (+0.84%)` | :arrow_up: |
| [client/daemon/peer/piece\_broker.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1159/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BpZWNlX2Jyb2tlci5nbw==) | `88.88% <0.00%> (+3.70%)` | :arrow_up: |


Patch:
diff --git a/cdn/rpcserver/rpcserver.go b/cdn/rpcserver/rpcserver.go
index f0f16d04..23089166 100644
--- a/cdn/rpcserver/rpcserver.go
+++ b/cdn/rpcserver/rpcserver.go
@@ -25,7 +25,9 @@ import (
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/peer"
+	"google.golang.org/grpc/status"
 
 	"d7y.io/dragonfly/v2/cdn/constants"
 	"d7y.io/dragonfly/v2/cdn/supervisor"
@@ -242,6 +244,10 @@ func (css *Server) GetPieceTasks(ctx context.Context, req *base.PieceTaskRequest
 	return pp, nil
 }
 
+func (css *Server) SyncPieceTasks(sync cdnsystem.Seeder_SyncPieceTasksServer) error {
+	return status.Error(codes.Unimplemented, "TODO")
+}
+
 func (css *Server) ListenAndServe() error {
 	// Generate GRPC listener
 	lis, _, err := rpc.ListenWithPortRange(css.config.AdvertiseIP, css.config.ListenPort, css.config.ListenPort)
diff --git a/cdn/supervisor/cdn/storage/disk/disk.go b/cdn/supervisor/cdn/storage/disk/disk.go
index 2b74ec40..12565dc2 100644
--- a/cdn/supervisor/cdn/storage/disk/disk.go
+++ b/cdn/supervisor/cdn/storage/disk/disk.go
@@ -26,7 +26,6 @@ import (
 	"strings"
 
 	"github.com/pkg/errors"
-	"github.com/sirupsen/logrus"
 	"go.uber.org/atomic"
 
 	"d7y.io/dragonfly/v2/cdn/gc"
@@ -203,6 +202,12 @@ func (s *diskStorageManager) ResetRepo(seedTask *task.SeedTask) error {
 }
 
 func (s *diskStorageManager) DeleteTask(taskID string) error {
+	if err := s.diskDriver.Remove(storage.GetUploadRaw(taskID)); err != nil && !os.IsNotExist(err) {
+		return err
+	}
+	if err := s.diskDriver.Remove(storage.GetUploadParentRaw(taskID)); err != nil && !os.IsNotExist(err) {
+		logger.StorageGCLogger.Warnf("taskID: %s failed remove upload parent bucket: %v", taskID, err)
+	}
 	if err := s.diskDriver.Remove(storage.GetTaskMetadataRaw(taskID)); err != nil && !os.IsNotExist(err) {
 		return err
 	}
@@ -212,12 +217,9 @@ func (s *diskStorageManager) DeleteTask(taskID string) error {
 	if err := s.diskDriver.Remove(storage.GetDownloadRaw(taskID)); err != nil && !os.IsNotExist(err) {
 		return err
 	}
-	if err := s.diskDriver.Remove(storage.GetUploadRaw(taskID)); err != nil && !os.IsNotExist(err) {
-		return err
-	}
 	// try to clean the parent bucket
-	if err := s.diskDriver.Remove(storage.GetParentRaw(taskID)); err != nil && !os.IsNotExist(err) {
-		logrus.Warnf("taskID: %s failed remove parent bucket: %v", taskID, err)
+	if err := s.diskDriver.Remove(storage.GetDownloadParentRaw(taskID)); err != nil && !os.IsNotExist(err) {
+		logger.StorageGCLogger.Warnf("taskID: %s failed remove download parent bucket: %v", taskID, err)
 	}
 	return nil
 }
@@ -280,22 +282,22 @@ func (s *diskStorageManager) TryFreeSpace(fileLength int64) (bool, error) {
 }
 
 func (s *diskStorageManager) GC() error {
-	logger.StorageGCLogger.With("type", "disk").Info("start the disk storage gc job")
+	logger.StorageGCLogger.With("type", "disk").Debug("start the disk storage gc job")
 	gcTaskIDs, err := s.diskCleaner.GC("disk", false)
 	if err != nil {
 		logger.StorageGCLogger.With("type", "disk").Error("failed to get gcTaskIDs")
 	}
-	var gcTasks []string
+	var actualGCTasks []string
 	var remainingTasks []string
 	for _, taskID := range gcTaskIDs {
 		synclock.Lock(taskID, false)
 		// try to ensure the taskID is not using again
 		if _, exist := s.taskManager.Exist(taskID); exist {
-			synclock.UnLock(taskID, false)
 			remainingTasks = append(remainingTasks, taskID)
+			synclock.UnLock(taskID, false)
 			continue
 		}
-		gcTasks = append(gcTasks, taskID)
+		actualGCTasks = append(actualGCTasks, taskID)
 		if err := s.DeleteTask(taskID); err != nil {
 			logger.StorageGCLogger.With("type", "disk").Errorf("failed to delete disk files with taskID(%s): %v", taskID, err)
 			synclock.UnLock(taskID, false)
@@ -303,7 +305,8 @@ func (s *diskStorageManager) GC() error {
 		}
 		synclock.UnLock(taskID, false)
 	}
-	logger.StorageGCLogger.With("type", "disk").Infof("at most %d tasks can be cleaned up, actual gc %d tasks", len(gcTaskIDs), len(remainingTasks))
-	logger.StorageGCLogger.With("type", "disk").Debugf("tasks %s were successfully cleared, leaving tasks %s remaining", gcTasks, remainingTasks)
+	logger.StorageGCLogger.With("type", "disk").Infof("at most %d tasks can be cleaned up, actual gc %d tasks, remaining %d tasks", len(gcTaskIDs),
+		len(actualGCTasks), len(remainingTasks))
+	logger.StorageGCLogger.With("type", "disk").Debugf("tasks %s were successfully cleared, leaving tasks %s remaining", actualGCTasks, remainingTasks)
 	return nil
 }
diff --git a/cdn/supervisor/cdn/storage/hybrid/hybrid.go b/cdn/supervisor/cdn/storage/hybrid/hybrid.go
index 77d2484a..2e6cf0eb 100644
--- a/cdn/supervisor/cdn/storage/hybrid/hybrid.go
+++ b/cdn/supervisor/cdn/storage/hybrid/hybrid.go
@@ -291,6 +291,14 @@ func (h *hybridStorageManager) deleteMemoryFiles(taskID string) error {
 }
 
 func (h *hybridStorageManager) deleteTaskFiles(taskID string, deleteHardLink bool) error {
+	// delete upload file
+	if err := h.diskDriver.Remove(storage.GetUploadRaw(taskID)); err != nil && !os.IsNotExist(err) {
+		return err
+	}
+	if err := h.diskDriver.Remove(storage.GetUploadParentRaw(taskID)); err != nil &&
+		!os.IsNotExist(err) {
+		logger.WithTaskID(taskID).Warnf("failed to remove upload parent bucket: %v", err)
+	}
 	// delete task file data
 	if err := h.diskDriver.Remove(storage.GetDownloadRaw(taskID)); err != nil && !os.IsNotExist(err) {
 		return err
@@ -299,10 +307,6 @@ func (h *hybridStorageManager) deleteTaskFiles(taskID string, deleteHardLink boo
 	if err := h.memoryDriver.Remove(storage.GetDownloadRaw(taskID)); err != nil && !os.IsNotExist(err) {
 		return err
 	}
-	// delete upload file
-	if err := h.diskDriver.Remove(storage.GetUploadRaw(taskID)); err != nil && !os.IsNotExist(err) {
-		return err
-	}
 	exists := h.diskDriver.Exits(getHardLinkRaw(taskID))
 	if !deleteHardLink && exists {
 		if err := h.diskDriver.MoveFile(h.diskDriver.GetPath(getHardLinkRaw(taskID)), h.diskDriver.GetPath(storage.GetDownloadRaw(taskID))); err != nil {
@@ -322,9 +326,9 @@ func (h *hybridStorageManager) deleteTaskFiles(taskID string, deleteHardLink boo
 		}
 	}
 	// try to clean the parent bucket
-	if err := h.diskDriver.Remove(storage.GetParentRaw(taskID)); err != nil &&
+	if err := h.diskDriver.Remove(storage.GetDownloadParentRaw(taskID)); err != nil &&
 		!os.IsNotExist(err) {
-		logger.WithTaskID(taskID).Warnf("failed to remove parent bucket: %v", err)
+		logger.WithTaskID(taskID).Warnf("failed to remove download parent bucket: %v", err)
 	}
 	return nil
 }
@@ -378,7 +382,7 @@ func (h *hybridStorageManager) tryShmSpace(url, taskID string, fileLength int64)
 }
 
 func (h *hybridStorageManager) GC() error {
-	logger.StorageGCLogger.With("type", "hybrid").Info("start the hybrid storage gc job")
+	logger.StorageGCLogger.With("type", "hybrid").Debug("start the hybrid storage gc job")
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
diff --git a/cdn/supervisor/cdn/storage/path_util.go b/cdn/supervisor/cdn/storage/path_util.go
index 9c10e8f4..d91074b2 100644
--- a/cdn/supervisor/cdn/storage/path_util.go
+++ b/cdn/supervisor/cdn/storage/path_util.go
@@ -86,13 +86,20 @@ func GetAppendPieceMetadataRaw(taskID string) *storedriver.Raw {
 	}
 }
 
-func GetParentRaw(taskID string) *storedriver.Raw {
+func GetDownloadParentRaw(taskID string) *storedriver.Raw {
 	return &storedriver.Raw{
 		Bucket: DownloadHome,
 		Key:    getParentKey(taskID),
 	}
 }
 
+func GetUploadParentRaw(taskID string) *storedriver.Raw {
+	return &storedriver.Raw{
+		Bucket: UploadHome,
+		Key:    getParentKey(taskID),
+	}
+}
+
 func GetDownloadHomeRaw() *storedriver.Raw {
 	return &storedriver.Raw{
 		Bucket: DownloadHome,
diff --git a/cdn/supervisor/cdn/storage/storage_gc.go b/cdn/supervisor/cdn/storage/storage_gc.go
index dc0395f9..e69bda72 100644
--- a/cdn/supervisor/cdn/storage/storage_gc.go
+++ b/cdn/supervisor/cdn/storage/storage_gc.go
@@ -87,8 +87,6 @@ func (cleaner *cleaner) GC(storagePattern string, force bool) ([]string, error)
 	walkTaskIds := make(map[string]bool)
 	var gcTaskIDs []string
 	walkFn := func(path string, info os.FileInfo, err error) error {
-		logger.StorageGCLogger.With("type", storagePattern).Debugf("start to walk path(%s)", path)
-
 		if err != nil {
 			logger.StorageGCLogger.With("type", storagePattern).Errorf("failed to access path(%s): %v", path, err)
 			return err
diff --git a/client/clientutil/keepalive.go b/client/clientutil/keepalive.go
index 6053f12d..bf5cb22a 100644
--- a/client/clientutil/keepalive.go
+++ b/client/clientutil/keepalive.go
@@ -19,6 +19,8 @@ package clientutil
 import (
 	"time"
 
+	"go.uber.org/atomic"
+
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 )
 
@@ -31,26 +33,28 @@ type KeepAlive interface {
 
 type keepAlive struct {
 	name   string
-	access time.Time
+	access atomic.Int64
 }
 
 var _ KeepAlive = (*keepAlive)(nil)
 
 func NewKeepAlive(name string) KeepAlive {
 	return &keepAlive{
-		name:   name,
-		access: time.Now(),
+		name: name,
 	}
 }
 
-func (k keepAlive) Keep() {
-	k.access = time.Now()
-	logger.Debugf("update %s keepalive access time: %s", k.name, k.access.Format(time.RFC3339))
+func (k *keepAlive) Keep() {
+	k.access.Store(time.Now().UnixNano())
 }
 
-func (k keepAlive) Alive(alive time.Duration) bool {
-	var now = time.Now()
+func (k *keepAlive) Alive(alive time.Duration) bool {
+	var (
+		now    = time.Now()
+		access = time.Unix(0, k.access.Load())
+	)
+
 	logger.Debugf("%s keepalive check, last access: %s, alive time: %f seconds, current time: %s",
-		k.name, k.access.Format(time.RFC3339), alive.Seconds(), now)
-	return k.access.Add(alive).After(now)
+		k.name, access.Format(time.RFC3339), alive.Seconds(), now)
+	return access.Add(alive).After(now)
 }
diff --git a/client/config/dfget.go b/client/config/dfget.go
index 4f47d848..3031d6c5 100644
--- a/client/config/dfget.go
+++ b/client/config/dfget.go
@@ -163,12 +163,22 @@ func (cfg *ClientOption) Validate() error {
 }
 
 func (cfg *ClientOption) Convert(args []string) error {
-	var err error
-
-	if cfg.Output, err = filepath.Abs(cfg.Output); err != nil {
-		return err
+	if stringutils.IsBlank(cfg.Output) {
+		url := strings.TrimRight(cfg.URL, "/")
+		idx := strings.LastIndexByte(url, '/')
+		if idx < 0 {
+			return fmt.Errorf("get output from url[%s] error", cfg.URL)
+		}
+		cfg.Output = url[idx+1:]
 	}
 
+	if !filepath.IsAbs(cfg.Output) {
+		absPath, err := filepath.Abs(cfg.Output)
+		if err != nil {
+			return fmt.Errorf("get absolute path[%s] error: %v", cfg.Output, err)
+		}
+		cfg.Output = absPath
+	}
 	if cfg.URL == "" && len(args) > 0 {
 		cfg.URL = args[0]
 	}
@@ -180,7 +190,6 @@ func (cfg *ClientOption) Convert(args []string) error {
 	if cfg.Console {
 		cfg.ShowProgress = false
 	}
-
 	return nil
 }
 
@@ -191,23 +200,9 @@ func (cfg *ClientOption) String() string {
 
 // This function must be called after checkURL
 func (cfg *ClientOption) checkOutput() error {
-	if stringutils.IsBlank(cfg.Output) {
-		url := strings.TrimRight(cfg.URL, "/")
-		idx := strings.LastIndexByte(url, '/')
-		if idx < 0 {
-			return fmt.Errorf("get output from url[%s] error", cfg.URL)
-		}
-		cfg.Output = url[idx+1:]
-	}
-
 	if !filepath.IsAbs(cfg.Output) {
-		absPath, err := filepath.Abs(cfg.Output)
-		if err != nil {
-			return fmt.Errorf("get absolute path[%s] error: %v", cfg.Output, err)
-		}
-		cfg.Output = absPath
+		return fmt.Errorf("path[%s] is not absolute path", cfg.Output)
 	}
-
 	outputDir, _ := path.Split(cfg.Output)
 	if err := MkdirAll(outputDir, 0777, basic.UserID, basic.UserGroup); err != nil {
 		return err
diff --git a/client/daemon/daemon.go b/client/daemon/daemon.go
index bb47b24d..3c1512df 100644
--- a/client/daemon/daemon.go
+++ b/client/daemon/daemon.go
@@ -161,7 +161,7 @@ func New(opt *config.DaemonOption, d dfpath.Dfpath) (Daemon, error) {
 		return nil, err
 	}
 
-	pieceManager, err := peer.NewPieceManager(storageManager,
+	pieceManager, err := peer.NewPieceManager(
 		opt.Download.PieceDownloadTimeout,
 		peer.WithLimiter(rate.NewLimiter(opt.Download.TotalRateLimit.Limit, int(opt.Download.TotalRateLimit.Limit))),
 		peer.WithCalculateDigest(opt.Download.CalculateDigest), peer.WithTransportOption(opt.Download.TransportOption),
diff --git a/client/daemon/peer/peerid_generator.go b/client/daemon/peer/peerid_generator.go
new file mode 100644
index 00000000..8c00cee1
--- /dev/null
+++ b/client/daemon/peer/peerid_generator.go
@@ -0,0 +1,51 @@
+/*
+ *     Copyright 2020 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 peer
+
+import "d7y.io/dragonfly/v2/pkg/idgen"
+
+const (
+	defaultPeerIDBufferSize = 1024
+)
+
+type IDGenerator interface {
+	PeerID() string
+}
+
+type peerIDGenerator struct {
+	ip string
+	ch chan string
+}
+
+func NewPeerIDGenerator(ip string) IDGenerator {
+	p := &peerIDGenerator{
+		ip: ip,
+		ch: make(chan string, defaultPeerIDBufferSize),
+	}
+	go p.run()
+	return p
+}
+
+func (p *peerIDGenerator) PeerID() string {
+	return <-p.ch
+}
+
+func (p *peerIDGenerator) run() {
+	for {
+		p.ch <- idgen.PeerID(p.ip)
+	}
+}
diff --git a/client/daemon/peer/peertask_conductor.go b/client/daemon/peer/peertask_conductor.go
index 1bf9a10f..0557fb80 100644
--- a/client/daemon/peer/peertask_conductor.go
+++ b/client/daemon/peer/peertask_conductor.go
@@ -382,20 +382,21 @@ func (pt *peerTaskConductor) cancel(code base.Code, reason string) {
 }
 
 func (pt *peerTaskConductor) backSource() {
-	backSourceCtx, backSourceSpan := tracer.Start(pt.ctx, config.SpanBackSource)
-	defer backSourceSpan.End()
+	ctx, span := tracer.Start(pt.ctx, config.SpanBackSource)
 	pt.contentLength.Store(-1)
-	err := pt.pieceManager.DownloadSource(backSourceCtx, pt, pt.request)
+	err := pt.pieceManager.DownloadSource(ctx, pt, pt.request)
 	if err != nil {
 		pt.Errorf("download from source error: %s", err)
-		backSourceSpan.SetAttributes(config.AttributePeerTaskSuccess.Bool(false))
-		backSourceSpan.RecordError(err)
+		span.SetAttributes(config.AttributePeerTaskSuccess.Bool(false))
+		span.RecordError(err)
 		pt.cancel(base.Code_ClientError, err.Error())
+		span.End()
 		return
 	}
 	pt.Done()
 	pt.Infof("download from source ok")
-	backSourceSpan.SetAttributes(config.AttributePeerTaskSuccess.Bool(true))
+	span.SetAttributes(config.AttributePeerTaskSuccess.Bool(true))
+	span.End()
 	return
 }
 
@@ -675,6 +676,7 @@ func (pt *peerTaskConductor) pullSinglePiece() {
 		pt.Infof("single piece download success")
 	} else {
 		// fallback to download from other peers
+		span.RecordError(err)
 		span.SetAttributes(config.AttributePieceSuccess.Bool(false))
 		span.End()
 
@@ -1075,6 +1077,7 @@ func (pt *peerTaskConductor) reportSuccessResult(request *DownloadPieceRequest,
 		})
 	if err != nil {
 		pt.Errorf("report piece task error: %v", err)
+		span.RecordError(err)
 	}
 
 	span.End()
@@ -1341,8 +1344,8 @@ func (pt *peerTaskConductor) PublishPieceInfo(pieceNum int32, size uint32) {
 		pt.Done()
 	}
 	pt.broker.Publish(
-		&pieceInfo{
-			num:      pieceNum,
-			finished: finished,
+		&PieceInfo{
+			Num:      pieceNum,
+			Finished: finished,
 		})
 }
diff --git a/client/daemon/peer/peertask_file.go b/client/daemon/peer/peertask_file.go
index 6f49fcc3..f7c86320 100644
--- a/client/daemon/peer/peertask_file.go
+++ b/client/daemon/peer/peertask_file.go
@@ -53,7 +53,7 @@ type fileTask struct {
 	ctx               context.Context
 	span              trace.Span
 	peerTaskConductor *peerTaskConductor
-	pieceCh           chan *pieceInfo
+	pieceCh           chan *PieceInfo
 
 	request *FileTaskRequest
 	// progressCh holds progress status
@@ -135,7 +135,7 @@ func (f *fileTask) syncProgress() {
 			return
 		case <-f.ctx.Done():
 		case piece := <-f.pieceCh:
-			if piece.finished {
+			if piece.Finished {
 				continue
 			}
 			pg := &FileTaskProgress{
diff --git a/client/daemon/peer/peertask_manager.go b/client/daemon/peer/peertask_manager.go
index a896cd86..0dca39a6 100644
--- a/client/daemon/peer/peertask_manager.go
+++ b/client/daemon/peer/peertask_manager.go
@@ -50,6 +50,8 @@ type TaskManager interface {
 	StartStreamTask(ctx context.Context, req *StreamTaskRequest) (
 		readCloser io.ReadCloser, attribute map[string]string, err error)
 
+	Subscribe(request *base.PieceTaskRequest) (*SubscribeResult, bool)
+
 	IsPeerTaskRunning(id string) bool
 
 	// Stop stops the PeerTaskManager
@@ -314,6 +316,28 @@ func (ptm *peerTaskManager) StartStreamTask(ctx context.Context, req *StreamTask
 	return readCloser, attribute, err
 }
 
+type SubscribeResult struct {
+	Storage          storage.TaskStorageDriver
+	PieceInfoChannel chan *PieceInfo
+	Success          chan struct{}
+	Fail             chan struct{}
+}
+
+func (ptm *peerTaskManager) Subscribe(request *base.PieceTaskRequest) (*SubscribeResult, bool) {
+	ptc, ok := ptm.findPeerTaskConductor(request.TaskId)
+	if !ok {
+		return nil, false
+	}
+
+	result := &SubscribeResult{
+		Storage:          ptc.storage,
+		PieceInfoChannel: ptc.broker.Subscribe(),
+		Success:          ptc.successCh,
+		Fail:             ptc.failCh,
+	}
+	return result, true
+}
+
 func (ptm *peerTaskManager) Stop(ctx context.Context) error {
 	// TODO
 	return nil
diff --git a/client/daemon/peer/peertask_piecetask_synchronizer.go b/client/daemon/peer/peertask_piecetask_synchronizer.go
new file mode 100644
index 00000000..305c4401
--- /dev/null
+++ b/client/daemon/peer/peertask_piecetask_synchronizer.go
@@ -0,0 +1,17 @@
+/*
+ *     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 peer
diff --git a/client/daemon/peer/peertask_stream.go b/client/daemon/peer/peertask_stream.go
index 099a76bf..3a691663 100644
--- a/client/daemon/peer/peertask_stream.go
+++ b/client/daemon/peer/peertask_stream.go
@@ -59,7 +59,7 @@ type streamTask struct {
 	ctx               context.Context
 	span              trace.Span
 	peerTaskConductor *peerTaskConductor
-	pieceCh           chan *pieceInfo
+	pieceCh           chan *PieceInfo
 }
 
 func (ptm *peerTaskManager) newStreamTask(
@@ -97,7 +97,7 @@ func (ptm *peerTaskManager) newStreamTask(
 
 func (s *streamTask) Start(ctx context.Context) (io.ReadCloser, map[string]string, error) {
 	// wait first piece to get content length and attribute (eg, response header for http/https)
-	var firstPiece *pieceInfo
+	var firstPiece *PieceInfo
 
 	attr := map[string]string{}
 	attr[config.HeaderDragonflyTask] = s.peerTaskConductor.taskID
@@ -163,22 +163,22 @@ func (s *streamTask) writeOnePiece(w io.Writer, pieceNum int32) (int64, error) {
 	return n, pc.Close()
 }
 
-func (s *streamTask) writeToPipe(firstPiece *pieceInfo, pw *io.PipeWriter) {
+func (s *streamTask) writeToPipe(firstPiece *PieceInfo, pw *io.PipeWriter) {
 	defer func() {
 		s.span.End()
 	}()
 	var (
 		desired int32
-		cur     *pieceInfo
+		cur     *PieceInfo
 		wrote   int64
 		err     error
 		cache   = make(map[int32]bool)
 	)
 	// update first piece to cache and check cur with desired
-	cache[firstPiece.num] = true
+	cache[firstPiece.Num] = true
 	cur = firstPiece
 	for {
-		if desired == cur.num {
+		if desired == cur.Num {
 			for {
 				delete(cache, desired)
 				_, span := tracer.Start(s.ctx, config.SpanWriteBackPiece)
@@ -202,9 +202,9 @@ func (s *streamTask) writeToPipe(firstPiece *pieceInfo, pw *io.PipeWriter) {
 			}
 		} else {
 			// not desired piece, cache it
-			cache[cur.num] = true
-			if cur.num < desired {
-				s.Warnf("piece number should be equal or greater than %d, received piece number: %d", desired, cur.num)
+			cache[cur.Num] = true
+			if cur.Num < desired {
+				s.Warnf("piece number should be equal or greater than %d, received piece number: %d", desired, cur.Num)
 			}
 		}
 
diff --git a/client/daemon/peer/piece_broker.go b/client/daemon/peer/piece_broker.go
index a32f8e7b..53372271 100644
--- a/client/daemon/peer/piece_broker.go
+++ b/client/daemon/peer/piece_broker.go
@@ -18,27 +18,27 @@ package peer
 
 type pieceBroker struct {
 	stopCh    chan struct{}
-	publishCh chan *pieceInfo
-	subCh     chan chan *pieceInfo
-	unsubCh   chan chan *pieceInfo
+	publishCh chan *PieceInfo
+	subCh     chan chan *PieceInfo
+	unsubCh   chan chan *PieceInfo
 }
 
-type pieceInfo struct {
-	num      int32
-	finished bool
+type PieceInfo struct {
+	Num      int32
+	Finished bool
 }
 
 func newPieceBroker() *pieceBroker {
 	return &pieceBroker{
 		stopCh:    make(chan struct{}),
-		publishCh: make(chan *pieceInfo, 10),
-		subCh:     make(chan chan *pieceInfo),
-		unsubCh:   make(chan chan *pieceInfo),
+		publishCh: make(chan *PieceInfo, 10),
+		subCh:     make(chan chan *PieceInfo),
+		unsubCh:   make(chan chan *PieceInfo),
 	}
 }
 
 func (b *pieceBroker) Start() {
-	subs := map[chan *pieceInfo]struct{}{}
+	subs := map[chan *PieceInfo]struct{}{}
 	for {
 		select {
 		case <-b.stopCh:
@@ -66,8 +66,8 @@ func (b *pieceBroker) Stop() {
 	close(b.stopCh)
 }
 
-func (b *pieceBroker) Subscribe() chan *pieceInfo {
-	msgCh := make(chan *pieceInfo, 5)
+func (b *pieceBroker) Subscribe() chan *PieceInfo {
+	msgCh := make(chan *PieceInfo, 5)
 	select {
 	case <-b.stopCh:
 	case b.subCh <- msgCh:
@@ -76,11 +76,11 @@ func (b *pieceBroker) Subscribe() chan *pieceInfo {
 	return msgCh
 }
 
-func (b *pieceBroker) Unsubscribe(msgCh chan *pieceInfo) {
+func (b *pieceBroker) Unsubscribe(msgCh chan *PieceInfo) {
 	b.unsubCh <- msgCh
 }
 
-func (b *pieceBroker) Publish(msg *pieceInfo) {
+func (b *pieceBroker) Publish(msg *PieceInfo) {
 	select {
 	case b.publishCh <- msg:
 	case <-b.stopCh:
diff --git a/client/daemon/peer/piece_downloader.go b/client/daemon/peer/piece_downloader.go
index dd2d1b1e..c7c07b0c 100644
--- a/client/daemon/peer/piece_downloader.go
+++ b/client/daemon/peer/piece_downloader.go
@@ -135,16 +135,27 @@ func WithTransport(rt http.RoundTripper) func(*pieceDownloader) error {
 }
 
 func (p *pieceDownloader) DownloadPiece(ctx context.Context, req *DownloadPieceRequest) (io.Reader, io.Closer, error) {
-	resp, err := p.httpClient.Do(buildDownloadPieceHTTPRequest(ctx, req))
+	httpRequest := buildDownloadPieceHTTPRequest(ctx, req)
+	resp, err := p.httpClient.Do(httpRequest)
 	if err != nil {
 		logger.Errorf("task id: %s, piece num: %d, dst: %s, download piece failed: %s",
 			req.TaskID, req.piece.PieceNum, req.DstAddr, err)
-		return nil, nil, &pieceDownloadError{err: err, connectionError: true}
+		return nil, nil, &pieceDownloadError{
+			target:          httpRequest.URL.String(),
+			err:             err,
+			connectionError: true,
+		}
 	}
 	if resp.StatusCode > 299 {
 		_, _ = io.Copy(io.Discard, resp.Body)
 		_ = resp.Body.Close()
-		return nil, nil, &pieceDownloadError{err: err, connectionError: false, status: resp.Status, statusCode: resp.StatusCode}
+		return nil, nil, &pieceDownloadError{
+			target:          httpRequest.URL.String(),
+			err:             err,
+			connectionError: false,
+			status:          resp.Status,
+			statusCode:      resp.StatusCode,
+		}
 	}
 	reader, closer := resp.Body.(io.Reader), resp.Body.(io.Closer)
 	if req.CalcDigest {
diff --git a/client/daemon/peer/piece_manager.go b/client/daemon/peer/piece_manager.go
index d8e8ac42..60876fa1 100644
--- a/client/daemon/peer/piece_manager.go
+++ b/client/daemon/peer/piece_manager.go
@@ -53,7 +53,7 @@ type pieceManager struct {
 
 var _ PieceManager = (*pieceManager)(nil)
 
-func NewPieceManager(s storage.TaskStorageDriver, pieceDownloadTimeout time.Duration, opts ...func(*pieceManager)) (PieceManager, error) {
+func NewPieceManager(pieceDownloadTimeout time.Duration, opts ...func(*pieceManager)) (PieceManager, error) {
 	pm := &pieceManager{
 		computePieceSize: util.ComputePieceSize,
 		calculateDigest:  true,
@@ -253,17 +253,22 @@ func (pm *pieceManager) DownloadSource(ctx context.Context, pt Task, request *sc
 	}
 	log := pt.Log()
 	log.Infof("start to download from source")
-	contentLengthRequest, err := source.NewRequestWithContext(ctx, request.Url, request.UrlMeta.Header)
+
+	// 1. download pieces from source
+	downloadRequest, err := source.NewRequestWithContext(ctx, request.Url, request.UrlMeta.Header)
 	if err != nil {
 		return err
 	}
-	contentLength, err := source.GetContentLength(contentLengthRequest)
+	response, err := source.Download(downloadRequest)
+	// TODO update expire info
 	if err != nil {
-		log.Warnf("get content length error: %s for %s", err, request.Url)
+		return err
 	}
+	contentLength := response.ContentLength
 	if contentLength < 0 {
 		log.Warnf("can not get content length for %s", request.Url)
 	} else {
+		log.Debugf("back source content length: %d", contentLength)
 		err = pt.GetStorage().UpdateTask(ctx,
 			&storage.UpdateTaskRequest{
 				PeerTaskMetadata: storage.PeerTaskMetadata{
@@ -276,17 +281,6 @@ func (pm *pieceManager) DownloadSource(ctx context.Context, pt Task, request *sc
 			return err
 		}
 	}
-	log.Debugf("get content length: %d", contentLength)
-	// 1. download piece from source
-	downloadRequest, err := source.NewRequestWithContext(ctx, request.Url, request.UrlMeta.Header)
-	if err != nil {
-		return err
-	}
-	response, err := source.Download(downloadRequest)
-	// TODO update expire info
-	if err != nil {
-		return err
-	}
 	defer response.Body.Close()
 	reader := response.Body.(io.Reader)
 
diff --git a/client/daemon/proxy/proxy.go b/client/daemon/proxy/proxy.go
index 1b1314d5..885d3e5b 100644
--- a/client/daemon/proxy/proxy.go
+++ b/client/daemon/proxy/proxy.go
@@ -79,6 +79,9 @@ type Proxy struct {
 	// directHandler are used to handle non-proxy requests
 	directHandler http.Handler
 
+	// transport is used to handle http proxy requests
+	transport http.RoundTripper
+
 	// peerTaskManager is the peer task manager
 	peerTaskManager peer.TaskManager
 
@@ -101,6 +104,8 @@ type Proxy struct {
 
 	// dumpHTTPContent indicates to dump http request header and response header
 	dumpHTTPContent bool
+
+	peerIDGenerator peer.IDGenerator
 }
 
 // Option is a functional option for configuring the proxy
@@ -114,6 +119,14 @@ func WithPeerHost(peerHost *scheduler.PeerHost) Option {
 	}
 }
 
+// WithPeerIDGenerator sets the *transport.PeerIDGenerator
+func WithPeerIDGenerator(peerIDGenerator peer.IDGenerator) Option {
+	return func(p *Proxy) *Proxy {
+		p.peerIDGenerator = peerIDGenerator
+		return p
+	}
+}
+
 // WithPeerTaskManager sets the peer.PeerTaskManager
 func WithPeerTaskManager(peerTaskManager peer.TaskManager) Option {
 	return func(p *Proxy) *Proxy {
@@ -230,6 +243,9 @@ func NewProxyWithOptions(options ...Option) (*Proxy, error) {
 		opt(proxy)
 	}
 
+	if proxy.transport == nil {
+		proxy.transport = proxy.newTransport(nil)
+	}
 	return proxy, nil
 }
 
@@ -246,9 +262,7 @@ func (proxy *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	span.SetAttributes(semconv.HTTPHostKey.String(r.Host))
 	span.SetAttributes(semconv.HTTPURLKey.String(r.URL.String()))
 	span.SetAttributes(semconv.HTTPMethodKey.String(r.Method))
-	defer func() {
-		span.End()
-	}()
+	defer span.End()
 
 	// update ctx to transfer trace id
 	r = r.WithContext(ctx)
@@ -258,6 +272,7 @@ func (proxy *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		user, pass, ok := proxyBasicAuth(r)
 		if !ok {
 			status := http.StatusProxyAuthRequired
+			span.SetAttributes(semconv.HTTPStatusCodeKey.Int(status))
 			http.Error(w, http.StatusText(status), status)
 			logger.Debugf("empty auth info: %s, url：%s", r.Host, r.URL.String())
 			return
@@ -265,6 +280,7 @@ func (proxy *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		// TODO dynamic auth config via manager
 		if user != proxy.basicAuth.Username || pass != proxy.basicAuth.Password {
 			status := http.StatusUnauthorized
+			span.SetAttributes(semconv.HTTPStatusCodeKey.Int(status))
 			http.Error(w, http.StatusText(status), status)
 			logger.Debugf("mismatch auth info (%s/%s): %s, url：%s", user, pass, r.Host, r.URL.String())
 			return
@@ -277,6 +293,7 @@ func (proxy *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	// check whiteList
 	if !directRequest && !proxy.checkWhiteList(r) {
 		status := http.StatusUnauthorized
+		span.SetAttributes(semconv.HTTPStatusCodeKey.Int(status))
 		http.Error(w, http.StatusText(status), status)
 		logger.Debugf("not in whitelist: %s, url：%s", r.Host, r.URL.String())
 		return
@@ -334,9 +351,9 @@ func parseBasicAuth(auth string) (username, password string, ok bool) {
 }
 
 func (proxy *Proxy) handleHTTP(span trace.Span, w http.ResponseWriter, req *http.Request) {
-	// FIXME did not need create a transport per request
-	resp, err := proxy.newTransport(nil).RoundTrip(req)
+	resp, err := proxy.transport.RoundTrip(req)
 	if err != nil {
+		span.RecordError(err)
 		http.Error(w, err.Error(), http.StatusServiceUnavailable)
 		return
 	}
@@ -458,7 +475,7 @@ func (proxy *Proxy) handleHTTPS(w http.ResponseWriter, r *http.Request) {
 
 func (proxy *Proxy) newTransport(tlsConfig *tls.Config) http.RoundTripper {
 	rt, _ := transport.New(
-		transport.WithPeerHost(proxy.peerHost),
+		transport.WithPeerIDGenerator(proxy.peerIDGenerator),
 		transport.WithPeerTaskManager(proxy.peerTaskManager),
 		transport.WithTLS(tlsConfig),
 		transport.WithCondition(proxy.shouldUseDragonfly),
@@ -472,7 +489,7 @@ func (proxy *Proxy) newTransport(tlsConfig *tls.Config) http.RoundTripper {
 func (proxy *Proxy) mirrorRegistry(w http.ResponseWriter, r *http.Request) {
 	reverseProxy := newReverseProxy(proxy.registry)
 	t, err := transport.New(
-		transport.WithPeerHost(proxy.peerHost),
+		transport.WithPeerIDGenerator(proxy.peerIDGenerator),
 		transport.WithPeerTaskManager(proxy.peerTaskManager),
 		transport.WithTLS(proxy.registry.TLSConfig()),
 		transport.WithCondition(proxy.shouldUseDragonflyForMirror),
diff --git a/client/daemon/proxy/proxy_manager.go b/client/daemon/proxy/proxy_manager.go
index 5c9cba65..8731515f 100644
--- a/client/daemon/proxy/proxy_manager.go
+++ b/client/daemon/proxy/proxy_manager.go
@@ -64,6 +64,7 @@ func NewProxyManager(peerHost *scheduler.PeerHost, peerTaskManager peer.TaskMana
 
 	options := []Option{
 		WithPeerHost(peerHost),
+		WithPeerIDGenerator(peer.NewPeerIDGenerator(peerHost.Ip)),
 		WithPeerTaskManager(peerTaskManager),
 		WithRules(proxies),
 		WithWhiteList(whiteList),
diff --git a/client/daemon/rpcserver/rpcserver.go b/client/daemon/rpcserver/rpcserver.go
index b3f91435..0da2ad97 100644
--- a/client/daemon/rpcserver/rpcserver.go
+++ b/client/daemon/rpcserver/rpcserver.go
@@ -71,23 +71,23 @@ func New(peerHost *scheduler.PeerHost, peerTaskManager peer.TaskManager, storage
 	return svr, nil
 }
 
-func (m *server) ServeDownload(listener net.Listener) error {
-	return m.downloadServer.Serve(listener)
+func (s *server) ServeDownload(listener net.Listener) error {
+	return s.downloadServer.Serve(listener)
 }
 
-func (m *server) ServePeer(listener net.Listener) error {
-	m.uploadAddr = fmt.Sprintf("%s:%d", m.peerHost.Ip, m.peerHost.DownPort)
-	return m.peerServer.Serve(listener)
+func (s *server) ServePeer(listener net.Listener) error {
+	s.uploadAddr = fmt.Sprintf("%s:%d", s.peerHost.Ip, s.peerHost.DownPort)
+	return s.peerServer.Serve(listener)
 }
 
-func (m *server) Stop() {
-	m.peerServer.GracefulStop()
-	m.downloadServer.GracefulStop()
+func (s *server) Stop() {
+	s.peerServer.GracefulStop()
+	s.downloadServer.GracefulStop()
 }
 
-func (m *server) GetPieceTasks(ctx context.Context, request *base.PieceTaskRequest) (*base.PiecePacket, error) {
-	m.Keep()
-	p, err := m.storageManager.GetPieces(ctx, request)
+func (s *server) GetPieceTasks(ctx context.Context, request *base.PieceTaskRequest) (*base.PiecePacket, error) {
+	s.Keep()
+	p, err := s.storageManager.GetPieces(ctx, request)
 	if err != nil {
 		code := base.Code_UnknownError
 		if err == dferrors.ErrInvalidArgument {
@@ -99,7 +99,7 @@ func (m *server) GetPieceTasks(ctx context.Context, request *base.PieceTaskReque
 			return nil, dferrors.New(code, err.Error())
 		}
 		// dst peer is not running
-		if !m.peerTaskManager.IsPeerTaskRunning(request.TaskId) {
+		if !s.peerTaskManager.IsPeerTaskRunning(request.TaskId) {
 			code = base.Code_PeerTaskNotFound
 			logger.Errorf("get piece tasks error: peer task not found, task id: %s, src peer: %s, dst peer: %s, piece num: %d, limit: %d",
 				request.TaskId, request.SrcPid, request.DstPid, request.StartNum, request.Limit)
@@ -115,7 +115,7 @@ func (m *server) GetPieceTasks(ctx context.Context, request *base.PieceTaskReque
 		return &base.PiecePacket{
 			TaskId:        request.TaskId,
 			DstPid:        request.DstPid,
-			DstAddr:       m.uploadAddr,
+			DstAddr:       s.uploadAddr,
 			PieceInfos:    nil,
 			TotalPiece:    -1,
 			ContentLength: -1,
@@ -125,18 +125,72 @@ func (m *server) GetPieceTasks(ctx context.Context, request *base.PieceTaskReque
 
 	logger.Debugf("receive get piece tasks request, task id: %s, src peer: %s, dst peer: %s, piece num: %d, limit: %d, length: %d",
 		request.TaskId, request.SrcPid, request.DstPid, request.StartNum, request.Limit, len(p.PieceInfos))
-	p.DstAddr = m.uploadAddr
+	p.DstAddr = s.uploadAddr
 	return p, nil
 }
 
-func (m *server) CheckHealth(context.Context) error {
-	m.Keep()
+// sendExistPieces will send as much as possible pieces
+func (s *server) sendExistPieces(request *base.PieceTaskRequest, sync dfdaemongrpc.Daemon_SyncPieceTasksServer, sentMap map[int32]struct{}) (total int32, sent int, err error) {
+	return sendExistPieces(sync.Context(), s.GetPieceTasks, request, sync, sentMap)
+}
+
+func (s *server) SyncPieceTasks(sync dfdaemongrpc.Daemon_SyncPieceTasksServer) error {
+	request, err := sync.Recv()
+	if err != nil {
+		return err
+	}
+	skipPieceCount := request.StartNum
+	var sentMap = make(map[int32]struct{})
+	total, sent, err := s.sendExistPieces(request, sync, sentMap)
+	if err != nil {
+		return err
+	}
+
+	// task is done, just return
+	if int(total) == sent {
+		return nil
+	}
+
+	// subscribe peer task message for remaining pieces
+	result, ok := s.peerTaskManager.Subscribe(request)
+	if !ok {
+		// task not found, double check for done task
+		total, sent, err = s.sendExistPieces(request, sync, sentMap)
+		if err != nil {
+			return err
+		}
+
+		if int(total) > sent {
+			return status.Errorf(codes.Unavailable, "peer task not finish, but no running task found")
+		}
+		return nil
+	}
+
+	var sub = &subscriber{
+		SubscribeResult: result,
+		sync:            sync,
+		request:         request,
+		skipPieceCount:  skipPieceCount,
+		totalPieces:     total,
+		sentMap:         sentMap,
+		done:            make(chan struct{}),
+		uploadAddr:      s.uploadAddr,
+		SugaredLoggerOnWith: logger.With("taskID", request.TaskId,
+			"localPeerID", request.DstPid, "remotePeerID", request.SrcPid),
+	}
+
+	go sub.receiveRemainingPieceTaskRequests()
+	return sub.sendRemainingPieceTasks()
+}
+
+func (s *server) CheckHealth(context.Context) error {
+	s.Keep()
 	return nil
 }
 
-func (m *server) Download(ctx context.Context,
+func (s *server) Download(ctx context.Context,
 	req *dfdaemongrpc.DownRequest, results chan<- *dfdaemongrpc.DownResult) error {
-	m.Keep()
+	s.Keep()
 	if req.UrlMeta == nil {
 		req.UrlMeta = &base.UrlMeta{}
 	}
@@ -145,8 +199,8 @@ func (m *server) Download(ctx context.Context,
 		PeerTaskRequest: scheduler.PeerTaskRequest{
 			Url:      req.Url,
 			UrlMeta:  req.UrlMeta,
-			PeerId:   idgen.PeerID(m.peerHost.Ip),
-			PeerHost: m.peerHost,
+			PeerId:   idgen.PeerID(s.peerHost.Ip),
+			PeerHost: s.peerHost,
 		},
 		Output:             req.Output,
 		Limit:              req.Limit,
@@ -168,7 +222,7 @@ func (m *server) Download(ctx context.Context,
 	}
 	log := logger.With("peer", peerTask.PeerId, "component", "downloadService")
 
-	peerTaskProgress, tiny, err := m.peerTaskManager.StartFileTask(ctx, peerTask)
+	peerTaskProgress, tiny, err := s.peerTaskManager.StartFileTask(ctx, peerTask)
 	if err != nil {
 		return dferrors.New(base.Code_UnknownError, fmt.Sprintf("%s", err))
 	}
diff --git a/client/daemon/rpcserver/subscriber.go b/client/daemon/rpcserver/subscriber.go
new file mode 100644
index 00000000..56923530
--- /dev/null
+++ b/client/daemon/rpcserver/subscriber.go
@@ -0,0 +1,202 @@
+/*
+ *     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 rpcserver
+
+import (
+	"context"
+	"io"
+	"sync"
+
+	"d7y.io/dragonfly/v2/client/daemon/peer"
+	"d7y.io/dragonfly/v2/internal/dferrors"
+	logger "d7y.io/dragonfly/v2/internal/dflog"
+	"d7y.io/dragonfly/v2/pkg/rpc/base"
+	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
+)
+
+type subscriber struct {
+	sync.Mutex
+	*logger.SugaredLoggerOnWith
+	*peer.SubscribeResult
+	sync           dfdaemon.Daemon_SyncPieceTasksServer
+	request        *base.PieceTaskRequest
+	skipPieceCount uint32
+	totalPieces    int32
+	sentMap        map[int32]struct{}
+	done           chan struct{}
+	uploadAddr     string
+}
+
+func (s *subscriber) getPieces(ctx context.Context, request *base.PieceTaskRequest) (*base.PiecePacket, error) {
+	p, err := s.Storage.GetPieces(ctx, request)
+	p.DstAddr = s.uploadAddr
+	return p, err
+}
+
+func sendExistPieces(
+	ctx context.Context,
+	get func(ctx context.Context, request *base.PieceTaskRequest) (*base.PiecePacket, error),
+	request *base.PieceTaskRequest,
+	sync dfdaemon.Daemon_SyncPieceTasksServer,
+	sendMap map[int32]struct{}) (total int32, sent int, err error) {
+	for {
+		pp, err := get(ctx, request)
+		if err != nil {
+			return -1, -1, err
+		}
+		if len(pp.PieceInfos) == 0 {
+			return pp.TotalPiece, sent, nil
+		}
+		if err = sync.Send(pp); err != nil {
+			return pp.TotalPiece, sent, err
+		}
+		for _, p := range pp.PieceInfos {
+			sendMap[p.PieceNum] = struct{}{}
+		}
+		sent += len(pp.PieceInfos)
+		if uint32(len(pp.PieceInfos)) < request.Limit {
+			return pp.TotalPiece, sent, nil
+		}
+		// the get piece func always return sorted pieces, use last piece num + 1 to get more pieces
+		request.StartNum = uint32(pp.PieceInfos[request.Limit-1].PieceNum + 1)
+	}
+}
+
+// sendExistPieces will send as much as possible pieces
+func (s *subscriber) sendExistPieces(startNum uint32) (total int32, sent int, err error) {
+	s.request.StartNum = startNum
+	return sendExistPieces(s.sync.Context(), s.getPieces, s.request, s.sync, s.sentMap)
+}
+
+func (s *subscriber) receiveRemainingPieceTaskRequests() {
+	defer close(s.done)
+	for {
+		request, err := s.sync.Recv()
+		if err == io.EOF {
+			s.Infof("SyncPieceTasks done, exit receiving")
+			return
+		}
+		if err != nil {
+			s.Errorf("SyncPieceTasks receive error: %s", err)
+			return
+		}
+		s.Debugf("receive request: %#v", request)
+		pp, err := s.getPieces(s.sync.Context(), request)
+		if err != nil {
+			s.Errorf("GetPieceTasks error: %s", err)
+			return
+		}
+
+		// TODO if not found, try to send to peer task conductor, then download it first
+		err = s.sync.Send(pp)
+		if err != nil {
+			s.Errorf("SyncPieceTasks send error: %s", err)
+			return
+		}
+		s.Lock()
+		for _, p := range pp.PieceInfos {
+			s.sentMap[p.PieceNum] = struct{}{}
+		}
+		s.Unlock()
+	}
+}
+
+func (s *subscriber) sendRemainingPieceTasks() error {
+	// nextPieceNum is the least piece num which did not send to remote peer
+	// may great then total piece count, check the total piece count when use it
+	var nextPieceNum uint32
+	s.Lock()
+	for i := int32(s.skipPieceCount); ; i++ {
+		if _, ok := s.sentMap[i]; !ok {
+			nextPieceNum = uint32(i)
+			break
+		}
+	}
+	s.Unlock()
+loop:
+	for {
+		select {
+		case <-s.done:
+			s.Infof("SyncPieceTasks done, exit sending, local task is running")
+			return nil
+		case info := <-s.PieceInfoChannel:
+			// not desired piece
+			s.Debugf("receive piece info, num: %d, %v", info.Num, info.Finished)
+			if s.totalPieces > -1 && uint32(info.Num) < nextPieceNum {
+				continue
+			}
+			s.Lock()
+			total, _, err := s.sendExistPieces(uint32(info.Num))
+			if err != nil {
+				s.Unlock()
+				s.Errorf("sent exist pieces error: %s", err)
+				return err
+			}
+			if total > -1 && s.totalPieces == -1 {
+				s.totalPieces = total
+			}
+			if s.totalPieces > -1 && len(s.sentMap)+int(s.skipPieceCount) == int(s.totalPieces) {
+				s.Unlock()
+				break loop
+			}
+			s.Unlock()
+			if info.Finished {
+				break loop
+			}
+			nextPieceNum = s.searchNextPieceNum(nextPieceNum)
+		case <-s.Success:
+			s.Lock()
+			// all pieces already sent
+			if s.totalPieces > -1 && nextPieceNum == uint32(s.totalPieces) {
+				s.Unlock()
+				break loop
+			}
+			total, _, err := s.sendExistPieces(nextPieceNum)
+			if err != nil {
+				s.Unlock()
+				s.Errorf("sent exist pieces error: %s", err)
+				return err
+			}
+			if total > -1 && s.totalPieces == -1 {
+				s.totalPieces = total
+			}
+			if s.totalPieces > -1 && len(s.sentMap)+int(s.skipPieceCount) != int(s.totalPieces) {
+				s.Unlock()
+				return dferrors.Newf(base.Code_ClientError, "peer task success, but can not send all pieces")
+			}
+			s.Unlock()
+			break loop
+		case <-s.Fail:
+			return dferrors.Newf(base.Code_ClientError, "peer task failed")
+		}
+	}
+	select {
+	case <-s.done:
+		s.Infof("SyncPieceTasks done, exit sending, local task is also done")
+		return nil
+	}
+}
+
+func (s *subscriber) searchNextPieceNum(cur uint32) (nextPieceNum uint32) {
+	for i := int32(cur); ; i++ {
+		if _, ok := s.sentMap[i]; !ok {
+			nextPieceNum = uint32(i)
+			break
+		}
+	}
+	return nextPieceNum
+}
diff --git a/client/daemon/storage/local_storage.go b/client/daemon/storage/local_storage.go
index 58d13bf3..59f2f5a4 100644
--- a/client/daemon/storage/local_storage.go
+++ b/client/daemon/storage/local_storage.go
@@ -294,7 +294,12 @@ func (t *localTaskStore) ReadAllPieces(ctx context.Context, req *ReadAllPiecesRe
 	}
 
 	if req.Range == nil {
-		return file, nil
+		// by jim: for some corner case, avoid the io.Copy call superfluous sendfile syscall
+		// then increase network latency
+		return &limitedReadFile{
+			reader: io.LimitReader(file, t.ContentLength),
+			closer: file,
+		}, nil
 	}
 
 	if _, err = file.Seek(req.Range.Start, io.SeekStart); err != nil {
@@ -394,7 +399,11 @@ func (t *localTaskStore) GetPieces(ctx context.Context, req *base.PieceTaskReque
 		t.Warnf("invalid start num: %d", req.StartNum)
 	}
 	for i := int32(0); i < int32(req.Limit); i++ {
-		if piece, ok := t.Pieces[int32(req.StartNum)+i]; ok {
+		num := int32(req.StartNum) + i
+		if t.TotalPieces > -1 && num >= t.TotalPieces {
+			break
+		}
+		if piece, ok := t.Pieces[num]; ok {
 			piecePacket.PieceInfos = append(piecePacket.PieceInfos, &base.PieceInfo{
 				PieceNum:    piece.Num,
 				RangeStart:  uint64(piece.Range.Start),
diff --git a/client/daemon/transport/transport.go b/client/daemon/transport/transport.go
index 2cbfd923..771a1dd7 100644
--- a/client/daemon/transport/transport.go
+++ b/client/daemon/transport/transport.go
@@ -39,9 +39,7 @@ import (
 	"d7y.io/dragonfly/v2/client/daemon/metrics"
 	"d7y.io/dragonfly/v2/client/daemon/peer"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
-	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	"d7y.io/dragonfly/v2/pkg/util/net/httputils"
 )
 
@@ -66,9 +64,6 @@ type transport struct {
 	// peerTaskManager is the peer task manager
 	peerTaskManager peer.TaskManager
 
-	// peerHost is the peer host info
-	peerHost *scheduler.PeerHost
-
 	// defaultFilter is used when http request without X-Dragonfly-Filter Header
 	defaultFilter string
 
@@ -77,15 +72,17 @@ type transport struct {
 
 	// dumpHTTPContent indicates to dump http request header and response header
 	dumpHTTPContent bool
+
+	peerIDGenerator peer.IDGenerator
 }
 
 // Option is functional config for transport.
 type Option func(rt *transport) *transport
 
-// WithPeerHost sets the peerHost for transport
-func WithPeerHost(peerHost *scheduler.PeerHost) Option {
+// WithPeerIDGenerator sets the peerIDGenerator for transport
+func WithPeerIDGenerator(peerIDGenerator peer.IDGenerator) Option {
 	return func(rt *transport) *transport {
-		rt.peerHost = peerHost
+		rt.peerIDGenerator = peerIDGenerator
 		return rt
 	}
 }
@@ -200,7 +197,7 @@ func NeedUseDragonfly(req *http.Request) bool {
 // the ctx has span info from transport, did not use the ctx from request
 func (rt *transport) download(ctx context.Context, req *http.Request) (*http.Response, error) {
 	url := req.URL.String()
-	peerID := idgen.PeerID(rt.peerHost.Ip)
+	peerID := rt.peerIDGenerator.PeerID()
 	log := logger.With("peer", peerID, "component", "transport")
 	log.Infof("start download with url: %s", url)
 
diff --git a/go.mod b/go.mod
index 0583cb78..5d4682b5 100644
--- a/go.mod
+++ b/go.mod
@@ -48,6 +48,7 @@ require (
 	github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.11.0
+	github.com/prometheus/common v0.28.0
 	github.com/schollz/progressbar/v3 v3.8.2
 	github.com/serialx/hashring v0.0.0-20200727003509-22c0c7ab6b1b
 	github.com/shirou/gopsutil/v3 v3.21.11
@@ -163,7 +164,6 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
-	github.com/prometheus/common v0.28.0 // indirect
 	github.com/prometheus/procfs v0.6.0 // indirect
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/robfig/cron/v3 v3.0.1 // indirect
diff --git a/internal/dflog/logcore.go b/internal/dflog/logcore.go
index 98d5c412..66e65608 100644
--- a/internal/dflog/logcore.go
+++ b/internal/dflog/logcore.go
@@ -36,8 +36,8 @@ var (
 )
 
 const (
-	defaultRotateMaxSize    = 300
-	defaultRotateMaxBackups = 50
+	defaultRotateMaxSize    = 100
+	defaultRotateMaxBackups = 10
 	defaultRotateMaxAge     = 7
 )
 
diff --git a/internal/dflog/logger.go b/internal/dflog/logger.go
index 39f97752..ddbc8cc5 100644
--- a/internal/dflog/logger.go
+++ b/internal/dflog/logger.go
@@ -35,6 +35,8 @@ var (
 	KeepAliveLogger  *zap.SugaredLogger
 	StatSeedLogger   *zap.Logger
 	DownloaderLogger *zap.Logger
+
+	coreLogLevelEnabler zapcore.LevelEnabler
 )
 
 func init() {
@@ -65,6 +67,7 @@ func SetLevel(level zapcore.Level) {
 
 func SetCoreLogger(log *zap.SugaredLogger) {
 	CoreLogger = log
+	coreLogLevelEnabler = log.Desugar().Core()
 }
 
 func SetGCLogger(log *zap.SugaredLogger) {
@@ -142,34 +145,58 @@ func WithHostnameAndIP(hostname, ip string) *SugaredLoggerOnWith {
 }
 
 func (log *SugaredLoggerOnWith) Infof(template string, args ...interface{}) {
+	if !coreLogLevelEnabler.Enabled(zap.InfoLevel) {
+		return
+	}
 	CoreLogger.Infow(fmt.Sprintf(template, args...), log.withArgs...)
 }
 
 func (log *SugaredLoggerOnWith) Info(args ...interface{}) {
+	if !coreLogLevelEnabler.Enabled(zap.InfoLevel) {
+		return
+	}
 	CoreLogger.Infow(fmt.Sprint(args...), log.withArgs...)
 }
 
 func (log *SugaredLoggerOnWith) Warnf(template string, args ...interface{}) {
+	if !coreLogLevelEnabler.Enabled(zap.WarnLevel) {
+		return
+	}
 	CoreLogger.Warnw(fmt.Sprintf(template, args...), log.withArgs...)
 }
 
 func (log *SugaredLoggerOnWith) Warn(args ...interface{}) {
+	if !coreLogLevelEnabler.Enabled(zap.WarnLevel) {
+		return
+	}
 	CoreLogger.Warnw(fmt.Sprint(args...), log.withArgs...)
 }
 
 func (log *SugaredLoggerOnWith) Errorf(template string, args ...interface{}) {
+	if !coreLogLevelEnabler.Enabled(zap.ErrorLevel) {
+		return
+	}
 	CoreLogger.Errorw(fmt.Sprintf(template, args...), log.withArgs...)
 }
 
 func (log *SugaredLoggerOnWith) Error(args ...interface{}) {
+	if !coreLogLevelEnabler.Enabled(zap.ErrorLevel) {
+		return
+	}
 	CoreLogger.Errorw(fmt.Sprint(args...), log.withArgs...)
 }
 
 func (log *SugaredLoggerOnWith) Debugf(template string, args ...interface{}) {
+	if !coreLogLevelEnabler.Enabled(zap.DebugLevel) {
+		return
+	}
 	CoreLogger.Debugw(fmt.Sprintf(template, args...), log.withArgs...)
 }
 
 func (log *SugaredLoggerOnWith) Debug(args ...interface{}) {
+	if !coreLogLevelEnabler.Enabled(zap.DebugLevel) {
+		return
+	}
 	CoreLogger.Debugw(fmt.Sprint(args...), log.withArgs...)
 }
 
diff --git a/manager/database/mysql.go b/manager/database/mysql.go
index b849e77f..0eb3f3c8 100644
--- a/manager/database/mysql.go
+++ b/manager/database/mysql.go
@@ -33,10 +33,13 @@ import (
 
 const (
 	// Default number of cdn load limit
-	defaultCDNLoadLimit = 300
+	DefaultCDNLoadLimit = 300
 
 	// Default number of client load limit
-	defaultClientLoadLimit = 50
+	DefaultClientLoadLimit = 50
+
+	// Default number of pieces to download in parallel
+	DefaultClientParallelCount = 4
 )
 
 func newMyqsl(cfg *config.MysqlConfig) (*gorm.DB, error) {
@@ -130,7 +133,7 @@ func seed(db *gorm.DB) error {
 			},
 			Name: "cdn-cluster-1",
 			Config: map[string]interface{}{
-				"load_limit": defaultCDNLoadLimit,
+				"load_limit": DefaultCDNLoadLimit,
 			},
 			IsDefault: true,
 		}).Error; err != nil {
@@ -150,7 +153,8 @@ func seed(db *gorm.DB) error {
 			Name:   "scheduler-cluster-1",
 			Config: map[string]interface{}{},
 			ClientConfig: map[string]interface{}{
-				"load_limit": defaultClientLoadLimit,
+				"load_limit":     DefaultClientLoadLimit,
+				"parallel_count": DefaultClientParallelCount,
 			},
 			Scopes:    map[string]interface{}{},
 			IsDefault: true,
diff --git a/pkg/idgen/peer_id.go b/pkg/idgen/peer_id.go
index c3b37a67..50f02c59 100644
--- a/pkg/idgen/peer_id.go
+++ b/pkg/idgen/peer_id.go
@@ -23,10 +23,16 @@ import (
 	"github.com/google/uuid"
 )
 
+var pid int
+
+func init() {
+	pid = os.Getpid()
+}
+
 func CDNPeerID(ip string) string {
 	return fmt.Sprintf("%s_%s", PeerID(ip), "CDN")
 }
 
 func PeerID(ip string) string {
-	return fmt.Sprintf("%s-%d-%s", ip, os.Getpid(), uuid.New())
+	return fmt.Sprintf("%s-%d-%s", ip, pid, uuid.New())
 }
diff --git a/pkg/rpc/cdnsystem/cdnsystem.pb.go b/pkg/rpc/cdnsystem/cdnsystem.pb.go
index 24732c2e..b5e280d1 100644
--- a/pkg/rpc/cdnsystem/cdnsystem.pb.go
+++ b/pkg/rpc/cdnsystem/cdnsystem.pb.go
@@ -16,7 +16,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.27.1
-// 	protoc        v3.17.3
+// 	protoc        v3.19.1
 // source: pkg/rpc/cdnsystem/cdnsystem.proto
 
 package cdnsystem
@@ -246,7 +246,7 @@ var file_pkg_rpc_cdnsystem_cdnsystem_proto_rawDesc = []byte{
 	0x6f, 0x75, 0x6e, 0x74, 0x12, 0x1d, 0x0a, 0x0a, 0x62, 0x65, 0x67, 0x69, 0x6e, 0x5f, 0x74, 0x69,
 	0x6d, 0x65, 0x18, 0x08, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x62, 0x65, 0x67, 0x69, 0x6e, 0x54,
 	0x69, 0x6d, 0x65, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x6e, 0x64, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18,
-	0x09, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07, 0x65, 0x6e, 0x64, 0x54, 0x69, 0x6d, 0x65, 0x32, 0x83,
+	0x09, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07, 0x65, 0x6e, 0x64, 0x54, 0x69, 0x6d, 0x65, 0x32, 0xc4,
 	0x01, 0x0a, 0x06, 0x53, 0x65, 0x65, 0x64, 0x65, 0x72, 0x12, 0x3d, 0x0a, 0x0b, 0x4f, 0x62, 0x74,
 	0x61, 0x69, 0x6e, 0x53, 0x65, 0x65, 0x64, 0x73, 0x12, 0x16, 0x2e, 0x63, 0x64, 0x6e, 0x73, 0x79,
 	0x73, 0x74, 0x65, 0x6d, 0x2e, 0x53, 0x65, 0x65, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
@@ -255,10 +255,14 @@ var file_pkg_rpc_cdnsystem_cdnsystem_proto_rawDesc = []byte{
 	0x69, 0x65, 0x63, 0x65, 0x54, 0x61, 0x73, 0x6b, 0x73, 0x12, 0x16, 0x2e, 0x62, 0x61, 0x73, 0x65,
 	0x2e, 0x50, 0x69, 0x65, 0x63, 0x65, 0x54, 0x61, 0x73, 0x6b, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
 	0x74, 0x1a, 0x11, 0x2e, 0x62, 0x61, 0x73, 0x65, 0x2e, 0x50, 0x69, 0x65, 0x63, 0x65, 0x50, 0x61,
-	0x63, 0x6b, 0x65, 0x74, 0x42, 0x27, 0x5a, 0x25, 0x64, 0x37, 0x79, 0x2e, 0x69, 0x6f, 0x2f, 0x64,
-	0x72, 0x61, 0x67, 0x6f, 0x6e, 0x66, 0x6c, 0x79, 0x2f, 0x76, 0x32, 0x2f, 0x70, 0x6b, 0x67, 0x2f,
-	0x72, 0x70, 0x63, 0x2f, 0x63, 0x64, 0x6e, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x62, 0x06, 0x70,
-	0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x63, 0x6b, 0x65, 0x74, 0x12, 0x3f, 0x0a, 0x0e, 0x53, 0x79, 0x6e, 0x63, 0x50, 0x69, 0x65, 0x63,
+	0x65, 0x54, 0x61, 0x73, 0x6b, 0x73, 0x12, 0x16, 0x2e, 0x62, 0x61, 0x73, 0x65, 0x2e, 0x50, 0x69,
+	0x65, 0x63, 0x65, 0x54, 0x61, 0x73, 0x6b, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x11,
+	0x2e, 0x62, 0x61, 0x73, 0x65, 0x2e, 0x50, 0x69, 0x65, 0x63, 0x65, 0x50, 0x61, 0x63, 0x6b, 0x65,
+	0x74, 0x28, 0x01, 0x30, 0x01, 0x42, 0x27, 0x5a, 0x25, 0x64, 0x37, 0x79, 0x2e, 0x69, 0x6f, 0x2f,
+	0x64, 0x72, 0x61, 0x67, 0x6f, 0x6e, 0x66, 0x6c, 0x79, 0x2f, 0x76, 0x32, 0x2f, 0x70, 0x6b, 0x67,
+	0x2f, 0x72, 0x70, 0x63, 0x2f, 0x63, 0x64, 0x6e, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x62, 0x06,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -287,10 +291,12 @@ var file_pkg_rpc_cdnsystem_cdnsystem_proto_depIdxs = []int32{
 	3, // 1: cdnsystem.PieceSeed.piece_info:type_name -> base.PieceInfo
 	0, // 2: cdnsystem.Seeder.ObtainSeeds:input_type -> cdnsystem.SeedRequest
 	4, // 3: cdnsystem.Seeder.GetPieceTasks:input_type -> base.PieceTaskRequest
-	1, // 4: cdnsystem.Seeder.ObtainSeeds:output_type -> cdnsystem.PieceSeed
-	5, // 5: cdnsystem.Seeder.GetPieceTasks:output_type -> base.PiecePacket
-	4, // [4:6] is the sub-list for method output_type
-	2, // [2:4] is the sub-list for method input_type
+	4, // 4: cdnsystem.Seeder.SyncPieceTasks:input_type -> base.PieceTaskRequest
+	1, // 5: cdnsystem.Seeder.ObtainSeeds:output_type -> cdnsystem.PieceSeed
+	5, // 6: cdnsystem.Seeder.GetPieceTasks:output_type -> base.PiecePacket
+	5, // 7: cdnsystem.Seeder.SyncPieceTasks:output_type -> base.PiecePacket
+	5, // [5:8] is the sub-list for method output_type
+	2, // [2:5] is the sub-list for method input_type
 	2, // [2:2] is the sub-list for extension type_name
 	2, // [2:2] is the sub-list for extension extendee
 	0, // [0:2] is the sub-list for field type_name
diff --git a/pkg/rpc/cdnsystem/cdnsystem.proto b/pkg/rpc/cdnsystem/cdnsystem.proto
index 5247c432..665ae4a1 100644
--- a/pkg/rpc/cdnsystem/cdnsystem.proto
+++ b/pkg/rpc/cdnsystem/cdnsystem.proto
@@ -56,4 +56,6 @@ service Seeder{
   rpc ObtainSeeds(SeedRequest)returns(stream PieceSeed);
   // Get piece tasks from cdn
   rpc GetPieceTasks(base.PieceTaskRequest)returns(base.PiecePacket);
+  // Sync piece tasks with other peers
+  rpc SyncPieceTasks(stream base.PieceTaskRequest)returns(stream base.PiecePacket);
 }
diff --git a/pkg/rpc/cdnsystem/cdnsystem_grpc.pb.go b/pkg/rpc/cdnsystem/cdnsystem_grpc.pb.go
index 515837aa..e6fb5631 100644
--- a/pkg/rpc/cdnsystem/cdnsystem_grpc.pb.go
+++ b/pkg/rpc/cdnsystem/cdnsystem_grpc.pb.go
@@ -1,4 +1,8 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.2.0
+// - protoc             v3.19.1
+// source: pkg/rpc/cdnsystem/cdnsystem.proto
 
 package cdnsystem
 
@@ -23,6 +27,8 @@ type SeederClient interface {
 	ObtainSeeds(ctx context.Context, in *SeedRequest, opts ...grpc.CallOption) (Seeder_ObtainSeedsClient, error)
 	// Get piece tasks from cdn
 	GetPieceTasks(ctx context.Context, in *base.PieceTaskRequest, opts ...grpc.CallOption) (*base.PiecePacket, error)
+	// Sync piece tasks with other peers
+	SyncPieceTasks(ctx context.Context, opts ...grpc.CallOption) (Seeder_SyncPieceTasksClient, error)
 }
 
 type seederClient struct {
@@ -74,6 +80,37 @@ func (c *seederClient) GetPieceTasks(ctx context.Context, in *base.PieceTaskRequ
 	return out, nil
 }
 
+func (c *seederClient) SyncPieceTasks(ctx context.Context, opts ...grpc.CallOption) (Seeder_SyncPieceTasksClient, error) {
+	stream, err := c.cc.NewStream(ctx, &Seeder_ServiceDesc.Streams[1], "/cdnsystem.Seeder/SyncPieceTasks", opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &seederSyncPieceTasksClient{stream}
+	return x, nil
+}
+
+type Seeder_SyncPieceTasksClient interface {
+	Send(*base.PieceTaskRequest) error
+	Recv() (*base.PiecePacket, error)
+	grpc.ClientStream
+}
+
+type seederSyncPieceTasksClient struct {
+	grpc.ClientStream
+}
+
+func (x *seederSyncPieceTasksClient) Send(m *base.PieceTaskRequest) error {
+	return x.ClientStream.SendMsg(m)
+}
+
+func (x *seederSyncPieceTasksClient) Recv() (*base.PiecePacket, error) {
+	m := new(base.PiecePacket)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 // SeederServer is the server API for Seeder service.
 // All implementations must embed UnimplementedSeederServer
 // for forward compatibility
@@ -82,6 +119,8 @@ type SeederServer interface {
 	ObtainSeeds(*SeedRequest, Seeder_ObtainSeedsServer) error
 	// Get piece tasks from cdn
 	GetPieceTasks(context.Context, *base.PieceTaskRequest) (*base.PiecePacket, error)
+	// Sync piece tasks with other peers
+	SyncPieceTasks(Seeder_SyncPieceTasksServer) error
 	mustEmbedUnimplementedSeederServer()
 }
 
@@ -95,6 +134,9 @@ func (UnimplementedSeederServer) ObtainSeeds(*SeedRequest, Seeder_ObtainSeedsSer
 func (UnimplementedSeederServer) GetPieceTasks(context.Context, *base.PieceTaskRequest) (*base.PiecePacket, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GetPieceTasks not implemented")
 }
+func (UnimplementedSeederServer) SyncPieceTasks(Seeder_SyncPieceTasksServer) error {
+	return status.Errorf(codes.Unimplemented, "method SyncPieceTasks not implemented")
+}
 func (UnimplementedSeederServer) mustEmbedUnimplementedSeederServer() {}
 
 // UnsafeSeederServer may be embedded to opt out of forward compatibility for this service.
@@ -147,6 +189,32 @@ func _Seeder_GetPieceTasks_Handler(srv interface{}, ctx context.Context, dec fun
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Seeder_SyncPieceTasks_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(SeederServer).SyncPieceTasks(&seederSyncPieceTasksServer{stream})
+}
+
+type Seeder_SyncPieceTasksServer interface {
+	Send(*base.PiecePacket) error
+	Recv() (*base.PieceTaskRequest, error)
+	grpc.ServerStream
+}
+
+type seederSyncPieceTasksServer struct {
+	grpc.ServerStream
+}
+
+func (x *seederSyncPieceTasksServer) Send(m *base.PiecePacket) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func (x *seederSyncPieceTasksServer) Recv() (*base.PieceTaskRequest, error) {
+	m := new(base.PieceTaskRequest)
+	if err := x.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 // Seeder_ServiceDesc is the grpc.ServiceDesc for Seeder service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -165,6 +233,12 @@ var Seeder_ServiceDesc = grpc.ServiceDesc{
 			Handler:       _Seeder_ObtainSeeds_Handler,
 			ServerStreams: true,
 		},
+		{
+			StreamName:    "SyncPieceTasks",
+			Handler:       _Seeder_SyncPieceTasks_Handler,
+			ServerStreams: true,
+			ClientStreams: true,
+		},
 	},
 	Metadata: "pkg/rpc/cdnsystem/cdnsystem.proto",
 }
diff --git a/pkg/rpc/cdnsystem/client/client.go b/pkg/rpc/cdnsystem/client/client.go
index 5fe5af32..16dba66e 100644
--- a/pkg/rpc/cdnsystem/client/client.go
+++ b/pkg/rpc/cdnsystem/client/client.go
@@ -69,6 +69,8 @@ type CdnClient interface {
 
 	GetPieceTasks(ctx context.Context, addr dfnet.NetAddr, req *base.PieceTaskRequest, opts ...grpc.CallOption) (*base.PiecePacket, error)
 
+	SyncPieceTasks(ctx context.Context, addr dfnet.NetAddr, ptr *base.PieceTaskRequest, opts ...grpc.CallOption) (cdnsystem.Seeder_SyncPieceTasksClient, error)
+
 	UpdateState(addrs []dfnet.NetAddr)
 
 	Close() error
@@ -114,3 +116,19 @@ func (cc *cdnClient) GetPieceTasks(ctx context.Context, addr dfnet.NetAddr, req
 	}
 	return res.(*base.PiecePacket), nil
 }
+
+func (cc *cdnClient) SyncPieceTasks(ctx context.Context, addr dfnet.NetAddr, req *base.PieceTaskRequest, opts ...grpc.CallOption) (cdnsystem.Seeder_SyncPieceTasksClient, error) {
+	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
+		client, err := cc.getSeederClientWithTarget(addr.GetEndpoint())
+		if err != nil {
+			return nil, err
+		}
+		return client.GetPieceTasks(ctx, req, opts...)
+	}, 0.2, 2.0, 3, nil)
+	if err != nil {
+		logger.WithTaskID(req.TaskId).Infof("SyncPieceTasks: invoke cdn node %s SyncPieceTasks failed: %v", addr.GetEndpoint(), err)
+		return nil, err
+	}
+	client := res.(cdnsystem.Seeder_SyncPieceTasksClient)
+	return client, client.Send(req)
+}
diff --git a/pkg/rpc/cdnsystem/server/server.go b/pkg/rpc/cdnsystem/server/server.go
index bd5898ec..a59dddf2 100644
--- a/pkg/rpc/cdnsystem/server/server.go
+++ b/pkg/rpc/cdnsystem/server/server.go
@@ -37,10 +37,12 @@ import (
 
 // SeederServer  refer to cdnsystem.SeederServer
 type SeederServer interface {
-	// Generate seeds and return to scheduler
+	// ObtainSeeds generate seeds and return to scheduler
 	ObtainSeeds(context.Context, *cdnsystem.SeedRequest, chan<- *cdnsystem.PieceSeed) error
-	// Get piece tasks from cdn
+	// GetPieceTasks get piece tasks from cdn
 	GetPieceTasks(context.Context, *base.PieceTaskRequest) (*base.PiecePacket, error)
+	// SyncPieceTasks sync piece tasks info with other peers
+	SyncPieceTasks(cdnsystem.Seeder_SyncPieceTasksServer) error
 }
 
 type proxy struct {
@@ -91,6 +93,10 @@ func (p *proxy) GetPieceTasks(ctx context.Context, ptr *base.PieceTaskRequest) (
 	return p.server.GetPieceTasks(ctx, ptr)
 }
 
+func (p *proxy) SyncPieceTasks(sync cdnsystem.Seeder_SyncPieceTasksServer) error {
+	return p.server.SyncPieceTasks(sync)
+}
+
 func send(psc chan *cdnsystem.PieceSeed, closePsc func(), stream cdnsystem.Seeder_ObtainSeedsServer, errChan chan error) {
 	err := safe.Call(func() {
 		defer closePsc()
diff --git a/pkg/rpc/dfdaemon/client/client.go b/pkg/rpc/dfdaemon/client/client.go
index 843719e7..47576cbb 100644
--- a/pkg/rpc/dfdaemon/client/client.go
+++ b/pkg/rpc/dfdaemon/client/client.go
@@ -75,6 +75,8 @@ type DaemonClient interface {
 
 	GetPieceTasks(ctx context.Context, addr dfnet.NetAddr, ptr *base.PieceTaskRequest, opts ...grpc.CallOption) (*base.PiecePacket, error)
 
+	SyncPieceTasks(ctx context.Context, addr dfnet.NetAddr, ptr *base.PieceTaskRequest, opts ...grpc.CallOption) (dfdaemon.Daemon_SyncPieceTasksClient, error)
+
 	CheckHealth(ctx context.Context, target dfnet.NetAddr, opts ...grpc.CallOption) error
 
 	Close() error
@@ -123,6 +125,23 @@ func (dc *daemonClient) GetPieceTasks(ctx context.Context, target dfnet.NetAddr,
 	return res.(*base.PiecePacket), nil
 }
 
+func (dc *daemonClient) SyncPieceTasks(ctx context.Context, target dfnet.NetAddr, ptr *base.PieceTaskRequest, opts ...grpc.CallOption) (dfdaemon.Daemon_SyncPieceTasksClient, error) {
+	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
+		client, err := dc.getDaemonClientWithTarget(target.GetEndpoint())
+		if err != nil {
+			return nil, err
+		}
+		return client.SyncPieceTasks(ctx, opts...)
+	}, 0.2, 2.0, 3, nil)
+	if err != nil {
+		logger.WithTaskID(ptr.TaskId).Infof("SyncPieceTasks: invoke daemon node %s SyncPieceTasks failed: %v", target, err)
+		return nil, err
+	}
+
+	client := res.(dfdaemon.Daemon_SyncPieceTasksClient)
+	return client, client.Send(ptr)
+}
+
 func (dc *daemonClient) CheckHealth(ctx context.Context, target dfnet.NetAddr, opts ...grpc.CallOption) (err error) {
 	_, err = rpc.ExecuteWithRetry(func() (interface{}, error) {
 		client, err := dc.getDaemonClientWithTarget(target.GetEndpoint())
diff --git a/pkg/rpc/dfdaemon/client/peer.go b/pkg/rpc/dfdaemon/client/peer.go
index cf69dcfa..c2b3fb13 100644
--- a/pkg/rpc/dfdaemon/client/peer.go
+++ b/pkg/rpc/dfdaemon/client/peer.go
@@ -28,10 +28,14 @@ import (
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/base/common"
 	cdnclient "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem/client"
+	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 )
 
-func GetPieceTasks(ctx context.Context, destPeer *scheduler.PeerPacket_DestPeer, ptr *base.PieceTaskRequest, opts ...grpc.CallOption) (*base.PiecePacket, error) {
+func GetPieceTasks(ctx context.Context,
+	destPeer *scheduler.PeerPacket_DestPeer,
+	ptr *base.PieceTaskRequest,
+	opts ...grpc.CallOption) (*base.PiecePacket, error) {
 	destAddr := fmt.Sprintf("%s:%d", destPeer.Ip, destPeer.RpcPort)
 	peerID := destPeer.PeerId
 	toCdn := strings.HasSuffix(peerID, common.CdnSuffix)
@@ -50,6 +54,28 @@ func GetPieceTasks(ctx context.Context, destPeer *scheduler.PeerPacket_DestPeer,
 	return client.(DaemonClient).GetPieceTasks(ctx, netAddr, ptr, opts...)
 }
 
+func SyncPieceTasks(ctx context.Context,
+	destPeer *scheduler.PeerPacket_DestPeer,
+	ptr *base.PieceTaskRequest,
+	opts ...grpc.CallOption) (dfdaemon.Daemon_SyncPieceTasksClient, error) {
+	destAddr := fmt.Sprintf("%s:%d", destPeer.Ip, destPeer.RpcPort)
+	peerID := destPeer.PeerId
+	toCdn := strings.HasSuffix(peerID, common.CdnSuffix)
+	var err error
+	netAddr := dfnet.NetAddr{
+		Type: dfnet.TCP,
+		Addr: destAddr,
+	}
+	client, err := getClient(netAddr, toCdn)
+	if err != nil {
+		return nil, err
+	}
+	if toCdn {
+		return client.(cdnclient.CdnClient).SyncPieceTasks(ctx, netAddr, ptr, opts...)
+	}
+	return client.(DaemonClient).SyncPieceTasks(ctx, netAddr, ptr, opts...)
+}
+
 func getClient(netAddr dfnet.NetAddr, toCdn bool) (rpc.Closer, error) {
 	if toCdn {
 		return cdnclient.GetElasticClientByAddrs([]dfnet.NetAddr{netAddr})
diff --git a/pkg/rpc/dfdaemon/dfdaemon.pb.go b/pkg/rpc/dfdaemon/dfdaemon.pb.go
index d43ea337..b8aeb5f1 100644
--- a/pkg/rpc/dfdaemon/dfdaemon.pb.go
+++ b/pkg/rpc/dfdaemon/dfdaemon.pb.go
@@ -303,7 +303,7 @@ var file_pkg_rpc_dfdaemon_dfdaemon_proto_rawDesc = []byte{
 	0x04, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x32, 0x02, 0x28, 0x00, 0x52, 0x0f, 0x63, 0x6f, 0x6d, 0x70,
 	0x6c, 0x65, 0x74, 0x65, 0x64, 0x4c, 0x65, 0x6e, 0x67, 0x74, 0x68, 0x12, 0x12, 0x0a, 0x04, 0x64,
 	0x6f, 0x6e, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x08, 0x52, 0x04, 0x64, 0x6f, 0x6e, 0x65, 0x32,
-	0xbe, 0x01, 0x0a, 0x06, 0x44, 0x61, 0x65, 0x6d, 0x6f, 0x6e, 0x12, 0x39, 0x0a, 0x08, 0x44, 0x6f,
+	0xff, 0x01, 0x0a, 0x06, 0x44, 0x61, 0x65, 0x6d, 0x6f, 0x6e, 0x12, 0x39, 0x0a, 0x08, 0x44, 0x6f,
 	0x77, 0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x12, 0x15, 0x2e, 0x64, 0x66, 0x64, 0x61, 0x65, 0x6d, 0x6f,
 	0x6e, 0x2e, 0x44, 0x6f, 0x77, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x14, 0x2e,
 	0x64, 0x66, 0x64, 0x61, 0x65, 0x6d, 0x6f, 0x6e, 0x2e, 0x44, 0x6f, 0x77, 0x6e, 0x52, 0x65, 0x73,
@@ -315,9 +315,14 @@ var file_pkg_rpc_dfdaemon_dfdaemon_proto_rawDesc = []byte{
 	0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62,
 	0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c,
 	0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79,
-	0x42, 0x26, 0x5a, 0x24, 0x64, 0x37, 0x79, 0x2e, 0x69, 0x6f, 0x2f, 0x64, 0x72, 0x61, 0x67, 0x6f,
-	0x6e, 0x66, 0x6c, 0x79, 0x2f, 0x76, 0x32, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x72, 0x70, 0x63, 0x2f,
-	0x64, 0x66, 0x64, 0x61, 0x65, 0x6d, 0x6f, 0x6e, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x12, 0x3f, 0x0a, 0x0e, 0x53, 0x79, 0x6e, 0x63, 0x50, 0x69, 0x65, 0x63, 0x65, 0x54, 0x61, 0x73,
+	0x6b, 0x73, 0x12, 0x16, 0x2e, 0x62, 0x61, 0x73, 0x65, 0x2e, 0x50, 0x69, 0x65, 0x63, 0x65, 0x54,
+	0x61, 0x73, 0x6b, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x11, 0x2e, 0x62, 0x61, 0x73,
+	0x65, 0x2e, 0x50, 0x69, 0x65, 0x63, 0x65, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x28, 0x01, 0x30,
+	0x01, 0x42, 0x26, 0x5a, 0x24, 0x64, 0x37, 0x79, 0x2e, 0x69, 0x6f, 0x2f, 0x64, 0x72, 0x61, 0x67,
+	0x6f, 0x6e, 0x66, 0x6c, 0x79, 0x2f, 0x76, 0x32, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x72, 0x70, 0x63,
+	0x2f, 0x64, 0x66, 0x64, 0x61, 0x65, 0x6d, 0x6f, 0x6e, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x33,
 }
 
 var (
@@ -346,11 +351,13 @@ var file_pkg_rpc_dfdaemon_dfdaemon_proto_depIdxs = []int32{
 	0, // 1: dfdaemon.Daemon.Download:input_type -> dfdaemon.DownRequest
 	3, // 2: dfdaemon.Daemon.GetPieceTasks:input_type -> base.PieceTaskRequest
 	4, // 3: dfdaemon.Daemon.CheckHealth:input_type -> google.protobuf.Empty
-	1, // 4: dfdaemon.Daemon.Download:output_type -> dfdaemon.DownResult
-	5, // 5: dfdaemon.Daemon.GetPieceTasks:output_type -> base.PiecePacket
-	4, // 6: dfdaemon.Daemon.CheckHealth:output_type -> google.protobuf.Empty
-	4, // [4:7] is the sub-list for method output_type
-	1, // [1:4] is the sub-list for method input_type
+	3, // 4: dfdaemon.Daemon.SyncPieceTasks:input_type -> base.PieceTaskRequest
+	1, // 5: dfdaemon.Daemon.Download:output_type -> dfdaemon.DownResult
+	5, // 6: dfdaemon.Daemon.GetPieceTasks:output_type -> base.PiecePacket
+	4, // 7: dfdaemon.Daemon.CheckHealth:output_type -> google.protobuf.Empty
+	5, // 8: dfdaemon.Daemon.SyncPieceTasks:output_type -> base.PiecePacket
+	5, // [5:9] is the sub-list for method output_type
+	1, // [1:5] is the sub-list for method input_type
 	1, // [1:1] is the sub-list for extension type_name
 	1, // [1:1] is the sub-list for extension extendee
 	0, // [0:1] is the sub-list for field type_name
diff --git a/pkg/rpc/dfdaemon/dfdaemon.proto b/pkg/rpc/dfdaemon/dfdaemon.proto
index 4c68a0da..5ee9db3a 100644
--- a/pkg/rpc/dfdaemon/dfdaemon.proto
+++ b/pkg/rpc/dfdaemon/dfdaemon.proto
@@ -65,4 +65,6 @@ service Daemon{
   rpc GetPieceTasks(base.PieceTaskRequest)returns(base.PiecePacket);
   // Check daemon health
   rpc CheckHealth(google.protobuf.Empty)returns(google.protobuf.Empty);
+  // Sync piece tasks with other peers
+  rpc SyncPieceTasks(stream base.PieceTaskRequest)returns(stream base.PiecePacket);
 }
diff --git a/pkg/rpc/dfdaemon/dfdaemon_grpc.pb.go b/pkg/rpc/dfdaemon/dfdaemon_grpc.pb.go
index 439a0b7d..75837c4d 100644
--- a/pkg/rpc/dfdaemon/dfdaemon_grpc.pb.go
+++ b/pkg/rpc/dfdaemon/dfdaemon_grpc.pb.go
@@ -30,6 +30,8 @@ type DaemonClient interface {
 	GetPieceTasks(ctx context.Context, in *base.PieceTaskRequest, opts ...grpc.CallOption) (*base.PiecePacket, error)
 	// Check daemon health
 	CheckHealth(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*emptypb.Empty, error)
+	// Sync piece tasks with other peers
+	SyncPieceTasks(ctx context.Context, opts ...grpc.CallOption) (Daemon_SyncPieceTasksClient, error)
 }
 
 type daemonClient struct {
@@ -90,6 +92,37 @@ func (c *daemonClient) CheckHealth(ctx context.Context, in *emptypb.Empty, opts
 	return out, nil
 }
 
+func (c *daemonClient) SyncPieceTasks(ctx context.Context, opts ...grpc.CallOption) (Daemon_SyncPieceTasksClient, error) {
+	stream, err := c.cc.NewStream(ctx, &Daemon_ServiceDesc.Streams[1], "/dfdaemon.Daemon/SyncPieceTasks", opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &daemonSyncPieceTasksClient{stream}
+	return x, nil
+}
+
+type Daemon_SyncPieceTasksClient interface {
+	Send(*base.PieceTaskRequest) error
+	Recv() (*base.PiecePacket, error)
+	grpc.ClientStream
+}
+
+type daemonSyncPieceTasksClient struct {
+	grpc.ClientStream
+}
+
+func (x *daemonSyncPieceTasksClient) Send(m *base.PieceTaskRequest) error {
+	return x.ClientStream.SendMsg(m)
+}
+
+func (x *daemonSyncPieceTasksClient) Recv() (*base.PiecePacket, error) {
+	m := new(base.PiecePacket)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 // DaemonServer is the server API for Daemon service.
 // All implementations must embed UnimplementedDaemonServer
 // for forward compatibility
@@ -100,6 +133,8 @@ type DaemonServer interface {
 	GetPieceTasks(context.Context, *base.PieceTaskRequest) (*base.PiecePacket, error)
 	// Check daemon health
 	CheckHealth(context.Context, *emptypb.Empty) (*emptypb.Empty, error)
+	// Sync piece tasks with other peers
+	SyncPieceTasks(Daemon_SyncPieceTasksServer) error
 	mustEmbedUnimplementedDaemonServer()
 }
 
@@ -116,6 +151,9 @@ func (UnimplementedDaemonServer) GetPieceTasks(context.Context, *base.PieceTaskR
 func (UnimplementedDaemonServer) CheckHealth(context.Context, *emptypb.Empty) (*emptypb.Empty, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method CheckHealth not implemented")
 }
+func (UnimplementedDaemonServer) SyncPieceTasks(Daemon_SyncPieceTasksServer) error {
+	return status.Errorf(codes.Unimplemented, "method SyncPieceTasks not implemented")
+}
 func (UnimplementedDaemonServer) mustEmbedUnimplementedDaemonServer() {}
 
 // UnsafeDaemonServer may be embedded to opt out of forward compatibility for this service.
@@ -186,6 +224,32 @@ func _Daemon_CheckHealth_Handler(srv interface{}, ctx context.Context, dec func(
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Daemon_SyncPieceTasks_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(DaemonServer).SyncPieceTasks(&daemonSyncPieceTasksServer{stream})
+}
+
+type Daemon_SyncPieceTasksServer interface {
+	Send(*base.PiecePacket) error
+	Recv() (*base.PieceTaskRequest, error)
+	grpc.ServerStream
+}
+
+type daemonSyncPieceTasksServer struct {
+	grpc.ServerStream
+}
+
+func (x *daemonSyncPieceTasksServer) Send(m *base.PiecePacket) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func (x *daemonSyncPieceTasksServer) Recv() (*base.PieceTaskRequest, error) {
+	m := new(base.PieceTaskRequest)
+	if err := x.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 // Daemon_ServiceDesc is the grpc.ServiceDesc for Daemon service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -208,6 +272,12 @@ var Daemon_ServiceDesc = grpc.ServiceDesc{
 			Handler:       _Daemon_Download_Handler,
 			ServerStreams: true,
 		},
+		{
+			StreamName:    "SyncPieceTasks",
+			Handler:       _Daemon_SyncPieceTasks_Handler,
+			ServerStreams: true,
+			ClientStreams: true,
+		},
 	},
 	Metadata: "pkg/rpc/dfdaemon/dfdaemon.proto",
 }
diff --git a/pkg/rpc/dfdaemon/server/server.go b/pkg/rpc/dfdaemon/server/server.go
index 58fcfae0..ac77d6af 100644
--- a/pkg/rpc/dfdaemon/server/server.go
+++ b/pkg/rpc/dfdaemon/server/server.go
@@ -32,13 +32,16 @@ import (
 	"d7y.io/dragonfly/v2/pkg/safe"
 )
 
+//go:generate mockgen -source server.go -destination ../../../../client/daemon/test/mock/daemon/daemon_server.go
 // DaemonServer refer to dfdaemon.DaemonServer
 type DaemonServer interface {
-	// Trigger client to download file
+	// Download triggers client to download file
 	Download(context.Context, *dfdaemon.DownRequest, chan<- *dfdaemon.DownResult) error
-	// Get piece tasks from other peers
+	// GetPieceTasks get piece tasks from other peers
 	GetPieceTasks(context.Context, *base.PieceTaskRequest) (*base.PiecePacket, error)
-	// Check daemon health
+	// SyncPieceTasks sync piece tasks info with other peers
+	SyncPieceTasks(dfdaemon.Daemon_SyncPieceTasksServer) error
+	// CheckHealth check daemon health
 	CheckHealth(context.Context) error
 }
 
@@ -89,6 +92,10 @@ func (p *proxy) GetPieceTasks(ctx context.Context, ptr *base.PieceTaskRequest) (
 	return p.server.GetPieceTasks(ctx, ptr)
 }
 
+func (p *proxy) SyncPieceTasks(sync dfdaemon.Daemon_SyncPieceTasksServer) error {
+	return p.server.SyncPieceTasks(sync)
+}
+
 func (p *proxy) CheckHealth(ctx context.Context, req *emptypb.Empty) (*emptypb.Empty, error) {
 	return new(emptypb.Empty), p.server.CheckHealth(ctx)
 }
diff --git a/pkg/source/httpprotocol/http_source_client.go b/pkg/source/httpprotocol/http_source_client.go
index 2a6cb3c4..6dfac626 100644
--- a/pkg/source/httpprotocol/http_source_client.go
+++ b/pkg/source/httpprotocol/http_source_client.go
@@ -196,6 +196,9 @@ func (client *httpSourceClient) Download(request *source.Request) (*source.Respo
 				ETag:         resp.Header.Get(headers.ETag),
 			},
 		))
+	if resp.ContentLength > 0 {
+		response.ContentLength = resp.ContentLength
+	}
 	return response, nil
 }
 
diff --git a/scheduler/config/dynconfig.go b/scheduler/config/dynconfig.go
index 95c5184d..abdd1b5a 100644
--- a/scheduler/config/dynconfig.go
+++ b/scheduler/config/dynconfig.go
@@ -214,6 +214,7 @@ func (d *dynconfig) Get() (*DynconfigData, error) {
 	if err := d.Unmarshal(&config); err != nil {
 		return nil, err
 	}
+
 	return &config, nil
 }
 
diff --git a/scheduler/resource/cdn.go b/scheduler/resource/cdn.go
index 4b7e1f50..7a169055 100644
--- a/scheduler/resource/cdn.go
+++ b/scheduler/resource/cdn.go
@@ -259,7 +259,7 @@ func cdnsToHosts(cdns []*config.CDN) map[string]*Host {
 	for _, cdn := range cdns {
 		var netTopology string
 		options := []HostOption{WithIsCDN(true)}
-		if config, ok := cdn.GetCDNClusterConfig(); ok {
+		if config, ok := cdn.GetCDNClusterConfig(); ok && config.LoadLimit > 0 {
 			options = append(options, WithUploadLoadLimit(int32(config.LoadLimit)))
 			netTopology = config.NetTopology
 		}
diff --git a/scheduler/resource/cdn_mock.go b/scheduler/resource/cdn_mock.go
index 7df1d973..67a461af 100644
--- a/scheduler/resource/cdn_mock.go
+++ b/scheduler/resource/cdn_mock.go
@@ -160,6 +160,26 @@ func (mr *MockCDNClientMockRecorder) OnNotify(arg0 interface{}) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnNotify", reflect.TypeOf((*MockCDNClient)(nil).OnNotify), arg0)
 }
 
+// SyncPieceTasks mocks base method.
+func (m *MockCDNClient) SyncPieceTasks(ctx context.Context, addr dfnet.NetAddr, ptr *base.PieceTaskRequest, opts ...grpc.CallOption) (cdnsystem.Seeder_SyncPieceTasksClient, error) {
+	m.ctrl.T.Helper()
+	varargs := []interface{}{ctx, addr, ptr}
+	for _, a := range opts {
+		varargs = append(varargs, a)
+	}
+	ret := m.ctrl.Call(m, "SyncPieceTasks", varargs...)
+	ret0, _ := ret[0].(cdnsystem.Seeder_SyncPieceTasksClient)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// SyncPieceTasks indicates an expected call of SyncPieceTasks.
+func (mr *MockCDNClientMockRecorder) SyncPieceTasks(ctx, addr, ptr interface{}, opts ...interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	varargs := append([]interface{}{ctx, addr, ptr}, opts...)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SyncPieceTasks", reflect.TypeOf((*MockCDNClient)(nil).SyncPieceTasks), varargs...)
+}
+
 // UpdateState mocks base method.
 func (m *MockCDNClient) UpdateState(addrs []dfnet.NetAddr) {
 	m.ctrl.T.Helper()
diff --git a/scheduler/resource/host.go b/scheduler/resource/host.go
index b374a992..3607f9b6 100644
--- a/scheduler/resource/host.go
+++ b/scheduler/resource/host.go
@@ -23,14 +23,10 @@ import (
 	"go.uber.org/atomic"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
+	"d7y.io/dragonfly/v2/manager/database"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 )
 
-const (
-	// Host default upload load limit
-	defaultUploadLoadLimit = 100
-)
-
 // HostOption is a functional option for configuring the host
 type HostOption func(h *Host) *Host
 
@@ -114,7 +110,7 @@ func NewHost(rawHost *scheduler.PeerHost, options ...HostOption) *Host {
 		IDC:             rawHost.Idc,
 		NetTopology:     rawHost.NetTopology,
 		Location:        rawHost.Location,
-		UploadLoadLimit: atomic.NewInt32(defaultUploadLoadLimit),
+		UploadLoadLimit: atomic.NewInt32(database.DefaultClientLoadLimit),
 		Peers:           &sync.Map{},
 		PeerCount:       atomic.NewInt32(0),
 		IsCDN:           false,
diff --git a/scheduler/resource/peer.go b/scheduler/resource/peer.go
index 32fb70f6..0ac917fb 100644
--- a/scheduler/resource/peer.go
+++ b/scheduler/resource/peer.go
@@ -35,11 +35,11 @@ import (
 )
 
 const (
+	// Default value of biz tag
+	DefaultBizTag = "unknow"
+
 	// Download tiny file timeout
 	downloadTinyFileContextTimeout = 2 * time.Minute
-
-	// Default value of biz tag
-	defaultBizTag = "unknow"
 )
 
 const (
@@ -159,7 +159,7 @@ type Peer struct {
 func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
 	p := &Peer{
 		ID:         id,
-		BizTag:     defaultBizTag,
+		BizTag:     DefaultBizTag,
 		Pieces:     &bitset.BitSet{},
 		pieceCosts: []int64{},
 		Stream:     &atomic.Value{},
@@ -183,7 +183,12 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
 			{Name: PeerEventRegisterNormal, Src: []string{PeerStatePending}, Dst: PeerStateReceivedNormal},
 			{Name: PeerEventDownload, Src: []string{PeerStateReceivedTiny, PeerStateReceivedSmall, PeerStateReceivedNormal}, Dst: PeerStateRunning},
 			{Name: PeerEventDownloadFromBackToSource, Src: []string{PeerStateReceivedTiny, PeerStateReceivedSmall, PeerStateReceivedNormal, PeerStateRunning}, Dst: PeerStateBackToSource},
-			{Name: PeerEventDownloadSucceeded, Src: []string{PeerStateRunning, PeerStateBackToSource}, Dst: PeerStateSucceeded},
+			{Name: PeerEventDownloadSucceeded, Src: []string{
+				// Since ReportPeerResult and ReportPieceResult are called in no order,
+				// the result may be reported after the register is successful.
+				PeerStateReceivedTiny, PeerStateReceivedSmall, PeerStateReceivedNormal,
+				PeerStateRunning, PeerStateBackToSource,
+			}, Dst: PeerStateSucceeded},
 			{Name: PeerEventDownloadFailed, Src: []string{
 				PeerStatePending, PeerStateReceivedTiny, PeerStateReceivedSmall, PeerStateReceivedNormal,
 				PeerStateRunning, PeerStateBackToSource, PeerStateSucceeded,
@@ -210,6 +215,7 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
 			PeerEventDownloadFromBackToSource: func(e *fsm.Event) {
 				p.Task.BackToSourcePeers.Add(p)
 				p.DeleteParent()
+				p.Host.DeletePeer(p.ID)
 				p.UpdateAt.Store(time.Now())
 				p.Log.Infof("peer state is %s", e.FSM.Current())
 			},
@@ -219,6 +225,7 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
 				}
 
 				p.DeleteParent()
+				p.Host.DeletePeer(p.ID)
 				p.UpdateAt.Store(time.Now())
 				p.Log.Infof("peer state is %s", e.FSM.Current())
 			},
@@ -228,11 +235,13 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
 				}
 
 				p.DeleteParent()
+				p.Host.DeletePeer(p.ID)
 				p.UpdateAt.Store(time.Now())
 				p.Log.Infof("peer state is %s", e.FSM.Current())
 			},
 			PeerEventLeave: func(e *fsm.Event) {
 				p.DeleteParent()
+				p.Host.DeletePeer(p.ID)
 				p.Log.Infof("peer state is %s", e.FSM.Current())
 			},
 		},
diff --git a/scheduler/scheduler/scheduler.go b/scheduler/scheduler/scheduler.go
index c320f0f5..9af5305f 100644
--- a/scheduler/scheduler/scheduler.go
+++ b/scheduler/scheduler/scheduler.go
@@ -23,6 +23,7 @@ import (
 	"sort"
 	"time"
 
+	"d7y.io/dragonfly/v2/manager/database"
 	"d7y.io/dragonfly/v2/pkg/container/set"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	rpcscheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
@@ -32,9 +33,6 @@ import (
 )
 
 const (
-	// Default number of pieces downloaded in parallel
-	defaultParallelCount = 4
-
 	// Default limit the number of filter traversals
 	defaultFilterParentLimit = 10
 
@@ -120,8 +118,6 @@ func (s *scheduler) ScheduleParent(ctx context.Context, peer *resource.Peer, blo
 				}
 			}
 
-			// If the peer downloads back-to-source, its parent needs to be deleted
-			peer.DeleteParent()
 			return
 		}
 
@@ -281,7 +277,8 @@ func (s *scheduler) filterParents(peer *resource.Peer, blocklist set.SafeSet) []
 		}
 
 		if parent.Host.FreeUploadLoad() <= 0 {
-			peer.Log.Debugf("parent %s is not selected because its free upload is empty", parent.ID)
+			peer.Log.Debugf("parent %s is not selected because its free upload is empty, upload limit is %d, peer count is %d",
+				parent.ID, parent.Host.UploadLoadLimit.Load(), parent.Host.PeerCount.Load())
 			return true
 		}
 
@@ -296,7 +293,7 @@ func (s *scheduler) filterParents(peer *resource.Peer, blocklist set.SafeSet) []
 
 // Construct peer successful packet
 func constructSuccessPeerPacket(dynconfig config.DynconfigInterface, peer *resource.Peer, parent *resource.Peer, candidateParents []*resource.Peer) *rpcscheduler.PeerPacket {
-	parallelCount := defaultParallelCount
+	parallelCount := database.DefaultClientParallelCount
 	if config, ok := dynconfig.GetSchedulerClusterClientConfig(); ok && config.ParallelCount > 0 {
 		parallelCount = int(config.ParallelCount)
 	}
diff --git a/scheduler/service/service.go b/scheduler/service/service.go
index dc989667..2343618b 100644
--- a/scheduler/service/service.go
+++ b/scheduler/service/service.go
@@ -78,7 +78,7 @@ func (s *Service) RegisterPeerTask(ctx context.Context, req *rpcscheduler.PeerTa
 	}
 	host := s.registerHost(ctx, req)
 	peer := s.registerPeer(ctx, req, task, host)
-	peer.Log.Infof("register peer task request: %#v", req)
+	peer.Log.Infof("register peer task request: %#v %#v %#v", req, req.UrlMeta, req.HostLoad)
 
 	// Task has been successful
 	if task.FSM.Is(resource.TaskStateSucceeded) {
@@ -376,7 +376,6 @@ func (s *Service) LeaveTask(ctx context.Context, req *rpcscheduler.PeerTarget) e
 		return true
 	})
 
-	peer.DeleteParent()
 	s.resource.PeerManager().Delete(peer.ID)
 	return nil
 }
@@ -420,7 +419,7 @@ func (s *Service) registerHost(ctx context.Context, req *rpcscheduler.PeerTaskRe
 	if !ok {
 		// Get scheduler cluster client config by manager
 		var options []resource.HostOption
-		if clientConfig, ok := s.dynconfig.GetSchedulerClusterClientConfig(); ok {
+		if clientConfig, ok := s.dynconfig.GetSchedulerClusterClientConfig(); ok && clientConfig.LoadLimit > 0 {
 			options = append(options, resource.WithUploadLoadLimit(int32(clientConfig.LoadLimit)))
 		}
 
@@ -436,7 +435,12 @@ func (s *Service) registerHost(ctx context.Context, req *rpcscheduler.PeerTaskRe
 
 // registerPeer creates a new peer or reuses a previous peer
 func (s *Service) registerPeer(ctx context.Context, req *rpcscheduler.PeerTaskRequest, task *resource.Task, host *resource.Host) *resource.Peer {
-	peer, loaded := s.resource.PeerManager().LoadOrStore(resource.NewPeer(req.PeerId, task, host, resource.WithBizTag(req.UrlMeta.Tag)))
+	var options []resource.PeerOption
+	if req.UrlMeta.Tag != "" {
+		options = append(options, resource.WithBizTag(req.UrlMeta.Tag))
+	}
+
+	peer, loaded := s.resource.PeerManager().LoadOrStore(resource.NewPeer(req.PeerId, task, host, options...))
 	if !loaded {
 		peer.Log.Info("create new peer")
 		return peer


Test patch:
diff --git a/cdn/supervisor/cdn/storage/disk/disk_test.go b/cdn/supervisor/cdn/storage/disk/disk_test.go
index be7faaf2..2c6de004 100644
--- a/cdn/supervisor/cdn/storage/disk/disk_test.go
+++ b/cdn/supervisor/cdn/storage/disk/disk_test.go
@@ -18,14 +18,19 @@ package disk
 
 import (
 	"fmt"
+	"os"
+	"path/filepath"
 	"testing"
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/suite"
 
 	"d7y.io/dragonfly/v2/cdn/storedriver"
+	"d7y.io/dragonfly/v2/cdn/storedriver/local"
 	"d7y.io/dragonfly/v2/cdn/supervisor/cdn/storage"
 	taskMock "d7y.io/dragonfly/v2/cdn/supervisor/mocks/task"
+	"d7y.io/dragonfly/v2/cdn/supervisor/task"
+	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/unit"
 )
 
@@ -132,3 +137,26 @@ func (suite *DiskStorageManagerSuite) TestTryFreeSpace() {
 		})
 	}
 }
+
+func (suite *DiskStorageManagerSuite) TestDeleteTask() {
+	workHome, _ := os.MkdirTemp("/tmp", "cdn-storageManager-")
+	diskDriver, err := local.NewStorageDriver(&storedriver.Config{BaseDir: workHome})
+	suite.Nil(err)
+	suite.m = &diskStorageManager{
+		diskDriver: diskDriver,
+	}
+	testTask := task.NewSeedTask("fooTask", "http://www.dragonfly", &base.UrlMeta{})
+	suite.Nil(suite.m.ResetRepo(testTask))
+	downloadFile, _ := os.Stat(filepath.Join(workHome, "download", "foo", "fooTask"))
+	uploadFile, _ := os.Stat(filepath.Join(workHome, "upload", "foo", "fooTask"))
+	suite.NotNil(downloadFile)
+	suite.NotNil(uploadFile)
+	suite.Nil(suite.m.DeleteTask(testTask.ID))
+	downloadFile, err = os.Stat(filepath.Join(workHome, "download", "foo", "fooTask"))
+	suite.True(os.IsNotExist(err))
+	uploadFile, err = os.Stat(filepath.Join(workHome, "upload", "foo", "fooTask"))
+	suite.True(os.IsNotExist(err))
+	suite.Nil(downloadFile)
+	suite.Nil(uploadFile)
+	os.RemoveAll(workHome)
+}
diff --git a/client/daemon/peer/peertask_stream_backsource_partial_test.go b/client/daemon/peer/peertask_stream_backsource_partial_test.go
index dfc38be0..80b93449 100644
--- a/client/daemon/peer/peertask_stream_backsource_partial_test.go
+++ b/client/daemon/peer/peertask_stream_backsource_partial_test.go
@@ -226,13 +226,11 @@ func TestStreamPeerTask_BackSource_Partial_WithContentLength(t *testing.T) {
 	source.UnRegister("http")
 	require.Nil(t, source.Register("http", sourceClient, httpprotocol.Adapter))
 	defer source.UnRegister("http")
-	sourceClient.EXPECT().GetContentLength(gomock.Any()).DoAndReturn(
-		func(request *source.Request) (int64, error) {
-			return int64(len(testBytes)), nil
-		})
 	sourceClient.EXPECT().Download(gomock.Any()).DoAndReturn(
 		func(request *source.Request) (*source.Response, error) {
-			return source.NewResponse(io.NopCloser(bytes.NewBuffer(testBytes))), nil
+			response := source.NewResponse(io.NopCloser(bytes.NewBuffer(testBytes)))
+			response.ContentLength = int64(len(testBytes))
+			return response, nil
 		})
 
 	pm := &pieceManager{
diff --git a/client/daemon/peer/piece_manager_test.go b/client/daemon/peer/piece_manager_test.go
index 6e351dc3..0d292251 100644
--- a/client/daemon/peer/piece_manager_test.go
+++ b/client/daemon/peer/piece_manager_test.go
@@ -197,7 +197,7 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 			}))
 			defer ts.Close()
 
-			pm, err := NewPieceManager(storageManager, pieceDownloadTimeout)
+			pm, err := NewPieceManager(pieceDownloadTimeout)
 			assert.Nil(err)
 			pm.(*pieceManager).computePieceSize = func(length int64) uint32 {
 				return tc.pieceSize
diff --git a/client/daemon/rpcserver/rpcserver_test.go b/client/daemon/rpcserver/rpcserver_test.go
index bc11b219..6f7401a3 100644
--- a/client/daemon/rpcserver/rpcserver_test.go
+++ b/client/daemon/rpcserver/rpcserver_test.go
@@ -22,8 +22,8 @@ import (
 	"io"
 	"net"
 	"os"
+	"sync"
 	"testing"
-	"time"
 
 	"github.com/distribution/distribution/v3/uuid"
 	"github.com/golang/mock/gomock"
@@ -81,27 +81,7 @@ func TestDownloadManager_ServeDownload(t *testing.T) {
 		peerTaskManager: mockPeerTaskManager,
 	}
 	m.downloadServer = dfdaemonserver.New(m)
-	port, err := freeport.GetFreePort()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	ln, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
-	assert.Nil(err, "get free port should be ok")
-	go func() {
-		if err := m.ServeDownload(ln); err != nil {
-			t.Error(err)
-		}
-	}()
-	time.Sleep(100 * time.Millisecond)
-
-	client, err := dfclient.GetClientByAddr([]dfnet.NetAddr{
-		{
-			Type: dfnet.TCP,
-			Addr: fmt.Sprintf(":%d", port),
-		},
-	})
-	assert.Nil(err, "grpc dial should be ok")
+	_, client := setupPeerServerAndClient(t, m, assert, m.ServeDownload)
 	request := &dfdaemongrpc.DownRequest{
 		Uuid:              uuid.Generate().String(),
 		Url:               "http://localhost/test",
@@ -164,33 +144,14 @@ func TestDownloadManager_ServePeer(t *testing.T) {
 			PieceMd5Sign:  "",
 		}, nil
 	})
-	m := &server{
+	s := &server{
 		KeepAlive:      clientutil.NewKeepAlive("test"),
 		peerHost:       &scheduler.PeerHost{},
 		storageManager: mockStorageManger,
 	}
-	m.peerServer = dfdaemonserver.New(m)
-	port, err := freeport.GetFreePort()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	ln, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
-	assert.Nil(err, "get free port should be ok")
-	go func() {
-		if err := m.ServePeer(ln); err != nil {
-			t.Error(err)
-		}
-	}()
-	time.Sleep(100 * time.Millisecond)
-
-	client, err := dfclient.GetClientByAddr([]dfnet.NetAddr{
-		{
-			Type: dfnet.TCP,
-			Addr: fmt.Sprintf(":%d", port),
-		},
-	})
-	assert.Nil(err, "grpc dial should be ok")
+	s.peerServer = dfdaemonserver.New(s)
+	port, client := setupPeerServerAndClient(t, s, assert, s.ServePeer)
+	defer s.peerServer.GracefulStop()
 
 	var tests = []struct {
 		request           *base.PieceTaskRequest
@@ -254,3 +215,326 @@ func TestDownloadManager_ServePeer(t *testing.T) {
 		assert.Equal(tc.responsePieceSize, len(response.PieceInfos))
 	}
 }
+
+func TestDownloadManager_SyncPieceTasks(t *testing.T) {
+	assert := testifyassert.New(t)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	var (
+		pieceSize = uint32(1024)
+	)
+
+	type pieceRange struct {
+		start int
+		end   int
+	}
+	var tests = []struct {
+		name            string
+		existTaskID     string // test for non-exists task
+		existPieces     []pieceRange
+		requestPieces   []int32
+		limit           uint32
+		totalPieces     uint32
+		followingPieces []pieceRange
+		success         bool
+		verify          func(t *testing.T, assert *testifyassert.Assertions)
+	}{
+		{
+			name: "already exists in storage",
+			existPieces: []pieceRange{
+				{
+					start: 0,
+					end:   10,
+				},
+			},
+			totalPieces: 11,
+			success:     true,
+			verify: func(t *testing.T, assert *testifyassert.Assertions) {
+			},
+		},
+		{
+			name: "already exists in storage - large",
+			existPieces: []pieceRange{
+				{
+					start: 0,
+					end:   1000,
+				},
+			},
+			totalPieces: 1001,
+			success:     true,
+			verify: func(t *testing.T, assert *testifyassert.Assertions) {
+			},
+		},
+		{
+			name: "partial exists in storage",
+			existPieces: []pieceRange{
+				{
+					start: 0,
+					end:   10,
+				},
+			},
+			followingPieces: []pieceRange{
+				{
+					start: 11,
+					end:   20,
+				},
+			},
+			totalPieces: 21,
+			success:     true,
+			verify: func(t *testing.T, assert *testifyassert.Assertions) {
+			},
+		},
+		{
+			name: "partial exists in storage - large",
+			existPieces: []pieceRange{
+				{
+					start: 0,
+					end:   1000,
+				},
+			},
+			followingPieces: []pieceRange{
+				{
+					start: 1001,
+					end:   2000,
+				},
+			},
+			totalPieces:   2001,
+			requestPieces: []int32{1000, 1010, 1020, 1040},
+			success:       true,
+			verify: func(t *testing.T, assert *testifyassert.Assertions) {
+			},
+		},
+		{
+			name: "not exists in storage",
+			followingPieces: []pieceRange{
+				{
+					start: 0,
+					end:   20,
+				},
+			},
+			totalPieces: 21,
+			success:     true,
+			verify: func(t *testing.T, assert *testifyassert.Assertions) {
+			},
+		},
+		{
+			name: "not exists in storage - large",
+			followingPieces: []pieceRange{
+				{
+					start: 0,
+					end:   2000,
+				},
+			},
+			totalPieces:   2001,
+			requestPieces: []int32{1000, 1010, 1020, 1040},
+			success:       true,
+			verify: func(t *testing.T, assert *testifyassert.Assertions) {
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			mockStorageManger := mock_storage.NewMockManager(ctrl)
+
+			if tc.limit == 0 {
+				tc.limit = 1024
+			}
+
+			var (
+				totalPieces []*base.PieceInfo
+				lock        sync.Mutex
+			)
+
+			var addedPieces = make(map[uint32]*base.PieceInfo)
+			for _, p := range tc.existPieces {
+				if p.end == 0 {
+					p.end = p.start
+				}
+				for i := p.start; i <= p.end; i++ {
+					if _, ok := addedPieces[uint32(i)]; ok {
+						continue
+					}
+					piece := &base.PieceInfo{
+						PieceNum:    int32(i),
+						RangeStart:  uint64(i) * uint64(pieceSize),
+						RangeSize:   pieceSize,
+						PieceOffset: uint64(i) * uint64(pieceSize),
+						PieceStyle:  base.PieceStyle_PLAIN,
+					}
+					totalPieces = append(totalPieces, piece)
+					addedPieces[uint32(i)] = piece
+				}
+			}
+
+			mockStorageManger.EXPECT().GetPieces(gomock.Any(),
+				gomock.Any()).AnyTimes().DoAndReturn(
+				func(ctx context.Context, req *base.PieceTaskRequest) (*base.PiecePacket, error) {
+					var pieces []*base.PieceInfo
+					lock.Lock()
+					for i := req.StartNum; i < tc.totalPieces; i++ {
+						if piece, ok := addedPieces[i]; ok {
+							if piece.PieceNum >= int32(req.StartNum) && len(pieces) < int(req.Limit) {
+								pieces = append(pieces, piece)
+							}
+						}
+					}
+					lock.Unlock()
+					return &base.PiecePacket{
+						TaskId:        req.TaskId,
+						DstPid:        req.DstPid,
+						DstAddr:       "",
+						PieceInfos:    pieces,
+						TotalPiece:    int32(tc.totalPieces),
+						ContentLength: int64(tc.totalPieces) * int64(pieceSize),
+						PieceMd5Sign:  "",
+					}, nil
+				})
+			mockTaskManager := mock_peer.NewMockTaskManager(ctrl)
+			mockTaskManager.EXPECT().Subscribe(gomock.Any()).AnyTimes().DoAndReturn(
+				func(request *base.PieceTaskRequest) (*peer.SubscribeResult, bool) {
+					ch := make(chan *peer.PieceInfo)
+					success := make(chan struct{})
+					fail := make(chan struct{})
+
+					go func(followingPieces []pieceRange) {
+						for i, p := range followingPieces {
+							if p.end == 0 {
+								p.end = p.start
+							}
+							for j := p.start; j <= p.end; j++ {
+								lock.Lock()
+								if _, ok := addedPieces[uint32(j)]; ok {
+									continue
+								}
+								piece := &base.PieceInfo{
+									PieceNum:    int32(j),
+									RangeStart:  uint64(j) * uint64(pieceSize),
+									RangeSize:   pieceSize,
+									PieceOffset: uint64(j) * uint64(pieceSize),
+									PieceStyle:  base.PieceStyle_PLAIN,
+								}
+								totalPieces = append(totalPieces, piece)
+								addedPieces[uint32(j)] = piece
+								lock.Unlock()
+
+								var finished bool
+								if i == len(followingPieces)-1 && j == p.end {
+									finished = true
+								}
+								ch <- &peer.PieceInfo{
+									Num:      int32(j),
+									Finished: finished,
+								}
+							}
+						}
+						close(success)
+					}(tc.followingPieces)
+
+					return &peer.SubscribeResult{
+						Storage:          mockStorageManger,
+						PieceInfoChannel: ch,
+						Success:          success,
+						Fail:             fail,
+					}, true
+				})
+
+			s := &server{
+				KeepAlive:       clientutil.NewKeepAlive("test"),
+				peerHost:        &scheduler.PeerHost{},
+				storageManager:  mockStorageManger,
+				peerTaskManager: mockTaskManager,
+			}
+
+			port, client := setupPeerServerAndClient(t, s, assert, s.ServePeer)
+			defer s.peerServer.GracefulStop()
+
+			syncClient, err := client.SyncPieceTasks(
+				context.Background(),
+				dfnet.NetAddr{
+					Type: dfnet.TCP,
+					Addr: fmt.Sprintf("127.0.0.1:%d", port),
+				},
+				&base.PieceTaskRequest{
+					TaskId:   tc.name,
+					SrcPid:   idgen.PeerID(iputils.IPv4),
+					DstPid:   idgen.PeerID(iputils.IPv4),
+					StartNum: 0,
+					Limit:    tc.limit,
+				})
+			assert.Nil(err, "client sync piece tasks grpc call should be ok")
+
+			var (
+				total       = make(map[int32]bool)
+				maxNum      int32
+				requestSent = make(chan bool)
+			)
+			if len(tc.requestPieces) == 0 {
+				close(requestSent)
+			} else {
+				go func() {
+					for _, n := range tc.requestPieces {
+						request := &base.PieceTaskRequest{
+							TaskId:   tc.name,
+							SrcPid:   idgen.PeerID(iputils.IPv4),
+							DstPid:   idgen.PeerID(iputils.IPv4),
+							StartNum: uint32(n),
+							Limit:    tc.limit,
+						}
+						assert.Nil(syncClient.Send(request))
+					}
+					close(requestSent)
+				}()
+			}
+			for {
+				p, err := syncClient.Recv()
+				if err == io.EOF {
+					break
+				}
+				for _, info := range p.PieceInfos {
+					total[info.PieceNum] = true
+					if info.PieceNum >= maxNum {
+						maxNum = info.PieceNum
+					}
+				}
+				if tc.success {
+					assert.Nil(err, "receive piece tasks should be ok")
+				}
+				if int(p.TotalPiece) == len(total) {
+					<-requestSent
+					err = syncClient.CloseSend()
+					assert.Nil(err)
+				}
+			}
+			if tc.success {
+				assert.Equal(int(maxNum+1), len(total))
+			}
+		})
+	}
+}
+
+func setupPeerServerAndClient(t *testing.T, srv *server, assert *testifyassert.Assertions, serveFunc func(listener net.Listener) error) (int, dfclient.DaemonClient) {
+	srv.peerServer = dfdaemonserver.New(srv)
+	port, err := freeport.GetFreePort()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ln, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
+	assert.Nil(err, "get free port should be ok")
+	go func() {
+		if err := serveFunc(ln); err != nil {
+			t.Error(err)
+		}
+	}()
+
+	client, err := dfclient.GetClientByAddr([]dfnet.NetAddr{
+		{
+			Type: dfnet.TCP,
+			Addr: fmt.Sprintf(":%d", port),
+		},
+	})
+	assert.Nil(err, "grpc dial should be ok")
+	return port, client
+}
diff --git a/client/daemon/test/mock/daemon/daemon_server.go b/client/daemon/test/mock/daemon/daemon_server.go
index d52550d4..3b36636e 100644
--- a/client/daemon/test/mock/daemon/daemon_server.go
+++ b/client/daemon/test/mock/daemon/daemon_server.go
@@ -1,5 +1,5 @@
 // Code generated by MockGen. DO NOT EDIT.
-// Source: ../../../../../pkg/rpc/dfdaemon/server/server.go
+// Source: server.go
 
 // Package mock_server is a generated GoMock package.
 package mock_server
@@ -78,3 +78,17 @@ func (mr *MockDaemonServerMockRecorder) GetPieceTasks(arg0, arg1 interface{}) *g
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPieceTasks", reflect.TypeOf((*MockDaemonServer)(nil).GetPieceTasks), arg0, arg1)
 }
+
+// SyncPieceTasks mocks base method.
+func (m *MockDaemonServer) SyncPieceTasks(arg0 dfdaemon.Daemon_SyncPieceTasksServer) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "SyncPieceTasks", arg0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// SyncPieceTasks indicates an expected call of SyncPieceTasks.
+func (mr *MockDaemonServerMockRecorder) SyncPieceTasks(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SyncPieceTasks", reflect.TypeOf((*MockDaemonServer)(nil).SyncPieceTasks), arg0)
+}
diff --git a/client/daemon/test/mock/peer/peertask_manager.go b/client/daemon/test/mock/peer/peertask_manager.go
index 2551a5d1..0b9f8f23 100644
--- a/client/daemon/test/mock/peer/peertask_manager.go
+++ b/client/daemon/test/mock/peer/peertask_manager.go
@@ -12,6 +12,7 @@ import (
 	peer "d7y.io/dragonfly/v2/client/daemon/peer"
 	storage "d7y.io/dragonfly/v2/client/daemon/storage"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
+	base "d7y.io/dragonfly/v2/pkg/rpc/base"
 	gomock "github.com/golang/mock/gomock"
 )
 
@@ -98,6 +99,21 @@ func (mr *MockTaskManagerMockRecorder) Stop(ctx interface{}) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Stop", reflect.TypeOf((*MockTaskManager)(nil).Stop), ctx)
 }
 
+// Subscribe mocks base method.
+func (m *MockTaskManager) Subscribe(request *base.PieceTaskRequest) (*peer.SubscribeResult, bool) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Subscribe", request)
+	ret0, _ := ret[0].(*peer.SubscribeResult)
+	ret1, _ := ret[1].(bool)
+	return ret0, ret1
+}
+
+// Subscribe indicates an expected call of Subscribe.
+func (mr *MockTaskManagerMockRecorder) Subscribe(request interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Subscribe", reflect.TypeOf((*MockTaskManager)(nil).Subscribe), request)
+}
+
 // MockTask is a mock of Task interface.
 type MockTask struct {
 	ctrl     *gomock.Controller
diff --git a/client/daemon/transport/transport_test.go b/client/daemon/transport/transport_test.go
index ba2a9c45..84a05e27 100644
--- a/client/daemon/transport/transport_test.go
+++ b/client/daemon/transport/transport_test.go
@@ -30,7 +30,6 @@ import (
 	"d7y.io/dragonfly/v2/client/daemon/peer"
 	"d7y.io/dragonfly/v2/client/daemon/test"
 	mock_peer "d7y.io/dragonfly/v2/client/daemon/test/mock/peer"
-	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 )
 
 func TestMain(m *testing.M) {
@@ -52,7 +51,7 @@ func TestTransport_RoundTrip(t *testing.T) {
 		},
 	)
 	rt, _ := New(
-		WithPeerHost(&scheduler.PeerHost{}),
+		WithPeerIDGenerator(peer.NewPeerIDGenerator("127.0.0.1")),
 		WithPeerTaskManager(peerTaskManager),
 		WithCondition(func(r *http.Request) bool {
 			return true
diff --git a/pkg/cache/cache_test.go b/pkg/cache/cache_test.go
index b1ebe10c..13fccddb 100644
--- a/pkg/cache/cache_test.go
+++ b/pkg/cache/cache_test.go
@@ -94,19 +94,19 @@ func TestCache(t *testing.T) {
 func TestCacheTimes(t *testing.T) {
 	var found bool
 
-	tc := New(50*time.Millisecond, 1*time.Millisecond)
+	tc := New(100*time.Millisecond, 1*time.Millisecond)
 	tc.Set("a", 1, DefaultExpiration)
 	tc.Set("b", 2, NoExpiration)
-	tc.Set("c", 3, 20*time.Millisecond)
-	tc.Set("d", 4, 70*time.Millisecond)
+	tc.Set("c", 3, 40*time.Millisecond)
+	tc.Set("d", 4, 140*time.Millisecond)
 
-	<-time.After(25 * time.Millisecond)
+	<-time.After(50 * time.Millisecond)
 	_, found = tc.Get("c")
 	if found {
 		t.Error("Found c when it should have been automatically deleted")
 	}
 
-	<-time.After(30 * time.Millisecond)
+	<-time.After(80 * time.Millisecond)
 	_, found = tc.Get("a")
 	if found {
 		t.Error("Found a when it should have been automatically deleted")
@@ -122,7 +122,7 @@ func TestCacheTimes(t *testing.T) {
 		t.Error("Did not find d even though it was set to expire later than the default")
 	}
 
-	<-time.After(20 * time.Millisecond)
+	<-time.After(40 * time.Millisecond)
 	_, found = tc.Get("d")
 	if found {
 		t.Error("Found d when it should have been automatically deleted (later than the default)")
diff --git a/scheduler/resource/cdn_test.go b/scheduler/resource/cdn_test.go
index 36c0d28d..ed6437ca 100644
--- a/scheduler/resource/cdn_test.go
+++ b/scheduler/resource/cdn_test.go
@@ -28,6 +28,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/manager/database"
 	"d7y.io/dragonfly/v2/manager/types"
 	rpcscheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	"d7y.io/dragonfly/v2/scheduler/config"
@@ -352,7 +353,7 @@ func TestCDNClient_cdnsToHosts(t *testing.T) {
 				assert.Equal(hosts[mockRawCDNHost.Uuid].IDC, mockRawCDNHost.Idc)
 				assert.Equal(hosts[mockRawCDNHost.Uuid].NetTopology, "")
 				assert.Equal(hosts[mockRawCDNHost.Uuid].Location, mockRawCDNHost.Location)
-				assert.Equal(hosts[mockRawCDNHost.Uuid].UploadLoadLimit.Load(), int32(defaultUploadLoadLimit))
+				assert.Equal(hosts[mockRawCDNHost.Uuid].UploadLoadLimit.Load(), int32(database.DefaultClientLoadLimit))
 				assert.Empty(hosts[mockRawCDNHost.Uuid].Peers)
 				assert.Equal(hosts[mockRawCDNHost.Uuid].IsCDN, true)
 				assert.NotEqual(hosts[mockRawCDNHost.Uuid].CreateAt.Load(), 0)
diff --git a/scheduler/resource/host_test.go b/scheduler/resource/host_test.go
index 559d162f..12664b8a 100644
--- a/scheduler/resource/host_test.go
+++ b/scheduler/resource/host_test.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
+	"d7y.io/dragonfly/v2/manager/database"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 )
@@ -72,7 +73,7 @@ func TestHost_NewHost(t *testing.T) {
 				assert.Equal(host.Location, mockRawHost.Location)
 				assert.Equal(host.IDC, mockRawHost.Idc)
 				assert.Equal(host.NetTopology, mockRawHost.NetTopology)
-				assert.Equal(host.UploadLoadLimit.Load(), int32(defaultUploadLoadLimit))
+				assert.Equal(host.UploadLoadLimit.Load(), int32(database.DefaultClientLoadLimit))
 				assert.Equal(host.PeerCount.Load(), int32(0))
 				assert.Equal(host.IsCDN, false)
 				assert.NotEqual(host.CreateAt.Load(), 0)
@@ -95,7 +96,7 @@ func TestHost_NewHost(t *testing.T) {
 				assert.Equal(host.Location, mockRawCDNHost.Location)
 				assert.Equal(host.IDC, mockRawCDNHost.Idc)
 				assert.Equal(host.NetTopology, mockRawCDNHost.NetTopology)
-				assert.Equal(host.UploadLoadLimit.Load(), int32(defaultUploadLoadLimit))
+				assert.Equal(host.UploadLoadLimit.Load(), int32(database.DefaultClientLoadLimit))
 				assert.Equal(host.PeerCount.Load(), int32(0))
 				assert.Equal(host.IsCDN, true)
 				assert.NotEqual(host.CreateAt.Load(), 0)
@@ -383,7 +384,7 @@ func TestHost_FreeUploadLoad(t *testing.T) {
 				host.StorePeer(mockPeer)
 				mockPeer.ID = idgen.PeerID("0.0.0.0")
 				host.StorePeer(mockPeer)
-				assert.Equal(host.FreeUploadLoad(), int32(defaultUploadLoadLimit-2))
+				assert.Equal(host.FreeUploadLoad(), int32(database.DefaultClientLoadLimit-2))
 			},
 		},
 		{
@@ -391,7 +392,7 @@ func TestHost_FreeUploadLoad(t *testing.T) {
 			rawHost: mockRawHost,
 			expect: func(t *testing.T, host *Host, mockPeer *Peer) {
 				assert := assert.New(t)
-				assert.Equal(host.FreeUploadLoad(), int32(defaultUploadLoadLimit))
+				assert.Equal(host.FreeUploadLoad(), int32(database.DefaultClientLoadLimit))
 			},
 		},
 	}
diff --git a/scheduler/scheduler/evaluator/evaluator_base_test.go b/scheduler/scheduler/evaluator/evaluator_base_test.go
index 22e98f0a..f480ea83 100644
--- a/scheduler/scheduler/evaluator/evaluator_base_test.go
+++ b/scheduler/scheduler/evaluator/evaluator_base_test.go
@@ -290,7 +290,7 @@ func TestEvaluatorBase_calculateFreeLoadScore(t *testing.T) {
 			},
 			expect: func(t *testing.T, score float64) {
 				assert := assert.New(t)
-				assert.Equal(score, float64(0.99))
+				assert.Equal(score, float64(0.98))
 			},
 		},
 		{
diff --git a/scheduler/service/service_test.go b/scheduler/service/service_test.go
index bd1820ce..34aecf11 100644
--- a/scheduler/service/service_test.go
+++ b/scheduler/service/service_test.go
@@ -1778,6 +1778,7 @@ func TestService_registerPeer(t *testing.T) {
 			expect: func(t *testing.T, peer *resource.Peer) {
 				assert := assert.New(t)
 				assert.Equal(peer.ID, mockPeerID)
+				assert.Equal(peer.BizTag, resource.DefaultBizTag)
 			},
 		},
 		{
@@ -1795,6 +1796,7 @@ func TestService_registerPeer(t *testing.T) {
 			expect: func(t *testing.T, peer *resource.Peer) {
 				assert := assert.New(t)
 				assert.Equal(peer.ID, mockPeerID)
+				assert.Equal(peer.BizTag, resource.DefaultBizTag)
 			},
 		},
 	}