Repository: dragonflyoss/Dragonfly2
Pull request #923 (2021-12-13 16:53:07)

Optimaze GC log
Signed-off-by: sunwp <244372610@qq.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description

<!--- Describe your changes in detail -->

## Related Issue

<!--- This project only accepts pull requests related to open issues -->
<!--- If suggesting a new feature or change, please discuss it in an issue first -->
<!--- If fixing a bug, there should be an issue describing it with steps to reproduce -->
<!--- Please link to the issue here: -->

## Motivation and Context

<!--- Why is this change required? What problem does it solve? -->

## Screenshots (if appropriate):

## Types of changes

<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)
- [ ] Documentation Update (if none of the other choices apply)

## Checklist:

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [ ] 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/923?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#923](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/923?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (ab30748) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/4a4debdea270755ef52c25bc8fe849938cc1a4f5?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (4a4debd) will **decrease** coverage by `0.02%`.
> The diff coverage is `0.00%`.

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

```diff
@@            Coverage Diff             @@
##             main     #923      +/-   ##
==========================================
- Coverage   43.27%   43.24%   -0.03%     
==========================================
  Files         106      106              
  Lines        8273     8273              
==========================================
- Hits         3580     3578       -2     
- Misses       4232     4234       +2     
  Partials      461      461              
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `43.24% <0.00%> (-0.03%)` | :arrow_down: |

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/923?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [cdn/supervisor/cdn/storage/disk/disk.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/923/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2RuL3N1cGVydmlzb3IvY2RuL3N0b3JhZ2UvZGlzay9kaXNrLmdv) | `18.04% <0.00%> (ø)` | |
| [cdn/supervisor/task/manager.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/923/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2RuL3N1cGVydmlzb3IvdGFzay9tYW5hZ2VyLmdv) | `26.00% <0.00%> (ø)` | |
| [client/daemon/peer/peertask\_base.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/923/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2Jhc2UuZ28=) | `47.29% <0.00%> (-1.21%)` | :arrow_down: |
| [cdn/supervisor/progress/progress.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/923/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2RuL3N1cGVydmlzb3IvcHJvZ3Jlc3MvcHJvZ3Jlc3MuZ28=) | `97.01% <0.00%> (+5.97%)` | :arrow_up: |


Patch:
diff --git a/cdn/rpcserver/rpcserver.go b/cdn/rpcserver/rpcserver.go
index 2eef1db7..8b976429 100644
--- a/cdn/rpcserver/rpcserver.go
+++ b/cdn/rpcserver/rpcserver.go
@@ -32,12 +32,11 @@ import (
 	"d7y.io/dragonfly/v2/cdn/supervisor/task"
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
 	cdnserver "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem/server"
-	"d7y.io/dragonfly/v2/pkg/source"
 	"d7y.io/dragonfly/v2/pkg/util/hostutils"
 )
 
@@ -78,7 +77,7 @@ func (css *Server) ObtainSeeds(ctx context.Context, req *cdnsystem.SeedRequest,
 		}
 	}()
 	// register seed task
-	pieceChan, err := css.service.RegisterSeedTask(ctx, clientAddr, task.NewSeedTask(req.TaskId, req.Url, req.UrlMeta))
+	registeredTask, pieceChan, err := css.service.RegisterSeedTask(ctx, clientAddr, task.NewSeedTask(req.TaskId, req.Url, req.UrlMeta))
 	if err != nil {
 		if supervisor.IsResourcesLacked(err) {
 			err = dferrors.Newf(base.Code_ResourceLacked, "resources lacked for task(%s): %v", req.TaskId, err)
@@ -104,8 +103,8 @@ func (css *Server) ObtainSeeds(ctx context.Context, req *cdnsystem.SeedRequest,
 				PieceStyle:  piece.PieceStyle,
 			},
 			Done:            false,
-			ContentLength:   source.UnknownSourceFileLen,
-			TotalPieceCount: task.UnknownTotalPieceCount,
+			ContentLength:   registeredTask.SourceFileLength,
+			TotalPieceCount: registeredTask.TotalPieceCount,
 		}
 		psc <- pieceSeed
 		jsonPiece, err := json.Marshal(pieceSeed)
diff --git a/cdn/supervisor/cdn/storage/disk/disk.go b/cdn/supervisor/cdn/storage/disk/disk.go
index 4b26b143..2b74ec40 100644
--- a/cdn/supervisor/cdn/storage/disk/disk.go
+++ b/cdn/supervisor/cdn/storage/disk/disk.go
@@ -280,27 +280,30 @@ func (s *diskStorageManager) TryFreeSpace(fileLength int64) (bool, error) {
 }
 
 func (s *diskStorageManager) GC() error {
-	logger.GcLogger.With("type", "disk").Info("start the disk storage gc job")
+	logger.StorageGCLogger.With("type", "disk").Info("start the disk storage gc job")
 	gcTaskIDs, err := s.diskCleaner.GC("disk", false)
 	if err != nil {
-		logger.GcLogger.With("type", "disk").Error("failed to get gcTaskIDs")
+		logger.StorageGCLogger.With("type", "disk").Error("failed to get gcTaskIDs")
 	}
-	var realGCCount int
+	var gcTasks []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)
 			continue
 		}
-		realGCCount++
+		gcTasks = append(gcTasks, taskID)
 		if err := s.DeleteTask(taskID); err != nil {
-			logger.GcLogger.With("type", "disk").Errorf("failed to delete disk files with taskID(%s): %v", taskID, err)
+			logger.StorageGCLogger.With("type", "disk").Errorf("failed to delete disk files with taskID(%s): %v", taskID, err)
 			synclock.UnLock(taskID, false)
 			continue
 		}
 		synclock.UnLock(taskID, false)
 	}
-	logger.GcLogger.With("type", "disk").Infof("at most %d tasks can be cleaned up, actual gc %d tasks", len(gcTaskIDs), realGCCount)
+	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)
 	return nil
 }
diff --git a/cdn/supervisor/cdn/storage/hybrid/hybrid.go b/cdn/supervisor/cdn/storage/hybrid/hybrid.go
index 1b5957c8..77d2484a 100644
--- a/cdn/supervisor/cdn/storage/hybrid/hybrid.go
+++ b/cdn/supervisor/cdn/storage/hybrid/hybrid.go
@@ -378,65 +378,69 @@ func (h *hybridStorageManager) tryShmSpace(url, taskID string, fileLength int64)
 }
 
 func (h *hybridStorageManager) GC() error {
-	logger.GcLogger.With("type", "hybrid").Info("start the hybrid storage gc job")
+	logger.StorageGCLogger.With("type", "hybrid").Info("start the hybrid storage gc job")
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		gcTaskIDs, err := h.diskDriverCleaner.GC("hybrid", false)
 		if err != nil {
-			logger.GcLogger.With("type", "hybrid").Error("gc disk: failed to get gcTaskIds")
+			logger.StorageGCLogger.With("type", "hybrid").Error("gc disk: failed to get gcTaskIDs")
 		}
-		realGCCount := h.gcTasks(gcTaskIDs, true)
-		logger.GcLogger.With("type", "hybrid").Infof("at most %d tasks can be cleaned up from disk, actual gc %d tasks", len(gcTaskIDs), realGCCount)
+		actualGCTasks := h.gcTasks(gcTaskIDs, true)
+		logger.StorageGCLogger.With("type", "hybrid").Infof("at most %d tasks can be cleaned up from disk, actual gc %d tasks", len(gcTaskIDs), len(actualGCTasks))
+		logger.StorageGCLogger.With("type", "hybrid").Debugf("actual gc %s tasks from disk", actualGCTasks)
 	}()
 	if h.hasShm {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
 			gcTaskIDs, err := h.memoryDriverCleaner.GC("hybrid", false)
-			logger.GcLogger.With("type", "hybrid").Infof("at most %d tasks can be cleaned up from memory", len(gcTaskIDs))
 			if err != nil {
-				logger.GcLogger.With("type", "hybrid").Error("gc memory: failed to get gcTaskIds")
+				logger.StorageGCLogger.With("type", "hybrid").Error("gc memory: failed to get gcTaskIds")
 			}
-			h.gcTasks(gcTaskIDs, false)
+			actualGCTasks := h.gcTasks(gcTaskIDs, false)
+			logger.StorageGCLogger.With("type", "hybrid").Infof("at most %d tasks can be cleaned up from memory, actual gc %d tasks", len(gcTaskIDs),
+				len(actualGCTasks))
+			logger.StorageGCLogger.With("type", "hybrid").Debugf("actual gc %s tasks from memory", actualGCTasks)
 		}()
 	}
 	wg.Wait()
 	return nil
 }
 
-func (h *hybridStorageManager) gcTasks(gcTaskIDs []string, isDisk bool) int {
-	var realGCCount int
+func (h *hybridStorageManager) gcTasks(gcTaskIDs []string, isDisk bool) (actualGCTasks []string) {
+	var realGCTasks []string
 	for _, taskID := range gcTaskIDs {
 		// try to ensure the taskID is not using again
 		if _, exist := h.taskManager.Exist(taskID); exist {
 			continue
 		}
-		realGCCount++
+		realGCTasks = append(realGCTasks, taskID)
 		synclock.Lock(taskID, false)
 		if isDisk {
 			if err := h.deleteDiskFiles(taskID); err != nil {
-				logger.GcLogger.With("type", "hybrid").Errorf("gc disk: failed to delete disk files with taskID(%s): %v", taskID, err)
+				logger.StorageGCLogger.With("type", "hybrid").Errorf("gc disk: failed to delete disk files with taskID(%s): %v", taskID, err)
 				synclock.UnLock(taskID, false)
 				continue
 			}
 		} else {
 			if err := h.deleteMemoryFiles(taskID); err != nil {
-				logger.GcLogger.With("type", "hybrid").Errorf("gc memory: failed to delete memory files with taskID(%s): %v", taskID, err)
+				logger.StorageGCLogger.With("type", "hybrid").Errorf("gc memory: failed to delete memory files with taskID(%s): %v", taskID, err)
 				synclock.UnLock(taskID, false)
 				continue
 			}
 		}
 		synclock.UnLock(taskID, false)
 	}
-	return realGCCount
+
+	return realGCTasks
 }
 
 func (h *hybridStorageManager) getMemoryUsableSpace() unit.Bytes {
 	totalSize, freeSize, err := h.memoryDriver.GetTotalAndFreeSpace()
 	if err != nil {
-		logger.GcLogger.With("type", "hybrid").Errorf("failed to get total and free space of memory: %v", err)
+		logger.StorageGCLogger.With("type", "hybrid").Errorf("failed to get total and free space of memory: %v", err)
 		return 0
 	}
 	// 如果内存总容量大于等于 72G，则返回内存的剩余可用空间
diff --git a/cdn/supervisor/cdn/storage/storage_gc.go b/cdn/supervisor/cdn/storage/storage_gc.go
index 7b045ed0..dc0395f9 100644
--- a/cdn/supervisor/cdn/storage/storage_gc.go
+++ b/cdn/supervisor/cdn/storage/storage_gc.go
@@ -77,7 +77,7 @@ func (cleaner *cleaner) GC(storagePattern string, force bool) ([]string, error)
 		}
 	}
 
-	logger.GcLogger.With("type", storagePattern).Debugf("storage is insufficient, start to exec gc with fullGC: %t", fullGC)
+	logger.StorageGCLogger.With("type", storagePattern).Debugf("storage is insufficient, start to exec gc with fullGC: %t", fullGC)
 
 	gapTasks := treemap.NewWith(godsutils.Int64Comparator)
 	intervalTasks := treemap.NewWith(godsutils.Int64Comparator)
@@ -87,10 +87,10 @@ 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.GcLogger.With("type", storagePattern).Debugf("start to walk path(%s)", path)
+		logger.StorageGCLogger.With("type", storagePattern).Debugf("start to walk path(%s)", path)
 
 		if err != nil {
-			logger.GcLogger.With("type", storagePattern).Errorf("failed to access path(%s): %v", path, err)
+			logger.StorageGCLogger.With("type", storagePattern).Errorf("failed to access path(%s): %v", path, err)
 			return err
 		}
 		if info.IsDir() {
@@ -116,13 +116,13 @@ func (cleaner *cleaner) GC(storagePattern string, force bool) ([]string, error)
 
 		metadata, err := cleaner.storageManager.ReadFileMetadata(taskID)
 		if err != nil || metadata == nil {
-			logger.GcLogger.With("type", storagePattern).Debugf("taskID: %s, failed to get metadata: %v", taskID, err)
+			logger.StorageGCLogger.With("type", storagePattern).Debugf("taskID: %s, failed to get metadata: %v", taskID, err)
 			gcTaskIDs = append(gcTaskIDs, taskID)
 			return nil
 		}
 		// put taskID into gapTasks or intervalTasks which will sort by some rules
 		if err := cleaner.sortInert(gapTasks, intervalTasks, metadata); err != nil {
-			logger.GcLogger.With("type", storagePattern).Errorf("failed to parse inert metadata(%#v): %v", metadata, err)
+			logger.StorageGCLogger.With("type", storagePattern).Errorf("failed to parse inert metadata(%#v): %v", metadata, err)
 		}
 
 		return nil
diff --git a/cdn/supervisor/service.go b/cdn/supervisor/service.go
index 751cc580..49ebf75c 100644
--- a/cdn/supervisor/service.go
+++ b/cdn/supervisor/service.go
@@ -40,7 +40,7 @@ func IsResourcesLacked(err error) bool {
 
 type CDNService interface {
 	// RegisterSeedTask registers seed task
-	RegisterSeedTask(ctx context.Context, clientAddr string, registerTask *task.SeedTask) (<-chan *task.PieceInfo, error)
+	RegisterSeedTask(ctx context.Context, clientAddr string, registerTask *task.SeedTask) (*task.SeedTask, <-chan *task.PieceInfo, error)
 
 	// GetSeedPieces returns pieces associated with taskID, which are sorted by pieceNum
 	GetSeedPieces(taskID string) (pieces []*task.PieceInfo, err error)
@@ -63,14 +63,17 @@ func NewCDNService(taskManager task.Manager, cdnManager cdn.Manager, progressMan
 	}, nil
 }
 
-func (service *cdnService) RegisterSeedTask(ctx context.Context, clientAddr string, registerTask *task.SeedTask) (<-chan *task.PieceInfo, error) {
-	if _, err := service.taskManager.AddOrUpdate(registerTask); err != nil {
-		return nil, err
+func (service *cdnService) RegisterSeedTask(ctx context.Context, clientAddr string, registerTask *task.SeedTask) (*task.SeedTask,
+	<-chan *task.PieceInfo, error) {
+	seedTask, err := service.taskManager.AddOrUpdate(registerTask)
+	if err != nil {
+		return nil, nil, err
 	}
-	if err := service.triggerCdnSyncAction(ctx, registerTask.ID); err != nil {
-		return nil, err
+	if err = service.triggerCdnSyncAction(ctx, registerTask.ID); err != nil {
+		return seedTask, nil, err
 	}
-	return service.progressManager.WatchSeedProgress(ctx, clientAddr, registerTask.ID)
+	pieceChan, err := service.progressManager.WatchSeedProgress(ctx, clientAddr, registerTask.ID)
+	return seedTask, pieceChan, err
 }
 
 // triggerCdnSyncAction trigger cdn sync action
diff --git a/cdn/supervisor/task/manager.go b/cdn/supervisor/task/manager.go
index 1bc5498c..0d83bd7a 100644
--- a/cdn/supervisor/task/manager.go
+++ b/cdn/supervisor/task/manager.go
@@ -151,19 +151,19 @@ func (tm *manager) AddOrUpdate(registerTask *SeedTask) (seedTask *SeedTask, err
 		return seedTask, err
 	}
 	seedTask.SourceFileLength = sourceFileLength
-	seedTask.Log().Debugf("success get file content length: %d", sourceFileLength)
-
 	// if success to get the information successfully with the req.Header then update the task.UrlMeta to registerTask.UrlMeta.
 	if registerTask.Header != nil {
 		seedTask.Header = registerTask.Header
 	}
 
 	// calculate piece size and update the PieceSize and PieceTotal
-	if registerTask.PieceSize <= 0 {
-		pieceSize := util.ComputePieceSize(registerTask.SourceFileLength)
-		seedTask.PieceSize = int32(pieceSize)
-		seedTask.Log().Debugf("piece size calculate result: %s", unit.ToBytes(int64(pieceSize)))
+	pieceSize := util.ComputePieceSize(registerTask.SourceFileLength)
+	seedTask.PieceSize = int32(pieceSize)
+	if sourceFileLength > 0 {
+		seedTask.TotalPieceCount = int32((registerTask.SourceFileLength + (int64(pieceSize) - 1)) / int64(pieceSize))
 	}
+	seedTask.Log().Infof("success get content length: %s, pieceSize: %s, totalPieceCount: %d", unit.ToBytes(seedTask.SourceFileLength),
+		unit.ToBytes(int64(seedTask.PieceSize)), seedTask.TotalPieceCount)
 	return seedTask, nil
 }
 
@@ -232,33 +232,32 @@ const (
 )
 
 func (tm *manager) GC() error {
-	logger.GcLogger.Info("start the task meta gc job")
+	logger.MetaGCLogger.Info("start the task meta gc job")
 	startTime := time.Now()
-
-	totalTaskNums := 0
-	removedTaskCount := 0
+	var gcTasks []string
+	var remainingTasks []string
 	tm.accessTimeMap.Range(func(key, value interface{}) bool {
-		totalTaskNums++
 		taskID := key.(string)
 		synclock.Lock(taskID, false)
 		defer synclock.UnLock(taskID, false)
 		atime := value.(time.Time)
 		if time.Since(atime) < tm.config.ExpireTime {
+			remainingTasks = append(remainingTasks, taskID)
 			return true
 		}
-
+		gcTasks = append(gcTasks, taskID)
 		// gc task memory data
-		logger.GcLogger.With("type", "meta").Infof("gc task: start to deal with task: %s", taskID)
+		logger.MetaGCLogger.With("type", "meta").Infof("gc task: %s", taskID)
 		tm.deleteTask(taskID)
-		removedTaskCount++
 		return true
 	})
 
 	// slow GC detected, report it with a log warning
 	if timeDuring := time.Since(startTime); timeDuring > gcTasksTimeout {
-		logger.GcLogger.With("type", "meta").Warnf("gc tasks: %d cost: %.3f", removedTaskCount, timeDuring.Seconds())
+		logger.MetaGCLogger.With("type", "meta").Warnf("gc tasks: %d cost: %.3f", len(gcTasks), timeDuring.Seconds())
 	}
-	logger.GcLogger.With("type", "meta").Infof("%d tasks were successfully cleared, leaving %d tasks remaining", removedTaskCount,
-		totalTaskNums-removedTaskCount)
+	logger.MetaGCLogger.With("type", "meta").Infof("%d tasks were successfully cleared, leaving %d tasks remaining", len(gcTasks),
+		len(remainingTasks))
+	logger.MetaGCLogger.With("type", "meta").Debugf("tasks %s were successfully cleared, leaving tasks %s remaining", gcTasks, remainingTasks)
 	return nil
 }
diff --git a/client/config/peerhost.go b/client/config/peerhost.go
index 46bbec73..cbb3f5b8 100644
--- a/client/config/peerhost.go
+++ b/client/config/peerhost.go
@@ -173,6 +173,7 @@ type DownloadOption struct {
 	PeerGRPC             ListenOption         `mapstructure:"peerGRPC" yaml:"peerGRPC"`
 	CalculateDigest      bool                 `mapstructure:"calculateDigest" yaml:"calculateDigest"`
 	TransportOption      *TransportOption     `mapstructure:"transportOption" yaml:"transportOption"`
+	GetPiecesMaxRetry    int                  `mapstructure:"getPiecesMaxRetry" yaml:"getPiecesMaxRetry"`
 }
 
 type TransportOption struct {
diff --git a/client/config/peerhost_darwin.go b/client/config/peerhost_darwin.go
index f5730541..193fdaa5 100644
--- a/client/config/peerhost_darwin.go
+++ b/client/config/peerhost_darwin.go
@@ -60,6 +60,7 @@ var peerHostConfig = DaemonOption{
 	Download: DownloadOption{
 		CalculateDigest:      true,
 		PieceDownloadTimeout: 30 * time.Second,
+		GetPiecesMaxRetry:    100,
 		TotalRateLimit: clientutil.RateLimit{
 			Limit: rate.Limit(DefaultTotalDownloadLimit),
 		},
diff --git a/client/config/peerhost_linux.go b/client/config/peerhost_linux.go
index b5795a5b..02a9d2b8 100644
--- a/client/config/peerhost_linux.go
+++ b/client/config/peerhost_linux.go
@@ -60,6 +60,7 @@ var peerHostConfig = DaemonOption{
 	Download: DownloadOption{
 		CalculateDigest:      true,
 		PieceDownloadTimeout: 30 * time.Second,
+		GetPiecesMaxRetry:    100,
 		TotalRateLimit: clientutil.RateLimit{
 			Limit: rate.Limit(DefaultTotalDownloadLimit),
 		},
diff --git a/client/daemon/daemon.go b/client/daemon/daemon.go
index 06daa811..a84e7e25 100644
--- a/client/daemon/daemon.go
+++ b/client/daemon/daemon.go
@@ -46,8 +46,8 @@ import (
 	"d7y.io/dragonfly/v2/client/daemon/upload"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/dfnet"
-	"d7y.io/dragonfly/v2/internal/dfpath"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/dfpath"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/reachable"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/manager"
@@ -168,7 +168,7 @@ func New(opt *config.DaemonOption, d dfpath.Dfpath) (Daemon, error) {
 		return nil, err
 	}
 	peerTaskManager, err := peer.NewPeerTaskManager(host, pieceManager, storageManager, sched, opt.Scheduler,
-		opt.Download.PerPeerRateLimit.Limit, opt.Storage.Multiplex, opt.Download.CalculateDigest)
+		opt.Download.PerPeerRateLimit.Limit, opt.Storage.Multiplex, opt.Download.CalculateDigest, opt.Download.GetPiecesMaxRetry)
 	if err != nil {
 		return nil, err
 	}
diff --git a/client/daemon/peer/peertask_base.go b/client/daemon/peer/peertask_base.go
index 5b570259..74c28445 100644
--- a/client/daemon/peer/peertask_base.go
+++ b/client/daemon/peer/peertask_base.go
@@ -103,6 +103,8 @@ type peerTask struct {
 	peerPacketReady chan bool
 	// pieceParallelCount stands the piece parallel count from peerPacket
 	pieceParallelCount *atomic.Int32
+	// getPiecesMaxRetry stands max retry to get pieces from one peer packet
+	getPiecesMaxRetry int
 
 	// done channel will be close when peer task is finished
 	done chan struct{}
@@ -321,7 +323,7 @@ func (pt *peerTask) isExitPeerPacketCode(pp *scheduler.PeerPacket) bool {
 		pt.failedCode = pp.Code
 		pt.failedReason = fmt.Sprintf("receive exit peer packet with code %d", pp.Code)
 		return true
-	case base.Code_SchedError:
+	case base.Code_SchedError, base.Code_SchedTaskStatusError:
 		// 5xxx
 		pt.failedCode = pp.Code
 		pt.failedReason = fmt.Sprintf("receive exit peer packet with code %d", pp.Code)
@@ -502,8 +504,8 @@ loop:
 
 func (pt *peerTask) init(piecePacket *base.PiecePacket, pieceBufferSize uint32) (chan *DownloadPieceRequest, bool) {
 	pt.contentLength.Store(piecePacket.ContentLength)
-	if pt.contentLength.Load() > 0 {
-		pt.span.SetAttributes(config.AttributeTaskContentLength.Int64(pt.contentLength.Load()))
+	if piecePacket.ContentLength > 0 {
+		pt.span.SetAttributes(config.AttributeTaskContentLength.Int64(piecePacket.ContentLength))
 	}
 	if err := pt.callback.Init(pt); err != nil {
 		pt.span.RecordError(err)
@@ -579,7 +581,7 @@ func (pt *peerTask) waitAvailablePeerPacket() (int32, bool) {
 			// preparePieceTasksByPeer func already send piece result with error
 			pt.Infof("new peer client ready, main peer: %s", pt.peerPacket.Load().(*scheduler.PeerPacket).MainPeer)
 			// research from piece 0
-			return pt.getNextPieceNum(0), true
+			return 0, true
 		}
 		// when scheduler says base.Code_SchedNeedBackSource, receivePeerPacket will close pt.peerPacketReady
 		pt.Infof("start download from source due to base.Code_SchedNeedBackSource")
@@ -605,6 +607,7 @@ func (pt *peerTask) waitAvailablePeerPacket() (int32, bool) {
 }
 
 func (pt *peerTask) dispatchPieceRequest(pieceRequestCh chan *DownloadPieceRequest, piecePacket *base.PiecePacket) {
+	pt.Debugf("dispatch piece request, piece count: %d", len(piecePacket.PieceInfos))
 	for _, piece := range piecePacket.PieceInfos {
 		pt.Infof("get piece %d from %s/%s, md5: %s, start: %d, size: %d",
 			piece.PieceNum, piecePacket.DstAddr, piecePacket.DstPid, piece.PieceMd5, piece.RangeStart, piece.RangeSize)
@@ -721,7 +724,9 @@ func (pt *peerTask) isCompleted() bool {
 
 func (pt *peerTask) preparePieceTasks(request *base.PieceTaskRequest) (p *base.PiecePacket, err error) {
 	defer pt.recoverFromPanic()
+	var retryCount int
 prepare:
+	retryCount++
 	peerPacket := pt.peerPacket.Load().(*scheduler.PeerPacket)
 	pt.pieceParallelCount.Store(peerPacket.ParallelCount)
 	request.DstPid = peerPacket.MainPeer.PeerId
@@ -730,6 +735,10 @@ prepare:
 		return
 	}
 	if err == errPeerPacketChanged {
+		if pt.getPiecesMaxRetry > 0 && retryCount > pt.getPiecesMaxRetry {
+			err = fmt.Errorf("get pieces max retry count reached")
+			return
+		}
 		goto prepare
 	}
 	for _, peer := range peerPacket.StealPeers {
@@ -739,6 +748,10 @@ prepare:
 			return
 		}
 		if err == errPeerPacketChanged {
+			if pt.getPiecesMaxRetry > 0 && retryCount > pt.getPiecesMaxRetry {
+				err = fmt.Errorf("get pieces max retry count reached")
+				return
+			}
 			goto prepare
 		}
 	}
@@ -788,7 +801,7 @@ retry:
 		code = de.Code
 	}
 	pt.Errorf("get piece task from peer %s error: %s, code: %d", peer.PeerId, err, code)
-	perr := pt.peerPacketStream.Send(&scheduler.PieceResult{
+	sendError := pt.peerPacketStream.Send(&scheduler.PieceResult{
 		TaskId:        pt.taskID,
 		SrcPid:        pt.peerID,
 		DstPid:        peer.PeerId,
@@ -798,9 +811,9 @@ retry:
 		HostLoad:      nil,
 		FinishedCount: -1,
 	})
-	if perr != nil {
-		span.RecordError(perr)
-		pt.Errorf("send piece result error: %s, code to send: %d", err, code)
+	if sendError != nil {
+		span.RecordError(sendError)
+		pt.Errorf("send piece result error: %s, code to send: %d", sendError, code)
 	}
 
 	if code == base.Code_CDNTaskNotFound && curPeerPacket == pt.peerPacket.Load().(*scheduler.PeerPacket) {
@@ -816,51 +829,75 @@ func (pt *peerTask) getPieceTasks(span trace.Span, curPeerPacket *scheduler.Peer
 		count             int
 	)
 	p, _, err := retry.Run(pt.ctx, func() (interface{}, bool, error) {
-		pp, getErr := dfclient.GetPieceTasks(pt.ctx, peer, request)
+		piecePacket, getError := dfclient.GetPieceTasks(pt.ctx, peer, request)
 		// when GetPieceTasks returns err, exit retry
-		if getErr != nil {
-			span.RecordError(getErr)
-			// fast way to exit retry
-			lastPeerPacket := pt.peerPacket.Load().(*scheduler.PeerPacket)
-			if curPeerPacket.MainPeer.PeerId != lastPeerPacket.MainPeer.PeerId {
-				pt.Warnf("get piece tasks with error: %s, but peer packet changed, switch to new peer packet, current destPeer %s, new destPeer %s", getErr,
-					curPeerPacket.MainPeer.PeerId, lastPeerPacket.MainPeer.PeerId)
-				peerPacketChanged = true
-				return nil, true, nil
-			}
-			return nil, true, getErr
-		}
-		// by santong: when peer return empty, retry later
-		if len(pp.PieceInfos) == 0 {
-			count++
-			er := pt.peerPacketStream.Send(&scheduler.PieceResult{
-				TaskId:        pt.taskID,
-				SrcPid:        pt.peerID,
-				DstPid:        peer.PeerId,
-				PieceInfo:     &base.PieceInfo{},
-				Success:       false,
-				Code:          base.Code_ClientWaitPieceReady,
-				HostLoad:      nil,
-				FinishedCount: pt.readyPieces.Settled(),
-			})
-			if er != nil {
-				span.RecordError(er)
-				pt.Errorf("send piece result with base.Code_ClientWaitPieceReady error: %s", er)
+		if getError != nil {
+			pt.Errorf("get piece tasks with error: %s", getError)
+			span.RecordError(getError)
+
+			// fast way 1 to exit retry
+			if de, ok := getError.(*dferrors.DfError); ok {
+				pt.Debugf("get piece task with grpc error, code: %d", de.Code)
+				// bad request, like invalid piece num, just exit
+				if de.Code == base.Code_BadRequest {
+					span.AddEvent("bad request")
+					pt.Warnf("get piece task from peer %s canceled: %s", peer.PeerId, getError)
+					return nil, true, getError
+				}
 			}
-			// fast way to exit retry
+
+			// fast way 2 to exit retry
 			lastPeerPacket := pt.peerPacket.Load().(*scheduler.PeerPacket)
 			if curPeerPacket.MainPeer.PeerId != lastPeerPacket.MainPeer.PeerId {
-				pt.Warnf("get empty pieces and peer packet changed, switch to new peer packet, current destPeer %s, new destPeer %s",
+				pt.Warnf("get piece tasks with error: %s, but peer packet changed, switch to new peer packet, current destPeer %s, new destPeer %s", getError,
 					curPeerPacket.MainPeer.PeerId, lastPeerPacket.MainPeer.PeerId)
 				peerPacketChanged = true
 				return nil, true, nil
 			}
-			span.AddEvent("retry due to empty pieces",
-				trace.WithAttributes(config.AttributeGetPieceRetry.Int(count)))
-			pt.Infof("peer %s returns success but with empty pieces, retry later", peer.PeerId)
-			return nil, false, dferrors.ErrEmptyValue
+			return nil, true, getError
 		}
-		return pp, false, nil
+		// got any pieces
+		if len(piecePacket.PieceInfos) > 0 {
+			return piecePacket, false, nil
+		}
+		// need update metadata
+		if piecePacket.ContentLength > pt.contentLength.Load() || piecePacket.TotalPiece > pt.totalPiece {
+			return piecePacket, false, nil
+		}
+		// invalid request num
+		if piecePacket.TotalPiece > -1 && uint32(piecePacket.TotalPiece) <= request.StartNum {
+			pt.Warnf("invalid start num: %d, total piece: %d", request.StartNum, piecePacket.TotalPiece)
+			return piecePacket, false, nil
+		}
+
+		// by santong: when peer return empty, retry later
+		sendError := pt.peerPacketStream.Send(&scheduler.PieceResult{
+			TaskId:        pt.taskID,
+			SrcPid:        pt.peerID,
+			DstPid:        peer.PeerId,
+			PieceInfo:     &base.PieceInfo{},
+			Success:       false,
+			Code:          base.Code_ClientWaitPieceReady,
+			HostLoad:      nil,
+			FinishedCount: pt.readyPieces.Settled(),
+		})
+		if sendError != nil {
+			span.RecordError(sendError)
+			pt.Errorf("send piece result with base.Code_ClientWaitPieceReady error: %s", sendError)
+		}
+		// fast way to exit retry
+		lastPeerPacket := pt.peerPacket.Load().(*scheduler.PeerPacket)
+		if curPeerPacket.MainPeer.PeerId != lastPeerPacket.MainPeer.PeerId {
+			pt.Warnf("get empty pieces and peer packet changed, switch to new peer packet, current destPeer %s, new destPeer %s",
+				curPeerPacket.MainPeer.PeerId, lastPeerPacket.MainPeer.PeerId)
+			peerPacketChanged = true
+			return nil, true, nil
+		}
+		count++
+		span.AddEvent("retry due to empty pieces",
+			trace.WithAttributes(config.AttributeGetPieceRetry.Int(count)))
+		pt.Infof("peer %s returns success but with empty pieces, retry later", peer.PeerId)
+		return nil, false, dferrors.ErrEmptyValue
 	}, 0.05, 0.2, 40, nil)
 	if peerPacketChanged {
 		return nil, errPeerPacketChanged
@@ -877,13 +914,14 @@ func (pt *peerTask) getNextPieceNum(cur int32) int32 {
 		return -1
 	}
 	i := cur
+	// try to find next not requested piece
 	for ; pt.requestedPieces.IsSet(i); i++ {
 	}
 	if pt.totalPiece > 0 && i >= pt.totalPiece {
 		// double check, re-search not success or not requested pieces
 		for i = int32(0); pt.requestedPieces.IsSet(i); i++ {
 		}
-		if pt.totalPiece > 0 && i >= int32(pt.totalPiece) {
+		if pt.totalPiece > 0 && i >= pt.totalPiece {
 			return -1
 		}
 	}
diff --git a/client/daemon/peer/peertask_file.go b/client/daemon/peer/peertask_file.go
index 4b5ecd4a..55b22bbf 100644
--- a/client/daemon/peer/peertask_file.go
+++ b/client/daemon/peer/peertask_file.go
@@ -28,7 +28,7 @@ import (
 
 	"d7y.io/dragonfly/v2/client/config"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	schedulerclient "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
@@ -86,7 +86,8 @@ func newFilePeerTask(ctx context.Context,
 	request *FilePeerTaskRequest,
 	schedulerClient schedulerclient.SchedulerClient,
 	schedulerOption config.SchedulerOption,
-	perPeerRateLimit rate.Limit) (context.Context, *filePeerTask, *TinyData, error) {
+	perPeerRateLimit rate.Limit,
+	getPiecesMaxRetry int) (context.Context, *filePeerTask, *TinyData, error) {
 	ctx, span := tracer.Start(ctx, config.SpanFilePeerTask, trace.WithSpanKind(trace.SpanKindClient))
 	span.SetAttributes(config.AttributePeerHost.String(host.Uuid))
 	span.SetAttributes(semconv.NetHostIPKey.String(host.Ip))
@@ -196,6 +197,7 @@ func newFilePeerTask(ctx context.Context,
 			contentLength:       atomic.NewInt64(-1),
 			pieceParallelCount:  atomic.NewInt32(0),
 			totalPiece:          -1,
+			getPiecesMaxRetry:   getPiecesMaxRetry,
 			schedulerOption:     schedulerOption,
 			schedulerClient:     schedulerClient,
 			limiter:             limiter,
diff --git a/client/daemon/peer/peertask_manager.go b/client/daemon/peer/peertask_manager.go
index 486453a5..964ec5a9 100644
--- a/client/daemon/peer/peertask_manager.go
+++ b/client/daemon/peer/peertask_manager.go
@@ -117,6 +117,8 @@ type peerTaskManager struct {
 	enableMultiplex bool
 
 	calculateDigest bool
+
+	getPiecesMaxRetry int
 }
 
 func NewPeerTaskManager(
@@ -127,18 +129,20 @@ func NewPeerTaskManager(
 	schedulerOption config.SchedulerOption,
 	perPeerRateLimit rate.Limit,
 	multiplex bool,
-	calculateDigest bool) (TaskManager, error) {
+	calculateDigest bool,
+	getPiecesMaxRetry int) (TaskManager, error) {
 
 	ptm := &peerTaskManager{
-		host:             host,
-		runningPeerTasks: sync.Map{},
-		pieceManager:     pieceManager,
-		storageManager:   storageManager,
-		schedulerClient:  schedulerClient,
-		schedulerOption:  schedulerOption,
-		perPeerRateLimit: perPeerRateLimit,
-		enableMultiplex:  multiplex,
-		calculateDigest:  calculateDigest,
+		host:              host,
+		runningPeerTasks:  sync.Map{},
+		pieceManager:      pieceManager,
+		storageManager:    storageManager,
+		schedulerClient:   schedulerClient,
+		schedulerOption:   schedulerOption,
+		perPeerRateLimit:  perPeerRateLimit,
+		enableMultiplex:   multiplex,
+		calculateDigest:   calculateDigest,
+		getPiecesMaxRetry: getPiecesMaxRetry,
 	}
 	return ptm, nil
 }
@@ -159,7 +163,7 @@ func (ptm *peerTaskManager) StartFilePeerTask(ctx context.Context, req *FilePeer
 		limit = rate.Limit(req.Limit)
 	}
 	ctx, pt, tiny, err := newFilePeerTask(ctx, ptm.host, ptm.pieceManager,
-		req, ptm.schedulerClient, ptm.schedulerOption, limit)
+		req, ptm.schedulerClient, ptm.schedulerOption, limit, ptm.getPiecesMaxRetry)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -217,8 +221,7 @@ func (ptm *peerTaskManager) StartStreamPeerTask(ctx context.Context, req *schedu
 	}
 
 	start := time.Now()
-	ctx, pt, tiny, err := newStreamPeerTask(ctx, ptm.host, ptm.pieceManager,
-		req, ptm.schedulerClient, ptm.schedulerOption, ptm.perPeerRateLimit)
+	ctx, pt, tiny, err := newStreamPeerTask(ctx, ptm, req)
 	if err != nil {
 		return nil, nil, err
 	}
diff --git a/client/daemon/peer/peertask_reuse.go b/client/daemon/peer/peertask_reuse.go
index 43ced0af..1bcf76f4 100644
--- a/client/daemon/peer/peertask_reuse.go
+++ b/client/daemon/peer/peertask_reuse.go
@@ -29,7 +29,7 @@ import (
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/client/daemon/storage"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 )
diff --git a/client/daemon/peer/peertask_stream.go b/client/daemon/peer/peertask_stream.go
index 82e4bf0e..7d8e0f59 100644
--- a/client/daemon/peer/peertask_stream.go
+++ b/client/daemon/peer/peertask_stream.go
@@ -31,10 +31,9 @@ import (
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/client/daemon/storage"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
-	schedulerclient "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
 )
 
 // StreamPeerTask represents a peer task with stream io for reading directly without once more disk io
@@ -55,36 +54,32 @@ type streamPeerTask struct {
 var _ StreamPeerTask = (*streamPeerTask)(nil)
 
 func newStreamPeerTask(ctx context.Context,
-	host *scheduler.PeerHost,
-	pieceManager PieceManager,
-	request *scheduler.PeerTaskRequest,
-	schedulerClient schedulerclient.SchedulerClient,
-	schedulerOption config.SchedulerOption,
-	perPeerRateLimit rate.Limit) (context.Context, *streamPeerTask, *TinyData, error) {
+	ptm *peerTaskManager,
+	request *scheduler.PeerTaskRequest) (context.Context, *streamPeerTask, *TinyData, error) {
 	ctx, span := tracer.Start(ctx, config.SpanStreamPeerTask, trace.WithSpanKind(trace.SpanKindClient))
-	span.SetAttributes(config.AttributePeerHost.String(host.Uuid))
-	span.SetAttributes(semconv.NetHostIPKey.String(host.Ip))
+	span.SetAttributes(config.AttributePeerHost.String(ptm.host.Uuid))
+	span.SetAttributes(semconv.NetHostIPKey.String(ptm.host.Ip))
 	span.SetAttributes(config.AttributePeerID.String(request.PeerId))
 	span.SetAttributes(semconv.HTTPURLKey.String(request.Url))
 
 	logger.Debugf("request overview, pid: %s, url: %s, filter: %s, meta: %s, tag: %s",
 		request.PeerId, request.Url, request.UrlMeta.Filter, request.UrlMeta, request.UrlMeta.Tag)
 	// trace register
-	regCtx, cancel := context.WithTimeout(ctx, schedulerOption.ScheduleTimeout.Duration)
+	regCtx, cancel := context.WithTimeout(ctx, ptm.schedulerOption.ScheduleTimeout.Duration)
 	defer cancel()
 	regCtx, regSpan := tracer.Start(regCtx, config.SpanRegisterTask)
 	logger.Infof("step 1: peer %s start to register", request.PeerId)
-	result, err := schedulerClient.RegisterPeerTask(regCtx, request)
+	result, err := ptm.schedulerClient.RegisterPeerTask(regCtx, request)
 	regSpan.RecordError(err)
 	regSpan.End()
 
 	var needBackSource bool
 	if err != nil {
 		if err == context.DeadlineExceeded {
-			logger.Errorf("scheduler did not response in %s", schedulerOption.ScheduleTimeout.Duration)
+			logger.Errorf("scheduler did not response in %s", ptm.schedulerOption.ScheduleTimeout.Duration)
 		}
 		logger.Errorf("step 1: peer %s register failed: %s", request.PeerId, err)
-		if schedulerOption.DisableAutoBackSource {
+		if ptm.schedulerOption.DisableAutoBackSource {
 			logger.Errorf("register peer task failed: %s, peer id: %s, auto back source disabled", err, request.PeerId)
 			span.RecordError(err)
 			span.End()
@@ -92,7 +87,7 @@ func newStreamPeerTask(ctx context.Context,
 		}
 		needBackSource = true
 		// can not detect source or scheduler error, create a new dummy scheduler client
-		schedulerClient = &dummySchedulerClient{}
+		ptm.schedulerClient = &dummySchedulerClient{}
 		result = &scheduler.RegisterResult{TaskId: idgen.TaskID(request.Url, request.UrlMeta)}
 		logger.Warnf("register peer task failed: %s, peer id: %s, try to back source", err, request.PeerId)
 	}
@@ -137,7 +132,7 @@ func newStreamPeerTask(ctx context.Context,
 		}
 	}
 
-	peerPacketStream, err := schedulerClient.ReportPieceResult(ctx, result.TaskId, request)
+	peerPacketStream, err := ptm.schedulerClient.ReportPieceResult(ctx, result.TaskId, request)
 	logger.Infof("step 2: start report peer %s piece result", request.PeerId)
 	if err != nil {
 		defer span.End()
@@ -145,19 +140,19 @@ func newStreamPeerTask(ctx context.Context,
 		return ctx, nil, nil, err
 	}
 	var limiter *rate.Limiter
-	if perPeerRateLimit > 0 {
-		limiter = rate.NewLimiter(perPeerRateLimit, int(perPeerRateLimit))
+	if ptm.perPeerRateLimit > 0 {
+		limiter = rate.NewLimiter(ptm.perPeerRateLimit, int(ptm.perPeerRateLimit))
 	}
 	pt := &streamPeerTask{
 		successPieceCh: make(chan int32),
 		streamDone:     make(chan struct{}),
 		peerTask: peerTask{
 			ctx:                 ctx,
-			host:                host,
+			host:                ptm.host,
 			needBackSource:      needBackSource,
 			request:             request,
 			peerPacketStream:    peerPacketStream,
-			pieceManager:        pieceManager,
+			pieceManager:        ptm.pieceManager,
 			peerPacketReady:     make(chan bool, 1),
 			peerID:              request.PeerId,
 			taskID:              result.TaskId,
@@ -172,8 +167,9 @@ func newStreamPeerTask(ctx context.Context,
 			contentLength:       atomic.NewInt64(-1),
 			pieceParallelCount:  atomic.NewInt32(0),
 			totalPiece:          -1,
-			schedulerOption:     schedulerOption,
-			schedulerClient:     schedulerClient,
+			getPiecesMaxRetry:   ptm.getPiecesMaxRetry,
+			schedulerOption:     ptm.schedulerOption,
+			schedulerClient:     ptm.schedulerClient,
 			limiter:             limiter,
 			completedLength:     atomic.NewInt64(0),
 			usedTraffic:         atomic.NewUint64(0),
diff --git a/client/daemon/rpcserver/rpcserver.go b/client/daemon/rpcserver/rpcserver.go
index faa50945..0c6c5c32 100644
--- a/client/daemon/rpcserver/rpcserver.go
+++ b/client/daemon/rpcserver/rpcserver.go
@@ -32,7 +32,7 @@ import (
 	"d7y.io/dragonfly/v2/client/daemon/storage"
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	dfdaemongrpc "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
 	dfdaemonserver "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/server"
@@ -88,6 +88,9 @@ func (m *server) GetPieceTasks(ctx context.Context, request *base.PieceTaskReque
 	p, err := m.storageManager.GetPieces(ctx, request)
 	if err != nil {
 		code := base.Code_UnknownError
+		if err == dferrors.ErrInvalidArgument {
+			code = base.Code_BadRequest
+		}
 		if err != storage.ErrTaskNotFound {
 			logger.Errorf("get piece tasks error: %s, task id: %s, src peer: %s, dst peer: %s, piece num: %d, limit: %d",
 				err, request.TaskId, request.SrcPid, request.DstPid, request.StartNum, request.Limit)
diff --git a/client/daemon/storage/local_storage.go b/client/daemon/storage/local_storage.go
index 5fc96670..c9c3e343 100644
--- a/client/daemon/storage/local_storage.go
+++ b/client/daemon/storage/local_storage.go
@@ -28,7 +28,6 @@ import (
 	"go.uber.org/atomic"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
-	"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/util/digestutils"
@@ -321,17 +320,22 @@ func (t *localTaskStore) GetPieces(ctx context.Context, req *base.PieceTaskReque
 		return nil, ErrInvalidDigest
 	}
 
-	var pieces []*base.PieceInfo
 	t.RLock()
 	defer t.RUnlock()
 	t.touch()
-	if t.TotalPieces > 0 && int32(req.StartNum) >= t.TotalPieces {
-		t.Errorf("invalid start num: %d", req.StartNum)
-		return nil, dferrors.ErrInvalidArgument
+	piecePacket := &base.PiecePacket{
+		TaskId:        req.TaskId,
+		DstPid:        t.PeerID,
+		TotalPiece:    t.TotalPieces,
+		ContentLength: t.ContentLength,
+		PieceMd5Sign:  t.PieceMd5Sign,
+	}
+	if t.TotalPieces > -1 && int32(req.StartNum) >= t.TotalPieces {
+		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 {
-			pieces = append(pieces, &base.PieceInfo{
+			piecePacket.PieceInfos = append(piecePacket.PieceInfos, &base.PieceInfo{
 				PieceNum:    piece.Num,
 				RangeStart:  uint64(piece.Range.Start),
 				RangeSize:   uint32(piece.Range.Length),
@@ -341,14 +345,7 @@ func (t *localTaskStore) GetPieces(ctx context.Context, req *base.PieceTaskReque
 			})
 		}
 	}
-	return &base.PiecePacket{
-		TaskId:        req.TaskId,
-		DstPid:        t.PeerID,
-		PieceInfos:    pieces,
-		TotalPiece:    t.TotalPieces,
-		ContentLength: t.ContentLength,
-		PieceMd5Sign:  t.PieceMd5Sign,
-	}, nil
+	return piecePacket, nil
 }
 
 func (t *localTaskStore) CanReclaim() bool {
diff --git a/client/daemon/transport/transport.go b/client/daemon/transport/transport.go
index 800398b7..0f007fa2 100644
--- a/client/daemon/transport/transport.go
+++ b/client/daemon/transport/transport.go
@@ -31,7 +31,7 @@ import (
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/client/daemon/peer"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"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"
diff --git a/cmd/cdn/cmd/root.go b/cmd/cdn/cmd/root.go
index dc4dd041..71c87d31 100644
--- a/cmd/cdn/cmd/root.go
+++ b/cmd/cdn/cmd/root.go
@@ -28,7 +28,7 @@ import (
 	"d7y.io/dragonfly/v2/cmd/dependency"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/dflog/logcore"
-	"d7y.io/dragonfly/v2/internal/dfpath"
+	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/source"
 	"d7y.io/dragonfly/v2/version"
 )
diff --git a/cmd/dependency/dependency.go b/cmd/dependency/dependency.go
index af6f3b95..bff19b71 100644
--- a/cmd/dependency/dependency.go
+++ b/cmd/dependency/dependency.go
@@ -49,7 +49,7 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/dflog/logcore"
 	"d7y.io/dragonfly/v2/internal/dfnet"
-	"d7y.io/dragonfly/v2/internal/dfpath"
+	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/unit"
 	"d7y.io/dragonfly/v2/pkg/util/hostutils"
 	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
diff --git a/cmd/dependency/plugin_cmd.go b/cmd/dependency/plugin_cmd.go
index d3edcadb..b0c5469a 100644
--- a/cmd/dependency/plugin_cmd.go
+++ b/cmd/dependency/plugin_cmd.go
@@ -23,8 +23,8 @@ import (
 
 	"github.com/spf13/cobra"
 
-	"d7y.io/dragonfly/v2/internal/dfpath"
 	"d7y.io/dragonfly/v2/internal/dfplugin"
+	"d7y.io/dragonfly/v2/pkg/dfpath"
 )
 
 var PluginCmd = &cobra.Command{
diff --git a/cmd/dfget/cmd/daemon.go b/cmd/dfget/cmd/daemon.go
index f084248b..bcbc28e5 100644
--- a/cmd/dfget/cmd/daemon.go
+++ b/cmd/dfget/cmd/daemon.go
@@ -33,7 +33,7 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/dflog/logcore"
 	"d7y.io/dragonfly/v2/internal/dfnet"
-	"d7y.io/dragonfly/v2/internal/dfpath"
+	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/client"
 	"d7y.io/dragonfly/v2/version"
 )
diff --git a/cmd/dfget/cmd/root.go b/cmd/dfget/cmd/root.go
index 8ea5895d..90d56af8 100644
--- a/cmd/dfget/cmd/root.go
+++ b/cmd/dfget/cmd/root.go
@@ -38,8 +38,8 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/dflog/logcore"
 	"d7y.io/dragonfly/v2/internal/dfnet"
-	"d7y.io/dragonfly/v2/internal/dfpath"
 	"d7y.io/dragonfly/v2/pkg/basic"
+	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/client"
 	"d7y.io/dragonfly/v2/pkg/unit"
 	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
diff --git a/cmd/manager/cmd/root.go b/cmd/manager/cmd/root.go
index 6f8e973a..48e4dce5 100644
--- a/cmd/manager/cmd/root.go
+++ b/cmd/manager/cmd/root.go
@@ -26,9 +26,9 @@ import (
 	"d7y.io/dragonfly/v2/cmd/dependency"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/dflog/logcore"
-	"d7y.io/dragonfly/v2/internal/dfpath"
 	"d7y.io/dragonfly/v2/manager"
 	"d7y.io/dragonfly/v2/manager/config"
+	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/version"
 )
 
diff --git a/cmd/scheduler/cmd/root.go b/cmd/scheduler/cmd/root.go
index 8e48086f..3dcc0387 100644
--- a/cmd/scheduler/cmd/root.go
+++ b/cmd/scheduler/cmd/root.go
@@ -26,7 +26,7 @@ import (
 	"d7y.io/dragonfly/v2/cmd/dependency"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/dflog/logcore"
-	"d7y.io/dragonfly/v2/internal/dfpath"
+	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/scheduler"
 	"d7y.io/dragonfly/v2/scheduler/config"
 	"d7y.io/dragonfly/v2/version"
diff --git a/internal/dflog/logcore/init.go b/internal/dflog/logcore/init.go
index d4fd49a0..c091b347 100644
--- a/internal/dflog/logcore/init.go
+++ b/internal/dflog/logcore/init.go
@@ -46,7 +46,7 @@ func InitManager(console bool, dir string) error {
 	if err != nil {
 		return err
 	}
-	logger.SetGcLogger(gcLogger.Sugar())
+	logger.SetGCLogger(gcLogger.Sugar())
 
 	jobLogger, err := CreateLogger(path.Join(logDir, JobLogFileName), false, false)
 	if err != nil {
@@ -80,7 +80,7 @@ func InitScheduler(console bool, dir string) error {
 	if err != nil {
 		return err
 	}
-	logger.SetGcLogger(gcLogger.Sugar())
+	logger.SetGCLogger(gcLogger.Sugar())
 
 	jobLogger, err := CreateLogger(path.Join(logDir, JobLogFileName), false, false)
 	if err != nil {
@@ -120,7 +120,13 @@ func InitCdnSystem(console bool, dir string) error {
 	if err != nil {
 		return err
 	}
-	logger.SetGcLogger(gcLogger.Sugar())
+	logger.SetGCLogger(gcLogger.Sugar())
+
+	storageGCLogger, err := CreateLogger(path.Join(logDir, StorageGCLogFileName), false, false)
+	if err != nil {
+		return err
+	}
+	logger.SetGCLogger(storageGCLogger.Sugar())
 
 	jobLogger, err := CreateLogger(path.Join(logDir, JobLogFileName), false, false)
 	if err != nil {
@@ -171,7 +177,7 @@ func InitDaemon(console bool, dir string) error {
 	if err != nil {
 		return err
 	}
-	logger.SetGcLogger(gcLogger.Sugar())
+	logger.SetGCLogger(gcLogger.Sugar())
 
 	return nil
 }
diff --git a/internal/dflog/logcore/logcore.go b/internal/dflog/logcore/logcore.go
index 662dd58f..8d7f7539 100644
--- a/internal/dflog/logcore/logcore.go
+++ b/internal/dflog/logcore/logcore.go
@@ -28,6 +28,7 @@ var (
 	CoreLogFileName       = "core.log"
 	GrpcLogFileName       = "grpc.log"
 	GCLogFileName         = "gc.log"
+	StorageGCLogFileName  = "storage-gc.log"
 	JobLogFileName        = "job.log"
 	StatPeerLogFileName   = "peer.log"
 	StatSeedLogFileName   = "stat/seed.log"
diff --git a/internal/dflog/logger.go b/internal/dflog/logger.go
index 5d2ee419..457ed194 100644
--- a/internal/dflog/logger.go
+++ b/internal/dflog/logger.go
@@ -26,7 +26,8 @@ import (
 var (
 	CoreLogger       *zap.SugaredLogger
 	GrpcLogger       *zap.SugaredLogger
-	GcLogger         *zap.SugaredLogger
+	MetaGCLogger     *zap.SugaredLogger
+	StorageGCLogger  *zap.SugaredLogger
 	JobLogger        *zap.SugaredLogger
 	KeepAliveLogger  *zap.SugaredLogger
 	StatPeerLogger   *zap.Logger
@@ -42,7 +43,8 @@ func init() {
 		sugar := log.Sugar()
 		SetCoreLogger(sugar)
 		SetGrpcLogger(sugar)
-		SetGcLogger(sugar)
+		SetGCLogger(sugar)
+		SetStorageGCLogger(sugar)
 		SetKeepAliveLogger(sugar)
 		SetStatPeerLogger(log)
 		SetStatSeedLogger(log)
@@ -55,8 +57,12 @@ func SetCoreLogger(log *zap.SugaredLogger) {
 	CoreLogger = log
 }
 
-func SetGcLogger(log *zap.SugaredLogger) {
-	GcLogger = log
+func SetGCLogger(log *zap.SugaredLogger) {
+	MetaGCLogger = log
+}
+
+func SetStorageGCLogger(log *zap.SugaredLogger) {
+	StorageGCLogger = log
 }
 
 func SetKeepAliveLogger(log *zap.SugaredLogger) {
diff --git a/internal/util/util.go b/internal/util/util.go
index 9a558c9d..7b55324c 100644
--- a/internal/util/util.go
+++ b/internal/util/util.go
@@ -29,7 +29,7 @@ const (
 // If the fileLength<0, which means failed to get fileLength
 // and then use the DefaultPieceSize.
 func ComputePieceSize(length int64) uint32 {
-	if length <= 0 || length <= 200*1024*1024 {
+	if length <= 200*1024*1024 {
 		return DefaultPieceSize
 	}
 
diff --git a/manager/manager.go b/manager/manager.go
index 76e82fa0..4483b35a 100644
--- a/manager/manager.go
+++ b/manager/manager.go
@@ -24,7 +24,6 @@ import (
 	"google.golang.org/grpc"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfpath"
 	"d7y.io/dragonfly/v2/manager/cache"
 	"d7y.io/dragonfly/v2/manager/config"
 	"d7y.io/dragonfly/v2/manager/database"
@@ -34,6 +33,7 @@ import (
 	"d7y.io/dragonfly/v2/manager/router"
 	"d7y.io/dragonfly/v2/manager/searcher"
 	"d7y.io/dragonfly/v2/manager/service"
+	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	grpc_manager_server "d7y.io/dragonfly/v2/pkg/rpc/manager/server"
 )
diff --git a/internal/dfpath/dfpath.go b/pkg/dfpath/dfpath.go
similarity index 100%
rename from internal/dfpath/dfpath.go
rename to pkg/dfpath/dfpath.go
diff --git a/internal/dfpath/dfpath_darwin.go b/pkg/dfpath/dfpath_darwin.go
similarity index 100%
rename from internal/dfpath/dfpath_darwin.go
rename to pkg/dfpath/dfpath_darwin.go
diff --git a/internal/dfpath/dfpath_linux.go b/pkg/dfpath/dfpath_linux.go
similarity index 100%
rename from internal/dfpath/dfpath_linux.go
rename to pkg/dfpath/dfpath_linux.go
diff --git a/internal/dfpath/dfpath_test.go b/pkg/dfpath/dfpath_test.go
similarity index 100%
rename from internal/dfpath/dfpath_test.go
rename to pkg/dfpath/dfpath_test.go
diff --git a/internal/idgen/host_id.go b/pkg/idgen/host_id.go
similarity index 100%
rename from internal/idgen/host_id.go
rename to pkg/idgen/host_id.go
diff --git a/internal/idgen/host_id_test.go b/pkg/idgen/host_id_test.go
similarity index 100%
rename from internal/idgen/host_id_test.go
rename to pkg/idgen/host_id_test.go
diff --git a/internal/idgen/peer_id.go b/pkg/idgen/peer_id.go
similarity index 100%
rename from internal/idgen/peer_id.go
rename to pkg/idgen/peer_id.go
diff --git a/internal/idgen/peer_id_test.go b/pkg/idgen/peer_id_test.go
similarity index 100%
rename from internal/idgen/peer_id_test.go
rename to pkg/idgen/peer_id_test.go
diff --git a/internal/idgen/task_id.go b/pkg/idgen/task_id.go
similarity index 87%
rename from internal/idgen/task_id.go
rename to pkg/idgen/task_id.go
index 8b117029..3287ec83 100644
--- a/internal/idgen/task_id.go
+++ b/pkg/idgen/task_id.go
@@ -27,15 +27,13 @@ import (
 // TaskID generates a taskId.
 // filter is separated by & character.
 func TaskID(url string, meta *base.UrlMeta) string {
-	var data []string
-
-	filter := ""
-	if meta != nil {
-		filter = meta.Filter
+	var filters []string
+	if meta != nil && meta.Filter != "" {
+		filters = strings.Split(meta.Filter, "&")
 	}
 
-	data = append(data, urlutils.FilterURLParam(url, strings.Split(filter, "&")))
-
+	var data []string
+	data = append(data, urlutils.FilterURLParam(url, filters))
 	if meta != nil {
 		if meta.Digest != "" {
 			data = append(data, meta.Digest)
diff --git a/internal/idgen/task_id_test.go b/pkg/idgen/task_id_test.go
similarity index 100%
rename from internal/idgen/task_id_test.go
rename to pkg/idgen/task_id_test.go
diff --git a/internal/idgen/uuid.go b/pkg/idgen/uuid.go
similarity index 100%
rename from internal/idgen/uuid.go
rename to pkg/idgen/uuid.go
diff --git a/internal/idgen/uuid_test.go b/pkg/idgen/uuid_test.go
similarity index 100%
rename from internal/idgen/uuid_test.go
rename to pkg/idgen/uuid_test.go
diff --git a/pkg/log/log.go b/pkg/log/log.go
index dc4a2c1f..1ba5cbb2 100644
--- a/pkg/log/log.go
+++ b/pkg/log/log.go
@@ -23,7 +23,7 @@ import (
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/dflog/logcore"
-	"d7y.io/dragonfly/v2/internal/dfpath"
+	"d7y.io/dragonfly/v2/pkg/dfpath"
 )
 
 // SetCoreLevel sets core log level, export internal SetCoreLevel for using dragonfly as library
@@ -70,7 +70,7 @@ func SetupDaemon(logDir string, console bool) error {
 	if err != nil {
 		return err
 	}
-	logger.SetGcLogger(gcLogger.Sugar())
+	logger.SetGCLogger(gcLogger.Sugar())
 
 	return nil
 }
diff --git a/pkg/rpc/dfdaemon/client/client.go b/pkg/rpc/dfdaemon/client/client.go
index c20074a3..3a8e40dd 100644
--- a/pkg/rpc/dfdaemon/client/client.go
+++ b/pkg/rpc/dfdaemon/client/client.go
@@ -29,7 +29,7 @@ import (
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/dfnet"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
diff --git a/pkg/rpc/scheduler/client/client.go b/pkg/rpc/scheduler/client/client.go
index 5cab4d95..2add25d5 100644
--- a/pkg/rpc/scheduler/client/client.go
+++ b/pkg/rpc/scheduler/client/client.go
@@ -27,7 +27,7 @@ import (
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/dfnet"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
diff --git a/pkg/sync/kmutex.go b/pkg/sync/kmutex.go
new file mode 100644
index 00000000..833bc570
--- /dev/null
+++ b/pkg/sync/kmutex.go
@@ -0,0 +1,48 @@
+/*
+ *     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 sync
+
+import (
+	"sync"
+)
+
+type Kmutex struct {
+	m *sync.Map
+}
+
+func NewKmutex() *Kmutex {
+	m := sync.Map{}
+	return &Kmutex{&m}
+}
+
+func (k *Kmutex) Lock(key interface{}) {
+	m := sync.Mutex{}
+	rm, _ := k.m.LoadOrStore(key, &m)
+	mu := rm.(*sync.Mutex)
+	mu.Lock()
+}
+
+func (k *Kmutex) Unlock(key interface{}) {
+	rm, ok := k.m.Load(key)
+	if !ok {
+		return
+	}
+
+	mu := rm.(*sync.Mutex)
+	k.m.Delete(key)
+	mu.Unlock()
+}
diff --git a/pkg/sync/krwmutex.go b/pkg/sync/krwmutex.go
new file mode 100644
index 00000000..fe3b5743
--- /dev/null
+++ b/pkg/sync/krwmutex.go
@@ -0,0 +1,66 @@
+/*
+ *     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 sync
+
+import (
+	"sync"
+)
+
+type Krwmutex struct {
+	m *sync.Map
+}
+
+func NewKrwmutex() *Krwmutex {
+	m := sync.Map{}
+	return &Krwmutex{&m}
+}
+
+func (k *Krwmutex) Lock(key interface{}) {
+	m := sync.RWMutex{}
+	rm, _ := k.m.LoadOrStore(key, &m)
+	mu := rm.(*sync.RWMutex)
+	mu.Lock()
+}
+
+func (k *Krwmutex) Unlock(key interface{}) {
+	rm, ok := k.m.Load(key)
+	if !ok {
+		return
+	}
+
+	mu := rm.(*sync.RWMutex)
+	k.m.Delete(key)
+	mu.Unlock()
+}
+
+func (k *Krwmutex) RLock(key interface{}) {
+	m := sync.RWMutex{}
+	rm, _ := k.m.LoadOrStore(key, &m)
+	mu := rm.(*sync.RWMutex)
+	mu.RLock()
+}
+
+func (k *Krwmutex) RUnlock(key interface{}) {
+	rm, ok := k.m.Load(key)
+	if !ok {
+		return
+	}
+
+	mu := rm.(*sync.RWMutex)
+	k.m.Delete(key)
+	mu.RUnlock()
+}
diff --git a/pkg/util/net/urlutils/url_utils.go b/pkg/util/net/urlutils/url_utils.go
index 4d358d57..44b35dc2 100644
--- a/pkg/util/net/urlutils/url_utils.go
+++ b/pkg/util/net/urlutils/url_utils.go
@@ -41,24 +41,11 @@ func FilterURLParam(str string, filters []string) string {
 	}
 
 	u.RawQuery = values.Encode()
-
 	return u.String()
 }
 
 // IsValidURL returns whether the string url is a valid URL.
 func IsValidURL(str string) bool {
 	u, err := url.Parse(str)
-	if err != nil {
-		return false
-	}
-
-	if stringutils.IsBlank(u.Scheme) {
-		return false
-	}
-
-	if stringutils.IsBlank(u.Host) {
-		return false
-	}
-
-	return true
+	return err == nil && u.Scheme != "" && u.Host != ""
 }
diff --git a/scheduler/core/service.go b/scheduler/core/service.go
index 4f5aaae8..048f27d4 100644
--- a/scheduler/core/service.go
+++ b/scheduler/core/service.go
@@ -34,7 +34,7 @@ import (
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/base/common"
 	schedulerRPC "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
-	"d7y.io/dragonfly/v2/pkg/synclock"
+	pkgsync "d7y.io/dragonfly/v2/pkg/sync"
 	"d7y.io/dragonfly/v2/scheduler/config"
 	"d7y.io/dragonfly/v2/scheduler/core/scheduler"
 	"d7y.io/dragonfly/v2/scheduler/metrics"
@@ -77,6 +77,7 @@ type SchedulerService struct {
 	monitor *monitor
 	done    chan struct{}
 	wg      sync.WaitGroup
+	kmu     *pkgsync.Krwmutex
 
 	config        *config.SchedulerConfig
 	dynconfig     config.DynconfigInterface
@@ -122,6 +123,8 @@ func NewSchedulerService(cfg *config.SchedulerConfig, pluginDir string, metricsC
 		metricsConfig: metricsConfig,
 		dynconfig:     dynConfig,
 		done:          make(chan struct{}),
+		wg:            sync.WaitGroup{},
+		kmu:           pkgsync.NewKrwmutex(),
 	}
 	if !ops.disableCDN {
 		var opts []grpc.DialOption
@@ -219,7 +222,7 @@ func (s *SchedulerService) GetPeer(id string) (*supervisor.Peer, bool) {
 	return s.peerManager.Get(id)
 }
 
-func (s *SchedulerService) RegisterPeerTask(req *schedulerRPC.PeerTaskRequest, task *supervisor.Task) *supervisor.Peer {
+func (s *SchedulerService) RegisterTask(req *schedulerRPC.PeerTaskRequest, task *supervisor.Task) *supervisor.Peer {
 	// get or create host
 	peerHost := req.PeerHost
 	host, ok := s.hostManager.Get(peerHost.Uuid)
@@ -246,26 +249,26 @@ func (s *SchedulerService) RegisterPeerTask(req *schedulerRPC.PeerTaskRequest, t
 	return peer
 }
 
-func (s *SchedulerService) GetOrCreateTask(ctx context.Context, task *supervisor.Task) *supervisor.Task {
+func (s *SchedulerService) GetOrAddTask(ctx context.Context, task *supervisor.Task) *supervisor.Task {
 	span := trace.SpanFromContext(ctx)
-	synclock.Lock(task.ID, true)
 
+	s.kmu.RLock(task.ID)
 	task, ok := s.taskManager.GetOrAdd(task)
 	if ok {
 		span.SetAttributes(config.AttributeTaskStatus.String(task.GetStatus().String()))
 		span.SetAttributes(config.AttributeLastTriggerTime.String(task.LastTriggerAt.Load().String()))
 		if task.LastTriggerAt.Load().Add(s.config.AccessWindow).After(time.Now()) || task.IsHealth() {
 			span.SetAttributes(config.AttributeNeedSeedCDN.Bool(false))
-			synclock.UnLock(task.ID, true)
+			s.kmu.RUnlock(task.ID)
 			return task
 		}
 	} else {
 		task.Log().Infof("add new task %s", task.ID)
 	}
-	synclock.UnLock(task.ID, true)
+	s.kmu.RUnlock(task.ID)
 
-	synclock.Lock(task.ID, false)
-	defer synclock.UnLock(task.ID, false)
+	s.kmu.Lock(task.ID)
+	defer s.kmu.Unlock(task.ID)
 
 	// do trigger
 	span.SetAttributes(config.AttributeTaskStatus.String(task.GetStatus().String()))
diff --git a/scheduler/job/job.go b/scheduler/job/job.go
index a9770949..0116ec1b 100644
--- a/scheduler/job/job.go
+++ b/scheduler/job/job.go
@@ -24,8 +24,8 @@ import (
 	"go.opentelemetry.io/otel/trace"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/idgen"
 	internaljob "d7y.io/dragonfly/v2/internal/job"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
 	"d7y.io/dragonfly/v2/scheduler/config"
diff --git a/scheduler/rpcserver/rpcserver.go b/scheduler/rpcserver/rpcserver.go
index 8e717ccb..d805f809 100644
--- a/scheduler/rpcserver/rpcserver.go
+++ b/scheduler/rpcserver/rpcserver.go
@@ -27,7 +27,7 @@ import (
 
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	schedulerserver "d7y.io/dragonfly/v2/pkg/rpc/scheduler/server"
@@ -53,66 +53,101 @@ func New(service *core.SchedulerService, opts ...grpc.ServerOption) (*grpc.Serve
 	return svr.Server, nil
 }
 
-func (s *server) RegisterPeerTask(ctx context.Context, request *scheduler.PeerTaskRequest) (resp *scheduler.RegisterResult, err error) {
-	defer func() {
-		logger.Debugf("peer %s register result %v, err: %v", request.PeerId, resp, err)
-	}()
+func (s *server) RegisterPeerTask(ctx context.Context, req *scheduler.PeerTaskRequest) (*scheduler.RegisterResult, error) {
+	taskID := idgen.TaskID(req.Url, req.UrlMeta)
+	log := logger.WithTaskAndPeerID(taskID, req.PeerId)
+	log.Infof("register peer task, req: %#v", req)
+
 	var span trace.Span
 	ctx, span = tracer.Start(ctx, config.SpanPeerRegister, trace.WithSpanKind(trace.SpanKindServer))
-	span.SetAttributes(config.AttributePeerRegisterRequest.String(request.String()))
 	defer span.End()
-	logger.Debugf("register peer task, req: %#v", request)
-	resp = new(scheduler.RegisterResult)
-
-	taskID := idgen.TaskID(request.Url, request.UrlMeta)
+	span.SetAttributes(config.AttributePeerRegisterRequest.String(req.String()))
 	span.SetAttributes(config.AttributeTaskID.String(taskID))
-	task := s.service.GetOrCreateTask(ctx, supervisor.NewTask(taskID, request.Url, request.UrlMeta))
+
+	// Get task or add new task
+	task := s.service.GetOrAddTask(ctx, supervisor.NewTask(taskID, req.Url, req.UrlMeta))
 	if task.IsFail() {
-		err = dferrors.New(base.Code_SchedTaskStatusError, "task status is fail")
-		logger.Errorf("task %s status is fail", task.ID)
-		span.RecordError(err)
-		return
+		dferr := dferrors.New(base.Code_SchedTaskStatusError, "task status is fail")
+		log.Error(dferr.Message)
+		span.RecordError(dferr)
+		return nil, dferr
 	}
-	resp.SizeScope = getTaskSizeScope(task)
-	span.SetAttributes(config.AttributeTaskSizeScope.String(resp.SizeScope.String()))
-	resp.TaskId = taskID
-	switch resp.SizeScope {
-	case base.SizeScope_TINY:
-		resp.DirectPiece = &scheduler.RegisterResult_PieceContent{
-			PieceContent: task.DirectPiece,
-		}
-		return
-	case base.SizeScope_SMALL:
-		peer := s.service.RegisterPeerTask(request, task)
-		parent, schErr := s.service.SelectParent(peer)
-		if schErr != nil {
-			span.AddEvent(config.EventSmallTaskSelectParentFail)
-			resp.SizeScope = base.SizeScope_NORMAL
-			resp.TaskId = taskID
-			return
-		}
-		firstPiece, _ := task.GetPiece(0)
-		singlePiece := &scheduler.SinglePiece{
-			DstPid:  parent.ID,
-			DstAddr: fmt.Sprintf("%s:%d", parent.Host.IP, parent.Host.DownloadPort),
-			PieceInfo: &base.PieceInfo{
-				PieceNum:    firstPiece.PieceNum,
-				RangeStart:  firstPiece.RangeStart,
-				RangeSize:   firstPiece.RangeSize,
-				PieceMd5:    firstPiece.PieceMd5,
-				PieceOffset: firstPiece.PieceOffset,
-				PieceStyle:  firstPiece.PieceStyle,
-			},
-		}
-		resp.DirectPiece = &scheduler.RegisterResult_SinglePiece{
-			SinglePiece: singlePiece,
+
+	// Task has been successful
+	if task.IsSuccess() {
+		log.Info("task has been successful")
+		sizeScope := task.GetSizeScope()
+		span.SetAttributes(config.AttributeTaskSizeScope.String(sizeScope.String()))
+		switch sizeScope {
+		case base.SizeScope_TINY:
+			log.Info("task size scope is tiny and return piece content directly")
+			return &scheduler.RegisterResult{
+				TaskId:    taskID,
+				SizeScope: sizeScope,
+				DirectPiece: &scheduler.RegisterResult_PieceContent{
+					PieceContent: task.DirectPiece,
+				},
+			}, nil
+		case base.SizeScope_SMALL:
+			log.Info("task size scope is small")
+			peer := s.service.RegisterTask(req, task)
+			parent, err := s.service.SelectParent(peer)
+			if err != nil {
+				log.Warn("task size scope is small and it can not select parent")
+				span.AddEvent(config.EventSmallTaskSelectParentFail)
+				return &scheduler.RegisterResult{
+					TaskId:    taskID,
+					SizeScope: sizeScope,
+				}, nil
+			}
+
+			firstPiece, ok := task.GetPiece(0)
+			if !ok {
+				log.Warn("task size scope is small and it can not get first piece")
+				return &scheduler.RegisterResult{
+					TaskId:    taskID,
+					SizeScope: sizeScope,
+				}, nil
+			}
+
+			singlePiece := &scheduler.SinglePiece{
+				DstPid:  parent.ID,
+				DstAddr: fmt.Sprintf("%s:%d", parent.Host.IP, parent.Host.DownloadPort),
+				PieceInfo: &base.PieceInfo{
+					PieceNum:    firstPiece.PieceNum,
+					RangeStart:  firstPiece.RangeStart,
+					RangeSize:   firstPiece.RangeSize,
+					PieceMd5:    firstPiece.PieceMd5,
+					PieceOffset: firstPiece.PieceOffset,
+					PieceStyle:  firstPiece.PieceStyle,
+				},
+			}
+			log.Infof("task size scope is small and return single piece %#v", sizeScope)
+			span.SetAttributes(config.AttributeSinglePiece.String(singlePiece.String()))
+			return &scheduler.RegisterResult{
+				TaskId:    taskID,
+				SizeScope: sizeScope,
+				DirectPiece: &scheduler.RegisterResult_SinglePiece{
+					SinglePiece: singlePiece,
+				},
+			}, nil
+		default:
+			log.Info("task size scope is normal and needs to be register")
+			s.service.RegisterTask(req, task)
+			return &scheduler.RegisterResult{
+				TaskId:    taskID,
+				SizeScope: sizeScope,
+			}, nil
 		}
-		span.SetAttributes(config.AttributeSinglePiece.String(singlePiece.String()))
-		return
-	default:
-		s.service.RegisterPeerTask(request, task)
-		return
 	}
+
+	// Task is unsuccessful
+	log.Info("task is unsuccessful and needs to be register")
+	s.service.RegisterTask(req, task)
+	return &scheduler.RegisterResult{
+		TaskId:    taskID,
+		SizeScope: base.SizeScope_NORMAL,
+	}, nil
 }
 
 func (s *server) ReportPieceResult(stream scheduler.Scheduler_ReportPieceResultServer) error {
@@ -206,15 +241,3 @@ func (s *server) LeaveTask(ctx context.Context, target *scheduler.PeerTarget) (e
 	}
 	return s.service.HandleLeaveTask(ctx, peer)
 }
-
-func getTaskSizeScope(task *supervisor.Task) base.SizeScope {
-	if task.IsSuccess() {
-		if task.ContentLength.Load() <= supervisor.TinyFileSize {
-			return base.SizeScope_TINY
-		}
-		if task.TotalPieceCount.Load() == 1 {
-			return base.SizeScope_SMALL
-		}
-	}
-	return base.SizeScope_NORMAL
-}
diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go
index 24a26411..fa8af51f 100644
--- a/scheduler/scheduler.go
+++ b/scheduler/scheduler.go
@@ -26,8 +26,8 @@ import (
 
 	"d7y.io/dragonfly/v2/cmd/dependency"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfpath"
 	"d7y.io/dragonfly/v2/internal/dynconfig"
+	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/gc"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/manager"
@@ -109,7 +109,7 @@ func New(cfg *config.Config, d dfpath.Dfpath) (*Server, error) {
 	s.dynconfig = dynConfig
 
 	// Initialize GC
-	s.gc = gc.New(gc.WithLogger(logger.GcLogger))
+	s.gc = gc.New(gc.WithLogger(logger.MetaGCLogger))
 
 	// Initialize scheduler service
 	var openTel bool
diff --git a/scheduler/supervisor/cdn.go b/scheduler/supervisor/cdn.go
index 8d813ef2..e83f6e9e 100644
--- a/scheduler/supervisor/cdn.go
+++ b/scheduler/supervisor/cdn.go
@@ -33,7 +33,7 @@ import (
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/dfnet"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
 	cdnclient "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem/client"
diff --git a/scheduler/supervisor/task.go b/scheduler/supervisor/task.go
index 3c53b90f..9ce5cc4b 100644
--- a/scheduler/supervisor/task.go
+++ b/scheduler/supervisor/task.go
@@ -292,6 +292,18 @@ func (task *Task) GetOrAddPiece(p *base.PieceInfo) (*base.PieceInfo, bool) {
 	return piece.(*base.PieceInfo), ok
 }
 
+func (task *Task) GetSizeScope() base.SizeScope {
+	if task.ContentLength.Load() <= TinyFileSize {
+		return base.SizeScope_TINY
+	}
+
+	if task.TotalPieceCount.Load() == 1 {
+		return base.SizeScope_SMALL
+	}
+
+	return base.SizeScope_NORMAL
+}
+
 func (task *Task) CanBackToSource() bool {
 	return task.BackToSourceWeight.Load() > 0
 }


Test patch:
diff --git a/cdn/supervisor/cdn/cache_writer_test.go b/cdn/supervisor/cdn/cache_writer_test.go
index b3c96fe0..9efb082a 100644
--- a/cdn/supervisor/cdn/cache_writer_test.go
+++ b/cdn/supervisor/cdn/cache_writer_test.go
@@ -27,9 +27,6 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/suite"
 
-	"d7y.io/dragonfly/v2/cdn/plugins"
-	"d7y.io/dragonfly/v2/cdn/storedriver"
-	"d7y.io/dragonfly/v2/cdn/storedriver/local"
 	"d7y.io/dragonfly/v2/cdn/supervisor/cdn/storage"
 	progressMock "d7y.io/dragonfly/v2/cdn/supervisor/mocks/progress"
 	taskMock "d7y.io/dragonfly/v2/cdn/supervisor/mocks/task"
@@ -47,30 +44,17 @@ type CacheWriterTestSuite struct {
 	suite.Suite
 }
 
-func NewPlugins(workHome string) map[plugins.PluginType][]*plugins.PluginProperties {
-	return map[plugins.PluginType][]*plugins.PluginProperties{
-		"storagedriver": {
-			{
-				Name:   local.DiskDriverName,
-				Enable: true,
-				Config: &storedriver.Config{
-					BaseDir: workHome,
-				},
-			},
-		},
-	}
-}
-
 func (suite *CacheWriterTestSuite) SetupSuite() {
 	suite.workHome, _ = os.MkdirTemp("/tmp", "cdn-CacheWriterDetectorTestSuite-")
 	suite.T().Log("workHome:", suite.workHome)
-	suite.Nil(plugins.Initialize(NewPlugins(suite.workHome)))
 	ctrl := gomock.NewController(suite.T())
 	progressManager := progressMock.NewMockManager(ctrl)
 	progressManager.EXPECT().PublishPiece(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
 	progressManager.EXPECT().PublishTask(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes()
 	taskManager := taskMock.NewMockManager(ctrl)
-	storageManager, err := storage.NewManager(storage.DefaultConfig(), taskManager)
+	storageConfig := storage.DefaultConfig()
+	storageConfig.DriverConfigs["disk"].BaseDir = suite.workHome
+	storageManager, err := storage.NewManager(storageConfig, taskManager)
 	suite.Require().Nil(err)
 	metadataManager := newMetadataManager(storageManager)
 	cdnReporter := newReporter(progressManager)
diff --git a/cdn/supervisor/cdn/manager_test.go b/cdn/supervisor/cdn/manager_test.go
index 70ee9113..14b14e34 100644
--- a/cdn/supervisor/cdn/manager_test.go
+++ b/cdn/supervisor/cdn/manager_test.go
@@ -27,13 +27,12 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/suite"
 
-	"d7y.io/dragonfly/v2/cdn/plugins"
 	"d7y.io/dragonfly/v2/cdn/supervisor/cdn/storage"
 	_ "d7y.io/dragonfly/v2/cdn/supervisor/cdn/storage/disk"
 	progressMock "d7y.io/dragonfly/v2/cdn/supervisor/mocks/progress"
 	taskMock "d7y.io/dragonfly/v2/cdn/supervisor/mocks/task"
 	"d7y.io/dragonfly/v2/cdn/supervisor/task"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/source"
 	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
@@ -56,10 +55,11 @@ type CDNManagerTestSuite struct {
 func (suite *CDNManagerTestSuite) SetupSuite() {
 	suite.workHome, _ = os.MkdirTemp("/tmp", "cdn-ManagerTestSuite-")
 	fmt.Printf("workHome: %s", suite.workHome)
-	suite.Nil(plugins.Initialize(NewPlugins(suite.workHome)))
 	ctrl := gomock.NewController(suite.T())
 	taskManager := taskMock.NewMockManager(ctrl)
-	storageManager, err := storage.NewManager(storage.DefaultConfig(), taskManager)
+	storageConfig := storage.DefaultConfig()
+	storageConfig.DriverConfigs["disk"].BaseDir = suite.workHome
+	storageManager, err := storage.NewManager(storageConfig, taskManager)
 	suite.Require().Nil(err)
 
 	progressManager := progressMock.NewMockManager(ctrl)
diff --git a/client/daemon/peer/peertask_file_test.go b/client/daemon/peer/peertask_file_test.go
index 32b2260d..430becd8 100644
--- a/client/daemon/peer/peertask_file_test.go
+++ b/client/daemon/peer/peertask_file_test.go
@@ -136,7 +136,7 @@ func TestFilePeerTask_BackSource_WithContentLength(t *testing.T) {
 		req,
 		ptm.schedulerClient,
 		ptm.schedulerOption,
-		0)
+		0, 10)
 	assert.Nil(err, "new file peer task")
 	pt.needBackSource = true
 
@@ -261,7 +261,7 @@ func TestFilePeerTask_BackSource_WithoutContentLength(t *testing.T) {
 		req,
 		ptm.schedulerClient,
 		ptm.schedulerOption,
-		0)
+		0, 10)
 	assert.Nil(err, "new file peer task")
 	pt.needBackSource = true
 
diff --git a/client/daemon/peer/peertask_stream_backsource_partial_test.go b/client/daemon/peer/peertask_stream_backsource_partial_test.go
index 5059c337..ab6801fe 100644
--- a/client/daemon/peer/peertask_stream_backsource_partial_test.go
+++ b/client/daemon/peer/peertask_stream_backsource_partial_test.go
@@ -256,8 +256,7 @@ func TestStreamPeerTask_BackSource_Partial_WithContentLength(t *testing.T) {
 		PeerHost: &scheduler.PeerHost{},
 	}
 	ctx := context.Background()
-	_, pt, _, err := newStreamPeerTask(ctx, ptm.host, pm, req,
-		ptm.schedulerClient, ptm.schedulerOption, 0)
+	_, pt, _, err := newStreamPeerTask(ctx, ptm, req)
 	assert.Nil(err, "new stream peer task")
 	pt.SetCallback(&streamPeerTaskCallback{
 		ptm:   ptm,
diff --git a/client/daemon/peer/peertask_stream_test.go b/client/daemon/peer/peertask_stream_test.go
index b755687b..4205b0f1 100644
--- a/client/daemon/peer/peertask_stream_test.go
+++ b/client/daemon/peer/peertask_stream_test.go
@@ -32,7 +32,6 @@ import (
 	"d7y.io/dragonfly/v2/client/clientutil"
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/client/daemon/test"
-	"d7y.io/dragonfly/v2/internal/util"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	"d7y.io/dragonfly/v2/pkg/source"
@@ -101,9 +100,11 @@ func TestStreamPeerTask_BackSource_WithContentLength(t *testing.T) {
 		},
 		runningPeerTasks: sync.Map{},
 		pieceManager: &pieceManager{
-			storageManager:   storageManager,
-			pieceDownloader:  downloader,
-			computePieceSize: util.ComputePieceSize,
+			storageManager:  storageManager,
+			pieceDownloader: downloader,
+			computePieceSize: func(contentLength int64) uint32 {
+				return uint32(pieceSize)
+			},
 		},
 		storageManager:  storageManager,
 		schedulerClient: schedulerClient,
@@ -120,19 +121,7 @@ func TestStreamPeerTask_BackSource_WithContentLength(t *testing.T) {
 		PeerHost: &scheduler.PeerHost{},
 	}
 	ctx := context.Background()
-	_, pt, _, err := newStreamPeerTask(ctx,
-		ptm.host,
-		&pieceManager{
-			storageManager:  storageManager,
-			pieceDownloader: downloader,
-			computePieceSize: func(contentLength int64) uint32 {
-				return uint32(pieceSize)
-			},
-		},
-		req,
-		ptm.schedulerClient,
-		ptm.schedulerOption,
-		0)
+	_, pt, _, err := newStreamPeerTask(ctx, ptm, req)
 	assert.Nil(err, "new stream peer task")
 	pt.SetCallback(&streamPeerTaskCallback{
 		ptm:   ptm,
@@ -211,9 +200,11 @@ func TestStreamPeerTask_BackSource_WithoutContentLength(t *testing.T) {
 		},
 		runningPeerTasks: sync.Map{},
 		pieceManager: &pieceManager{
-			storageManager:   storageManager,
-			pieceDownloader:  downloader,
-			computePieceSize: util.ComputePieceSize,
+			storageManager:  storageManager,
+			pieceDownloader: downloader,
+			computePieceSize: func(contentLength int64) uint32 {
+				return uint32(pieceSize)
+			},
 		},
 		storageManager:  storageManager,
 		schedulerClient: schedulerClient,
@@ -230,19 +221,7 @@ func TestStreamPeerTask_BackSource_WithoutContentLength(t *testing.T) {
 		PeerHost: &scheduler.PeerHost{},
 	}
 	ctx := context.Background()
-	_, pt, _, err := newStreamPeerTask(ctx,
-		ptm.host,
-		&pieceManager{
-			storageManager:  storageManager,
-			pieceDownloader: downloader,
-			computePieceSize: func(contentLength int64) uint32 {
-				return uint32(pieceSize)
-			},
-		},
-		req,
-		schedulerClient,
-		ptm.schedulerOption,
-		0)
+	_, pt, _, err := newStreamPeerTask(ctx, ptm, req)
 	assert.Nil(err, "new stream peer task")
 	pt.SetCallback(&streamPeerTaskCallback{
 		ptm:   ptm,
diff --git a/client/daemon/rpcserver/rpcserver_test.go b/client/daemon/rpcserver/rpcserver_test.go
index 75ed9a51..5048784a 100644
--- a/client/daemon/rpcserver/rpcserver_test.go
+++ b/client/daemon/rpcserver/rpcserver_test.go
@@ -35,7 +35,7 @@ import (
 	mock_peer "d7y.io/dragonfly/v2/client/daemon/test/mock/peer"
 	mock_storage "d7y.io/dragonfly/v2/client/daemon/test/mock/storage"
 	"d7y.io/dragonfly/v2/internal/dfnet"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	dfdaemongrpc "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
 	dfclient "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/client"
diff --git a/client/dfget/dfget_test.go b/client/dfget/dfget_test.go
index 3d8bc4df..90a88f6b 100644
--- a/client/dfget/dfget_test.go
+++ b/client/dfget/dfget_test.go
@@ -29,7 +29,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"d7y.io/dragonfly/v2/client/config"
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/source"
 	sourcemock "d7y.io/dragonfly/v2/pkg/source/mock"
 	"d7y.io/dragonfly/v2/pkg/util/digestutils"
diff --git a/pkg/sync/kmutex_test.go b/pkg/sync/kmutex_test.go
new file mode 100644
index 00000000..6154fce3
--- /dev/null
+++ b/pkg/sync/kmutex_test.go
@@ -0,0 +1,87 @@
+/*
+ *     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 sync
+
+import (
+	"testing"
+	"time"
+)
+
+func TestKmutex_Lock(t *testing.T) {
+	k := NewKmutex()
+	done := time.After(time.Second)
+
+	k.Lock("foo")
+	foo := make(chan struct{})
+	go func() {
+		k.Lock("foo")
+		close(foo)
+	}()
+
+	k.Lock("bar")
+	bar := make(chan struct{})
+	go func() {
+		k.Lock("bar")
+		close(bar)
+	}()
+
+	k.Lock("baz")
+	baz := make(chan struct{})
+	go func() {
+		k.Lock("baz")
+		close(baz)
+	}()
+
+	select {
+	case <-bar:
+	case <-baz:
+	case <-foo:
+	case <-done:
+		t.Fatal("kmutex is not unlocked")
+	default:
+	}
+
+	k.Unlock("foo")
+	select {
+	case <-bar:
+	case <-baz:
+	case <-done:
+		t.Fatal("bar and baz are not unlocked")
+	case <-foo:
+		k.Unlock("foo")
+	}
+
+	k.Unlock("bar")
+	select {
+	case <-foo:
+	case <-baz:
+	case <-done:
+		t.Fatal("foo and baz are not unlocked")
+	case <-bar:
+		k.Unlock("bar")
+	}
+
+	k.Unlock("baz")
+	select {
+	case <-foo:
+	case <-bar:
+	case <-done:
+		t.Fatal("foo and bar are not unlocked")
+	case <-baz:
+		k.Unlock("baz")
+	}
+}
diff --git a/pkg/sync/krwmutex_test.go b/pkg/sync/krwmutex_test.go
new file mode 100644
index 00000000..615f51bf
--- /dev/null
+++ b/pkg/sync/krwmutex_test.go
@@ -0,0 +1,152 @@
+/*
+ *     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 sync
+
+import (
+	"testing"
+	"time"
+)
+
+func TestKrwmutex_Lock(t *testing.T) {
+	k := NewKrwmutex()
+	done := time.After(time.Second)
+
+	k.Lock("foo")
+	foo := make(chan struct{})
+	go func() {
+		k.Lock("foo")
+		close(foo)
+	}()
+
+	k.Lock("bar")
+	bar := make(chan struct{})
+	go func() {
+		k.Lock("bar")
+		close(bar)
+	}()
+
+	k.Lock("baz")
+	baz := make(chan struct{})
+	go func() {
+		k.Lock("baz")
+		close(baz)
+	}()
+
+	select {
+	case <-bar:
+	case <-baz:
+	case <-foo:
+	case <-done:
+		t.Fatal("kmutex is not unlocked")
+	default:
+	}
+
+	k.Unlock("foo")
+	select {
+	case <-bar:
+	case <-baz:
+	case <-done:
+		t.Fatal("bar and baz are not unlocked")
+	case <-foo:
+		k.Unlock("foo")
+	}
+
+	k.Unlock("bar")
+	select {
+	case <-foo:
+	case <-baz:
+	case <-done:
+		t.Fatal("foo and baz are not unlocked")
+	case <-bar:
+		k.Unlock("bar")
+	}
+
+	k.Unlock("baz")
+	select {
+	case <-foo:
+	case <-bar:
+	case <-done:
+		t.Fatal("foo and bar are not unlocked")
+	case <-baz:
+		k.Unlock("baz")
+	}
+}
+
+func TestKrwmutex_RLock(t *testing.T) {
+	k := NewKrwmutex()
+	done := time.After(time.Second)
+
+	k.Lock("foo")
+	foo := make(chan struct{})
+	go func() {
+		k.RLock("foo")
+		close(foo)
+	}()
+
+	k.Lock("bar")
+	bar := make(chan struct{})
+	go func() {
+		k.RLock("bar")
+		close(bar)
+	}()
+
+	k.Lock("baz")
+	baz := make(chan struct{})
+	go func() {
+		k.RLock("baz")
+		close(baz)
+	}()
+
+	select {
+	case <-bar:
+	case <-baz:
+	case <-foo:
+	case <-done:
+		t.Fatal("kmutex is not unlocked")
+	default:
+	}
+
+	k.Unlock("foo")
+	select {
+	case <-bar:
+	case <-baz:
+	case <-done:
+		t.Fatal("bar and baz are not unlocked")
+	case <-foo:
+		k.RUnlock("foo")
+	}
+
+	k.Unlock("bar")
+	select {
+	case <-foo:
+	case <-baz:
+	case <-done:
+		t.Fatal("foo and baz are not unlocked")
+	case <-bar:
+		k.RUnlock("bar")
+	}
+
+	k.Unlock("baz")
+	select {
+	case <-foo:
+	case <-bar:
+	case <-done:
+		t.Fatal("foo and bar are not unlocked")
+	case <-baz:
+		k.RUnlock("baz")
+	}
+}
diff --git a/scheduler/core/evaluator/evaluator_base_test.go b/scheduler/core/evaluator/evaluator_base_test.go
index 6cb898e3..c5753897 100644
--- a/scheduler/core/evaluator/evaluator_base_test.go
+++ b/scheduler/core/evaluator/evaluator_base_test.go
@@ -22,7 +22,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 
-	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/util/mathutils"
 	"d7y.io/dragonfly/v2/scheduler/supervisor"
 )
diff --git a/test/README.md b/test/README.md
index 02925426..c482fcc2 100644
--- a/test/README.md
+++ b/test/README.md
@@ -12,7 +12,7 @@ Unit tests is in the project directory.
 make test
 ```
 
-### Runing uint tests with coverage reports
+### Running uint tests with coverage reports
 
 ```bash
 make test-coverage
@@ -28,7 +28,7 @@ E2E tests is in `test/e2e` path.
 make e2e-test
 ```
 
-### Runing E2E tests with coverage reports
+### Running E2E tests with coverage reports
 
 ```bash
 make e2e-test-coverage
diff --git a/test/e2e/manager/preheat.go b/test/e2e/manager/preheat.go
index 049d05fe..c0f7fbc7 100644
--- a/test/e2e/manager/preheat.go
+++ b/test/e2e/manager/preheat.go
@@ -28,10 +28,10 @@ import (
 	. "github.com/onsi/ginkgo" //nolint
 	. "github.com/onsi/gomega" //nolint
 
-	"d7y.io/dragonfly/v2/internal/idgen"
 	internaljob "d7y.io/dragonfly/v2/internal/job"
 	"d7y.io/dragonfly/v2/manager/model"
 	"d7y.io/dragonfly/v2/manager/types"
+	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/util/structutils"
 	"d7y.io/dragonfly/v2/test/e2e/e2eutil"