Repository: dragonflyoss/Dragonfly2
Pull request #585 (2021-08-26 09:00:49)

Check free space when Registering Task
In the hybridStorageMgr, the remaining space of the memory is checked. I want to introduced this feature in other situations.


Patch:
diff --git a/cdnsystem/cdn.go b/cdnsystem/cdn.go
index ef7e80b3..eebb3639 100644
--- a/cdnsystem/cdn.go
+++ b/cdnsystem/cdn.go
@@ -84,9 +84,6 @@ func New(cfg *config.Config) (*Server, error) {
 
 	// Initialize storage manager
 	storageMgr.Initialize(taskMgr)
-	if err != nil {
-		return nil, errors.Wrapf(err, "create storage manager")
-	}
 
 	// Initialize storage manager
 	cdnSeedServer, err := rpcserver.NewCdnSeedServer(cfg, taskMgr)
diff --git a/cdnsystem/errors/errors.go b/cdnsystem/errors/errors.go
index 0de97f0e..d35bae1e 100644
--- a/cdnsystem/errors/errors.go
+++ b/cdnsystem/errors/errors.go
@@ -92,6 +92,9 @@ var (
 
 	// ErrConvertFailed represents failed to convert.
 	ErrConvertFailed = errors.New("convert failed")
+
+	// ErrResourcesLacked represents a lack of resources, for example, the disk does not have enough space.
+	ErrResourcesLacked = errors.New("resources lacked")
 )
 
 // IsSystemError checks the error is a system error or not.
@@ -152,3 +155,7 @@ func IsFileNotExist(err error) bool {
 	_, ok := err.(ErrFileNotExist)
 	return ok
 }
+
+func IsResourcesLacked(err error) bool {
+	return errors.Cause(err) == ErrResourcesLacked
+}
diff --git a/cdnsystem/rpcserver/rpcserver.go b/cdnsystem/rpcserver/rpcserver.go
index 6bf9aa77..33c4188d 100644
--- a/cdnsystem/rpcserver/rpcserver.go
+++ b/cdnsystem/rpcserver/rpcserver.go
@@ -143,6 +143,11 @@ func (css *CdnSeedServer) ObtainSeeds(ctx context.Context, req *cdnsystem.SeedRe
 	// register task
 	pieceChan, err := css.taskMgr.Register(ctx, registerRequest)
 	if err != nil {
+		if cdnerrors.IsResourcesLacked(err) {
+			err = dferrors.Newf(dfcodes.ResourceLacked, "resources lacked for task(%s): %v", req.TaskId, err)
+			span.RecordError(err)
+			return err
+		}
 		err = dferrors.Newf(dfcodes.CdnTaskRegistryFail, "failed to register seed task(%s): %v", req.TaskId, err)
 		span.RecordError(err)
 		return err
diff --git a/cdnsystem/supervisor/cdn/manager.go b/cdnsystem/supervisor/cdn/manager.go
index e5fc5787..3cb61cf2 100644
--- a/cdnsystem/supervisor/cdn/manager.go
+++ b/cdnsystem/supervisor/cdn/manager.go
@@ -174,6 +174,10 @@ func (cm *Manager) Delete(taskID string) error {
 	return nil
 }
 
+func (cm *Manager) TryFreeSpace(fileLength int64) (bool, error) {
+	return cm.cacheStore.TryFreeSpace(fileLength)
+}
+
 func (cm *Manager) handleCDNResult(task *types.SeedTask, sourceDigest string, downloadMetadata *downloadMetadata) (bool, error) {
 	logger.WithTaskID(task.TaskID).Debugf("handle cdn result, downloadMetaData: %+v", downloadMetadata)
 	var isSuccess = true
diff --git a/cdnsystem/supervisor/cdn/storage/disk/disk.go b/cdnsystem/supervisor/cdn/storage/disk/disk.go
index f058c5c8..5645bd72 100644
--- a/cdnsystem/supervisor/cdn/storage/disk/disk.go
+++ b/cdnsystem/supervisor/cdn/storage/disk/disk.go
@@ -20,9 +20,13 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"os"
+	"path"
 	"strings"
 	"time"
 
+	"go.uber.org/atomic"
+
 	cdnerrors "d7y.io/dragonfly/v2/cdnsystem/errors"
 	"d7y.io/dragonfly/v2/cdnsystem/storedriver"
 	"d7y.io/dragonfly/v2/cdnsystem/storedriver/local"
@@ -131,7 +135,7 @@ func (s *diskStorageMgr) GC() error {
 	for _, taskID := range gcTaskIDs {
 		synclock.Lock(taskID, false)
 		// try to ensure the taskID is not using again
-		if s.taskMgr.Exist(taskID) {
+		if _, exist := s.taskMgr.Exist(taskID); exist {
 			synclock.UnLock(taskID, false)
 			continue
 		}
@@ -230,3 +234,55 @@ func (s *diskStorageMgr) DeleteTask(taskID string) error {
 func (s *diskStorageMgr) ResetRepo(task *types.SeedTask) error {
 	return s.DeleteTask(task.TaskID)
 }
+
+func (s *diskStorageMgr) TryFreeSpace(fileLength int64) (bool, error) {
+	freeSpace, err := s.diskDriver.GetFreeSpace()
+	if err != nil {
+		return false, err
+	}
+	if freeSpace > 500*unit.GB && freeSpace.ToNumber() > fileLength {
+		return true, nil
+	}
+
+	remainder := atomic.NewInt64(0)
+	r := &storedriver.Raw{
+		WalkFn: func(filePath string, info os.FileInfo, err error) error {
+			if fileutils.IsRegular(filePath) {
+				taskID := strings.Split(path.Base(filePath), ".")[0]
+				task, exist := s.taskMgr.Exist(taskID)
+				if exist {
+					var totalLen int64 = 0
+					if task.CdnFileLength > 0 {
+						totalLen = task.CdnFileLength
+					} else {
+						totalLen = task.SourceFileLength
+					}
+					if totalLen > 0 {
+						remainder.Add(totalLen - info.Size())
+					}
+				} else {
+					logger.Warnf("failed to get task: %s", taskID)
+				}
+			}
+			return nil
+		},
+	}
+	s.diskDriver.Walk(r)
+
+	enoughSpace := freeSpace.ToNumber()-remainder.Load() > fileLength
+	if !enoughSpace {
+		s.cleaner.GC("disk", true)
+		remainder.Store(0)
+		s.diskDriver.Walk(r)
+		freeSpace, err = s.diskDriver.GetFreeSpace()
+		if err != nil {
+			return false, err
+		}
+		enoughSpace = freeSpace.ToNumber()-remainder.Load() > fileLength
+	}
+	if !enoughSpace {
+		return false, nil
+	}
+
+	return true, nil
+}
diff --git a/cdnsystem/supervisor/cdn/storage/hybrid/hybrid.go b/cdnsystem/supervisor/cdn/storage/hybrid/hybrid.go
index 5aca00ab..11329a08 100644
--- a/cdnsystem/supervisor/cdn/storage/hybrid/hybrid.go
+++ b/cdnsystem/supervisor/cdn/storage/hybrid/hybrid.go
@@ -177,10 +177,7 @@ func (h *hybridStorageMgr) gcTasks(gcTaskIDs []string, isDisk bool) int {
 	for _, taskID := range gcTaskIDs {
 		synclock.Lock(taskID, false)
 		// try to ensure the taskID is not using again
-		if _, err := h.taskMgr.Get(taskID); err == nil || !cdnerrors.IsDataNotFound(err) {
-			if err != nil {
-				logger.GcLogger.With("type", "hybrid").Errorf("gc disk: failed to get taskID(%s): %v", taskID, err)
-			}
+		if _, exist := h.taskMgr.Exist(taskID); exist {
 			synclock.UnLock(taskID, false)
 			continue
 		}
@@ -297,6 +294,58 @@ func (h *hybridStorageMgr) StatDownloadFile(taskID string) (*storedriver.Storage
 	return h.diskDriver.Stat(storage.GetDownloadRaw(taskID))
 }
 
+func (h *hybridStorageMgr) TryFreeSpace(fileLength int64) (bool, error) {
+	diskFreeSpace, err := h.diskDriver.GetFreeSpace()
+	if err != nil {
+		return false, err
+	}
+	if diskFreeSpace > 500*unit.GB && diskFreeSpace.ToNumber() > fileLength {
+		return true, nil
+	}
+
+	remainder := atomic.NewInt64(0)
+	r := &storedriver.Raw{
+		WalkFn: func(filePath string, info os.FileInfo, err error) error {
+			if fileutils.IsRegular(filePath) {
+				taskID := strings.Split(path.Base(filePath), ".")[0]
+				task, exist := h.taskMgr.Exist(taskID)
+				if exist {
+					var totalLen int64 = 0
+					if task.CdnFileLength > 0 {
+						totalLen = task.CdnFileLength
+					} else {
+						totalLen = task.SourceFileLength
+					}
+					if totalLen > 0 {
+						remainder.Add(totalLen - info.Size())
+					}
+				} else {
+					logger.Warnf("failed to get task: %s", taskID)
+				}
+			}
+			return nil
+		},
+	}
+	h.diskDriver.Walk(r)
+
+	enoughSpace := diskFreeSpace.ToNumber()-remainder.Load() > fileLength
+	if !enoughSpace {
+		h.diskDriverCleaner.GC("hybrid", true)
+		remainder.Store(0)
+		h.diskDriver.Walk(r)
+		diskFreeSpace, err = h.diskDriver.GetFreeSpace()
+		if err != nil {
+			return false, err
+		}
+		enoughSpace = diskFreeSpace.ToNumber()-remainder.Load() > fileLength
+	}
+	if !enoughSpace {
+		return false, nil
+	}
+
+	return true, nil
+}
+
 func (h *hybridStorageMgr) deleteDiskFiles(taskID string) error {
 	return h.deleteTaskFiles(taskID, true, true)
 }
@@ -352,9 +401,9 @@ func (h *hybridStorageMgr) tryShmSpace(url, taskID string, fileLength int64) (st
 		h.memoryDriver.Walk(&storedriver.Raw{
 			WalkFn: func(filePath string, info os.FileInfo, err error) error {
 				if fileutils.IsRegular(filePath) {
-					taskID := path.Base(filePath)
-					task, err := h.taskMgr.Get(taskID)
-					if err == nil {
+					taskID := strings.Split(path.Base(filePath), ".")[0]
+					task, exist := h.taskMgr.Exist(taskID)
+					if exist {
 						var totalLen int64 = 0
 						if task.CdnFileLength > 0 {
 							totalLen = task.CdnFileLength
@@ -365,7 +414,7 @@ func (h *hybridStorageMgr) tryShmSpace(url, taskID string, fileLength int64) (st
 							remainder.Add(totalLen - info.Size())
 						}
 					} else {
-						logger.Warnf("failed to get task: %s: %v", taskID, err)
+						logger.Warnf("failed to get task: %s", taskID)
 					}
 				}
 				return nil
diff --git a/cdnsystem/supervisor/cdn/storage/mock/mock_storage_mgr.go b/cdnsystem/supervisor/cdn/storage/mock/mock_storage_mgr.go
index 0e7720c5..eafa7a3c 100644
--- a/cdnsystem/supervisor/cdn/storage/mock/mock_storage_mgr.go
+++ b/cdnsystem/supervisor/cdn/storage/mock/mock_storage_mgr.go
@@ -166,6 +166,21 @@ func (mr *MockManagerMockRecorder) StatDownloadFile(arg0 interface{}) *gomock.Ca
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StatDownloadFile", reflect.TypeOf((*MockManager)(nil).StatDownloadFile), arg0)
 }
 
+// TryFreeSpace mocks base method.
+func (m *MockManager) TryFreeSpace(arg0 int64) (bool, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "TryFreeSpace", arg0)
+	ret0, _ := ret[0].(bool)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// TryFreeSpace indicates an expected call of TryFreeSpace.
+func (mr *MockManagerMockRecorder) TryFreeSpace(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TryFreeSpace", reflect.TypeOf((*MockManager)(nil).TryFreeSpace), arg0)
+}
+
 // WriteDownloadFile mocks base method.
 func (m *MockManager) WriteDownloadFile(arg0 string, arg1, arg2 int64, arg3 io.Reader) error {
 	m.ctrl.T.Helper()
diff --git a/cdnsystem/supervisor/cdn/storage/storage_gc.go b/cdnsystem/supervisor/cdn/storage/storage_gc.go
index 6938f3ad..053e382d 100644
--- a/cdnsystem/supervisor/cdn/storage/storage_gc.go
+++ b/cdnsystem/supervisor/cdn/storage/storage_gc.go
@@ -100,7 +100,7 @@ func (cleaner *Cleaner) GC(storagePattern string, force bool) ([]string, error)
 		walkTaskIds[taskID] = true
 
 		// we should return directly when we success to get info which means it is being used
-		if cleaner.taskMgr.Exist(taskID) {
+		if _, exist := cleaner.taskMgr.Exist(taskID); exist {
 			return nil
 		}
 
diff --git a/cdnsystem/supervisor/cdn/storage/storage_mgr.go b/cdnsystem/supervisor/cdn/storage/storage_mgr.go
index 3d63aad8..03b1f790 100644
--- a/cdnsystem/supervisor/cdn/storage/storage_mgr.go
+++ b/cdnsystem/supervisor/cdn/storage/storage_mgr.go
@@ -71,6 +71,9 @@ type Manager interface {
 
 	// DeleteTask delete task from storage
 	DeleteTask(taskID string) error
+
+	// TryFreeSpace checks if there is enough space for the file, return true while we are sure that there is enough space.
+	TryFreeSpace(fileLength int64) (bool, error)
 }
 
 // FileMetaData meta data of task
diff --git a/cdnsystem/supervisor/cdn_mgr.go b/cdnsystem/supervisor/cdn_mgr.go
index 2e4b22bf..fa036730 100644
--- a/cdnsystem/supervisor/cdn_mgr.go
+++ b/cdnsystem/supervisor/cdn_mgr.go
@@ -34,4 +34,7 @@ type CDNMgr interface {
 	// Delete the cdn meta with specified taskID.
 	// The file on the disk will be deleted when the force is true.
 	Delete(string) error
+
+	// TryFreeSpace checks if the free space of the storage is larger than the fileLength.
+	TryFreeSpace(fileLength int64) (bool, error)
 }
diff --git a/cdnsystem/supervisor/mock/mock_cdn_mgr.go b/cdnsystem/supervisor/mock/mock_cdn_mgr.go
index 058ae937..31a7bcf3 100644
--- a/cdnsystem/supervisor/mock/mock_cdn_mgr.go
+++ b/cdnsystem/supervisor/mock/mock_cdn_mgr.go
@@ -63,3 +63,18 @@ func (mr *MockCDNMgrMockRecorder) TriggerCDN(arg0, arg1 interface{}) *gomock.Cal
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TriggerCDN", reflect.TypeOf((*MockCDNMgr)(nil).TriggerCDN), arg0, arg1)
 }
+
+// TryFreeSpace mocks base method.
+func (m *MockCDNMgr) TryFreeSpace(arg0 int64) (bool, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "TryFreeSpace", arg0)
+	ret0, _ := ret[0].(bool)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// TryFreeSpace indicates an expected call of TryFreeSpace.
+func (mr *MockCDNMgrMockRecorder) TryFreeSpace(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TryFreeSpace", reflect.TypeOf((*MockCDNMgr)(nil).TryFreeSpace), arg0)
+}
diff --git a/cdnsystem/supervisor/mock/mock_task_mgr.go b/cdnsystem/supervisor/mock/mock_task_mgr.go
index ab451779..f4d3dc40 100644
--- a/cdnsystem/supervisor/mock/mock_task_mgr.go
+++ b/cdnsystem/supervisor/mock/mock_task_mgr.go
@@ -9,7 +9,6 @@ import (
 	reflect "reflect"
 
 	types "d7y.io/dragonfly/v2/cdnsystem/types"
-	syncmap "d7y.io/dragonfly/v2/pkg/structure/syncmap"
 	gomock "github.com/golang/mock/gomock"
 )
 
@@ -51,11 +50,12 @@ func (mr *MockSeedTaskMgrMockRecorder) Delete(arg0 interface{}) *gomock.Call {
 }
 
 // Exist mocks base method.
-func (m *MockSeedTaskMgr) Exist(arg0 string) bool {
+func (m *MockSeedTaskMgr) Exist(arg0 string) (*types.SeedTask, bool) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "Exist", arg0)
-	ret0, _ := ret[0].(bool)
-	return ret0
+	ret0, _ := ret[0].(*types.SeedTask)
+	ret1, _ := ret[1].(bool)
+	return ret0, ret1
 }
 
 // Exist indicates an expected call of Exist.
@@ -79,21 +79,6 @@ func (mr *MockSeedTaskMgrMockRecorder) Get(arg0 interface{}) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockSeedTaskMgr)(nil).Get), arg0)
 }
 
-// GetAccessTime mocks base method.
-func (m *MockSeedTaskMgr) GetAccessTime() (*syncmap.SyncMap, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetAccessTime")
-	ret0, _ := ret[0].(*syncmap.SyncMap)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// GetAccessTime indicates an expected call of GetAccessTime.
-func (mr *MockSeedTaskMgrMockRecorder) GetAccessTime() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAccessTime", reflect.TypeOf((*MockSeedTaskMgr)(nil).GetAccessTime))
-}
-
 // GetPieces mocks base method.
 func (m *MockSeedTaskMgr) GetPieces(arg0 context.Context, arg1 string) ([]*types.SeedPiece, error) {
 	m.ctrl.T.Helper()
diff --git a/cdnsystem/supervisor/task/manager.go b/cdnsystem/supervisor/task/manager.go
index d9be8db2..85b1b789 100644
--- a/cdnsystem/supervisor/task/manager.go
+++ b/cdnsystem/supervisor/task/manager.go
@@ -19,7 +19,6 @@ package task
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"time"
 
 	"d7y.io/dragonfly/v2/cdnsystem/config"
@@ -181,13 +180,9 @@ func (tm Manager) Get(taskID string) (*types.SeedTask, error) {
 	return task, err
 }
 
-func (tm Manager) Exist(taskID string) bool {
-	_, err := tm.getTask(taskID)
-	return err == nil || !cdnerrors.IsDataNotFound(err)
-}
-
-func (tm Manager) GetAccessTime() (*syncmap.SyncMap, error) {
-	return tm.accessTimeMap, nil
+func (tm Manager) Exist(taskID string) (*types.SeedTask, bool) {
+	task, err := tm.getTask(taskID)
+	return task, err == nil
 }
 
 func (tm Manager) Delete(taskID string) error {
@@ -215,10 +210,7 @@ func (tm *Manager) GC() error {
 	var removedTaskCount int
 	startTime := time.Now()
 	// get all taskIDs and the corresponding accessTime
-	taskAccessMap, err := tm.GetAccessTime()
-	if err != nil {
-		return fmt.Errorf("gc tasks: failed to get task accessTime map for GC: %v", err)
-	}
+	taskAccessMap := tm.accessTimeMap
 
 	// range all tasks and determine whether they are expired
 	taskIDs := taskAccessMap.ListKeyAsStringSlice()
diff --git a/cdnsystem/supervisor/task/manager_util.go b/cdnsystem/supervisor/task/manager_util.go
index 78948968..824726eb 100644
--- a/cdnsystem/supervisor/task/manager_util.go
+++ b/cdnsystem/supervisor/task/manager_util.go
@@ -111,6 +111,14 @@ func (tm *Manager) addOrUpdateTask(ctx context.Context, request *types.TaskRegis
 	// if not support file length header request ,return -1
 	task.SourceFileLength = sourceFileLength
 	logger.WithTaskID(taskID).Debugf("get file content length: %d", sourceFileLength)
+	if task.SourceFileLength > 0 {
+		ok, err := tm.cdnMgr.TryFreeSpace(task.SourceFileLength)
+		if err != nil {
+			logger.Errorf("failed to try free space: %v", err)
+		} else if !ok {
+			return nil, cdnerrors.ErrResourcesLacked
+		}
+	}
 
 	// if success to get the information successfully with the req.Header then update the task.Header to req.Header.
 	if request.Header != nil {
diff --git a/cdnsystem/supervisor/task_mgr.go b/cdnsystem/supervisor/task_mgr.go
index 1b383ec3..41d57a04 100644
--- a/cdnsystem/supervisor/task_mgr.go
+++ b/cdnsystem/supervisor/task_mgr.go
@@ -21,7 +21,6 @@ import (
 	"context"
 
 	"d7y.io/dragonfly/v2/cdnsystem/types"
-	"d7y.io/dragonfly/v2/pkg/structure/syncmap"
 )
 
 // SeedTaskMgr as an interface defines all operations against SeedTask.
@@ -36,10 +35,7 @@ type SeedTaskMgr interface {
 	Get(string) (*types.SeedTask, error)
 
 	// Exist check task existence with specified taskId.
-	Exist(string) bool
-
-	// GetAccessTime get all tasks accessTime.
-	GetAccessTime() (*syncmap.SyncMap, error)
+	Exist(string) (*types.SeedTask, bool)
 
 	// Delete delete a task.
 	Delete(string) error
diff --git a/pkg/rpc/cdnsystem/client/client.go b/pkg/rpc/cdnsystem/client/client.go
index 8e9bb6be..b26f1a59 100644
--- a/pkg/rpc/cdnsystem/client/client.go
+++ b/pkg/rpc/cdnsystem/client/client.go
@@ -101,17 +101,15 @@ func (cc *cdnClient) ObtainSeeds(ctx context.Context, sr *cdnsystem.SeedRequest,
 
 func (cc *cdnClient) GetPieceTasks(ctx context.Context, addr dfnet.NetAddr, req *base.PieceTaskRequest, opts ...grpc.CallOption) (*base.PiecePacket, error) {
 	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		defer func() {
-			logger.WithTaskID(req.TaskId).Infof("invoke cdn node %s GetPieceTasks", addr.GetEndpoint())
-		}()
 		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 {
-		return res.(*base.PiecePacket), nil
+	if err != nil {
+		logger.WithTaskID(req.TaskId).Infof("GetPieceTasks: invoke cdn node %s GetPieceTasks failed: %v", addr.GetEndpoint(), err)
+		return nil, err
 	}
-	return nil, err
+	return res.(*base.PiecePacket), nil
 }
diff --git a/pkg/rpc/cdnsystem/client/piece_seed_stream.go b/pkg/rpc/cdnsystem/client/piece_seed_stream.go
index 21fca81c..781426fa 100644
--- a/pkg/rpc/cdnsystem/client/piece_seed_stream.go
+++ b/pkg/rpc/cdnsystem/client/piece_seed_stream.go
@@ -23,7 +23,6 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
-	"github.com/pkg/errors"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -64,22 +63,23 @@ func newPieceSeedStream(ctx context.Context, sc *cdnClient, hashKey string, sr *
 }
 
 func (pss *PieceSeedStream) initStream() error {
+	var target string
 	stream, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, cdnServerNode, err := pss.sc.getCdnClient(pss.hashKey, false)
+		var client cdnsystem.SeederClient
+		var err error
+		client, target, err = pss.sc.getCdnClient(pss.hashKey, false)
 		if err != nil {
 			return nil, err
 		}
-		logger.WithTaskID(pss.hashKey).Infof("invoke cdn node %s ObtainSeeds", cdnServerNode)
 		return client.ObtainSeeds(pss.ctx, pss.sr, pss.opts...)
 	}, pss.InitBackoff, pss.MaxBackOff, pss.MaxAttempts, nil)
-	if err == nil {
-		pss.stream = stream.(cdnsystem.Seeder_ObtainSeedsClient)
-		pss.StreamTimes = 1
-	}
 	if err != nil {
-		err = pss.replaceClient(pss.hashKey, err)
+		logger.WithTaskID(pss.hashKey).Infof("initStream: invoke cdn node %s ObtainSeeds failed: %v", target, err)
+		return pss.replaceClient(pss.hashKey, err)
 	}
-	return err
+	pss.stream = stream.(cdnsystem.Seeder_ObtainSeedsClient)
+	pss.StreamTimes = 1
+	return nil
 }
 
 func (pss *PieceSeedStream) Recv() (ps *cdnsystem.PieceSeed, err error) {
@@ -91,58 +91,61 @@ func (pss *PieceSeedStream) Recv() (ps *cdnsystem.PieceSeed, err error) {
 }
 
 func (pss *PieceSeedStream) retryRecv(cause error) (*cdnsystem.PieceSeed, error) {
-	if status.Code(cause) == codes.DeadlineExceeded {
+	if status.Code(cause) == codes.DeadlineExceeded || status.Code(cause) == codes.Canceled {
 		return nil, cause
 	}
-
-	if err := pss.replaceStream(pss.hashKey, cause); err != nil {
-		if err := pss.replaceClient(pss.hashKey, cause); err != nil {
-			return nil, cause
-		}
+	if err := pss.replaceStream(cause); err != nil {
+		return nil, err
 	}
-
 	return pss.Recv()
 }
 
-func (pss *PieceSeedStream) replaceStream(key string, cause error) error {
+func (pss *PieceSeedStream) replaceStream(cause error) error {
 	if pss.StreamTimes >= pss.MaxAttempts {
-		return errors.New("times of replacing stream reaches limit")
+		logger.WithTaskID(pss.hashKey).Info("replace stream reach max attempt")
+		return cause
 	}
-
+	var target string
 	stream, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, _, err := pss.sc.getCdnClient(key, true)
+		var client cdnsystem.SeederClient
+		var err error
+		client, target, err = pss.sc.getCdnClient(pss.hashKey, true)
 		if err != nil {
 			return nil, err
 		}
 		return client.ObtainSeeds(pss.ctx, pss.sr, pss.opts...)
 	}, pss.InitBackoff, pss.MaxBackOff, pss.MaxAttempts, cause)
-	if err == nil {
-		pss.stream = stream.(cdnsystem.Seeder_ObtainSeedsClient)
-		pss.StreamTimes++
+	if err != nil {
+		logger.WithTaskID(pss.hashKey).Infof("replaceStream: invoke cdn node %s ObtainSeeds failed: %v", target, err)
+		return pss.replaceStream(cause)
 	}
-	return err
+	pss.stream = stream.(cdnsystem.Seeder_ObtainSeedsClient)
+	pss.StreamTimes++
+	return nil
 }
 
 func (pss *PieceSeedStream) replaceClient(key string, cause error) error {
 	preNode, err := pss.sc.TryMigrate(key, cause, pss.failedServers)
 	if err != nil {
-		return err
+		logger.WithTaskID(pss.hashKey).Infof("replaceClient: tryMigrate cdn node failed: %v", err)
+		return cause
 	}
 	pss.failedServers = append(pss.failedServers, preNode)
-
+	var target string
 	stream, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, _, err := pss.sc.getCdnClient(key, true)
+		var client cdnsystem.SeederClient
+		var err error
+		client, target, err = pss.sc.getCdnClient(key, true)
 		if err != nil {
 			return nil, err
 		}
 		return client.ObtainSeeds(pss.ctx, pss.sr, pss.opts...)
 	}, pss.InitBackoff, pss.MaxBackOff, pss.MaxAttempts, cause)
-	if err == nil {
-		pss.stream = stream.(cdnsystem.Seeder_ObtainSeedsClient)
-		pss.StreamTimes = 1
-	}
 	if err != nil {
-		err = pss.replaceClient(key, cause)
+		logger.WithTaskID(pss.hashKey).Infof("replaceClient: invoke cdn node %s ObtainSeeds failed: %v", target, err)
+		return pss.replaceClient(key, cause)
 	}
-	return err
+	pss.stream = stream.(cdnsystem.Seeder_ObtainSeedsClient)
+	pss.StreamTimes = 1
+	return nil
 }
diff --git a/pkg/rpc/cdnsystem/server/server.go b/pkg/rpc/cdnsystem/server/server.go
index dba2f0b7..ff01f88f 100644
--- a/pkg/rpc/cdnsystem/server/server.go
+++ b/pkg/rpc/cdnsystem/server/server.go
@@ -39,7 +39,7 @@ func init() {
 	})
 }
 
-// see cdnsystem.SeederServer
+// SeederServer see cdnsystem.SeederServer
 type SeederServer interface {
 	ObtainSeeds(context.Context, *cdnsystem.SeedRequest, chan<- *cdnsystem.PieceSeed) error
 	GetPieceTasks(context.Context, *base.PieceTaskRequest) (*base.PiecePacket, error)
diff --git a/pkg/rpc/client.go b/pkg/rpc/client.go
index 438f1bce..bdf2bbbd 100644
--- a/pkg/rpc/client.go
+++ b/pkg/rpc/client.go
@@ -25,7 +25,10 @@ import (
 	"github.com/pkg/errors"
 	"github.com/serialx/hashring"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/status"
+	"k8s.io/apimachinery/pkg/util/sets"
 
 	"d7y.io/dragonfly/v2/internal/dfcodes"
 	"d7y.io/dragonfly/v2/internal/dferrors"
@@ -204,16 +207,10 @@ func (conn *Connection) AddServerNodes(addrs []dfnet.NetAddr) error {
 }
 
 // findCandidateClientConn find candidate node client conn other than exclusiveNodes
-func (conn *Connection) findCandidateClientConn(key string, exclusiveNodes ...string) (*candidateClient, error) {
+func (conn *Connection) findCandidateClientConn(key string, exclusiveNodes sets.String) (*candidateClient, error) {
 	if node, ok := conn.key2NodeMap.Load(key); ok {
 		candidateNode := node.(string)
-		selected := true
-		for _, exclusiveNode := range exclusiveNodes {
-			if exclusiveNode == candidateNode {
-				selected = false
-			}
-		}
-		if selected {
+		if !exclusiveNodes.Has(candidateNode) {
 			if client, ok := conn.node2ClientMap.Load(node); ok {
 				return &candidateClient{
 					node: candidateNode,
@@ -232,23 +229,17 @@ func (conn *Connection) findCandidateClientConn(key string, exclusiveNodes ...st
 	}
 	candidateNodes := make([]string, 0)
 	for _, ringNode := range ringNodes {
-		candidate := true
-		for _, exclusiveNode := range exclusiveNodes {
-			if exclusiveNode == ringNode {
-				candidate = false
-			}
-		}
-		if candidate {
+		if !exclusiveNodes.Has(ringNode) {
 			candidateNodes = append(candidateNodes, ringNode)
 		}
 	}
 	logger.With("conn", conn.name).Infof("candidate result for hash key %s: all server node list: %v, exclusiveNodes node list: %v, candidate node list: %v",
-		key, ringNodes, exclusiveNodes, candidateNodes)
+		key, ringNodes, exclusiveNodes.List(), candidateNodes)
 	for _, candidateNode := range candidateNodes {
 		// Check whether there is a corresponding mapping client in the node2ClientMap
 		// TODO 下面部分可以直接调用loadOrCreate方法，但是日志没有这么调用打印全
 		if client, ok := conn.node2ClientMap.Load(candidateNode); ok {
-			logger.With("conn", conn.name).Infof("hit cache candidateNode %s for hash key %s", candidateNode, key)
+			logger.With("conn", conn.name).Debugf("hit cache candidateNode %s for hash key %s", candidateNode, key)
 			return &candidateClient{
 				node: candidateNode,
 				Ref:  client,
@@ -334,6 +325,7 @@ func (conn *Connection) loadOrCreateClientConnByNode(node string) (clientConn *g
 // stick whether hash key need already associated with specify node
 func (conn *Connection) GetClientConn(hashKey string, stick bool) (*grpc.ClientConn, error) {
 	logger.With("conn", conn.name).Debugf("start to get client conn hashKey %s, stick %t", hashKey, stick)
+	defer logger.With("conn", conn.name).Debugf("get client conn done, hashKey %s, stick %t end", hashKey, stick)
 	conn.rwMutex.RLock()
 	node, ok := conn.key2NodeMap.Load(hashKey)
 	if stick && !ok {
@@ -351,12 +343,12 @@ func (conn *Connection) GetClientConn(hashKey string, stick bool) (*grpc.ClientC
 		}
 		return clientConn, nil
 	}
-	logger.With("conn", conn.name).Infof("no server node associated with hash key %s was found, start find candidate", hashKey)
+	logger.With("conn", conn.name).Infof("no server node associated with hash key %s was found, start find candidate server", hashKey)
 	conn.rwMutex.RUnlock()
 	// if absence
 	conn.rwMutex.Lock()
 	defer conn.rwMutex.Unlock()
-	client, err := conn.findCandidateClientConn(hashKey)
+	client, err := conn.findCandidateClientConn(hashKey, sets.NewString())
 	if err != nil {
 		return nil, errors.Wrapf(err, "prob candidate client conn for hash key %s", hashKey)
 	}
@@ -370,6 +362,10 @@ func (conn *Connection) GetClientConn(hashKey string, stick bool) (*grpc.ClientC
 // preNode node before the migration
 func (conn *Connection) TryMigrate(key string, cause error, exclusiveNodes []string) (preNode string, err error) {
 	logger.With("conn", conn.name).Infof("start try migrate server node for key %s, cause err: %v", key, cause)
+	if status.Code(cause) == codes.DeadlineExceeded || status.Code(cause) == codes.Canceled {
+		logger.With("conn", conn.name).Infof("migrate server node for key %s failed, cause err: %v", key, cause)
+		return "", cause
+	}
 	// TODO recover findCandidateClientConn error
 	if e, ok := cause.(*dferrors.DfError); ok {
 		if e.Code != dfcodes.ResourceLacked && e.Code != dfcodes.UnknownError {
@@ -378,16 +374,17 @@ func (conn *Connection) TryMigrate(key string, cause error, exclusiveNodes []str
 	}
 	currentNode := ""
 	conn.rwMutex.RLock()
-	if currentNode, ok := conn.key2NodeMap.Load(key); ok {
-		preNode = currentNode.(string)
-		exclusiveNodes = append(exclusiveNodes, currentNode.(string))
+	if node, ok := conn.key2NodeMap.Load(key); ok {
+		currentNode = node.(string)
+		preNode = currentNode
+		exclusiveNodes = append(exclusiveNodes, preNode)
 	} else {
 		logger.With("conn", conn.name).Warnf("failed to find server node for hash key %s", key)
 	}
 	conn.rwMutex.RUnlock()
 	conn.rwMutex.Lock()
 	defer conn.rwMutex.Unlock()
-	client, err := conn.findCandidateClientConn(key, exclusiveNodes...)
+	client, err := conn.findCandidateClientConn(key, sets.NewString(exclusiveNodes...))
 	if err != nil {
 		return "", errors.Wrapf(err, "find candidate client conn for hash key %s", key)
 	}
diff --git a/pkg/rpc/client_util.go b/pkg/rpc/client_util.go
index d6ddf92e..2d3a47a1 100644
--- a/pkg/rpc/client_util.go
+++ b/pkg/rpc/client_util.go
@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
 	"d7y.io/dragonfly/v2/internal/dfcodes"
@@ -190,6 +191,9 @@ func ExecuteWithRetry(f func() (interface{}, error), initBackoff float64, maxBac
 				return res, cause
 			}
 		}
+		if status.Code(cause) == codes.DeadlineExceeded || status.Code(cause) == codes.Canceled {
+			return res, cause
+		}
 		if i > 0 {
 			time.Sleep(mathutils.RandBackoff(initBackoff, maxBackoff, 2.0, i))
 		}
diff --git a/pkg/rpc/dfdaemon/client/client.go b/pkg/rpc/dfdaemon/client/client.go
index a23f2a0a..bfd842b3 100644
--- a/pkg/rpc/dfdaemon/client/client.go
+++ b/pkg/rpc/dfdaemon/client/client.go
@@ -22,6 +22,7 @@ import (
 	"sync"
 	"time"
 
+	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/idgen"
 	"d7y.io/dragonfly/v2/pkg/basic/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc"
@@ -67,7 +68,7 @@ func GetElasticClientByAddrs(addrs []dfnet.NetAddr, opts ...grpc.DialOption) (Da
 	return elasticDaemonClient, nil
 }
 
-// see dfdaemon.DaemonClient
+// DaemonClient see dfdaemon.DaemonClient
 type DaemonClient interface {
 	Download(ctx context.Context, req *dfdaemon.DownRequest, opts ...grpc.CallOption) (*DownResultStream, error)
 
@@ -114,12 +115,11 @@ func (dc *daemonClient) GetPieceTasks(ctx context.Context, target dfnet.NetAddr,
 		}
 		return client.GetPieceTasks(ctx, ptr, opts...)
 	}, 0.2, 2.0, 3, nil)
-
-	if err == nil {
-		return res.(*base.PiecePacket), nil
+	if err != nil {
+		logger.WithTaskID(ptr.TaskId).Infof("GetPieceTasks: invoke daemon node %s GetPieceTasks failed: %v", target, err)
+		return nil, err
 	}
-
-	return nil, err
+	return res.(*base.PiecePacket), nil
 }
 
 func (dc *daemonClient) CheckHealth(ctx context.Context, target dfnet.NetAddr, opts ...grpc.CallOption) (err error) {
@@ -130,6 +130,9 @@ func (dc *daemonClient) CheckHealth(ctx context.Context, target dfnet.NetAddr, o
 		}
 		return client.CheckHealth(ctx, new(empty.Empty), opts...)
 	}, 0.2, 2.0, 3, nil)
-
+	if err != nil {
+		logger.Infof("CheckHealth: invoke daemon node %s CheckHealth failed: %v", target, err)
+		return
+	}
 	return
 }
diff --git a/pkg/rpc/dfdaemon/client/down_result_stream.go b/pkg/rpc/dfdaemon/client/down_result_stream.go
index 66d9eafc..91a12fd8 100644
--- a/pkg/rpc/dfdaemon/client/down_result_stream.go
+++ b/pkg/rpc/dfdaemon/client/down_result_stream.go
@@ -23,7 +23,6 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
-	"github.com/pkg/errors"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -63,22 +62,23 @@ func newDownResultStream(ctx context.Context, dc *daemonClient, hashKey string,
 }
 
 func (drs *DownResultStream) initStream() error {
+	var target string
 	stream, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, _, err := drs.dc.getDaemonClient(drs.hashKey, false)
+		var client dfdaemon.DaemonClient
+		var err error
+		client, target, err = drs.dc.getDaemonClient(drs.hashKey, false)
 		if err != nil {
 			return nil, err
 		}
 		return client.Download(drs.ctx, drs.req, drs.opts...)
 	}, drs.InitBackoff, drs.MaxBackOff, drs.MaxAttempts, nil)
-	if err == nil {
-		drs.stream = stream.(dfdaemon.Daemon_DownloadClient)
-		drs.StreamTimes = 1
-	}
 	if err != nil {
-		err = drs.replaceClient(err)
+		logger.WithTaskID(drs.hashKey).Infof("initStream: invoke daemon node %s Download failed: %v", target, err)
+		return drs.replaceClient(err)
 	}
-
-	return err
+	drs.stream = stream.(dfdaemon.Daemon_DownloadClient)
+	drs.StreamTimes = 1
+	return nil
 }
 
 func (drs *DownResultStream) Recv() (dr *dfdaemon.DownResult, err error) {
@@ -100,14 +100,12 @@ func (drs *DownResultStream) Recv() (dr *dfdaemon.DownResult, err error) {
 }
 
 func (drs *DownResultStream) retryRecv(cause error) (*dfdaemon.DownResult, error) {
-	if status.Code(cause) == codes.DeadlineExceeded {
+	if status.Code(cause) == codes.DeadlineExceeded || status.Code(cause) == codes.Canceled {
 		return nil, cause
 	}
 
 	if err := drs.replaceStream(cause); err != nil {
-		if err := drs.replaceClient(cause); err != nil {
-			return nil, cause
-		}
+		return nil, err
 	}
 
 	return drs.Recv()
@@ -115,43 +113,51 @@ func (drs *DownResultStream) retryRecv(cause error) (*dfdaemon.DownResult, error
 
 func (drs *DownResultStream) replaceStream(cause error) error {
 	if drs.StreamTimes >= drs.MaxAttempts {
-		return errors.New("times of replacing stream reaches limit")
+		logger.WithTaskID(drs.hashKey).Info("replace stream reach max attempt")
+		return cause
 	}
-
+	var target string
 	stream, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, _, err := drs.dc.getDaemonClient(drs.hashKey, true)
+		var client dfdaemon.DaemonClient
+		var err error
+		client, target, err = drs.dc.getDaemonClient(drs.hashKey, true)
 		if err != nil {
 			return nil, err
 		}
 		return client.Download(drs.ctx, drs.req, drs.opts...)
 	}, drs.InitBackoff, drs.MaxBackOff, drs.MaxAttempts, cause)
-	if err == nil {
-		drs.stream = stream.(dfdaemon.Daemon_DownloadClient)
-		drs.StreamTimes++
+	if err != nil {
+		logger.WithTaskID(drs.hashKey).Infof("replaceStream: invoke daemon node %s Download failed: %v", target, err)
+		return drs.replaceClient(cause)
 	}
-	return err
+	drs.stream = stream.(dfdaemon.Daemon_DownloadClient)
+	drs.StreamTimes++
+	return nil
 }
 
 func (drs *DownResultStream) replaceClient(cause error) error {
 	preNode, err := drs.dc.TryMigrate(drs.hashKey, cause, drs.failedServers)
 	if err != nil {
-		return err
+		logger.WithTaskID(drs.hashKey).Infof("replaceClient: tryMigrate daemon node failed: %v", err)
+		return cause
 	}
 	drs.failedServers = append(drs.failedServers, preNode)
 
+	var target string
 	stream, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, _, err := drs.dc.getDaemonClient(drs.hashKey, true)
+		var client dfdaemon.DaemonClient
+		var err error
+		client, target, err = drs.dc.getDaemonClient(drs.hashKey, true)
 		if err != nil {
 			return nil, err
 		}
 		return client.Download(drs.ctx, drs.req, drs.opts...)
 	}, drs.InitBackoff, drs.MaxBackOff, drs.MaxAttempts, cause)
-	if err == nil {
-		drs.stream = stream.(dfdaemon.Daemon_DownloadClient)
-		drs.StreamTimes = 1
-	}
 	if err != nil {
-		err = drs.replaceClient(cause)
+		logger.WithTaskID(drs.hashKey).Infof("replaceClient: invoke daemon node %s Download failed: %v", target, err)
+		return drs.replaceClient(cause)
 	}
-	return err
+	drs.stream = stream.(dfdaemon.Daemon_DownloadClient)
+	drs.StreamTimes = 1
+	return nil
 }
diff --git a/pkg/rpc/scheduler/client/client.go b/pkg/rpc/scheduler/client/client.go
index 58c792f2..f600b55b 100644
--- a/pkg/rpc/scheduler/client/client.go
+++ b/pkg/rpc/scheduler/client/client.go
@@ -20,17 +20,17 @@ import (
 	"context"
 	"time"
 
-	"github.com/pkg/errors"
-	"google.golang.org/grpc"
-
 	"d7y.io/dragonfly/v2/internal/dfcodes"
-	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/idgen"
 	"d7y.io/dragonfly/v2/pkg/basic/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
+	"github.com/pkg/errors"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 func GetClientByAddr(addrs []dfnet.NetAddr, opts ...grpc.DialOption) (SchedulerClient, error) {
@@ -47,10 +47,11 @@ func GetClientByAddr(addrs []dfnet.NetAddr, opts ...grpc.DialOption) (SchedulerC
 	return sc, nil
 }
 
-// see scheduler.SchedulerClient
+// SchedulerClient see scheduler.SchedulerClient
 type SchedulerClient interface {
+	// RegisterPeerTask register peer task to scheduler
 	RegisterPeerTask(context.Context, *scheduler.PeerTaskRequest, ...grpc.CallOption) (*scheduler.RegisterResult, error)
-	// IsMigrating of ptr will be set to true
+	// ReportPieceResult IsMigrating of ptr will be set to true
 	ReportPieceResult(context.Context, string, *scheduler.PeerTaskRequest, ...grpc.CallOption) (PeerPacketStream, error)
 
 	ReportPeerResult(context.Context, *scheduler.PeerResult, ...grpc.CallOption) error
@@ -72,16 +73,9 @@ func (sc *schedulerClient) getSchedulerClient(key string, stick bool) (scheduler
 	return scheduler.NewSchedulerClient(clientConn), clientConn.Target(), nil
 }
 
-func (sc *schedulerClient) RegisterPeerTask(ctx context.Context, ptr *scheduler.PeerTaskRequest, opts ...grpc.CallOption) (rr *scheduler.RegisterResult,
-	err error) {
-	return sc.doRegisterPeerTask(ctx, ptr, []string{}, opts)
-}
-
-func (sc *schedulerClient) doRegisterPeerTask(ctx context.Context, ptr *scheduler.PeerTaskRequest, exclusiveNodes []string,
-	opts []grpc.CallOption) (rr *scheduler.RegisterResult, err error) {
+func (sc *schedulerClient) RegisterPeerTask(ctx context.Context, ptr *scheduler.PeerTaskRequest, opts ...grpc.CallOption) (*scheduler.RegisterResult, error) {
 	var (
 		taskID        string
-		code          base.Code
 		schedulerNode string
 		res           interface{}
 	)
@@ -90,65 +84,66 @@ func (sc *schedulerClient) doRegisterPeerTask(ctx context.Context, ptr *schedule
 		ptr.Url)
 	reg := func() (interface{}, error) {
 		var client scheduler.SchedulerClient
+		var err error
 		client, schedulerNode, err = sc.getSchedulerClient(key, false)
 		if err != nil {
-			code = dfcodes.ServerUnavailable
 			return nil, err
 		}
 		return client.RegisterPeerTask(ctx, ptr, opts...)
 	}
-	res, err = rpc.ExecuteWithRetry(reg, 0.5, 5.0, 5, nil)
-	if err == nil {
-		rr = res.(*scheduler.RegisterResult)
-		taskID = rr.TaskId
-		code = dfcodes.Success
-		if taskID != key {
-			logger.WithTaskAndPeerID(taskID, ptr.PeerId).Warnf("register peer task correct taskId from %s to %s", key, taskID)
-			sc.Connection.CorrectKey2NodeRelation(key, taskID)
-		}
-		logger.With("peerId", ptr.PeerId).
-			Infof("register peer task result success for taskId: %s, url: %s, scheduler: %s",
-				taskID, ptr.Url, schedulerNode)
-		return
+	res, err := rpc.ExecuteWithRetry(reg, 0.2, 2.0, 3, nil)
+	if err != nil {
+		logger.WithTaskAndPeerID(key, ptr.PeerId).Errorf("RegisterPeerTask: register peer task to scheduler %s failed: %v", schedulerNode, err)
+		return sc.retryRegisterPeerTask(ctx, key, ptr, []string{schedulerNode}, err, opts)
 	}
-
-	if de, ok := err.(*dferrors.DfError); ok {
-		code = de.Code
+	rr := res.(*scheduler.RegisterResult)
+	taskID = rr.TaskId
+	if taskID != key {
+		logger.WithTaskAndPeerID(taskID, ptr.PeerId).Warnf("register peer task correct taskId from %s to %s", key, taskID)
+		sc.Connection.CorrectKey2NodeRelation(key, taskID)
 	}
-	logger.With("peerId", ptr.PeerId, "errMsg", err).
-		Errorf("register peer task result failed, code: [%d] for taskId: %s, url: %s, scheduler: %s",
-			int32(code), taskID, ptr.Url, schedulerNode)
+	logger.WithTaskAndPeerID(taskID, ptr.PeerId).
+		Infof("register peer task result success url: %s, scheduler: %s", ptr.Url, schedulerNode)
+	return rr, err
+}
 
-	// previous schedule failed, report peer task to free load and other resources
-	var client scheduler.SchedulerClient
-	client, schedulerNode, err = sc.getSchedulerClient(key, false)
+func (sc *schedulerClient) retryRegisterPeerTask(ctx context.Context, hashKey string, ptr *scheduler.PeerTaskRequest, exclusiveNodes []string, cause error,
+	opts []grpc.CallOption) (*scheduler.RegisterResult, error) {
+	if status.Code(cause) == codes.Canceled || status.Code(cause) == codes.DeadlineExceeded {
+		return nil, cause
+	}
+	var (
+		taskID        string
+		schedulerNode string
+	)
+	preNode, err := sc.TryMigrate(hashKey, cause, exclusiveNodes)
 	if err != nil {
-		logger.With("peerId", ptr.PeerId, "errMsg", err).Errorf("get scheduler client failed")
-	} else {
-		_, e := client.ReportPeerResult(
-			context.Background(),
-			&scheduler.PeerResult{
-				TaskId:         taskID,
-				PeerId:         ptr.PeerId,
-				SrcIp:          ptr.PeerHost.Ip,
-				SecurityDomain: ptr.PeerHost.SecurityDomain,
-				Idc:            ptr.PeerHost.Idc,
-				Url:            ptr.Url,
-				ContentLength:  -1,
-				Traffic:        -1,
-				Cost:           0,
-				Success:        false,
-				Code:           dfcodes.UnknownError,
-			})
-		logger.With("peerId", ptr.PeerId, "errMsg", e).Warnf("report failed peer result")
-	}
-
-	var preNode string
-	if preNode, err = sc.TryMigrate(key, err, exclusiveNodes); err == nil {
-		exclusiveNodes = append(exclusiveNodes, preNode)
-		return sc.doRegisterPeerTask(ctx, ptr, exclusiveNodes, opts)
+		return nil, cause
 	}
-	return
+	exclusiveNodes = append(exclusiveNodes, preNode)
+	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
+		var client scheduler.SchedulerClient
+		var err error
+		client, schedulerNode, err = sc.getSchedulerClient(hashKey, true)
+		if err != nil {
+			return nil, err
+		}
+		return client.RegisterPeerTask(ctx, ptr, opts...)
+	}, 0.2, 2.0, 3, cause)
+	if err != nil {
+		logger.WithTaskAndPeerID(hashKey, ptr.PeerId).Errorf("retryRegisterPeerTask: register peer task to scheduler %s failed: %v", schedulerNode, err)
+		return sc.retryRegisterPeerTask(ctx, hashKey, ptr, exclusiveNodes, err, opts)
+	}
+	rr := res.(*scheduler.RegisterResult)
+	taskID = rr.TaskId
+	if taskID != hashKey {
+		logger.WithTaskAndPeerID(taskID, ptr.PeerId).Warnf("register peer task correct taskId from %s to %s", hashKey, taskID)
+		sc.Connection.CorrectKey2NodeRelation(hashKey, taskID)
+	}
+	logger.WithTaskAndPeerID(taskID, ptr.PeerId).
+		Infof("register peer task result success url: %s, scheduler: %s", ptr.Url, schedulerNode)
+	return rr, nil
+
 }
 
 func (sc *schedulerClient) ReportPieceResult(ctx context.Context, taskID string, ptr *scheduler.PeerTaskRequest, opts ...grpc.CallOption) (PeerPacketStream, error) {
@@ -162,16 +157,40 @@ func (sc *schedulerClient) ReportPieceResult(ctx context.Context, taskID string,
 }
 
 func (sc *schedulerClient) ReportPeerResult(ctx context.Context, pr *scheduler.PeerResult, opts ...grpc.CallOption) error {
-	return sc.doReportPeerResult(ctx, pr, []string{}, opts)
+	var (
+		schedulerNode string
+	)
+	_, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
+		var client scheduler.SchedulerClient
+		var err error
+		client, schedulerNode, err = sc.getSchedulerClient(pr.TaskId, true)
+		if err != nil {
+			return nil, err
+		}
+		return client.ReportPeerResult(ctx, pr, opts...)
+	}, 0.2, 2.0, 3, nil)
+	if err != nil {
+		logger.WithTaskAndPeerID(pr.TaskId, pr.PeerId).Errorf("ReportPeerResult: report peer result to scheduler %s failed: %v", schedulerNode, err)
+		return sc.retryReportPeerResult(ctx, pr, []string{schedulerNode}, err, opts)
+	}
+	return nil
 }
 
-func (sc *schedulerClient) doReportPeerResult(ctx context.Context, pr *scheduler.PeerResult, exclusiveNodes []string, opts []grpc.CallOption) (err error) {
+func (sc *schedulerClient) retryReportPeerResult(ctx context.Context, pr *scheduler.PeerResult, exclusiveNodes []string,
+	cause error, opts []grpc.CallOption) (err error) {
+	if status.Code(cause) == codes.Canceled || status.Code(cause) == codes.DeadlineExceeded {
+		return cause
+	}
 	var (
 		schedulerNode string
 		suc           bool
 		code          base.Code
 	)
-
+	preNode, err := sc.TryMigrate(pr.TaskId, err, exclusiveNodes)
+	if err != nil {
+		return cause
+	}
+	exclusiveNodes = append(exclusiveNodes, preNode)
 	_, err = rpc.ExecuteWithRetry(func() (interface{}, error) {
 		var client scheduler.SchedulerClient
 		client, schedulerNode, err = sc.getSchedulerClient(pr.TaskId, true)
@@ -180,24 +199,16 @@ func (sc *schedulerClient) doReportPeerResult(ctx context.Context, pr *scheduler
 			return nil, err
 		}
 		return client.ReportPeerResult(ctx, pr, opts...)
-	}, 0.5, 5.0, 5, nil)
-	if err == nil {
-		suc = true
-		code = dfcodes.Success
+	}, 0.2, 2.0, 3, nil)
+	if err != nil {
+		logger.WithTaskAndPeerID(pr.TaskId, pr.PeerId).Errorf("retryReportPeerResult: report peer result to scheduler %s failed: %v", schedulerNode, err)
+		return sc.retryReportPeerResult(ctx, pr, exclusiveNodes, cause, opts)
 	}
 
 	logger.With("peerId", pr.PeerId, "errMsg", err).
 		Infof("report peer result: %t[%d], peer task down result: %t[%d] for taskId: %s, url: %s, scheduler: %s, length: %d, traffic: %d, cost: %d", suc, int32(code),
 			pr.Success, int32(pr.Code), pr.TaskId, pr.Url, schedulerNode, pr.ContentLength, pr.Traffic, pr.Cost)
 
-	if err != nil {
-		var preNode string
-		if preNode, err = sc.TryMigrate(pr.TaskId, err, exclusiveNodes); err == nil {
-			exclusiveNodes = append(exclusiveNodes, preNode)
-			return sc.doReportPeerResult(ctx, pr, exclusiveNodes, opts)
-		}
-	}
-
 	return
 }
 
@@ -207,21 +218,22 @@ func (sc *schedulerClient) LeaveTask(ctx context.Context, pt *scheduler.PeerTarg
 		suc           bool
 	)
 	defer func() {
-		logger.With("peerId", pt.PeerId, "errMsg", err).Infof("leave from task result: %t for taskId: %s, scheduler: %s", suc, pt.TaskId, schedulerNode)
+		logger.With("peerId", pt.PeerId, "errMsg", err).Infof("leave from task result: %t for taskId: %s, scheduler server node: %s, err:%v", suc, pt.TaskId,
+			schedulerNode, err)
 	}()
 
-	_, err = rpc.ExecuteWithRetry(func() (interface{}, error) {
+	leaveFun := func() (interface{}, error) {
 		var client scheduler.SchedulerClient
-		client, schedulerNode, err = sc.getSchedulerClient(pt.TaskId, true)
+		client, schedulerNode, err = sc.getSchedulerClient(pt.TaskId, false)
 		if err != nil {
 			return nil, err
 		}
 		return client.LeaveTask(ctx, pt, opts...)
-	}, 0.5, 5.0, 3, nil)
+	}
+	_, err = rpc.ExecuteWithRetry(leaveFun, 0.2, 2.0, 3, nil)
 	if err == nil {
 		suc = true
 	}
-
 	return
 }
 
diff --git a/pkg/rpc/scheduler/client/peer_packet_stream.go b/pkg/rpc/scheduler/client/peer_packet_stream.go
index 8c88a2c1..dc15efdf 100644
--- a/pkg/rpc/scheduler/client/peer_packet_stream.go
+++ b/pkg/rpc/scheduler/client/peer_packet_stream.go
@@ -20,7 +20,7 @@ import (
 	"context"
 	"io"
 
-	"github.com/pkg/errors"
+	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -103,21 +103,19 @@ func (pps *peerPacketStream) Recv() (pp *scheduler.PeerPacket, err error) {
 }
 
 func (pps *peerPacketStream) retrySend(pr *scheduler.PieceResult, cause error) error {
-	if status.Code(cause) == codes.DeadlineExceeded {
+	if status.Code(cause) == codes.DeadlineExceeded || status.Code(cause) == codes.Canceled {
 		return cause
 	}
 
 	if err := pps.replaceStream(cause); err != nil {
-		if err := pps.replaceClient(cause); err != nil {
-			return cause
-		}
+		return err
 	}
 
 	return pps.Send(pr)
 }
 
 func (pps *peerPacketStream) retryRecv(cause error) (*scheduler.PeerPacket, error) {
-	if status.Code(cause) == codes.DeadlineExceeded {
+	if status.Code(cause) == codes.DeadlineExceeded || status.Code(cause) == codes.Canceled {
 		return nil, cause
 	}
 	_, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
@@ -150,67 +148,74 @@ func (pps *peerPacketStream) retryRecv(cause error) (*scheduler.PeerPacket, erro
 }
 
 func (pps *peerPacketStream) initStream() error {
+	var target string
 	stream, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, _, err := pps.sc.getSchedulerClient(pps.hashKey, true)
+		var client scheduler.SchedulerClient
+		var err error
+		client, target, err = pps.sc.getSchedulerClient(pps.hashKey, true)
 		if err != nil {
 			return nil, err
 		}
 		return client.ReportPieceResult(pps.ctx, pps.opts...)
 	}, pps.retryMeta.InitBackoff, pps.retryMeta.MaxBackOff, pps.retryMeta.MaxAttempts, nil)
-	if err == nil {
-		pps.stream = stream.(scheduler.Scheduler_ReportPieceResultClient)
-		pps.retryMeta.StreamTimes = 1
-	}
 	if err != nil {
-		err = pps.replaceClient(err)
+		logger.WithTaskID(pps.hashKey).Infof("initStream: invoke scheduler node %s ReportPieceResult failed: %v", target, err)
+		return pps.replaceClient(err)
 	}
-	return err
+	pps.stream = stream.(scheduler.Scheduler_ReportPieceResultClient)
+	pps.retryMeta.StreamTimes = 1
+	return nil
 }
 
 func (pps *peerPacketStream) replaceStream(cause error) error {
 	if pps.retryMeta.StreamTimes >= pps.retryMeta.MaxAttempts {
-		return errors.New("times of replacing stream reaches limit")
+		return cause
 	}
+	var target string
 	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, _, err := pps.sc.getSchedulerClient(pps.hashKey, true)
+		var client scheduler.SchedulerClient
+		var err error
+		client, target, err = pps.sc.getSchedulerClient(pps.hashKey, true)
 		if err != nil {
 			return nil, err
 		}
 		return client.ReportPieceResult(pps.ctx, pps.opts...)
 	}, pps.retryMeta.InitBackoff, pps.retryMeta.MaxBackOff, pps.retryMeta.MaxAttempts, cause)
-	if err == nil {
-		pps.stream = res.(scheduler.Scheduler_ReportPieceResultClient)
-		pps.retryMeta.StreamTimes++
+	if err != nil {
+		logger.WithTaskID(pps.hashKey).Infof("replaceStream: invoke scheduler node %s ReportPieceResult failed: %v", target, err)
+		return pps.replaceStream(cause)
 	}
-	return err
+	pps.stream = res.(scheduler.Scheduler_ReportPieceResultClient)
+	pps.retryMeta.StreamTimes++
+	return nil
 }
 
 func (pps *peerPacketStream) replaceClient(cause error) error {
 	preNode, err := pps.sc.TryMigrate(pps.hashKey, cause, pps.failedServers)
 	if err != nil {
-		return err
+		logger.WithTaskID(pps.hashKey).Infof("replaceClient: tryMigrate scheduler node failed: %v", err)
+		return cause
 	}
 	pps.failedServers = append(pps.failedServers, preNode)
-
+	var target string
 	stream, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, _, err := pps.sc.getSchedulerClient(pps.hashKey, true)
+		var client scheduler.SchedulerClient
+		var err error
+		client, target, err = pps.sc.getSchedulerClient(pps.hashKey, true)
 		if err != nil {
 			return nil, err
 		}
-		//timeCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-		//defer cancel()
 		_, err = client.RegisterPeerTask(pps.ctx, pps.ptr)
 		if err != nil {
 			return nil, err
 		}
 		return client.ReportPieceResult(pps.ctx, pps.opts...)
 	}, pps.retryMeta.InitBackoff, pps.retryMeta.MaxBackOff, pps.retryMeta.MaxAttempts, cause)
-	if err == nil {
-		pps.stream = stream.(scheduler.Scheduler_ReportPieceResultClient)
-		pps.retryMeta.StreamTimes = 1
-	}
 	if err != nil {
-		err = pps.replaceClient(cause)
+		logger.WithTaskID(pps.hashKey).Infof("replaceClient: invoke scheduler node %s ReportPieceResult failed: %v", target, err)
+		return pps.replaceClient(cause)
 	}
-	return err
+	pps.stream = stream.(scheduler.Scheduler_ReportPieceResultClient)
+	pps.retryMeta.StreamTimes = 1
+	return nil
 }
diff --git a/scheduler/core/scheduler/basic/basic_scheduler.go b/scheduler/core/scheduler/basic/basic_scheduler.go
index 289649ec..0c1f4589 100644
--- a/scheduler/core/scheduler/basic/basic_scheduler.go
+++ b/scheduler/core/scheduler/basic/basic_scheduler.go
@@ -146,6 +146,7 @@ func (s *Scheduler) selectCandidateChildren(peer *supervisor.Peer, limit int) (l
 			logger.WithTaskAndPeerID(peer.Task.TaskID, peer.PeerID).Debugf("******candidate child peer is not selected because it is nil")
 			return false
 		}
+		// TODO IsWaiting
 		if candidateNode.IsDone() {
 			logger.WithTaskAndPeerID(peer.Task.TaskID, peer.PeerID).Debugf("******candidate child peer %s is not selected because it has done",
 				candidateNode.PeerID)
diff --git a/scheduler/core/service.go b/scheduler/core/service.go
index 04597461..5cf659cc 100644
--- a/scheduler/core/service.go
+++ b/scheduler/core/service.go
@@ -179,14 +179,12 @@ func (s *SchedulerService) GetPeerTask(peerTaskID string) (peerTask *supervisor.
 }
 
 func (s *SchedulerService) RegisterPeerTask(req *schedulerRPC.PeerTaskRequest, task *supervisor.Task) (*supervisor.Peer, error) {
-	// get or create host
-	reqPeerHost := req.PeerHost
 	var (
 		peer     *supervisor.Peer
 		ok       bool
 		peerHost *supervisor.PeerHost
 	)
-
+	reqPeerHost := req.PeerHost
 	if peerHost, ok = s.hostManager.Get(reqPeerHost.Uuid); !ok {
 		peerHost = supervisor.NewClientPeerHost(reqPeerHost.Uuid, reqPeerHost.Ip, reqPeerHost.HostName, reqPeerHost.RpcPort, reqPeerHost.DownPort,
 			reqPeerHost.SecurityDomain, reqPeerHost.Location, reqPeerHost.Idc, reqPeerHost.NetTopology, s.config.ClientLoad)


Test patch:
diff --git a/cdnsystem/supervisor/cdn/storage/disk/disk_test.go b/cdnsystem/supervisor/cdn/storage/disk/disk_test.go
new file mode 100644
index 00000000..b9048efe
--- /dev/null
+++ b/cdnsystem/supervisor/cdn/storage/disk/disk_test.go
@@ -0,0 +1,115 @@
+package disk
+
+import (
+	"fmt"
+	"testing"
+
+	"d7y.io/dragonfly/v2/cdnsystem/storedriver"
+	"d7y.io/dragonfly/v2/cdnsystem/supervisor/cdn/storage"
+	"d7y.io/dragonfly/v2/cdnsystem/supervisor/mock"
+	"github.com/golang/mock/gomock"
+
+	"d7y.io/dragonfly/v2/pkg/unit"
+
+	"github.com/stretchr/testify/suite"
+)
+
+func TestDiskStorageMgrSuite(t *testing.T) {
+	suite.Run(t, new(DiskStorageMgrSuite))
+}
+
+type DiskStorageMgrSuite struct {
+	m *diskStorageMgr
+	suite.Suite
+}
+
+func (suite *DiskStorageMgrSuite) TestTryFreeSpace() {
+	ctrl := gomock.NewController(suite.T())
+	diskDriver := storedriver.NewMockDriver(ctrl)
+	taskMgr := mock.NewMockSeedTaskMgr(ctrl)
+	suite.m = &diskStorageMgr{
+		diskDriver: diskDriver,
+		taskMgr:    taskMgr,
+	}
+	diskDriver.EXPECT().GetTotalSpace().Return(100*unit.GB, nil)
+	cleaner, _ := storage.NewStorageCleaner(suite.m.getDefaultGcConfig(), diskDriver, suite.m, taskMgr)
+	suite.m.cleaner = cleaner
+
+	tests := []struct {
+		name       string
+		setupSuite func()
+		fileLength int64
+		success    func(bool, error) bool
+	}{
+		{
+			name: "very large free space",
+			setupSuite: func() {
+				// call GetFreeSpace 1 time in TryFreeSpace and return
+				diskDriver.EXPECT().GetFreeSpace().Return(unit.TB, nil)
+			},
+			fileLength: unit.MB.ToNumber(),
+			success: func(ok bool, err error) bool {
+				return ok == true && err == nil
+			},
+		},
+		{
+			name: "try a small file",
+			setupSuite: func() {
+				// call GetFreeSpace 1 time in TryFreeSpace
+				diskDriver.EXPECT().GetFreeSpace().Return(100*unit.GB, nil)
+				// call Walk 1 time in TryFreeSpace
+				diskDriver.EXPECT().Walk(gomock.Any())
+			},
+			fileLength: unit.KB.ToNumber(),
+			success: func(ok bool, err error) bool {
+				return ok == true && err == nil
+			},
+		},
+		{
+			name: "try a very large file",
+			setupSuite: func() {
+				// call GetFreeSpace 2 times in TryFreeSpace, 1 time in GC
+				diskDriver.EXPECT().GetFreeSpace().Return(100*unit.GB, nil).Times(3)
+				// call Walk 2 times in TryFreeSpace, 1 time in GC
+				diskDriver.EXPECT().Walk(gomock.Any()).Times(3)
+			},
+			fileLength: unit.TB.ToNumber(),
+			success: func(ok bool, err error) bool {
+				return ok == false && err == nil
+			},
+		},
+		{
+			name: "if get free space meets error",
+			setupSuite: func() {
+				// call GetFreeSpace 1 times in TryFreeSpace and return
+				diskDriver.EXPECT().GetFreeSpace().Return(unit.ToBytes(0), fmt.Errorf("a error for test"))
+			},
+			fileLength: unit.MB.ToNumber(),
+			success: func(ok bool, err error) bool {
+				return ok == false && err != nil && err.Error() == "a error for test"
+			},
+		},
+		{
+			name: "ok after gc",
+			setupSuite: func() {
+				// first call GetFreeSpace 1 times in TryFreeSpace, 1 time in GC
+				diskDriver.EXPECT().GetFreeSpace().Return(100*unit.MB, nil).Times(2)
+				// then call GetFreeSpace 1 times in TryFreeSpace, get another value
+				diskDriver.EXPECT().GetFreeSpace().Return(100*unit.GB, nil)
+				// call Walk 2 times in TryFreeSpace, 1 time in GC
+				diskDriver.EXPECT().Walk(gomock.Any()).Times(3)
+			},
+			fileLength: unit.GB.ToNumber(),
+			success: func(ok bool, err error) bool {
+				return ok == true && err == nil
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		suite.Run(tt.name, func() {
+			tt.setupSuite()
+			suite.True(tt.success(suite.m.TryFreeSpace(tt.fileLength)))
+		})
+	}
+}
diff --git a/test/e2e/concurrency_test.go b/test/e2e/concurrency_test.go
index 33c87c21..8cc00996 100644
--- a/test/e2e/concurrency_test.go
+++ b/test/e2e/concurrency_test.go
@@ -24,11 +24,6 @@ import (
 	. "github.com/onsi/gomega" //nolint
 )
 
-const (
-	hostnameFilePath = "/etc/hostname"
-	proxy            = "localhost:65001"
-)
-
 var _ = Describe("Download concurrency", func() {
 	Context("ab", func() {
 		It("concurrent 100 should be ok", func() {
diff --git a/test/e2e/dfget_test.go b/test/e2e/dfget_test.go
index ba264a29..45dce33d 100644
--- a/test/e2e/dfget_test.go
+++ b/test/e2e/dfget_test.go
@@ -25,10 +25,6 @@ import (
 	. "github.com/onsi/gomega" //nolint
 )
 
-const (
-	dragonflyNamespace = "dragonfly-system"
-)
-
 var _ = Describe("Download with dfget", func() {
 	Context("dfget", func() {
 		It("dfget download should be ok", func() {
diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go
index 33f6b51d..b1c72d4c 100644
--- a/test/e2e/e2e_test.go
+++ b/test/e2e/e2e_test.go
@@ -24,6 +24,14 @@ import (
 	"d7y.io/dragonfly/v2/test/e2e/e2eutil"
 	. "github.com/onsi/ginkgo" //nolint
 	. "github.com/onsi/gomega" //nolint
+
+	_ "d7y.io/dragonfly/v2/test/e2e/manager"
+)
+
+const (
+	proxy              = "localhost:65001"
+	hostnameFilePath   = "/etc/hostname"
+	dragonflyNamespace = "dragonfly-system"
 )
 
 var _ = BeforeSuite(func() {
diff --git a/test/e2e/e2eutil/exec.go b/test/e2e/e2eutil/exec.go
index e93c226f..904962de 100644
--- a/test/e2e/e2eutil/exec.go
+++ b/test/e2e/e2eutil/exec.go
@@ -1,6 +1,8 @@
 package e2eutil
 
 import (
+	"encoding/json"
+	"fmt"
 	"os/exec"
 )
 
@@ -55,6 +57,23 @@ func (p *PodExec) Command(arg ...string) *exec.Cmd {
 	return KubeCtlCommand(extArgs...)
 }
 
+func (p *PodExec) CurlCommand(method string, header map[string]string, data map[string]interface{}, target string) *exec.Cmd {
+	extArgs := []string{"/usr/bin/curl", target, "-s"}
+	if method != "" {
+		extArgs = append(extArgs, "-X", method)
+	}
+	if header != nil {
+		for k, v := range header {
+			extArgs = append(extArgs, "-H", fmt.Sprintf("%s:%s", k, v))
+		}
+	}
+	if data != nil {
+		jsonData, _ := json.Marshal(data)
+		extArgs = append(extArgs, "-d", string(jsonData))
+	}
+	return p.Command(extArgs...)
+}
+
 func KubeCtlCopyCommand(ns, pod, source, target string) *exec.Cmd {
 	return KubeCtlCommand("-n", ns, "cp", pod+":"+source, target)
 }
diff --git a/test/e2e/manager/constants.go b/test/e2e/manager/constants.go
new file mode 100644
index 00000000..7b97c6de
--- /dev/null
+++ b/test/e2e/manager/constants.go
@@ -0,0 +1,16 @@
+package manager
+
+const (
+	cdnCachePath = "/tmp/cdn/download"
+
+	managerService = "dragonfly-manager.dragonfly-system.svc"
+	managerPort    = "8080"
+	preheatPath    = "api/v1/preheats"
+	managerTag     = "d7y/manager"
+
+	dragonflyNamespace = "dragonfly-system"
+	e2eNamespace       = "dragonfly-e2e"
+
+	proxy            = "localhost:65001"
+	hostnameFilePath = "/etc/hostname"
+)
diff --git a/test/e2e/manager/preheat.go b/test/e2e/manager/preheat.go
new file mode 100644
index 00000000..3bbb4b50
--- /dev/null
+++ b/test/e2e/manager/preheat.go
@@ -0,0 +1,245 @@
+package manager
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"d7y.io/dragonfly/v2/internal/idgen"
+	"d7y.io/dragonfly/v2/manager/types"
+	"d7y.io/dragonfly/v2/pkg/rpc/base"
+	"d7y.io/dragonfly/v2/test/e2e/e2eutil"
+	machineryv1tasks "github.com/RichardKnop/machinery/v1/tasks"
+	. "github.com/onsi/ginkgo" //nolint
+	. "github.com/onsi/gomega" //nolint
+)
+
+var _ = Describe("Preheat with manager", func() {
+	Context("preheat", func() {
+		It("preheat files should be ok", func() {
+			var cdnPods [3]*e2eutil.PodExec
+			for i := 0; i < 3; i++ {
+				cdnPods[i] = getCDNExec(i)
+			}
+			fsPod := getFileServerExec()
+
+			for _, v := range e2eutil.GetFileList() {
+				url := e2eutil.GetFileURL(v)
+				fmt.Println("download url " + url)
+
+				// get original file digest
+				out, err := e2eutil.DockerCommand("sha256sum", v).CombinedOutput()
+				fmt.Println(string(out))
+				Expect(err).NotTo(HaveOccurred())
+				sha256sum1 := strings.Split(string(out), " ")[0]
+
+				// preheat file
+				out, err = fsPod.CurlCommand("POST", map[string]string{"Content-Type": "application/json"},
+					map[string]interface{}{"type": "file", "url": url},
+					fmt.Sprintf("http://%s:%s/%s", managerService, managerPort, preheatPath)).CombinedOutput()
+				fmt.Println(string(out))
+				Expect(err).NotTo(HaveOccurred())
+
+				// wait for success
+				preheatJob := &types.Preheat{}
+				err = json.Unmarshal(out, preheatJob)
+				Expect(err).NotTo(HaveOccurred())
+				done := waitForDone(preheatJob, fsPod)
+				Expect(done).Should(BeTrue())
+
+				// generate task_id, also the filename
+				cdnTaskID := idgen.TaskID(url, &base.UrlMeta{Tag: managerTag})
+				fmt.Println(cdnTaskID)
+
+				sha256sum2 := checkPreheatResult(cdnPods, cdnTaskID)
+				if sha256sum2 == "" {
+					fmt.Println("preheat file not found")
+				}
+				Expect(sha256sum1).To(Equal(sha256sum2))
+			}
+		})
+
+		It("preheat image should be ok", func() {
+			url := "https://registry-1.docker.io/v2/library/alpine/manifests/3.14"
+			fmt.Println("download image " + url)
+
+			var (
+				cdnTaskIDs = []string{
+					"effb4ac6e36d9a2a425ab142ba0a21fd0d49feea67a839fbd776ebb04e6f9eb7",
+					"ceaaf57ceba7221c2d54c62d77860e28b091837f235ba802c0722c522d6c7a8a",
+				}
+				sha256sum1 = []string{
+					"14119a10abf4669e8cdbdff324a9f9605d99697215a0d21c360fe8dfa8471bab",
+					"a0d0a0d46f8b52473982a3c466318f479767577551a53ffc9074c9fa7035982e",
+				}
+			)
+
+			var cdnPods [3]*e2eutil.PodExec
+			for i := 0; i < 3; i++ {
+				cdnPods[i] = getCDNExec(i)
+			}
+			fsPod := getFileServerExec()
+
+			// preheat file
+			out, err := fsPod.CurlCommand("POST", map[string]string{"Content-Type": "application/json"},
+				map[string]interface{}{"type": "image", "url": url},
+				fmt.Sprintf("http://%s:%s/%s", managerService, managerPort, preheatPath)).CombinedOutput()
+			fmt.Println(string(out))
+			Expect(err).NotTo(HaveOccurred())
+
+			// wait for success
+			preheatJob := &types.Preheat{}
+			err = json.Unmarshal(out, preheatJob)
+			Expect(err).NotTo(HaveOccurred())
+			done := waitForDone(preheatJob, fsPod)
+			Expect(done).Should(BeTrue())
+
+			for i, cdnTaskID := range cdnTaskIDs {
+				sha256sum2 := checkPreheatResult(cdnPods, cdnTaskID)
+				if sha256sum2 == "" {
+					fmt.Println("preheat file not found")
+				}
+				Expect(sha256sum1[i]).To(Equal(sha256sum2))
+			}
+		})
+
+		It("concurrency 100 preheat should be ok", func() {
+			// generate the data file
+			url := e2eutil.GetFileURL(hostnameFilePath)
+			fmt.Println("download url " + url)
+			dataFilePath := "post_data"
+			fd, err := os.Create(dataFilePath)
+			Expect(err).NotTo(HaveOccurred())
+			_, err = fd.WriteString(fmt.Sprintf(`{"type":"file","url":"%s"}`, url))
+			fd.Close()
+			Expect(err).NotTo(HaveOccurred())
+
+			// use ab to post the data file to manager concurrently
+			out, err := e2eutil.ABCommand("-c", "100", "-n", "200", "-T", "application/json", "-p", dataFilePath, "-X", proxy, fmt.Sprintf("http://%s:%s/%s", managerService, managerPort, preheatPath)).CombinedOutput()
+			fmt.Println(string(out))
+			Expect(err).NotTo(HaveOccurred())
+			os.Remove(dataFilePath)
+			Expect(err).NotTo(HaveOccurred())
+
+			// get original file digest
+			out, err = e2eutil.DockerCommand("sha256sum", hostnameFilePath).CombinedOutput()
+			fmt.Println(string(out))
+			Expect(err).NotTo(HaveOccurred())
+			sha256sum1 := strings.Split(string(out), " ")[0]
+
+			var cdnPods [3]*e2eutil.PodExec
+			for i := 0; i < 3; i++ {
+				cdnPods[i] = getCDNExec(i)
+			}
+			fsPod := getFileServerExec()
+
+			// use a curl to preheat the same file, git a id to wait for success
+			out, err = fsPod.CurlCommand("POST", map[string]string{"Content-Type": "application/json"},
+				map[string]interface{}{"type": "file", "url": url},
+				fmt.Sprintf("http://%s:%s/%s", managerService, managerPort, preheatPath)).CombinedOutput()
+			fmt.Println(string(out))
+			Expect(err).NotTo(HaveOccurred())
+
+			// wait for success
+			preheatJob := &types.Preheat{}
+			err = json.Unmarshal(out, preheatJob)
+			Expect(err).NotTo(HaveOccurred())
+			done := waitForDone(preheatJob, fsPod)
+			Expect(done).Should(BeTrue())
+
+			// generate task id to find the file
+			cdnTaskID := idgen.TaskID(url, &base.UrlMeta{Tag: managerTag})
+			fmt.Println(cdnTaskID)
+
+			sha256sum2 := checkPreheatResult(cdnPods, cdnTaskID)
+			if sha256sum2 == "" {
+				fmt.Println("preheat file not found")
+			}
+			Expect(sha256sum1).To(Equal(sha256sum2))
+		})
+	})
+})
+
+func waitForDone(preheat *types.Preheat, pod *e2eutil.PodExec) bool {
+	ticker := time.NewTicker(time.Second * 3)
+	defer ticker.Stop()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute*5)
+	defer cancel()
+	for {
+		select {
+		case <-ctx.Done():
+			return false
+		case <-ticker.C:
+			out, err := pod.CurlCommand("", nil, nil,
+				fmt.Sprintf("http://%s:%s/%s/%s", managerService, managerPort, preheatPath, preheat.ID)).CombinedOutput()
+			fmt.Println(string(out))
+			Expect(err).NotTo(HaveOccurred())
+			err = json.Unmarshal(out, preheat)
+			Expect(err).NotTo(HaveOccurred())
+			switch preheat.Status {
+			case machineryv1tasks.StateSuccess:
+				return true
+			case machineryv1tasks.StateFailure:
+				return false
+			default:
+			}
+		}
+	}
+}
+
+func checkPreheatResult(cdnPods [3]*e2eutil.PodExec, cdnTaskID string) string {
+	var sha256sum2 string
+	for _, cdn := range cdnPods {
+		out, err := cdn.Command("ls", cdnCachePath).CombinedOutput()
+		if err != nil {
+			// if the directory does not exist, skip this cdn
+			continue
+		}
+		// directory name is the first three characters of the task id
+		dir := cdnTaskID[0:3]
+		if !strings.Contains(string(out), dir) {
+			continue
+		}
+
+		out, err = cdn.Command("ls", fmt.Sprintf("%s/%s", cdnCachePath, dir)).CombinedOutput()
+		Expect(err).NotTo(HaveOccurred())
+		// file name is the same as task id
+		file := cdnTaskID
+		if !strings.Contains(string(out), file) {
+			continue
+		}
+
+		// calculate digest of downloaded file
+		out, err = cdn.Command("sha256sum", fmt.Sprintf("%s/%s/%s", cdnCachePath, dir, file)).CombinedOutput()
+		fmt.Println(string(out))
+		Expect(err).NotTo(HaveOccurred())
+		sha256sum2 = strings.Split(string(out), " ")[0]
+		break
+	}
+	return sha256sum2
+}
+
+// getCDNExec get cdn pods
+func getCDNExec(n int) *e2eutil.PodExec {
+	out, err := e2eutil.KubeCtlCommand("-n", dragonflyNamespace, "get", "pod", "-l", "component=cdn",
+		"-o", fmt.Sprintf("jsonpath='{range .items[%d]}{.metadata.name}{end}'", n)).CombinedOutput()
+	podName := strings.Trim(string(out), "'")
+	Expect(err).NotTo(HaveOccurred())
+	fmt.Println(podName)
+	Expect(strings.HasPrefix(podName, "dragonfly-cdn-")).Should(BeTrue())
+	return e2eutil.NewPodExec(dragonflyNamespace, podName, "cdn")
+}
+
+// getFileServerExec get the file-server pod for curl
+func getFileServerExec() *e2eutil.PodExec {
+	out, err := e2eutil.KubeCtlCommand("-n", e2eNamespace, "get", "pod", "-l", "component=file-server",
+		"-o", "jsonpath='{range .items[*]}{.metadata.name}{end}'").CombinedOutput()
+	podName := strings.Trim(string(out), "'")
+	Expect(err).NotTo(HaveOccurred())
+	fmt.Println(podName)
+	Expect(strings.HasPrefix(podName, "file-server-")).Should(BeTrue())
+	return e2eutil.NewPodExec(e2eNamespace, podName, "")
+}