Repository: dragonflyoss/Dragonfly2
Pull request #517 (2021-08-04 02:43:08)

let cdn support sha256
let cdn support sha256
fix #517 


Patch:
diff --git a/build/images/base/Dockerfile b/build/images/base/Dockerfile
new file mode 100644
index 00000000..b35d2a00
--- /dev/null
+++ b/build/images/base/Dockerfile
@@ -0,0 +1,24 @@
+FROM quay.io/iovisor/bpftrace:v0.13.0
+
+ENV DEBIAN_FRONTEND noninteractive
+
+RUN apt-get update \
+    && apt-get install -y binutils bpfcc-tools curl git iproute2 iputils-ping jq socat tree vim wget \
+    && apt-get install -y openssh-client openssh-server \
+    && rm -rf /var/lib/apt/lists/* \
+    && mkdir -p /run/sshd
+
+ENV PATH=/go/bin:/usr/local/go/bin:$PATH
+
+ENV GOPATH=/go
+ENV GOLANG_VERSION 1.16.6
+ENV GOLANG_DOWNLOAD_URL https://golang.org/dl/go$GOLANG_VERSION.linux-amd64.tar.gz
+ENV GOLANG_DOWNLOAD_SHA256 be333ef18b3016e9d7cb7b1ff1fdb0cac800ca0be4cf2290fe613b3d069dfe0d
+
+ARG GOPROXY
+
+RUN curl -fsSL "$GOLANG_DOWNLOAD_URL" -o golang.tar.gz \
+	&& echo "$GOLANG_DOWNLOAD_SHA256  golang.tar.gz" | sha256sum -c - \
+	&& tar -C /usr/local -xzf golang.tar.gz \
+	&& rm golang.tar.gz \
+	&& go install github.com/go-delve/delve/cmd/dlv@v1.7.0
\ No newline at end of file
diff --git a/build/images/base/README.md b/build/images/base/README.md
new file mode 100644
index 00000000..172e5f79
--- /dev/null
+++ b/build/images/base/README.md
@@ -0,0 +1,26 @@
+# Base Image for Debugging
+
+## Build Base Image
+
+```shell
+BASE_IMAGE=dragonflyoss/base:bpftrace-v0.13.0-go-v1.16.6
+
+docker run -ti -v /usr/src:/usr/src:ro \
+    -v /lib/modules/:/lib/modules:ro \
+    -v /sys/kernel/debug/:/sys/kernel/debug:rw \
+    --net=host --pid=host --privileged \
+    ${BASE_IMAGE}
+```
+
+## Build Dragonfly Images
+
+```shell
+# tags for condition compiling
+export GOTAGS="debug"
+# gcflags for dlv
+export GOGCFLAGS="all=-N -l"
+# base image
+export BASE_IMAGE=dragonflyoss/base:bpftrace-v0.13.0-go-v1.16.6
+
+make docker-build
+```
\ No newline at end of file
diff --git a/build/images/cdn/Dockerfile b/build/images/cdn/Dockerfile
index 33abe492..1a0adab6 100644
--- a/build/images/cdn/Dockerfile
+++ b/build/images/cdn/Dockerfile
@@ -1,4 +1,4 @@
-FROM golang:1.15.10-alpine as builder
+FROM golang:1.16.6-alpine as builder
 
 WORKDIR /go/src/d7y.io/dragonfly/v2
 
@@ -8,9 +8,11 @@ COPY . /go/src/d7y.io/dragonfly/v2
 
 ARG GOPROXY
 ARG GOTAGS
+ARG GOGCFLAGS
 
 RUN make build-cdn && make install-cdn
 
+# TODO support BASE_IMAGE
 FROM nginx:1.19-alpine
 
 COPY --from=builder /go/src/d7y.io/dragonfly/v2/hack/start-cdn.sh /root/start.sh
diff --git a/build/images/dfdaemon/Dockerfile b/build/images/dfdaemon/Dockerfile
index d8cf7acb..df5fa1d6 100644
--- a/build/images/dfdaemon/Dockerfile
+++ b/build/images/dfdaemon/Dockerfile
@@ -1,4 +1,6 @@
-FROM golang:1.15.10-alpine as builder
+ARG BASE_IMAGE=alpine:3.12
+
+FROM golang:1.16.6-alpine as builder
 
 WORKDIR /go/src/d7y.io/dragonfly/v2
 
@@ -8,12 +10,13 @@ COPY . /go/src/d7y.io/dragonfly/v2
 
 ARG GOPROXY
 ARG GOTAGS
+ARG GOGCFLAGS
 
 RUN make build-dfget && make install-dfget
 
-FROM alpine:3.12
+FROM ${BASE_IMAGE}
 
-ENV PATH=/opt/dragonfly/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
+ENV PATH=/opt/dragonfly/bin:$PATH
 
 COPY --from=builder /opt/dragonfly/bin/dfget /opt/dragonfly/bin/dfget
 
diff --git a/build/images/manager/Dockerfile b/build/images/manager/Dockerfile
index 23c7f786..bcaabfdd 100644
--- a/build/images/manager/Dockerfile
+++ b/build/images/manager/Dockerfile
@@ -1,4 +1,6 @@
-FROM golang:1.15.10-alpine as builder
+ARG BASE_IMAGE=alpine:3.12
+
+FROM golang:1.16.6-alpine as builder
 
 WORKDIR /go/src/d7y.io/dragonfly/v2
 
@@ -8,12 +10,13 @@ COPY . /go/src/d7y.io/dragonfly/v2
 
 ARG GOPROXY
 ARG GOTAGS
+ARG GOGCFLAGS
 
 RUN make build-manager && make install-manager
 
-FROM alpine:3.12
+FROM ${BASE_IMAGE}
 
-ENV PATH=/opt/dragonfly/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
+ENV PATH=/opt/dragonfly/bin:$PATH
 
 COPY --from=builder /opt/dragonfly/bin/manager /opt/dragonfly/bin/manager
 
diff --git a/build/images/scheduler/Dockerfile b/build/images/scheduler/Dockerfile
index 595b0993..aa581c2a 100644
--- a/build/images/scheduler/Dockerfile
+++ b/build/images/scheduler/Dockerfile
@@ -1,4 +1,6 @@
-FROM golang:1.15.10-alpine as builder
+ARG BASE_IMAGE=alpine:3.12
+
+FROM golang:1.16.6-alpine as builder
 
 WORKDIR /go/src/d7y.io/dragonfly/v2
 
@@ -8,12 +10,13 @@ COPY . /go/src/d7y.io/dragonfly/v2
 
 ARG GOPROXY
 ARG GOTAGS
+ARG GOGCFLAGS
 
 RUN make build-scheduler && make install-scheduler
 
-FROM alpine:3.12
+FROM ${BASE_IMAGE}
 
-ENV PATH=/opt/dragonfly/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
+ENV PATH=/opt/dragonfly/bin:$PATH
 
 COPY --from=builder /opt/dragonfly/bin/scheduler /opt/dragonfly/bin/scheduler
 
diff --git a/cdnsystem/daemon/cdn/cache_data_mgr.go b/cdnsystem/daemon/cdn/cache_data_mgr.go
index c73ebc1c..50457754 100644
--- a/cdnsystem/daemon/cdn/cache_data_mgr.go
+++ b/cdnsystem/daemon/cdn/cache_data_mgr.go
@@ -119,8 +119,8 @@ func (mm *cacheDataManager) updateStatusAndResult(taskID string, metaData *stora
 		if metaData.TotalPieceCount > 0 {
 			originMetaData.TotalPieceCount = metaData.TotalPieceCount
 		}
-		if !stringutils.IsBlank(metaData.SourceRealMd5) {
-			originMetaData.SourceRealMd5 = metaData.SourceRealMd5
+		if !stringutils.IsBlank(metaData.SourceRealDigest) {
+			originMetaData.SourceRealDigest = metaData.SourceRealDigest
 		}
 		if !stringutils.IsBlank(metaData.PieceMd5Sign) {
 			originMetaData.PieceMd5Sign = metaData.PieceMd5Sign
diff --git a/cdnsystem/daemon/cdn/cache_detector.go b/cdnsystem/daemon/cdn/cache_detector.go
index dd1beaa5..860c7135 100644
--- a/cdnsystem/daemon/cdn/cache_detector.go
+++ b/cdnsystem/daemon/cdn/cache_detector.go
@@ -59,12 +59,12 @@ func newCacheDetector(cacheDataManager *cacheDataManager) *cacheDetector {
 	}
 }
 
-func (cd *cacheDetector) detectCache(task *types.SeedTask, fileMd5 hash.Hash) (*cacheResult, error) {
+func (cd *cacheDetector) detectCache(task *types.SeedTask, fileDigest hash.Hash) (*cacheResult, error) {
 	//err := cd.cacheStore.CreateUploadLink(ctx, task.TaskId)
 	//if err != nil {
 	//	return nil, errors.Wrapf(err, "failed to create upload symbolic link")
 	//}
-	result, err := cd.doDetect(task, fileMd5)
+	result, err := cd.doDetect(task, fileDigest)
 	if err != nil {
 		logger.WithTaskID(task.TaskID).Infof("failed to detect cache, reset cache: %v", err)
 		metaData, err := cd.resetCache(task)
@@ -83,7 +83,7 @@ func (cd *cacheDetector) detectCache(task *types.SeedTask, fileMd5 hash.Hash) (*
 }
 
 // doDetect the actual detect action which detects file metaData and pieces metaData of specific task
-func (cd *cacheDetector) doDetect(task *types.SeedTask, fileMd5 hash.Hash) (result *cacheResult, err error) {
+func (cd *cacheDetector) doDetect(task *types.SeedTask, fileDigest hash.Hash) (result *cacheResult, err error) {
 	fileMetaData, err := cd.cacheDataManager.readFileMetaData(task.TaskID)
 	if err != nil {
 		return nil, errors.Wrapf(err, "read file meta data of task %s", task.TaskID)
@@ -118,7 +118,7 @@ func (cd *cacheDetector) doDetect(task *types.SeedTask, fileMd5 hash.Hash) (resu
 	if !supportRange {
 		return nil, cdnerrors.ErrResourceNotSupportRangeRequest{URL: task.URL}
 	}
-	return cd.parseByReadFile(task.TaskID, fileMetaData, fileMd5)
+	return cd.parseByReadFile(task.TaskID, fileMetaData, fileDigest)
 }
 
 // parseByReadMetaFile detect cache by read meta and pieceMeta files of task
@@ -154,7 +154,7 @@ func (cd *cacheDetector) parseByReadMetaFile(taskID string, fileMetaData *storag
 }
 
 // parseByReadFile detect cache by read pieceMeta and data files of task
-func (cd *cacheDetector) parseByReadFile(taskID string, metaData *storage.FileMetaData, fileMd5 hash.Hash) (*cacheResult, error) {
+func (cd *cacheDetector) parseByReadFile(taskID string, metaData *storage.FileMetaData, fileDigest hash.Hash) (*cacheResult, error) {
 	reader, err := cd.cacheDataManager.readDownloadFile(taskID)
 	if err != nil {
 		return nil, errors.Wrapf(err, "read data file")
@@ -177,7 +177,7 @@ func (cd *cacheDetector) parseByReadFile(taskID string, metaData *storage.FileMe
 			break
 		}
 		// read content
-		if err := checkPieceContent(reader, tempRecords[index], fileMd5); err != nil {
+		if err := checkPieceContent(reader, tempRecords[index], fileDigest); err != nil {
 			logger.WithTaskID(taskID).Errorf("read content of pieceNum %d failed: %v", tempRecords[index].PieceNum, err)
 			break
 		}
@@ -237,19 +237,19 @@ func checkSameFile(task *types.SeedTask, metaData *storage.FileMetaData) error {
 	if metaData.TaskURL != task.TaskURL {
 		return errors.Errorf("meta task taskUrl(%s) is not equals with task taskUrl(%s)", metaData.TaskURL, task.URL)
 	}
-	if !stringutils.IsBlank(metaData.SourceRealMd5) && !stringutils.IsBlank(task.RequestMd5) &&
-		metaData.SourceRealMd5 != task.RequestMd5 {
-		return errors.Errorf("meta task source md5(%s) is not equals with task request md5(%s)",
-			metaData.SourceRealMd5, task.RequestMd5)
+	if !stringutils.IsBlank(metaData.SourceRealDigest) && !stringutils.IsBlank(task.RequestDigest) &&
+		metaData.SourceRealDigest != task.RequestDigest {
+		return errors.Errorf("meta task source digest(%s) is not equals with task request digest(%s)",
+			metaData.SourceRealDigest, task.RequestDigest)
 	}
 	return nil
 }
 
 //checkPieceContent read piece content from reader and check data integrity by pieceMetaRecord
-func checkPieceContent(reader io.Reader, pieceRecord *storage.PieceMetaRecord, fileMd5 hash.Hash) error {
+func checkPieceContent(reader io.Reader, pieceRecord *storage.PieceMetaRecord, fileDigest hash.Hash) error {
 	// TODO Analyze the original data for the slice format to calculate fileMd5
 	pieceMd5 := md5.New()
-	tee := io.TeeReader(io.TeeReader(io.LimitReader(reader, int64(pieceRecord.PieceLen)), pieceMd5), fileMd5)
+	tee := io.TeeReader(io.TeeReader(io.LimitReader(reader, int64(pieceRecord.PieceLen)), pieceMd5), fileDigest)
 	if n, err := io.Copy(ioutil.Discard, tee); n != int64(pieceRecord.PieceLen) || err != nil {
 		return errors.Wrap(err, "read piece content")
 	}
diff --git a/cdnsystem/daemon/cdn/manager.go b/cdnsystem/daemon/cdn/manager.go
index 55f6f006..d3e21687 100644
--- a/cdnsystem/daemon/cdn/manager.go
+++ b/cdnsystem/daemon/cdn/manager.go
@@ -17,10 +17,12 @@
 package cdn
 
 import (
+	"crypto/md5"
 	"time"
 
+	"d7y.io/dragonfly/v2/pkg/util/digestutils"
+
 	"context"
-	"crypto/md5"
 	"fmt"
 
 	"d7y.io/dragonfly/v2/cdnsystem/daemon"
@@ -85,9 +87,16 @@ func (cm *Manager) TriggerCDN(ctx context.Context, task *types.SeedTask) (seedTa
 	// obtain taskId write lock
 	cm.cdnLocker.Lock(task.TaskID, false)
 	defer cm.cdnLocker.UnLock(task.TaskID, false)
+
+	var fileDigest = md5.New()
+	var digestType = digestutils.Md5Hash.String()
+	if !stringutils.IsBlank(task.RequestDigest) {
+		requestDigest := digestutils.Parse(task.RequestDigest)
+		digestType = requestDigest[0]
+		fileDigest = digestutils.CreateHash(digestType)
+	}
 	// first: detect Cache
-	fileMd5 := md5.New()
-	detectResult, err := cm.detector.detectCache(task, fileMd5)
+	detectResult, err := cm.detector.detectCache(task, fileDigest)
 	if err != nil {
 		seedTask.UpdateStatus(types.TaskInfoCdnStatusFailed)
 		return seedTask, errors.Wrapf(err, "failed to detect cache")
@@ -101,7 +110,7 @@ func (cm *Manager) TriggerCDN(ctx context.Context, task *types.SeedTask) (seedTa
 	// full cache
 	if detectResult.breakPoint == -1 {
 		logger.WithTaskID(task.TaskID).Infof("cache full hit on local")
-		seedTask.UpdateTaskInfo(types.TaskInfoCdnStatusSuccess, detectResult.fileMetaData.SourceRealMd5, detectResult.fileMetaData.PieceMd5Sign,
+		seedTask.UpdateTaskInfo(types.TaskInfoCdnStatusSuccess, detectResult.fileMetaData.SourceRealDigest, detectResult.fileMetaData.PieceMd5Sign,
 			detectResult.fileMetaData.SourceFileLen, detectResult.fileMetaData.CdnFileLength)
 		return seedTask, nil
 	}
@@ -116,8 +125,8 @@ func (cm *Manager) TriggerCDN(ctx context.Context, task *types.SeedTask) (seedTa
 		return seedTask, err
 	}
 	defer body.Close()
+	reader := limitreader.NewLimitReaderWithLimiterAndDigest(body, cm.limiter, fileDigest, digestutils.Algorithms[digestType])
 
-	reader := limitreader.NewLimitReaderWithLimiterAndMD5Sum(body, cm.limiter, fileMd5)
 	// forth: write to storage
 	downloadMetadata, err := cm.writer.startWriter(reader, task, detectResult)
 	if err != nil {
@@ -129,14 +138,14 @@ func (cm *Manager) TriggerCDN(ctx context.Context, task *types.SeedTask) (seedTa
 	}
 	server.StatSeedFinish(task.TaskID, task.URL, true, nil, start.Nanosecond(), time.Now().Nanosecond(), downloadMetadata.backSourceLength,
 		downloadMetadata.realSourceFileLength)
-	sourceMD5 := reader.Md5()
+	sourceDigest := reader.Digest()
 	// fifth: handle CDN result
-	success, err := cm.handleCDNResult(task, sourceMD5, downloadMetadata)
+	success, err := cm.handleCDNResult(task, sourceDigest, downloadMetadata)
 	if err != nil || !success {
 		seedTask.UpdateStatus(types.TaskInfoCdnStatusFailed)
 		return seedTask, err
 	}
-	seedTask.UpdateTaskInfo(types.TaskInfoCdnStatusSuccess, sourceMD5, downloadMetadata.pieceMd5Sign,
+	seedTask.UpdateTaskInfo(types.TaskInfoCdnStatusSuccess, sourceDigest, downloadMetadata.pieceMd5Sign,
 		downloadMetadata.realSourceFileLength, downloadMetadata.realCdnFileLength)
 	return seedTask, nil
 }
@@ -149,13 +158,13 @@ func (cm *Manager) Delete(taskID string) error {
 	return nil
 }
 
-func (cm *Manager) handleCDNResult(task *types.SeedTask, sourceMd5 string, downloadMetadata *downloadMetadata) (bool, error) {
+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
 	var errorMsg string
 	// check md5
-	if !stringutils.IsBlank(task.RequestMd5) && task.RequestMd5 != sourceMd5 {
-		errorMsg = fmt.Sprintf("file md5 not match expected: %s real: %s", task.RequestMd5, sourceMd5)
+	if !stringutils.IsBlank(task.RequestDigest) && task.RequestDigest != sourceDigest {
+		errorMsg = fmt.Sprintf("file digest not match expected: %s real: %s", task.RequestDigest, sourceDigest)
 		isSuccess = false
 	}
 	// check source length
@@ -178,13 +187,13 @@ func (cm *Manager) handleCDNResult(task *types.SeedTask, sourceMd5 string, downl
 		cdnFileLength = 0
 	}
 	if err := cm.cacheDataManager.updateStatusAndResult(task.TaskID, &storage.FileMetaData{
-		Finish:          true,
-		Success:         isSuccess,
-		SourceRealMd5:   sourceMd5,
-		PieceMd5Sign:    pieceMd5Sign,
-		CdnFileLength:   cdnFileLength,
-		SourceFileLen:   sourceFileLen,
-		TotalPieceCount: downloadMetadata.pieceTotalCount,
+		Finish:           true,
+		Success:          isSuccess,
+		SourceRealDigest: sourceDigest,
+		PieceMd5Sign:     pieceMd5Sign,
+		CdnFileLength:    cdnFileLength,
+		SourceFileLen:    sourceFileLen,
+		TotalPieceCount:  downloadMetadata.pieceTotalCount,
 	}); err != nil {
 		return false, errors.Wrap(err, "failed to update task status and result")
 	}
@@ -193,7 +202,7 @@ func (cm *Manager) handleCDNResult(task *types.SeedTask, sourceMd5 string, downl
 		return false, errors.New(errorMsg)
 	}
 
-	logger.WithTaskID(task.TaskID).Infof("success to get task, downloadMetadata: %+v realMd5: %s", downloadMetadata, sourceMd5)
+	logger.WithTaskID(task.TaskID).Infof("success to get task, downloadMetadata: %+v realDigest: %s", downloadMetadata, sourceDigest)
 
 	return true, nil
 }
diff --git a/cdnsystem/daemon/cdn/storage/storage_mgr.go b/cdnsystem/daemon/cdn/storage/storage_mgr.go
index bd9024d2..50a1acf8 100644
--- a/cdnsystem/daemon/cdn/storage/storage_mgr.go
+++ b/cdnsystem/daemon/cdn/storage/storage_mgr.go
@@ -75,19 +75,19 @@ type Manager interface {
 
 // FileMetaData
 type FileMetaData struct {
-	TaskID          string            `json:"taskId"`
-	TaskURL         string            `json:"taskUrl"`
-	PieceSize       int32             `json:"pieceSize"`
-	SourceFileLen   int64             `json:"sourceFileLen"`
-	AccessTime      int64             `json:"accessTime"`
-	Interval        int64             `json:"interval"`
-	CdnFileLength   int64             `json:"cdnFileLength"`
-	SourceRealMd5   string            `json:"sourceRealMd5"`
-	PieceMd5Sign    string            `json:"pieceMd5Sign"`
-	ExpireInfo      map[string]string `json:"expireInfo"`
-	Finish          bool              `json:"finish"`
-	Success         bool              `json:"success"`
-	TotalPieceCount int32             `json:"totalPieceCount"`
+	TaskID           string            `json:"taskId"`
+	TaskURL          string            `json:"taskUrl"`
+	PieceSize        int32             `json:"pieceSize"`
+	SourceFileLen    int64             `json:"sourceFileLen"`
+	AccessTime       int64             `json:"accessTime"`
+	Interval         int64             `json:"interval"`
+	CdnFileLength    int64             `json:"cdnFileLength"`
+	SourceRealDigest string            `json:"sourceRealDigest"`
+	PieceMd5Sign     string            `json:"pieceMd5Sign"`
+	ExpireInfo       map[string]string `json:"expireInfo"`
+	Finish           bool              `json:"finish"`
+	Success          bool              `json:"success"`
+	TotalPieceCount  int32             `json:"totalPieceCount"`
 	//PieceMetaDataSign string            `json:"pieceMetaDataSign"`
 }
 
diff --git a/cdnsystem/daemon/task/manager_util.go b/cdnsystem/daemon/task/manager_util.go
index 3caa56f3..374137af 100644
--- a/cdnsystem/daemon/task/manager_util.go
+++ b/cdnsystem/daemon/task/manager_util.go
@@ -62,7 +62,7 @@ func (tm *Manager) addOrUpdateTask(ctx context.Context, request *types.TaskRegis
 	newTask := &types.SeedTask{
 		TaskID:           taskID,
 		Header:           request.Header,
-		RequestMd5:       request.Md5,
+		RequestDigest:    request.Digest,
 		URL:              request.URL,
 		TaskURL:          taskURL,
 		CdnStatus:        types.TaskInfoCdnStatusWaiting,
@@ -153,8 +153,8 @@ func (tm *Manager) updateTask(taskID string, updateTaskInfo *types.SeedTask) (*t
 		task.CdnFileLength = updateTaskInfo.CdnFileLength
 	}
 
-	if !stringutils.IsBlank(updateTaskInfo.SourceRealMd5) {
-		task.SourceRealMd5 = updateTaskInfo.SourceRealMd5
+	if !stringutils.IsBlank(updateTaskInfo.SourceRealDigest) {
+		task.SourceRealDigest = updateTaskInfo.SourceRealDigest
 	}
 
 	if !stringutils.IsBlank(updateTaskInfo.PieceMd5Sign) {
@@ -181,14 +181,14 @@ func isSameTask(task1, task2 *types.SeedTask) bool {
 		return false
 	}
 
-	if !stringutils.IsBlank(task1.RequestMd5) && !stringutils.IsBlank(task2.RequestMd5) {
-		if task1.RequestMd5 != task2.RequestMd5 {
+	if !stringutils.IsBlank(task1.RequestDigest) && !stringutils.IsBlank(task2.RequestDigest) {
+		if task1.RequestDigest != task2.RequestDigest {
 			return false
 		}
 	}
 
-	if !stringutils.IsBlank(task1.RequestMd5) && !stringutils.IsBlank(task2.SourceRealMd5) {
-		return task1.SourceRealMd5 == task2.RequestMd5
+	if !stringutils.IsBlank(task1.RequestDigest) && !stringutils.IsBlank(task2.SourceRealDigest) {
+		return task1.SourceRealDigest == task2.RequestDigest
 	}
 
 	return true
diff --git a/cdnsystem/server/server.go b/cdnsystem/server/server.go
index 7b7011f4..d2855a4f 100644
--- a/cdnsystem/server/server.go
+++ b/cdnsystem/server/server.go
@@ -110,7 +110,6 @@ func New(cfg *config.Config) (*Server, error) {
 		if err := s.register(context.Background()); err != nil {
 			return nil, err
 		}
-		logger.Info("cdn register to manager")
 	}
 
 	return s, nil
@@ -180,17 +179,14 @@ func (s *Server) register(ctx context.Context) error {
 	logger.Infof("update cdn %s to manager successfully", cdn.HostName)
 
 	cdnClusterID := s.config.Manager.CDNClusterID
-	if cdnClusterID != 0 {
-		if _, err := s.managerClient.AddCDNToCDNCluster(ctx, &manager.AddCDNToCDNClusterRequest{
-			CdnId:        cdn.Id,
-			CdnClusterId: cdnClusterID,
-		}); err != nil {
-			logger.Warnf("add cdn %s to cdn cluster %s failed %v", cdn.HostName, cdnClusterID, err)
-			return err
-		}
-		logger.Infof("add cdn %s to cdn cluster %s successfully", cdn.HostName, cdnClusterID)
+	if _, err := s.managerClient.AddCDNToCDNCluster(ctx, &manager.AddCDNToCDNClusterRequest{
+		CdnId:        cdn.Id,
+		CdnClusterId: cdnClusterID,
+	}); err != nil {
+		logger.Warnf("add cdn %s to cdn cluster %s failed %v", cdn.HostName, cdnClusterID, err)
+		return err
 	}
-
+	logger.Infof("add cdn %s to cdn cluster %s successfully", cdn.HostName, cdnClusterID)
 	return nil
 }
 
diff --git a/cdnsystem/server/service/cdn_seed_server.go b/cdnsystem/server/service/cdn_seed_server.go
index d59b3f6a..21c86289 100644
--- a/cdnsystem/server/service/cdn_seed_server.go
+++ b/cdnsystem/server/service/cdn_seed_server.go
@@ -32,6 +32,7 @@ import (
 	"d7y.io/dragonfly/v2/internal/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
+	"d7y.io/dragonfly/v2/pkg/util/digestutils"
 	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
 	"d7y.io/dragonfly/v2/pkg/util/net/urlutils"
 	"d7y.io/dragonfly/v2/pkg/util/stringutils"
@@ -60,7 +61,11 @@ func constructRegisterRequest(req *cdnsystem.SeedRequest) (*types.TaskRegisterRe
 	header := make(map[string]string)
 	if meta != nil {
 		if !stringutils.IsBlank(meta.Digest) {
-			header["md5"] = meta.Digest
+			digest := digestutils.Parse(meta.Digest)
+			if _, ok := digestutils.Algorithms[digest[0]]; !ok {
+				return nil, errors.Errorf("unsupported digest algorithm")
+			}
+			header["digest"] = meta.Digest
 		}
 		if !stringutils.IsBlank(meta.Range) {
 			header["range"] = meta.Range
@@ -72,7 +77,7 @@ func constructRegisterRequest(req *cdnsystem.SeedRequest) (*types.TaskRegisterRe
 	return &types.TaskRegisterRequest{
 		Header: header,
 		URL:    req.Url,
-		Md5:    header["md5"],
+		Digest: header["digest"],
 		TaskID: req.TaskId,
 		Filter: strings.Split(req.UrlMeta.Filter, "&"),
 	}, nil
diff --git a/cdnsystem/types/seed_task_info.go b/cdnsystem/types/seed_task_info.go
index 73b865db..f5509fbc 100644
--- a/cdnsystem/types/seed_task_info.go
+++ b/cdnsystem/types/seed_task_info.go
@@ -26,8 +26,8 @@ type SeedTask struct {
 	Header           map[string]string `json:"header,omitempty"`
 	CdnStatus        string            `json:"cdnStatus,omitempty"`
 	PieceTotal       int32             `json:"pieceTotal,omitempty"`
-	RequestMd5       string            `json:"requestMd5,omitempty"`
-	SourceRealMd5    string            `json:"sourceRealMd5,omitempty"`
+	RequestDigest    string            `json:"requestDigest,omitempty"`
+	SourceRealDigest string            `json:"sourceRealDigest,omitempty"`
 	PieceMd5Sign     string            `json:"pieceMd5Sign,omitempty"`
 }
 
@@ -59,10 +59,10 @@ func (task *SeedTask) UpdateStatus(cdnStatus string) {
 	task.CdnStatus = cdnStatus
 }
 
-func (task *SeedTask) UpdateTaskInfo(cdnStatus, realMD5, pieceMd5Sign string, sourceFileLength, cdnFileLength int64) {
+func (task *SeedTask) UpdateTaskInfo(cdnStatus, realDigest, pieceMd5Sign string, sourceFileLength, cdnFileLength int64) {
 	task.CdnStatus = cdnStatus
 	task.PieceMd5Sign = pieceMd5Sign
-	task.SourceRealMd5 = realMD5
+	task.SourceRealDigest = realDigest
 	task.SourceFileLength = sourceFileLength
 	task.CdnFileLength = cdnFileLength
 }
diff --git a/cdnsystem/types/task_register_request.go b/cdnsystem/types/task_register_request.go
index 5cdd31c2..970549f7 100644
--- a/cdnsystem/types/task_register_request.go
+++ b/cdnsystem/types/task_register_request.go
@@ -20,7 +20,7 @@ package types
 type TaskRegisterRequest struct {
 	URL    string            `json:"rawURL,omitempty"`
 	TaskID string            `json:"taskId,omitempty"`
-	Md5    string            `json:"md5,omitempty"`
+	Digest string            `json:"digest,omitempty"`
 	Filter []string          `json:"filter,omitempty"`
 	Header map[string]string `json:"header,omitempty"`
 }
diff --git a/client/config/peerhost.go b/client/config/peerhost.go
index 343f75f4..49fbf447 100644
--- a/client/config/peerhost.go
+++ b/client/config/peerhost.go
@@ -30,11 +30,12 @@ import (
 	"strings"
 	"time"
 
+	"github.com/pkg/errors"
+	"gopkg.in/yaml.v3"
+
 	"d7y.io/dragonfly/v2/cmd/dependency/base"
 	"d7y.io/dragonfly/v2/pkg/unit"
 	"d7y.io/dragonfly/v2/pkg/util/stringutils"
-	"github.com/pkg/errors"
-	"gopkg.in/yaml.v3"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
 	"d7y.io/dragonfly/v2/pkg/basic/dfnet"
@@ -120,6 +121,9 @@ type SchedulerOption struct {
 
 	// ScheduleTimeout is request timeout.
 	ScheduleTimeout clientutil.Duration `mapstructure:"scheduleTimeout" yaml:"scheduleTimeout"`
+
+	// DisableAutoBackSource indicates not back source normally, only scheduler says back source
+	DisableAutoBackSource bool `mapstructure:"disableAutoBackSource" yaml:"disableAutoBackSource"`
 }
 
 type HostOption struct {
diff --git a/client/daemon/peer/peertask_base.go b/client/daemon/peer/peertask_base.go
index 2376aae0..03f37a0a 100644
--- a/client/daemon/peer/peertask_base.go
+++ b/client/daemon/peer/peertask_base.go
@@ -60,9 +60,14 @@ type peerTask struct {
 	ctx    context.Context
 	cancel context.CancelFunc
 
-	// backSource indicates downloading resource from instead of other peers
-	backSource bool
-	request    *scheduler.PeerTaskRequest
+	// needBackSource indicates downloading resource from instead of other peers
+	needBackSource bool
+
+	backSourceFunc        func()
+	reportPieceResultFunc func(pieceTask *base.PieceInfo, pieceResult *scheduler.PieceResult) error
+	setContentLengthFunc  func(i int64) error
+
+	request *scheduler.PeerTaskRequest
 
 	// pieceManager will be used for downloading piece
 	pieceManager PieceManager
@@ -73,12 +78,13 @@ type peerTask struct {
 
 	// schedule options
 	schedulerOption config.SchedulerOption
+	schedulerClient schedulerclient.SchedulerClient
 
 	// peer task meta info
 	peerID          string
 	taskID          string
-	contentLength   int64
 	totalPiece      int32
+	contentLength   *atomic.Int64
 	completedLength *atomic.Int64
 	usedTraffic     *atomic.Int64
 
@@ -88,11 +94,11 @@ type peerTask struct {
 	// TODO peerPacketStream
 	peerPacketStream schedulerclient.PeerPacketStream
 	// peerPacket is the latest available peers from peerPacketCh
-	peerPacket *scheduler.PeerPacket
+	peerPacket atomic.Value // *scheduler.PeerPacket
 	// peerPacketReady will receive a ready signal for peerPacket ready
 	peerPacketReady chan bool
 	// pieceParallelCount stands the piece parallel count from peerPacket
-	pieceParallelCount int32
+	pieceParallelCount *atomic.Int32
 
 	// done channel will be close when peer task is finished
 	done chan struct{}
@@ -125,7 +131,7 @@ type peerTask struct {
 var _ Task = (*peerTask)(nil)
 
 func (pt *peerTask) ReportPieceResult(pieceTask *base.PieceInfo, pieceResult *scheduler.PieceResult) error {
-	panic("implement me")
+	return pt.reportPieceResultFunc(pieceTask, pieceResult)
 }
 
 func (pt *peerTask) SetCallback(callback TaskCallback) {
@@ -141,11 +147,11 @@ func (pt *peerTask) GetTaskID() string {
 }
 
 func (pt *peerTask) GetContentLength() int64 {
-	return pt.contentLength
+	return pt.contentLength.Load()
 }
 
 func (pt *peerTask) SetContentLength(i int64) error {
-	panic("implement me")
+	return pt.setContentLengthFunc(i)
 }
 
 func (pt *peerTask) AddTraffic(n int64) {
@@ -168,13 +174,17 @@ func (pt *peerTask) Log() *logger.SugaredLoggerOnWith {
 	return pt.SugaredLoggerOnWith
 }
 
-func (pt *peerTask) pullPieces(pti Task, cleanUnfinishedFunc func()) {
+func (pt *peerTask) backSource() {
+	pt.backSourceFunc()
+}
+
+func (pt *peerTask) pullPieces(cleanUnfinishedFunc func()) {
 	// when there is a single piece, try to download first
 	if pt.singlePiece != nil {
-		go pt.pullSinglePiece(pti, cleanUnfinishedFunc)
+		go pt.pullSinglePiece(cleanUnfinishedFunc)
 	} else {
 		go pt.receivePeerPacket()
-		go pt.pullPiecesFromPeers(pti, cleanUnfinishedFunc)
+		go pt.pullPiecesFromPeers(cleanUnfinishedFunc)
 	}
 }
 
@@ -212,6 +222,11 @@ loop:
 		if err != nil {
 			pt.failedCode = dfcodes.UnknownError
 			if de, ok := err.(*dferrors.DfError); ok {
+				if de.Code == dfcodes.SchedNeedBackSource {
+					pt.needBackSource = true
+					close(pt.peerPacketReady)
+					return
+				}
 				pt.failedCode = de.Code
 				pt.failedReason = de.Message
 				pt.Errorf("receive peer packet failed: %s", pt.failedReason)
@@ -259,8 +274,8 @@ loop:
 			firstPeerSpan.End()
 		}
 
-		pt.peerPacket = peerPacket
-		pt.pieceParallelCount = pt.peerPacket.ParallelCount
+		pt.peerPacket.Store(peerPacket)
+		pt.pieceParallelCount.Store(peerPacket.ParallelCount)
 		select {
 		case pt.peerPacketReady <- true:
 		case <-pt.ctx.Done():
@@ -299,15 +314,15 @@ func (pt *peerTask) isExitPeerPacketCode(pp *scheduler.PeerPacket) bool {
 	return false
 }
 
-func (pt *peerTask) pullSinglePiece(pti Task, cleanUnfinishedFunc func()) {
+func (pt *peerTask) pullSinglePiece(cleanUnfinishedFunc func()) {
 	pt.Infof("single piece, dest peer id: %s, piece num: %d, size: %d",
 		pt.singlePiece.DstPid, pt.singlePiece.PieceInfo.PieceNum, pt.singlePiece.PieceInfo.RangeSize)
 
 	ctx, span := tracer.Start(pt.ctx, fmt.Sprintf(config.SpanDownloadPiece, pt.singlePiece.PieceInfo.PieceNum))
 	span.SetAttributes(config.AttributePiece.Int(int(pt.singlePiece.PieceInfo.PieceNum)))
 
-	pt.contentLength = int64(pt.singlePiece.PieceInfo.RangeSize)
-	if err := pt.callback.Init(pti); err != nil {
+	pt.contentLength.Store(int64(pt.singlePiece.PieceInfo.RangeSize))
+	if err := pt.callback.Init(pt); err != nil {
 		pt.failedReason = err.Error()
 		pt.failedCode = dfcodes.ClientError
 		cleanUnfinishedFunc()
@@ -323,7 +338,7 @@ func (pt *peerTask) pullSinglePiece(pti Task, cleanUnfinishedFunc func()) {
 		DstAddr: pt.singlePiece.DstAddr,
 		piece:   pt.singlePiece.PieceInfo,
 	}
-	if pt.pieceManager.DownloadPiece(ctx, pti, request) {
+	if pt.pieceManager.DownloadPiece(ctx, pt, request) {
 		pt.Infof("single piece download success")
 		span.SetAttributes(config.AttributePieceSuccess.Bool(true))
 		span.End()
@@ -333,49 +348,33 @@ func (pt *peerTask) pullSinglePiece(pti Task, cleanUnfinishedFunc func()) {
 		span.End()
 		pt.Warnf("single piece download failed, switch to download from other peers")
 		go pt.receivePeerPacket()
-		pt.pullPiecesFromPeers(pti, cleanUnfinishedFunc)
+		pt.pullPiecesFromPeers(cleanUnfinishedFunc)
 	}
 }
 
-// TODO when main peer is not available, switch to steel peers
-// piece manager need peer task interface, pti make it compatibility for stream peer task
-func (pt *peerTask) pullPiecesFromPeers(pti Task, cleanUnfinishedFunc func()) {
+func (pt *peerTask) pullPiecesFromPeers(cleanUnfinishedFunc func()) {
 	defer func() {
 		close(pt.failedPieceCh)
 		cleanUnfinishedFunc()
 	}()
-	// wait first available peer
-	select {
-	case <-pt.ctx.Done():
-		err := pt.ctx.Err()
-		pt.Errorf("context done due to %s", err)
-		if pt.failedReason == failedReasonNotSet && err != nil {
-			pt.failedReason = err.Error()
-		}
-		pt.span.AddEvent(fmt.Sprintf("pulling pieces end due to %s", err))
-		return
-	case <-pt.peerPacketReady:
-		// preparePieceTasksByPeer func already send piece result with error
-		pt.Infof("new peer client ready, scheduler time cost: %dus, main peer: %s",
-			time.Now().Sub(pt.callback.GetStartTime()).Microseconds(), pt.peerPacket.MainPeer)
-	case <-time.After(pt.schedulerOption.ScheduleTimeout.Duration):
-		pt.failedReason = reasonScheduleTimeout
-		pt.failedCode = dfcodes.ClientScheduleTimeout
-		pt.Errorf(pt.failedReason)
+
+	if !pt.waitFirstPeerPacket() {
+		pt.Errorf("wait first peer packet error")
 		return
 	}
 	var (
 		num            int32
+		ok             bool
 		limit          int32
 		initialized    bool
 		pieceRequestCh chan *DownloadPieceRequest
-		// keep same size with pt.failedPieceCh for avoiding dead lock
+		// keep same size with pt.failedPieceCh for avoiding dead-lock
 		pieceBufferSize = int32(config.DefaultPieceChanSize)
 	)
+	limit = pieceBufferSize
 loop:
 	for {
-		limit = pieceBufferSize
-		// check whether catch exit signal or get a failed piece
+		// 1, check whether catch exit signal or get a failed piece
 		// if nothing got, process normal pieces
 		select {
 		case <-pt.done:
@@ -394,12 +393,13 @@ loop:
 			}
 			break loop
 		case failed := <-pt.failedPieceCh:
-			pt.Warnf("download piece/%d failed, retry", failed)
+			pt.Warnf("download piece %d failed, retry", failed)
 			num = failed
 			limit = 1
 		default:
 		}
 
+		// 2, try to get pieces
 		pt.Debugf("try to get pieces, number: %d, limit: %d", num, limit)
 		piecePacket, err := pt.preparePieceTasks(
 			&base.PieceTaskRequest{
@@ -412,50 +412,16 @@ loop:
 		if err != nil {
 			pt.Warnf("get piece task error: %s, wait available peers from scheduler", err)
 			pt.span.RecordError(err)
-			select {
-			// when peer task without content length or total pieces count, match here
-			case <-pt.done:
-				pt.Infof("peer task done, stop get pieces from peer")
-			case <-pt.ctx.Done():
-				pt.Debugf("context done due to %s", pt.ctx.Err())
-				if !pt.peerTaskDone {
-					if pt.failedCode == failedCodeNotSet {
-						pt.failedReason = reasonContextCanceled
-						pt.failedCode = dfcodes.ClientContextCanceled
-					}
-				}
-			case <-pt.peerPacketReady:
-				// preparePieceTasksByPeer func already send piece result with error
-				pt.Infof("new peer client ready, main peer: %s", pt.peerPacket.MainPeer)
-				// research from piece 0
-				num = pt.getNextPieceNum(0)
-				continue loop
-			case <-time.After(pt.schedulerOption.ScheduleTimeout.Duration):
-				pt.failedReason = reasonReScheduleTimeout
-				pt.failedCode = dfcodes.ClientScheduleTimeout
-				pt.Errorf(pt.failedReason)
+			if num, ok = pt.waitAvailablePeerPacket(); !ok {
+				break loop
 			}
-			// only <-pt.peerPacketReady continue loop, others break
-			break loop
 		}
 
 		if !initialized {
-			pt.contentLength = piecePacket.ContentLength
-			if pt.contentLength > 0 {
-				pt.span.SetAttributes(config.AttributeTaskContentLength.Int64(pt.contentLength))
-			}
 			initialized = true
-			if err = pt.callback.Init(pt); err != nil {
-				pt.span.RecordError(err)
-				pt.failedReason = err.Error()
-				pt.failedCode = dfcodes.ClientError
+			if pieceRequestCh, ok = pt.init(piecePacket, pieceBufferSize); !ok {
 				break loop
 			}
-			pc := pt.peerPacket.ParallelCount
-			pieceRequestCh = make(chan *DownloadPieceRequest, pieceBufferSize)
-			for i := int32(0); i < pc; i++ {
-				go pt.downloadPieceWorker(i, pti, pieceRequestCh)
-			}
 		}
 
 		// update total piece
@@ -465,61 +431,168 @@ loop:
 			pt.Debugf("update total piece count: %d", pt.totalPiece)
 		}
 
-		// trigger DownloadPiece
-		for _, piece := range piecePacket.PieceInfos {
-			pt.Infof("get piece %d from %s/%s", piece.PieceNum, piecePacket.DstAddr, piecePacket.DstPid)
-			if !pt.requestedPieces.IsSet(piece.PieceNum) {
-				pt.requestedPieces.Set(piece.PieceNum)
-			}
-			req := &DownloadPieceRequest{
-				TaskID:  pt.GetTaskID(),
-				DstPid:  piecePacket.DstPid,
-				DstAddr: piecePacket.DstAddr,
-				piece:   piece,
+		// 3. dispatch piece request to all workers
+		pt.dispatchPieceRequest(pieceRequestCh, piecePacket)
+
+		// 4. get next piece
+		num = pt.getNextPieceNum(num)
+		if num != -1 {
+			// get next piece success
+			limit = pieceBufferSize
+			continue
+		}
+		pt.Infof("all pieces requests send, just wait failed pieces")
+		// just need one piece
+		limit = 1
+		// get failed piece
+		if num, ok = pt.waitFailedPiece(); !ok {
+			break loop
+		}
+	}
+}
+
+func (pt *peerTask) init(piecePacket *base.PiecePacket, pieceBufferSize int32) (chan *DownloadPieceRequest, bool) {
+	pt.contentLength.Store(piecePacket.ContentLength)
+	if pt.contentLength.Load() > 0 {
+		pt.span.SetAttributes(config.AttributeTaskContentLength.Int64(pt.contentLength.Load()))
+	}
+	if err := pt.callback.Init(pt); err != nil {
+		pt.span.RecordError(err)
+		pt.failedReason = err.Error()
+		pt.failedCode = dfcodes.ClientError
+		return nil, false
+	}
+	pc := pt.peerPacket.Load().(*scheduler.PeerPacket).ParallelCount
+	pieceRequestCh := make(chan *DownloadPieceRequest, pieceBufferSize)
+	for i := int32(0); i < pc; i++ {
+		go pt.downloadPieceWorker(i, pt, pieceRequestCh)
+	}
+	return pieceRequestCh, true
+}
+
+func (pt *peerTask) waitFirstPeerPacket() bool {
+	// wait first available peer
+	select {
+	case <-pt.ctx.Done():
+		err := pt.ctx.Err()
+		pt.Errorf("context done due to %s", err)
+		if pt.failedReason == failedReasonNotSet && err != nil {
+			pt.failedReason = err.Error()
+		}
+		pt.span.AddEvent(fmt.Sprintf("pulling pieces end due to %s", err))
+	case <-pt.peerPacketReady:
+		// preparePieceTasksByPeer func already send piece result with error
+		pt.Infof("new peer client ready, scheduler time cost: %dus, main peer: %s",
+			time.Now().Sub(pt.callback.GetStartTime()).Microseconds(), pt.peerPacket.Load().(*scheduler.PeerPacket).MainPeer)
+		return true
+	case <-time.After(pt.schedulerOption.ScheduleTimeout.Duration):
+		if pt.schedulerOption.DisableAutoBackSource {
+			pt.failedReason = reasonScheduleTimeout
+			pt.failedCode = dfcodes.ClientScheduleTimeout
+			logger.Errorf("%s, auto back source disabled", pt.failedReason)
+		}
+		pt.Errorf("start download from source due to %s", reasonScheduleTimeout)
+		pt.needBackSource = true
+		pt.backSource()
+	}
+	return false
+}
+
+func (pt *peerTask) waitAvailablePeerPacket() (int32, bool) {
+	// only <-pt.peerPacketReady continue loop, others break
+	select {
+	// when peer task without content length or total pieces count, match here
+	case <-pt.done:
+		pt.Infof("peer task done, stop get pieces from peer")
+	case <-pt.ctx.Done():
+		pt.Debugf("context done due to %s", pt.ctx.Err())
+		if !pt.peerTaskDone {
+			if pt.failedCode == failedCodeNotSet {
+				pt.failedReason = reasonContextCanceled
+				pt.failedCode = dfcodes.ClientContextCanceled
 			}
-			select {
-			case pieceRequestCh <- req:
-			case <-pt.done:
-				pt.Warnf("peer task done, but still some piece request not process")
-			case <-pt.ctx.Done():
-				pt.Warnf("context done due to %s", pt.ctx.Err())
-				if !pt.peerTaskDone {
-					if pt.failedCode == failedCodeNotSet {
-						pt.failedReason = reasonContextCanceled
-						pt.failedCode = dfcodes.ClientContextCanceled
-					}
+		}
+	case _, ok := <-pt.peerPacketReady:
+		if ok {
+			// 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
+		}
+		// when schedule timeout, receivePeerPacket will close pt.peerPacketReady
+		if pt.schedulerOption.DisableAutoBackSource {
+			pt.failedReason = reasonReScheduleTimeout
+			pt.failedCode = dfcodes.ClientScheduleTimeout
+			logger.Errorf("%s, auto back source disabled", pt.failedReason)
+		} else {
+			pt.Errorf("start download from source due to dfcodes.SchedNeedBackSource")
+			pt.needBackSource = true
+			pt.backSource()
+		}
+	case <-time.After(pt.schedulerOption.ScheduleTimeout.Duration):
+		if pt.schedulerOption.DisableAutoBackSource {
+			pt.failedReason = reasonReScheduleTimeout
+			pt.failedCode = dfcodes.ClientScheduleTimeout
+			logger.Errorf("%s, auto back source disabled", pt.failedReason)
+		}
+		pt.Errorf("start download from source due to %s", reasonReScheduleTimeout)
+		pt.needBackSource = true
+		pt.backSource()
+	}
+	return -1, false
+}
+
+func (pt *peerTask) dispatchPieceRequest(pieceRequestCh chan *DownloadPieceRequest, piecePacket *base.PiecePacket) {
+	for _, piece := range piecePacket.PieceInfos {
+		pt.Infof("get piece %d from %s/%s", piece.PieceNum, piecePacket.DstAddr, piecePacket.DstPid)
+		if !pt.requestedPieces.IsSet(piece.PieceNum) {
+			pt.requestedPieces.Set(piece.PieceNum)
+		}
+		req := &DownloadPieceRequest{
+			TaskID:  pt.GetTaskID(),
+			DstPid:  piecePacket.DstPid,
+			DstAddr: piecePacket.DstAddr,
+			piece:   piece,
+		}
+		select {
+		case pieceRequestCh <- req:
+		case <-pt.done:
+			pt.Warnf("peer task done, but still some piece request not process")
+		case <-pt.ctx.Done():
+			pt.Warnf("context done due to %s", pt.ctx.Err())
+			if !pt.peerTaskDone {
+				if pt.failedCode == failedCodeNotSet {
+					pt.failedReason = reasonContextCanceled
+					pt.failedCode = dfcodes.ClientContextCanceled
 				}
 			}
 		}
+	}
+}
 
-		num = pt.getNextPieceNum(num)
-		if num == -1 {
-			pt.Infof("all pieces requests send, just wait failed pieces")
-			if pt.isCompleted() {
-				break loop
-			}
-			// use no default branch select to wait failed piece or exit
-			select {
-			case <-pt.done:
-				pt.Infof("peer task done, stop get pieces from peer")
-				break loop
-			case <-pt.ctx.Done():
-				if !pt.peerTaskDone {
-					if pt.failedCode == failedCodeNotSet {
-						pt.failedReason = reasonContextCanceled
-						pt.failedCode = dfcodes.ClientContextCanceled
-					}
-					pt.Errorf("context done due to %s, progress is not done", pt.ctx.Err())
-				} else {
-					pt.Debugf("context done due to %s, progress is already done", pt.ctx.Err())
-				}
-				break loop
-			case failed := <-pt.failedPieceCh:
-				pt.Warnf("download piece/%d failed, retry", failed)
-				num = failed
-				limit = 1
+func (pt *peerTask) waitFailedPiece() (int32, bool) {
+	if pt.isCompleted() {
+		return -1, false
+	}
+	// use no default branch select to wait failed piece or exit
+	select {
+	case <-pt.done:
+		pt.Infof("peer task done, stop get pieces from peer")
+		return -1, false
+	case <-pt.ctx.Done():
+		if !pt.peerTaskDone {
+			if pt.failedCode == failedCodeNotSet {
+				pt.failedReason = reasonContextCanceled
+				pt.failedCode = dfcodes.ClientContextCanceled
 			}
+			pt.Errorf("context done due to %s, progress is not done", pt.ctx.Err())
+		} else {
+			pt.Debugf("context done due to %s, progress is already done", pt.ctx.Err())
 		}
+		return -1, false
+	case failed := <-pt.failedPieceCh:
+		pt.Warnf("download piece/%d failed, retry", failed)
+		return -1, false
 	}
 }
 
@@ -581,24 +654,25 @@ func (pt *peerTask) downloadPieceWorker(id int32, pti Task, requests chan *Downl
 }
 
 func (pt *peerTask) isCompleted() bool {
-	return pt.completedLength.Load() == pt.contentLength
+	return pt.completedLength.Load() == pt.contentLength.Load()
 }
 
 func (pt *peerTask) preparePieceTasks(request *base.PieceTaskRequest) (p *base.PiecePacket, err error) {
 	defer pt.recoverFromPanic()
 prepare:
-	pt.pieceParallelCount = pt.peerPacket.ParallelCount
-	request.DstPid = pt.peerPacket.MainPeer.PeerId
-	p, err = pt.preparePieceTasksByPeer(pt.peerPacket, pt.peerPacket.MainPeer, request)
+	peerPacket := pt.peerPacket.Load().(*scheduler.PeerPacket)
+	pt.pieceParallelCount.Store(peerPacket.ParallelCount)
+	request.DstPid = peerPacket.MainPeer.PeerId
+	p, err = pt.preparePieceTasksByPeer(peerPacket, peerPacket.MainPeer, request)
 	if err == nil {
 		return
 	}
 	if err == errPeerPacketChanged {
 		goto prepare
 	}
-	for _, peer := range pt.peerPacket.StealPeers {
+	for _, peer := range peerPacket.StealPeers {
 		request.DstPid = peer.PeerId
-		p, err = pt.preparePieceTasksByPeer(pt.peerPacket, peer, request)
+		p, err = pt.preparePieceTasksByPeer(peerPacket, peer, request)
 		if err == nil {
 			return
 		}
@@ -666,7 +740,7 @@ retry:
 		pt.Errorf("send piece result error: %s, code: %d", err, code)
 	}
 
-	if code == dfcodes.CdnTaskNotFound && curPeerPacket == pt.peerPacket {
+	if code == dfcodes.CdnTaskNotFound && curPeerPacket == pt.peerPacket.Load().(*scheduler.PeerPacket) {
 		span.AddEvent("retry for CdnTaskNotFound")
 		goto retry
 	}
@@ -684,9 +758,10 @@ func (pt *peerTask) getPieceTasks(span trace.Span, curPeerPacket *scheduler.Peer
 		if getErr != nil {
 			span.RecordError(getErr)
 			// fast way to exit retry
-			if curPeerPacket.MainPeer.PeerId != pt.peerPacket.MainPeer.PeerId {
+			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, pt.peerPacket.MainPeer.PeerId)
+					curPeerPacket.MainPeer.PeerId, lastPeerPacket.MainPeer.PeerId)
 				peerPacketChanged = true
 				return nil, true, nil
 			}
@@ -709,9 +784,10 @@ func (pt *peerTask) getPieceTasks(span trace.Span, curPeerPacket *scheduler.Peer
 				pt.Errorf("send piece result error: %s, code: %d", peer.PeerId, er)
 			}
 			// fast way to exit retry
-			if curPeerPacket.MainPeer.PeerId != pt.peerPacket.MainPeer.PeerId {
+			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, pt.peerPacket.MainPeer.PeerId)
+					curPeerPacket.MainPeer.PeerId, lastPeerPacket.MainPeer.PeerId)
 				peerPacketChanged = true
 				return nil, true, nil
 			}
diff --git a/client/daemon/peer/peertask_dummy.go b/client/daemon/peer/peertask_dummy.go
new file mode 100644
index 00000000..eccb20bd
--- /dev/null
+++ b/client/daemon/peer/peertask_dummy.go
@@ -0,0 +1,61 @@
+/*
+ *     Copyright 2020 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package peer
+
+import (
+	"context"
+
+	"d7y.io/dragonfly/v2/internal/dfcodes"
+	"d7y.io/dragonfly/v2/internal/dferrors"
+	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
+	schedulerclient "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
+	"google.golang.org/grpc"
+)
+
+type dummySchedulerClient struct {
+}
+
+func (d *dummySchedulerClient) RegisterPeerTask(ctx context.Context, request *scheduler.PeerTaskRequest, option ...grpc.CallOption) (*scheduler.RegisterResult, error) {
+	panic("should not call this function")
+}
+
+func (d *dummySchedulerClient) ReportPieceResult(ctx context.Context, s string, request *scheduler.PeerTaskRequest, option ...grpc.CallOption) (schedulerclient.PeerPacketStream, error) {
+	return &dummyPeerPacketStream{}, nil
+}
+
+func (d *dummySchedulerClient) ReportPeerResult(ctx context.Context, result *scheduler.PeerResult, option ...grpc.CallOption) error {
+	return nil
+}
+
+func (d *dummySchedulerClient) LeaveTask(ctx context.Context, target *scheduler.PeerTarget, option ...grpc.CallOption) error {
+	return nil
+}
+
+func (d *dummySchedulerClient) Close() error {
+	return nil
+}
+
+type dummyPeerPacketStream struct {
+}
+
+func (d *dummyPeerPacketStream) Recv() (pp *scheduler.PeerPacket, err error) {
+	return nil, dferrors.New(dfcodes.SchedNeedBackSource, "")
+}
+
+func (d *dummyPeerPacketStream) Send(pr *scheduler.PieceResult) (err error) {
+	return nil
+}
diff --git a/client/daemon/peer/peertask_file.go b/client/daemon/peer/peertask_file.go
index 7d625656..d63d36e6 100644
--- a/client/daemon/peer/peertask_file.go
+++ b/client/daemon/peer/peertask_file.go
@@ -20,7 +20,6 @@ import (
 	"context"
 	"sync"
 
-	"d7y.io/dragonfly/v2/internal/dferrors"
 	"github.com/pkg/errors"
 	"go.opentelemetry.io/otel/semconv"
 	"go.opentelemetry.io/otel/trace"
@@ -78,7 +77,7 @@ func newFilePeerTask(ctx context.Context,
 	request *scheduler.PeerTaskRequest,
 	schedulerClient schedulerclient.SchedulerClient,
 	schedulerOption config.SchedulerOption,
-	perPeerRateLimit rate.Limit) (context.Context, FilePeerTask, *TinyData, error) {
+	perPeerRateLimit rate.Limit) (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))
@@ -93,19 +92,19 @@ func newFilePeerTask(ctx context.Context,
 	regSpan.RecordError(err)
 	regSpan.End()
 
-	var backSource bool
+	var needBackSource bool
 	if err != nil {
 		logger.Errorf("step 1: peer %s register failed: err", request.PeerId, err)
-		// check if it is back source error
-		if de, ok := err.(*dferrors.DfError); ok && de.Code == dfcodes.SchedNeedBackSource {
-			backSource = true
-		}
-		// not back source
-		if !backSource {
+		if schedulerOption.DisableAutoBackSource {
+			logger.Errorf("register peer task failed: %s, peer id: %s, auto back source disabled", err, request.PeerId)
 			span.RecordError(err)
 			span.End()
 			return ctx, nil, nil, err
 		}
+		needBackSource = true
+		// can not detect source or scheduler error, create a new dummy scheduler client
+		schedulerClient = &dummySchedulerClient{}
+		logger.Warnf("register peer task failed: %s, peer id: %s, try to back source", err, request.PeerId)
 	}
 	if result == nil {
 		defer span.End()
@@ -118,7 +117,7 @@ func newFilePeerTask(ctx context.Context,
 		result.TaskId, request.PeerId, base.SizeScope_name[int32(result.SizeScope)])
 
 	var singlePiece *scheduler.SinglePiece
-	if !backSource {
+	if !needBackSource {
 		switch result.SizeScope {
 		case base.SizeScope_SMALL:
 			span.SetAttributes(config.AttributePeerTaskSizeScope.String("small"))
@@ -160,12 +159,12 @@ func newFilePeerTask(ctx context.Context,
 	if perPeerRateLimit > 0 {
 		limiter = rate.NewLimiter(perPeerRateLimit, int(perPeerRateLimit))
 	}
-	return ctx, &filePeerTask{
+	pt := &filePeerTask{
 		progressCh:     make(chan *FilePeerTaskProgress),
 		progressStopCh: make(chan bool),
 		peerTask: peerTask{
 			host:                host,
-			backSource:          backSource,
+			needBackSource:      needBackSource,
 			request:             request,
 			peerPacketStream:    peerPacketStream,
 			pieceManager:        pieceManager,
@@ -181,36 +180,35 @@ func newFilePeerTask(ctx context.Context,
 			failedPieceCh:       make(chan int32, config.DefaultPieceChanSize),
 			failedReason:        failedReasonNotSet,
 			failedCode:          dfcodes.UnknownError,
-			contentLength:       -1,
+			contentLength:       atomic.NewInt64(-1),
+			pieceParallelCount:  atomic.NewInt32(0),
 			totalPiece:          -1,
 			schedulerOption:     schedulerOption,
+			schedulerClient:     schedulerClient,
 			limiter:             limiter,
 			completedLength:     atomic.NewInt64(0),
 			usedTraffic:         atomic.NewInt64(0),
 			SugaredLoggerOnWith: logger.With("peer", request.PeerId, "task", result.TaskId, "component", "filePeerTask"),
 		},
-	}, nil, nil
+	}
+	// bind func that base peer task did not implement
+	pt.backSourceFunc = pt.backSource
+	pt.setContentLengthFunc = pt.SetContentLength
+	pt.reportPieceResultFunc = pt.ReportPieceResult
+	return ctx, pt, nil, nil
 }
 
 func (pt *filePeerTask) Start(ctx context.Context) (chan *FilePeerTaskProgress, error) {
 	pt.ctx, pt.cancel = context.WithCancel(ctx)
-	if pt.backSource {
-		pt.contentLength = -1
+
+	if pt.needBackSource {
+		pt.contentLength.Store(-1)
 		_ = pt.callback.Init(pt)
-		go func() {
-			defer pt.cleanUnfinished()
-			err := pt.pieceManager.DownloadSource(ctx, pt, pt.request)
-			if err != nil {
-				pt.Errorf("download from source error: %s", err)
-				return
-			}
-			pt.Infof("download from source ok")
-			pt.finish()
-		}()
+		go pt.backSource()
 		return pt.progressCh, nil
 	}
 
-	pt.pullPieces(pt, pt.cleanUnfinished)
+	pt.pullPieces(pt.cleanUnfinished)
 
 	// return a progress channel for request download progress
 	return pt.progressCh, nil
@@ -252,7 +250,7 @@ func (pt *filePeerTask) ReportPieceResult(piece *base.PieceInfo, pieceResult *sc
 		},
 		TaskID:          pt.taskID,
 		PeerID:          pt.peerID,
-		ContentLength:   pt.contentLength,
+		ContentLength:   pt.contentLength.Load(),
 		CompletedLength: pt.completedLength.Load(),
 		PeerTaskDone:    false,
 	}
@@ -306,7 +304,7 @@ func (pt *filePeerTask) finish() error {
 			},
 			TaskID:          pt.taskID,
 			PeerID:          pt.peerID,
-			ContentLength:   pt.contentLength,
+			ContentLength:   pt.contentLength.Load(),
 			CompletedLength: pt.completedLength.Load(),
 			PeerTaskDone:    true,
 			DoneCallback: func() {
@@ -361,7 +359,7 @@ func (pt *filePeerTask) cleanUnfinished() {
 			},
 			TaskID:          pt.taskID,
 			PeerID:          pt.peerID,
-			ContentLength:   pt.contentLength,
+			ContentLength:   pt.contentLength.Load(),
 			CompletedLength: pt.completedLength.Load(),
 			PeerTaskDone:    true,
 			DoneCallback: func() {
@@ -406,10 +404,23 @@ func (pt *filePeerTask) cleanUnfinished() {
 }
 
 func (pt *filePeerTask) SetContentLength(i int64) error {
-	pt.contentLength = i
+	pt.contentLength.Store(i)
 	if !pt.isCompleted() {
 		return errors.New("SetContentLength should call after task completed")
 	}
 
 	return pt.finish()
 }
+
+func (pt *filePeerTask) backSource() {
+	defer pt.cleanUnfinished()
+	err := pt.pieceManager.DownloadSource(pt.ctx, pt, pt.request)
+	if err != nil {
+		pt.Errorf("download from source error: %s", err)
+		pt.failedReason = err.Error()
+		return
+	}
+	pt.Infof("download from source ok")
+	_ = pt.finish()
+	return
+}
diff --git a/client/daemon/peer/peertask_file_callback.go b/client/daemon/peer/peertask_file_callback.go
index ffadc4f6..f4382aa1 100644
--- a/client/daemon/peer/peertask_file_callback.go
+++ b/client/daemon/peer/peertask_file_callback.go
@@ -29,6 +29,7 @@ import (
 type filePeerTaskCallback struct {
 	ctx   context.Context
 	ptm   *peerTaskManager
+	pt    *filePeerTask
 	req   *FilePeerTaskRequest
 	start time.Time
 }
@@ -92,7 +93,7 @@ func (p *filePeerTaskCallback) Done(pt Task) error {
 		return e
 	}
 	p.ptm.PeerTaskDone(p.req.PeerId)
-	err := p.ptm.schedulerClient.ReportPeerResult(context.Background(), &scheduler.PeerResult{
+	err := p.pt.schedulerClient.ReportPeerResult(context.Background(), &scheduler.PeerResult{
 		TaskId:         pt.GetTaskID(),
 		PeerId:         pt.GetPeerID(),
 		SrcIp:          p.ptm.host.Ip,
@@ -117,7 +118,7 @@ func (p *filePeerTaskCallback) Fail(pt Task, code base.Code, reason string) erro
 	p.ptm.PeerTaskDone(p.req.PeerId)
 	var end = time.Now()
 	pt.Log().Errorf("file peer task failed, code: %d, reason: %s", code, reason)
-	err := p.ptm.schedulerClient.ReportPeerResult(context.Background(), &scheduler.PeerResult{
+	err := p.pt.schedulerClient.ReportPeerResult(context.Background(), &scheduler.PeerResult{
 		TaskId:         pt.GetTaskID(),
 		PeerId:         pt.GetPeerID(),
 		SrcIp:          p.ptm.host.Ip,
diff --git a/client/daemon/peer/peertask_manager.go b/client/daemon/peer/peertask_manager.go
index 919c6fa3..7dafae93 100644
--- a/client/daemon/peer/peertask_manager.go
+++ b/client/daemon/peer/peertask_manager.go
@@ -26,12 +26,12 @@ import (
 	"sync"
 	"time"
 
-	"d7y.io/dragonfly/v2/client/clientutil"
 	"github.com/go-http-utils/headers"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/time/rate"
 
+	"d7y.io/dragonfly/v2/client/clientutil"
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/client/daemon/storage"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
@@ -67,7 +67,7 @@ type Task interface {
 	GetTaskID() string
 	GetTotalPieces() int32
 	GetContentLength() int64
-	// SetContentLength will called after download completed, when download from source without content length
+	// SetContentLength will be called after download completed, when download from source without content length
 	SetContentLength(int64) error
 	SetCallback(TaskCallback)
 	AddTraffic(int64)
@@ -186,6 +186,7 @@ func (ptm *peerTaskManager) StartFilePeerTask(ctx context.Context, req *FilePeer
 	pt.SetCallback(&filePeerTaskCallback{
 		ctx:   ctx,
 		ptm:   ptm,
+		pt:    pt,
 		req:   req,
 		start: start,
 	})
@@ -221,12 +222,14 @@ func (ptm *peerTaskManager) StartStreamPeerTask(ctx context.Context, req *schedu
 		}, nil
 	}
 
-	pt.SetCallback(&streamPeerTaskCallback{
-		ctx:   ctx,
-		ptm:   ptm,
-		req:   req,
-		start: start,
-	})
+	pt.SetCallback(
+		&streamPeerTaskCallback{
+			ctx:   ctx,
+			ptm:   ptm,
+			pt:    pt,
+			req:   req,
+			start: start,
+		})
 
 	ptm.runningPeerTasks.Store(req.PeerId, pt)
 
diff --git a/client/daemon/peer/peertask_stream.go b/client/daemon/peer/peertask_stream.go
index 359fe0dd..69e3b5aa 100644
--- a/client/daemon/peer/peertask_stream.go
+++ b/client/daemon/peer/peertask_stream.go
@@ -21,7 +21,6 @@ import (
 	"fmt"
 	"io"
 
-	"d7y.io/dragonfly/v2/internal/dferrors"
 	"github.com/go-http-utils/headers"
 	"github.com/pkg/errors"
 	"go.opentelemetry.io/otel/semconv"
@@ -60,7 +59,7 @@ func newStreamPeerTask(ctx context.Context,
 	request *scheduler.PeerTaskRequest,
 	schedulerClient schedulerclient.SchedulerClient,
 	schedulerOption config.SchedulerOption,
-	perPeerRateLimit rate.Limit) (context.Context, StreamPeerTask, *TinyData, error) {
+	perPeerRateLimit rate.Limit) (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))
@@ -76,17 +75,19 @@ func newStreamPeerTask(ctx context.Context,
 	regSpan.RecordError(err)
 	regSpan.End()
 
-	var backSource bool
+	var needBackSource bool
 	if err != nil {
-		// check if it is back source error
-		if de, ok := err.(*dferrors.DfError); ok && de.Code == dfcodes.SchedNeedBackSource {
-			backSource = true
-		}
-		// not back source
-		if !backSource {
-			logger.Errorf("register peer task failed: %s, peer id: %s", err, request.PeerId)
+		logger.Errorf("step 1: peer %s register failed: err", request.PeerId, err)
+		if schedulerOption.DisableAutoBackSource {
+			logger.Errorf("register peer task failed: %s, peer id: %s, auto back source disabled", err, request.PeerId)
+			span.RecordError(err)
+			span.End()
 			return ctx, nil, nil, err
 		}
+		needBackSource = true
+		// can not detect source or scheduler error, create a new dummy scheduler client
+		schedulerClient = &dummySchedulerClient{}
+		logger.Warnf("register peer task failed: %s, peer id: %s, try to back source", err, request.PeerId)
 	}
 	if result == nil {
 		defer span.End()
@@ -99,7 +100,7 @@ func newStreamPeerTask(ctx context.Context,
 		result.TaskId, request.PeerId, base.SizeScope_name[int32(result.SizeScope)])
 
 	var singlePiece *scheduler.SinglePiece
-	if !backSource {
+	if !needBackSource {
 		switch result.SizeScope {
 		case base.SizeScope_SMALL:
 			span.SetAttributes(config.AttributePeerTaskSizeScope.String("small"))
@@ -139,11 +140,11 @@ func newStreamPeerTask(ctx context.Context,
 	if perPeerRateLimit > 0 {
 		limiter = rate.NewLimiter(perPeerRateLimit, int(perPeerRateLimit))
 	}
-	return ctx, &streamPeerTask{
+	pt := &streamPeerTask{
 		peerTask: peerTask{
 			ctx:                 ctx,
 			host:                host,
-			backSource:          backSource,
+			needBackSource:      needBackSource,
 			request:             request,
 			peerPacketStream:    peerPacketStream,
 			pieceManager:        pieceManager,
@@ -158,16 +159,23 @@ func newStreamPeerTask(ctx context.Context,
 			failedPieceCh:       make(chan int32, config.DefaultPieceChanSize),
 			failedReason:        failedReasonNotSet,
 			failedCode:          dfcodes.UnknownError,
-			contentLength:       -1,
+			contentLength:       atomic.NewInt64(-1),
+			pieceParallelCount:  atomic.NewInt32(0),
 			totalPiece:          -1,
 			schedulerOption:     schedulerOption,
+			schedulerClient:     schedulerClient,
 			limiter:             limiter,
 			completedLength:     atomic.NewInt64(0),
 			usedTraffic:         atomic.NewInt64(0),
 			SugaredLoggerOnWith: logger.With("peer", request.PeerId, "task", result.TaskId, "component", "streamPeerTask"),
 		},
-		successPieceCh: make(chan int32, 4),
-	}, nil, nil
+		successPieceCh: make(chan int32),
+	}
+	// bind func that base peer task did not implement
+	pt.backSourceFunc = pt.backSource
+	pt.setContentLengthFunc = pt.SetContentLength
+	pt.reportPieceResultFunc = pt.ReportPieceResult
+	return ctx, pt, nil, nil
 }
 
 func (s *streamPeerTask) ReportPieceResult(piece *base.PieceInfo, pieceResult *scheduler.PieceResult) error {
@@ -210,21 +218,10 @@ func (s *streamPeerTask) ReportPieceResult(piece *base.PieceInfo, pieceResult *s
 
 func (s *streamPeerTask) Start(ctx context.Context) (io.Reader, map[string]string, error) {
 	s.ctx, s.cancel = context.WithCancel(ctx)
-	if s.backSource {
-		go func() {
-			s.contentLength = -1
-			_ = s.callback.Init(s)
-			err := s.pieceManager.DownloadSource(ctx, s, s.request)
-			if err != nil {
-				s.Errorf("download from source error: %s", err)
-				s.cleanUnfinished()
-				return
-			}
-			s.Debugf("download from source ok")
-			_ = s.finish()
-		}()
+	if s.needBackSource {
+		go s.backSource()
 	} else {
-		s.pullPieces(s, s.cleanUnfinished)
+		s.pullPieces(s.cleanUnfinished)
 	}
 
 	// wait first piece to get content length and attribute (eg, response header for http/https)
@@ -259,10 +256,6 @@ func (s *streamPeerTask) Start(ctx context.Context) (io.Reader, map[string]strin
 		attr[config.HeaderDragonflyPeer] = s.peerID
 		return nil, attr, err
 	case first := <-s.successPieceCh:
-		//if !ok {
-		//	s.Warnf("successPieceCh closed unexpect")
-		//	return nil, nil, errors.NewDaemonConfig("early done")
-		//}
 		firstPiece = first
 	}
 
@@ -270,7 +263,7 @@ func (s *streamPeerTask) Start(ctx context.Context) (io.Reader, map[string]strin
 	attr := map[string]string{}
 	var reader io.Reader = pr
 	var writer io.Writer = pw
-	if s.contentLength != -1 {
+	if s.contentLength.Load() != -1 {
 		attr[headers.ContentLength] = fmt.Sprintf("%d", s.contentLength)
 	} else {
 		attr[headers.TransferEncoding] = "chunked"
@@ -321,8 +314,7 @@ func (s *streamPeerTask) Start(ctx context.Context) (io.Reader, map[string]strin
 				// not desired piece, cache it
 				cache[cur] = true
 				if cur < desired {
-					s.Warnf("piece number should be equal or greater than %d, received piece number: %d",
-						desired, cur)
+					s.Warnf("piece number should be equal or greater than %d, received piece number: %d", desired, cur)
 				}
 			}
 
@@ -404,7 +396,7 @@ func (s *streamPeerTask) cleanUnfinished() {
 }
 
 func (s *streamPeerTask) SetContentLength(i int64) error {
-	s.contentLength = i
+	s.contentLength.Store(i)
 	if !s.isCompleted() {
 		return nil
 	}
@@ -431,3 +423,18 @@ func (s *streamPeerTask) writeTo(w io.Writer, pieceNum int32) (int64, error) {
 	}
 	return n, pc.Close()
 }
+
+func (s *streamPeerTask) backSource() {
+	s.contentLength.Store(-1)
+	_ = s.callback.Init(s)
+	err := s.pieceManager.DownloadSource(s.ctx, s, s.request)
+	if err != nil {
+		s.Errorf("download from source error: %s", err)
+		s.failedReason = err.Error()
+		s.cleanUnfinished()
+		return
+	}
+	s.Debugf("download from source ok")
+	_ = s.finish()
+	return
+}
diff --git a/client/daemon/peer/peertask_stream_callback.go b/client/daemon/peer/peertask_stream_callback.go
index 09dcf28f..7984cb42 100644
--- a/client/daemon/peer/peertask_stream_callback.go
+++ b/client/daemon/peer/peertask_stream_callback.go
@@ -29,6 +29,7 @@ import (
 type streamPeerTaskCallback struct {
 	ctx   context.Context
 	ptm   *peerTaskManager
+	pt    *streamPeerTask
 	req   *scheduler.PeerTaskRequest
 	start time.Time
 }
@@ -90,7 +91,7 @@ func (p *streamPeerTaskCallback) Done(pt Task) error {
 		return e
 	}
 	p.ptm.PeerTaskDone(p.req.PeerId)
-	err := p.ptm.schedulerClient.ReportPeerResult(context.Background(), &scheduler.PeerResult{
+	err := p.pt.schedulerClient.ReportPeerResult(context.Background(), &scheduler.PeerResult{
 		TaskId:         pt.GetTaskID(),
 		PeerId:         pt.GetPeerID(),
 		SrcIp:          p.ptm.host.Ip,
@@ -115,7 +116,7 @@ func (p *streamPeerTaskCallback) Fail(pt Task, code base.Code, reason string) er
 	p.ptm.PeerTaskDone(p.req.PeerId)
 	var end = time.Now()
 	pt.Log().Errorf("stream peer task failed, code: %d, reason: %s", code, reason)
-	err := p.ptm.schedulerClient.ReportPeerResult(context.Background(), &scheduler.PeerResult{
+	err := p.pt.schedulerClient.ReportPeerResult(context.Background(), &scheduler.PeerResult{
 		TaskId:         pt.GetTaskID(),
 		PeerId:         pt.GetPeerID(),
 		SrcIp:          p.ptm.host.Ip,
diff --git a/client/daemon/peer/piece_manager.go b/client/daemon/peer/piece_manager.go
index 0f6a00a5..da1004d4 100644
--- a/client/daemon/peer/piece_manager.go
+++ b/client/daemon/peer/piece_manager.go
@@ -22,11 +22,9 @@ import (
 	"math"
 	"time"
 
-	"d7y.io/dragonfly/v2/pkg/source"
 	"golang.org/x/time/rate"
 
 	cdnconfig "d7y.io/dragonfly/v2/cdnsystem/config"
-
 	"d7y.io/dragonfly/v2/client/clientutil"
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/client/daemon/storage"
@@ -34,6 +32,7 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
+	"d7y.io/dragonfly/v2/pkg/source"
 	"d7y.io/dragonfly/v2/pkg/util/digestutils"
 )
 
diff --git a/client/dfget/dfget.go b/client/dfget/dfget.go
index 241c29d3..01fd112a 100644
--- a/client/dfget/dfget.go
+++ b/client/dfget/dfget.go
@@ -158,9 +158,9 @@ func downloadFromSource(ctx context.Context, cfg *config.DfgetConfig, hdr map[st
 
 	if !stringutils.IsBlank(cfg.Digest) {
 		parsedHash := digestutils.Parse(cfg.Digest)
-		realHash := digestutils.HashFile(target.Name(), parsedHash[0])
+		realHash := digestutils.HashFile(target.Name(), digestutils.Algorithms[parsedHash[0]])
 
-		if realHash != parsedHash[1] {
+		if realHash != "" && realHash != parsedHash[1] {
 			return errors.Errorf("%s digest is not matched: real[%s] expected[%s]", parsedHash[0], realHash, parsedHash[1])
 		}
 	}
diff --git a/cmd/dependency/version_cmd.go b/cmd/dependency/version_cmd.go
index 09a8e58e..a414aef2 100644
--- a/cmd/dependency/version_cmd.go
+++ b/cmd/dependency/version_cmd.go
@@ -31,8 +31,14 @@ var VersionCmd = &cobra.Command{
 	DisableAutoGenTag: true,
 	SilenceUsage:      true,
 	Run: func(cmd *cobra.Command, args []string) {
-		fmt.Printf("Major: %s Minor: %s\n", version.Major, version.Minor)
-		fmt.Printf("GitVersion: %s\n", version.GitVersion)
-		fmt.Printf("Platform: %s GoVersion: %s BuildDay: %s\n", version.Platform, version.GoVersion, version.BuildDay)
+		fmt.Printf("MajorVersion:\t%s\n", version.Major)
+		fmt.Printf("MinorVersion:\t%s\n", version.Minor)
+		fmt.Printf("GitVersion:\t%s\n", version.GitVersion)
+		fmt.Printf("GitCommit:\t%s\n", version.GitCommit)
+		fmt.Printf("Platform:\t%s\n", version.Platform)
+		fmt.Printf("BuildTime:\t%s\n", version.BuildTime)
+		fmt.Printf("GoVersion:\t%s\n", version.GoVersion)
+		fmt.Printf("Gotags:   \t%s\n", version.Gotags)
+		fmt.Printf("Gogcflags:\t%s\n", version.Gogcflags)
 	},
 }
diff --git a/deploy/charts/dragonfly/values.yaml b/deploy/charts/dragonfly/values.yaml
index 91b19c1f..569a87e5 100644
--- a/deploy/charts/dragonfly/values.yaml
+++ b/deploy/charts/dragonfly/values.yaml
@@ -57,7 +57,7 @@ cdn:
   image: dragonflyoss/cdn
   tag: v0.1.0-beta
   kubectlImage: bitnami/kubectl
-  pullPolicy: Always
+  pullPolicy: IfNotPresent
   resources:
     requests:
       cpu: "0"
diff --git a/docs/en/config/dfget.yaml b/docs/en/config/dfget.yaml
index 33486783..322b85a1 100644
--- a/docs/en/config/dfget.yaml
+++ b/docs/en/config/dfget.yaml
@@ -27,6 +27,8 @@ keepStorage: true
 scheduler:
   # schedule timeout
   scheduleTimeout: 30s
+  # when true, only scheduler says back source, daemon can back source
+  disableAutoBackSource: false
   # below example is a stand address
   netAddrs:
     - type: tcp
diff --git a/go.mod b/go.mod
index ce93a779..db18b6eb 100644
--- a/go.mod
+++ b/go.mod
@@ -33,7 +33,6 @@ require (
 	github.com/jarcoal/httpmock v1.0.8
 	github.com/klauspost/compress v1.13.1 // indirect
 	github.com/kr/text v0.2.0 // indirect
-	github.com/mattn/go-sqlite3 v2.0.1+incompatible // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
 	github.com/mcuadros/go-gin-prometheus v0.1.0
 	github.com/mitchellh/mapstructure v1.4.1
@@ -73,9 +72,9 @@ require (
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 	gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776
-	gorm.io/datatypes v1.0.1
 	gorm.io/driver/mysql v1.0.5
 	gorm.io/gorm v1.21.9
+	gorm.io/plugin/soft_delete v1.0.2
 	k8s.io/apimachinery v0.20.6 // indirect
 	k8s.io/client-go v11.0.0+incompatible
 )
diff --git a/go.sum b/go.sum
index c8e587a3..bfa9b965 100644
--- a/go.sum
+++ b/go.sum
@@ -124,9 +124,8 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/denisenkom/go-mssqldb v0.0.0-20200428022330-06a60b6afbbc h1:VRRKCwnzqk8QCaRC4os14xoKDdbHqqlJtJA0oc1ZAjg=
 github.com/denisenkom/go-mssqldb v0.0.0-20200428022330-06a60b6afbbc/go.mod h1:xbL0rPBG9cCiLr28tMa8zpbdarY27NDyej4t/EjAShU=
-github.com/denisenkom/go-mssqldb v0.9.0 h1:RSohk2RsiZqLZ0zCjtfn3S4Gp4exhpBWHyQ7D0yGjAk=
-github.com/denisenkom/go-mssqldb v0.9.0/go.mod h1:xbL0rPBG9cCiLr28tMa8zpbdarY27NDyej4t/EjAShU=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
@@ -525,9 +524,8 @@ github.com/mattn/go-isatty v0.0.13/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Ky
 github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
 github.com/mattn/go-runewidth v0.0.13 h1:lTGmDsbAYt5DmK6OnoV7EuIF1wEIFAcxld6ypU4OSgU=
 github.com/mattn/go-runewidth v0.0.13/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
-github.com/mattn/go-sqlite3 v1.14.5/go.mod h1:WVKg1VTActs4Qso6iwGbiFih2UIHo0ENGwNd0Lj+XmI=
-github.com/mattn/go-sqlite3 v2.0.1+incompatible h1:xQ15muvnzGBHpIpdrNi1DA5x0+TcBZzsIDwmw9uTHzw=
-github.com/mattn/go-sqlite3 v2.0.1+incompatible/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsOqkbpncsNc=
+github.com/mattn/go-sqlite3 v1.14.3 h1:j7a/xn1U6TKA/PHHxqZuzh64CdtRc7rU9M+AvkOl5bA=
+github.com/mattn/go-sqlite3 v1.14.3/go.mod h1:WVKg1VTActs4Qso6iwGbiFih2UIHo0ENGwNd0Lj+XmI=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 h1:I0XW9+e1XWDxdcEniV4rQAIOPUGDq67JSCiRCgGCZLI=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
@@ -800,7 +798,6 @@ golang.org/x/crypto v0.0.0-20200117160349-530e935923ad/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
-golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e h1:gsTQYXdTw2Gq7RBsWvlQ91b+aEQ6bXFUngBGuR8sPpI=
 golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -1222,30 +1219,28 @@ gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776 h1:tQIYjPdBoyREyB9XMu+nnTclpTYkz2zFM+lzLJFO4gQ=
 gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gorm.io/datatypes v1.0.1 h1:6npnXbBtjpSb7FFVA2dG/llyTN8tvZfbUqs+WyLrYgQ=
-gorm.io/datatypes v1.0.1/go.mod h1:HEHoUU3/PO5ZXfAJcVWl11+zWlE16+O0X2DgJEb4Ixs=
 gorm.io/driver/mysql v1.0.3/go.mod h1:twGxftLBlFgNVNakL7F+P/x9oYqoymG3YYT8cAfI9oI=
 gorm.io/driver/mysql v1.0.5 h1:WAAmvLK2rG0tCOqrf5XcLi2QUwugd4rcVJ/W3aoon9o=
 gorm.io/driver/mysql v1.0.5/go.mod h1:N1OIhHAIhx5SunkMGqWbGFVeh4yTNWKmMo1GOAsohLI=
 gorm.io/driver/postgres v1.0.8 h1:PAgM+PaHOSAeroTjHkCHCBIHHoBIf9RgPWGo8dF2DA8=
 gorm.io/driver/postgres v1.0.8/go.mod h1:4eOzrI1MUfm6ObJU/UcmbXyiHSs8jSwH95G5P5dxcAg=
-gorm.io/driver/sqlite v1.1.4 h1:PDzwYE+sI6De2+mxAneV9Xs11+ZyKV6oxD3wDGkaNvM=
-gorm.io/driver/sqlite v1.1.4/go.mod h1:mJCeTFr7+crvS+TRnWc5Z3UvwxUN1BGBLMrf5LA9DYw=
+gorm.io/driver/sqlite v1.1.3 h1:BYfdVuZB5He/u9dt4qDpZqiqDJ6KhPqs5QUqsr/Eeuc=
+gorm.io/driver/sqlite v1.1.3/go.mod h1:AKDgRWk8lcSQSw+9kxCJnX/yySj8G3rdwYlU57cB45c=
+gorm.io/driver/sqlserver v1.0.4 h1:V15fszi0XAo7fbx3/cF50ngshDSN4QT0MXpWTylyPTY=
 gorm.io/driver/sqlserver v1.0.4/go.mod h1:ciEo5btfITTBCj9BkoUVDvgQbUdLWQNqdFY5OGuGnRg=
-gorm.io/driver/sqlserver v1.0.7 h1:uwUtb0kdFwW5PkRbd2KJ2h4wlsqvLSjox1XVg/RnzRE=
-gorm.io/driver/sqlserver v1.0.7/go.mod h1:ng66aHI47ZIKz/vvnxzDoonzmTS8HXP+JYlgg67wOog=
 gorm.io/gorm v1.20.0/go.mod h1:0HFTzE/SqkGTzK6TlDPPQbAYCluiVvhzoA1+aVyzenw=
+gorm.io/gorm v1.20.1/go.mod h1:0HFTzE/SqkGTzK6TlDPPQbAYCluiVvhzoA1+aVyzenw=
 gorm.io/gorm v1.20.4/go.mod h1:0HFTzE/SqkGTzK6TlDPPQbAYCluiVvhzoA1+aVyzenw=
-gorm.io/gorm v1.20.7/go.mod h1:0HFTzE/SqkGTzK6TlDPPQbAYCluiVvhzoA1+aVyzenw=
 gorm.io/gorm v1.20.11/go.mod h1:0HFTzE/SqkGTzK6TlDPPQbAYCluiVvhzoA1+aVyzenw=
 gorm.io/gorm v1.20.12/go.mod h1:0HFTzE/SqkGTzK6TlDPPQbAYCluiVvhzoA1+aVyzenw=
 gorm.io/gorm v1.21.3/go.mod h1:0HFTzE/SqkGTzK6TlDPPQbAYCluiVvhzoA1+aVyzenw=
-gorm.io/gorm v1.21.4/go.mod h1:0HFTzE/SqkGTzK6TlDPPQbAYCluiVvhzoA1+aVyzenw=
-gorm.io/gorm v1.21.6/go.mod h1:F+OptMscr0P2F2qU97WT1WimdH9GaQPoDW7AYd5i2Y0=
+gorm.io/gorm v1.21.7/go.mod h1:F+OptMscr0P2F2qU97WT1WimdH9GaQPoDW7AYd5i2Y0=
 gorm.io/gorm v1.21.9 h1:INieZtn4P2Pw6xPJ8MzT0G4WUOsHq3RhfuDF1M6GW0E=
 gorm.io/gorm v1.21.9/go.mod h1:F+OptMscr0P2F2qU97WT1WimdH9GaQPoDW7AYd5i2Y0=
 gorm.io/plugin/dbresolver v1.1.0 h1:cegr4DeprR6SkLIQlKhJLYxH8muFbJ4SmnojXvoeb00=
 gorm.io/plugin/dbresolver v1.1.0/go.mod h1:tpImigFAEejCALOttyhWqsy4vfa2Uh/vAUVnL5IRF7Y=
+gorm.io/plugin/soft_delete v1.0.2 h1:BrchTwbT4qjAJrTjXeAOhdqRnWgRhaIxnQMCTgNYVoQ=
+gorm.io/plugin/soft_delete v1.0.2/go.mod h1:gBRnGiHKEXQIST8E/EWXkzQQFzJ09Rk+ZNqmHrrLPEA=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
diff --git a/hack/build.sh b/hack/build.sh
index e20b36c5..4c1e81f5 100755
--- a/hack/build.sh
+++ b/hack/build.sh
@@ -11,9 +11,18 @@ MANAGER_BINARY_NAME=manager
 
 PKG=d7y.io/dragonfly/v2
 BUILD_IMAGE=golang:1.15.8
+
+VERSION=$(git rev-parse --short HEAD)
+BUILD_TIME=$(date -u '+%Y-%m-%dT%H:%M:%SZ')
+
 GOPROXY=${GOPROXY:-}
-DATE=$(date "+%Y%m%d-%H:%M:%S")
 GOTAGS=${GOTAGS:-}
+GOGCFLAGS=${GOGCFLAGS:-}
+GOLDFLAGS="-X d7y.io/dragonfly/v2/version.GitCommit=${VERSION}"
+GOLDFLAGS="${GOLDFLAGS} -X d7y.io/dragonfly/v2/version.BuildTime=${BUILD_TIME}"
+GOLDFLAGS="${GOLDFLAGS} -X \"d7y.io/dragonfly/v2/version.Gotags=${GOTAGS:-none}\""
+GOLDFLAGS="${GOLDFLAGS} -X \"d7y.io/dragonfly/v2/version.GoVersion=$(go version | grep -o 'go[^ ].*')\""
+GOLDFLAGS="${GOLDFLAGS} -X \"d7y.io/dragonfly/v2/version.Gogcflags=${GOGCFLAGS:-none}\""
 
 curDir=$(cd "$(dirname "$0")" && pwd)
 cd "${curDir}" || return
@@ -33,7 +42,7 @@ create-dirs() {
 build-local() {
     test -f "${BUILD_SOURCE_HOME}/${BUILD_PATH}/$1" && rm -f "${BUILD_SOURCE_HOME}/${BUILD_PATH}/$1"
     cd "${BUILD_SOURCE_HOME}/cmd/$2" || return
-    go build -tags="${GOTAGS}" -o="${BUILD_SOURCE_HOME}/${BUILD_PATH}/$1"
+    go build -tags="${GOTAGS}" -ldflags="${GOLDFLAGS}" -gcflags="${GOGCFLAGS}" -o="${BUILD_SOURCE_HOME}/${BUILD_PATH}/$1"
     chmod a+x "${BUILD_SOURCE_HOME}/${BUILD_PATH}/$1"
     echo "BUILD: $2 in ${BUILD_SOURCE_HOME}/${BUILD_PATH}/$1"
 }
diff --git a/hack/docker-build.sh b/hack/docker-build.sh
index 34c48dbd..8c675580 100755
--- a/hack/docker-build.sh
+++ b/hack/docker-build.sh
@@ -4,44 +4,49 @@ set -o nounset
 set -o errexit
 set -o pipefail
 
+curDir=$(cd "$(dirname "$0")" && pwd)
+cd "${curDir}/../" || return
+
 D7Y_VERSION=${D7Y_VERSION:-"latest"}
 D7Y_REGISTRY=${D7Y_REGISTRY:-d7yio}
-GOPROXY=${GOPROXY:-}
-curDir=$(cd "$(dirname "$0")" && pwd)
 IMAGES_DIR="build/images"
-cd "${curDir}/../" || return
+BASE_IMAGE=${BASE_IMAGE:-alpine:3.12}
+
 GOPROXY=${GOPROXY:-`go env GOPROXY`}
 GOTAGS=${GOTAGS:-}
-
-docker-build::build-cdn() {
-    docker build --build-arg GOPROXY="${GOPROXY}" -t "${D7Y_REGISTRY}"/cdn:"${D7Y_VERSION}" -f "${IMAGES_DIR}/cdn/Dockerfile" .
-}
-
-docker-build::build-dfdaemon() {
-    docker build --build-arg GOPROXY="${GOPROXY}" --build-arg GOTAGS="${GOTAGS}" -t "${D7Y_REGISTRY}"/dfdaemon:"${D7Y_VERSION}" -f "${IMAGES_DIR}/dfdaemon/Dockerfile" .
-}
-
-docker-build::build-scheduler() {
-    docker build --build-arg GOPROXY="${GOPROXY}" -t "${D7Y_REGISTRY}"/scheduler:"${D7Y_VERSION}" -f "${IMAGES_DIR}/scheduler/Dockerfile" .
-}
-
-docker-build::build-manager() {
-    docker build --build-arg GOPROXY="${GOPROXY}" -t "${D7Y_REGISTRY}"/manager:"${D7Y_VERSION}" -f "${IMAGES_DIR}/manager/Dockerfile" .
+GOGCFLAGS=${GOGCFLAGS:-}
+
+# enable bash debug output
+DEBUG=${DEBUG:-}
+
+if [[ -n "$DEBUG" ]]; then
+    set -x
+fi
+
+docker-build() {
+    name=$1
+    docker build \
+      --build-arg GOPROXY="${GOPROXY}" \
+      --build-arg GOTAGS="${GOTAGS}" \
+      --build-arg GOGCFLAGS="${GOGCFLAGS}" \
+      --build-arg BASE_IMAGE="${BASE_IMAGE}" \
+      -t "${D7Y_REGISTRY}/${name}:${D7Y_VERSION}" \
+      -f "${IMAGES_DIR}/${name}/Dockerfile" .
 }
 
 main() {
     case "${1-}" in
     cdn)
-        docker-build::build-cdn
+        docker-build cdn
         ;;
     dfdaemon)
-        docker-build::build-dfdaemon
+        docker-build dfdaemon
         ;;
     scheduler)
-        docker-build::build-scheduler
+        docker-build scheduler
         ;;
     manager)
-        docker-build::build-manager
+        docker-build manager
     esac
 }
 
diff --git a/internal/constants/constants.go b/internal/constants/constants.go
index 3c647a68..cb63bb3a 100644
--- a/internal/constants/constants.go
+++ b/internal/constants/constants.go
@@ -21,7 +21,4 @@ const (
 	SourcePattern = "source"
 	CDNPattern    = "cdn"
 	P2PPattern    = "p2p"
-
-	Sha256Hash = "sha256"
-	Md5Hash    = "md5"
 )
diff --git a/manager/config/config.go b/manager/config/config.go
index 58bc4f74..675dcc3b 100644
--- a/manager/config/config.go
+++ b/manager/config/config.go
@@ -31,6 +31,7 @@ type MysqlConfig struct {
 	Host     string `yaml:"host" mapstructure:"host"`
 	Port     int    `yaml:"port" mapstructure:"port"`
 	DBName   string `yaml:"dbname" mapstructure:"dbname"`
+	Migrate  bool   `yaml:"migrate" mapstructure:"migrate"`
 }
 
 type RedisConfig struct {
@@ -93,6 +94,9 @@ func New() *Config {
 				BrokerDB:  1,
 				BackendDB: 2,
 			},
+			Mysql: &MysqlConfig{
+				Migrate: true,
+			},
 		},
 		Cache: &CacheConfig{
 			Redis: &RedisCacheConfig{
@@ -139,6 +143,9 @@ func (cfg *Config) Validate() error {
 		}
 
 		if cfg.Database.Mysql == nil {
+			if cfg.Database.Mysql.Host == "" {
+				return errors.New("empty cache mysql host is not specified")
+			}
 			return errors.New("empty cache mysql config is not specified")
 		}
 	}
diff --git a/manager/database/database.go b/manager/database/database.go
index 96dd1fcf..88ab3eda 100644
--- a/manager/database/database.go
+++ b/manager/database/database.go
@@ -44,14 +44,17 @@ func newMyqsl(cfg *config.MysqlConfig) (*gorm.DB, error) {
 		NamingStrategy: schema.NamingStrategy{
 			SingularTable: true,
 		},
+		DisableForeignKeyConstraintWhenMigrating: true,
 	})
 	if err != nil {
 		return nil, err
 	}
 
 	// Run migration
-	if err := migrate(db); err != nil {
-		return nil, err
+	if cfg.Migrate {
+		if err := migrate(db); err != nil {
+			return nil, err
+		}
 	}
 
 	// Run seed
@@ -63,7 +66,7 @@ func newMyqsl(cfg *config.MysqlConfig) (*gorm.DB, error) {
 }
 
 func migrate(db *gorm.DB) error {
-	return db.AutoMigrate(
+	return db.Set("gorm:table_options", "DEFAULT CHARSET=utf8mb4 ROW_FORMAT=Dynamic").AutoMigrate(
 		&model.CDNCluster{},
 		&model.CDN{},
 		&model.SchedulerCluster{},
@@ -87,26 +90,6 @@ func seed(db *gorm.DB) error {
 		}
 	}
 
-	var adminUserCount int64
-	var adminUserName = "admin"
-	if err := db.Model(model.User{}).Where("name = ?", adminUserName).Count(&adminUserCount).Error; err != nil {
-		return err
-	}
-	if adminUserCount <= 0 {
-		encryptedPasswordBytes, err := bcrypt.GenerateFromPassword([]byte("Dragonfly2"), bcrypt.MinCost)
-		if err != nil {
-			return err
-		}
-		if err := db.Create(&model.User{
-			EncryptedPassword: string(encryptedPasswordBytes),
-			Name:              adminUserName,
-			Email:             fmt.Sprintf("%s@Dragonfly2.com", adminUserName),
-			State:             model.UserStateEnabled,
-		}).Error; err != nil {
-			return err
-		}
-	}
-
 	var schedulerClusterCount int64
 	if err := db.Model(model.SchedulerCluster{}).Count(&schedulerClusterCount).Error; err != nil {
 		return err
@@ -138,5 +121,25 @@ func seed(db *gorm.DB) error {
 		}
 	}
 
+	var adminUserCount int64
+	var adminUserName = "admin"
+	if err := db.Model(model.User{}).Where("name = ?", adminUserName).Count(&adminUserCount).Error; err != nil {
+		return err
+	}
+	if adminUserCount <= 0 {
+		encryptedPasswordBytes, err := bcrypt.GenerateFromPassword([]byte("Dragonfly2"), bcrypt.MinCost)
+		if err != nil {
+			return err
+		}
+		if err := db.Create(&model.User{
+			EncryptedPassword: string(encryptedPasswordBytes),
+			Name:              adminUserName,
+			Email:             fmt.Sprintf("%s@Dragonfly2.com", adminUserName),
+			State:             model.UserStateEnabled,
+		}).Error; err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
diff --git a/manager/model/casbin_rule.go b/manager/model/casbin_rule.go
new file mode 100644
index 00000000..632be50a
--- /dev/null
+++ b/manager/model/casbin_rule.go
@@ -0,0 +1,12 @@
+package model
+
+type CasbinRule struct {
+	ID    uint   `gorm:"primaryKey;autoIncrement;comment:id"`
+	Ptype string `gorm:"type:varchar(100);default:NULL;uniqueIndex:uk_casbin_rule;comment:policy type"`
+	V0    string `gorm:"type:varchar(100);default:NULL;uniqueIndex:uk_casbin_rule;comment:v0"`
+	V1    string `gorm:"type:varchar(100);default:NULL;uniqueIndex:uk_casbin_rule;comment:v1"`
+	V2    string `gorm:"type:varchar(100);default:NULL;uniqueIndex:uk_casbin_rule;comment:v2"`
+	V3    string `gorm:"type:varchar(100);default:NULL;uniqueIndex:uk_casbin_rule;comment:v3"`
+	V4    string `gorm:"type:varchar(100);default:NULL;uniqueIndex:uk_casbin_rule;comment:v4"`
+	V5    string `gorm:"type:varchar(100);default:NULL;uniqueIndex:uk_casbin_rule;comment:v5"`
+}
diff --git a/manager/model/cdn.go b/manager/model/cdn.go
index 92bf4138..ccf22a6a 100644
--- a/manager/model/cdn.go
+++ b/manager/model/cdn.go
@@ -7,13 +7,13 @@ const (
 
 type CDN struct {
 	Model
-	HostName     string `gorm:"column:host_name;size:256;uniqueIndex;not null" json:"host_name"`
-	IDC          string `gorm:"column:idc;size:1024" json:"idc"`
-	Location     string `gorm:"column:location;size:1024" json:"location"`
-	IP           string `gorm:"column:ip;size:256;not null" json:"ip"`
-	Port         int32  `gorm:"column:port;not null" json:"port"`
-	DownloadPort int32  `gorm:"column:download_port;not null" json:"download_port"`
-	Status       string `gorm:"type:enum('active', 'inactive');default:'inactive'" json:"status"`
-	CDNClusterID *uint
+	HostName     string     `gorm:"column:host_name;type:varchar(256);index:uk_scheduler_cluster_name,unique;not null;comment:hostname" json:"host_name"`
+	IDC          string     `gorm:"column:idc;type:varchar(1024);comment:internet data center" json:"idc"`
+	Location     string     `gorm:"column:location;type:varchar(1024);comment:location" json:"location"`
+	IP           string     `gorm:"column:ip;type:varchar(256);not null;comment:ip address" json:"ip"`
+	Port         int32      `gorm:"column:port;not null;comment:grpc service listening port" json:"port"`
+	DownloadPort int32      `gorm:"column:download_port;not null;comment:download service listening port" json:"download_port"`
+	Status       string     `gorm:"column:status;type:varchar(256);default:'inactive';comment:service status" json:"status"`
+	CDNClusterID *uint      `gorm:"index:idx_cdn_cluster_cd_ns;comment:cdn cluster id"`
 	CDNCluster   CDNCluster `json:"-"`
 }
diff --git a/manager/model/cdn_cluster.go b/manager/model/cdn_cluster.go
index a2c7765c..b030a19f 100644
--- a/manager/model/cdn_cluster.go
+++ b/manager/model/cdn_cluster.go
@@ -1,16 +1,12 @@
 package model
 
-import (
-	"gorm.io/datatypes"
-)
-
 type CDNCluster struct {
 	Model
-	Name              string             `gorm:"column:name;size:256;uniqueIndex;not null" json:"name"`
-	BIO               string             `gorm:"column:bio;size:1024" json:"bio"`
-	Config            datatypes.JSONMap  `gorm:"column:config;not null" json:"config"`
+	Name              string             `gorm:"column:name;type:varchar(256);index:uk_cdn_cluster_name,unique;not null;comment:name" json:"name"`
+	BIO               string             `gorm:"column:bio;type:varchar(1024);comment:biography" json:"bio"`
+	Config            JSONMap            `gorm:"column:config;not null;comment:configuration" json:"config"`
 	SchedulerClusters []SchedulerCluster `gorm:"many2many:cdn_cluster_scheduler_cluster;" json:"-"`
 	CDNs              []CDN              `json:"-"`
-	SecurityGroupID   *uint
-	SecurityGroup     SecurityGroup `json:"-"`
+	SecurityGroupID   *uint              `gorm:"comment:security group id"`
+	SecurityGroup     SecurityGroup      `json:"-"`
 }
diff --git a/manager/model/models.go b/manager/model/models.go
index 40a0d42a..3776465e 100644
--- a/manager/model/models.go
+++ b/manager/model/models.go
@@ -1,16 +1,22 @@
 package model
 
 import (
+	"database/sql/driver"
+	"encoding/json"
+	"errors"
+	"fmt"
 	"time"
 
 	"gorm.io/gorm"
+	"gorm.io/gorm/schema"
+	"gorm.io/plugin/soft_delete"
 )
 
 type Model struct {
-	ID        uint           `gorm:"primarykey" json:"id"`
-	CreatedAt time.Time      `json:"created_at"`
-	UpdatedAt time.Time      `json:"updated_at"`
-	DeletedAt gorm.DeletedAt `gorm:"index" json:"deleted_at"`
+	ID        uint                  `gorm:"primarykey;comment:id" json:"id"`
+	CreatedAt time.Time             `gorm:"column:created_at;type:timestamp;comment:created timestamp" json:"created_at"`
+	UpdatedAt time.Time             `gorm:"column:updated_at;type:timestamp;comment:updated timestamp" json:"updated_at"`
+	IsDel     soft_delete.DeletedAt `gorm:"softDelete:flag;comment:soft delete flag"`
 }
 
 func Paginate(page, perPage int) func(db *gorm.DB) *gorm.DB {
@@ -19,3 +25,52 @@ func Paginate(page, perPage int) func(db *gorm.DB) *gorm.DB {
 		return db.Offset(offset).Limit(perPage)
 	}
 }
+
+type JSONMap map[string]interface{}
+
+func (m JSONMap) Value() (driver.Value, error) {
+	if m == nil {
+		return nil, nil
+	}
+	ba, err := m.MarshalJSON()
+	return string(ba), err
+}
+
+func (m *JSONMap) Scan(val interface{}) error {
+	var ba []byte
+	switch v := val.(type) {
+	case []byte:
+		ba = v
+	case string:
+		ba = []byte(v)
+	default:
+		return errors.New(fmt.Sprint("Failed to unmarshal JSONB value:", val))
+	}
+	t := map[string]interface{}{}
+	err := json.Unmarshal(ba, &t)
+	*m = JSONMap(t)
+	return err
+}
+
+func (m JSONMap) MarshalJSON() ([]byte, error) {
+	if m == nil {
+		return []byte("null"), nil
+	}
+	t := (map[string]interface{})(m)
+	return json.Marshal(t)
+}
+
+func (m *JSONMap) UnmarshalJSON(b []byte) error {
+	t := map[string]interface{}{}
+	err := json.Unmarshal(b, &t)
+	*m = JSONMap(t)
+	return err
+}
+
+func (m JSONMap) GormDataType() string {
+	return "jsonmap"
+}
+
+func (JSONMap) GormDBDataType(db *gorm.DB, field *schema.Field) string {
+	return "text"
+}
diff --git a/manager/model/scheduler.go b/manager/model/scheduler.go
index 16c7343c..80f6666a 100644
--- a/manager/model/scheduler.go
+++ b/manager/model/scheduler.go
@@ -1,9 +1,5 @@
 package model
 
-import (
-	"gorm.io/datatypes"
-)
-
 const (
 	SchedulerStatusActive   = "active"
 	SchedulerStatusInactive = "inactive"
@@ -11,14 +7,14 @@ const (
 
 type Scheduler struct {
 	Model
-	HostName           string            `gorm:"column:host_name;size:256;uniqueIndex;not null" json:"host_name"`
-	VIPs               string            `gorm:"column:vips;size:1024" json:"vips"`
-	IDC                string            `gorm:"column:idc;size:1024" json:"idc"`
-	Location           string            `gorm:"column:location;size:1024" json:"location"`
-	NetConfig          datatypes.JSONMap `gorm:"column:net_config" json:"net_config"`
-	IP                 string            `gorm:"column:ip;size:256;not null" json:"ip"`
-	Port               int32             `gorm:"column:port;not null" json:"port"`
-	Status             string            `gorm:"type:enum('active', 'inactive');default:'inactive'" json:"status"`
-	SchedulerClusterID *uint
+	HostName           string           `gorm:"column:host_name;type:varchar(256);index:uk_scheduler_host_name,unique;not null;comment:hostname" json:"host_name"`
+	VIPs               string           `gorm:"column:vips;type:varchar(1024);comment:virtual ip address" json:"vips"`
+	IDC                string           `gorm:"column:idc;type:varchar(1024);comment:internet data center" json:"idc"`
+	Location           string           `gorm:"column:location;type:varchar(1024);comment:location" json:"location"`
+	NetConfig          JSONMap          `gorm:"column:net_config;comment:network configuration" json:"net_config"`
+	IP                 string           `gorm:"column:ip;type:varchar(256);not null;comment:ip address" json:"ip"`
+	Port               int32            `gorm:"column:port;not null;comment:grpc service listening port" json:"port"`
+	Status             string           `gorm:"column:status;type:varchar(256);default:'inactive';comment:service status" json:"status"`
+	SchedulerClusterID *uint            `gorm:"comment:scheduler cluster id"`
 	SchedulerCluster   SchedulerCluster `json:"-"`
 }
diff --git a/manager/model/scheduler_cluster.go b/manager/model/scheduler_cluster.go
index 2c7b3683..77bfed7b 100644
--- a/manager/model/scheduler_cluster.go
+++ b/manager/model/scheduler_cluster.go
@@ -1,19 +1,15 @@
 package model
 
-import (
-	"gorm.io/datatypes"
-)
-
 type SchedulerCluster struct {
 	Model
-	Name            string            `gorm:"column:name;size:256;uniqueIndex;not null" json:"name"`
-	BIO             string            `gorm:"column:bio;size:1024" json:"bio"`
-	Config          datatypes.JSONMap `gorm:"column:config;not null" json:"config"`
-	ClientConfig    datatypes.JSONMap `gorm:"column:client_config;not null" json:"client_config"`
-	Scopes          datatypes.JSONMap `gorm:"column:scopes" json:"scopes"`
-	IsDefault       bool              `gorm:"column:is_default;not null;default:false" json:"is_default"`
-	CDNClusters     []CDNCluster      `gorm:"many2many:cdn_cluster_scheduler_cluster;" json:"-"`
-	Schedulers      []Scheduler       `json:"-"`
-	SecurityGroupID *uint
+	Name            string        `gorm:"column:name;type:varchar(256);index:uk_scheduler_cluster_name,unique;not null;comment:name" json:"name"`
+	BIO             string        `gorm:"column:bio;type:varchar(1024);comment:biography" json:"bio"`
+	Config          JSONMap       `gorm:"column:config;not null;comment:configuration" json:"config"`
+	ClientConfig    JSONMap       `gorm:"column:client_config;not null;comment:client configuration" json:"client_config"`
+	Scopes          JSONMap       `gorm:"column:scopes;comment:match scopes" json:"scopes"`
+	IsDefault       bool          `gorm:"column:is_default;not null;default:false;comment:default scheduler" json:"is_default"`
+	CDNClusters     []CDNCluster  `gorm:"many2many:cdn_cluster_scheduler_cluster;" json:"-"`
+	Schedulers      []Scheduler   `json:"-"`
+	SecurityGroupID *uint         `gorm:"comment:security group id"`
 	SecurityGroup   SecurityGroup `json:"-"`
 }
diff --git a/manager/model/security_group.go b/manager/model/security_group.go
index 108b251e..38efe37d 100644
--- a/manager/model/security_group.go
+++ b/manager/model/security_group.go
@@ -2,10 +2,10 @@ package model
 
 type SecurityGroup struct {
 	Model
-	Name              string             `gorm:"column:name;size:256;uniqueIndex;not null" json:"name"`
-	BIO               string             `gorm:"column:bio;size:1024" json:"bio"`
-	Domain            string             `gorm:"column:domain;size:256;uniqueIndex;not null" json:"domain"`
-	ProxyDomain       string             `gorm:"column:proxy_domain;size:1024" json:"proxy_domain"`
+	Name              string             `gorm:"column:name;type:varchar(256);index:uk_security_group_name,unique;not null;comment:name" json:"name"`
+	BIO               string             `gorm:"column:bio;type:varchar(1024);comment:biography" json:"bio"`
+	Domain            string             `gorm:"column:domain;type:varchar(256);index:uk_security_group_domain,unique;not null;comment:domain" json:"domain"`
+	ProxyDomain       string             `gorm:"column:proxy_domain;type:varchar(1024);comment:proxy domain" json:"proxy_domain"`
 	SchedulerClusters []SchedulerCluster `json:"-"`
 	CDNClusters       []CDNCluster       `json:"-"`
 }
diff --git a/manager/model/user.go b/manager/model/user.go
index 6fc39100..5a04a0d6 100644
--- a/manager/model/user.go
+++ b/manager/model/user.go
@@ -7,13 +7,13 @@ const (
 
 type User struct {
 	Model
-	Email             string `gorm:"column:email;size:256;uniqueIndex;not null" json:"email"`
-	Name              string `gorm:"column:name;size:256;uniqueIndex;not null" json:"name"`
-	EncryptedPassword string `gorm:"column:encrypted_password;not null;size:1024" json:"-"`
-	Avatar            string `gorm:"column:avatar;size:256" json:"avatar"`
-	Phone             string `gorm:"column:phone;size:256" json:"phone"`
-	PrivateToken      string `gorm:"column:private_token;size:256" json:"private_token"`
-	State             string `gorm:"type:enum('enable', 'disable');default:'enable'" json:"state"`
-	Location          string `gorm:"column:location;size:1024" json:"location"`
-	BIO               string `gorm:"column:bio;size:1024" json:"bio"`
+	Email             string `gorm:"column:email;type:varchar(256);index:uk_user_email,unique;not null;comment:email address" json:"email"`
+	Name              string `gorm:"column:name;type:varchar(256);index:uk_user_name,unique;not null;comment:name" json:"name"`
+	EncryptedPassword string `gorm:"column:encrypted_password;not null;size:1024;comment:encrypted password" json:"-"`
+	Avatar            string `gorm:"column:avatar;type:varchar(256);comment:avatar address" json:"avatar"`
+	Phone             string `gorm:"column:phone;type:varchar(256);comment:phone number" json:"phone"`
+	PrivateToken      string `gorm:"column:private_token;type:varchar(256);comment:private token" json:"private_token"`
+	State             string `gorm:"column:state;type:varchar(256);default:'enable';comment:state" json:"state"`
+	Location          string `gorm:"column:location;type:varchar(256);comment:location" json:"location"`
+	BIO               string `gorm:"column:bio;type:varchar(256);comment:biography" json:"bio"`
 }
diff --git a/manager/permission/rbac/rbac.go b/manager/permission/rbac/rbac.go
index c8d366c0..ed847327 100644
--- a/manager/permission/rbac/rbac.go
+++ b/manager/permission/rbac/rbac.go
@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
+	managermodel "d7y.io/dragonfly/v2/manager/model"
 	"d7y.io/dragonfly/v2/pkg/util/stringutils"
 	"github.com/casbin/casbin/v2"
 	"github.com/casbin/casbin/v2/model"
@@ -34,7 +35,7 @@ m = g(r.sub, p.sub) && r.obj == p.obj && (r.act == p.act || p.act == "*") || r.s
 `
 
 func NewEnforcer(gdb *gorm.DB) (*casbin.Enforcer, error) {
-	gormAdapter, err := gormadapter.NewAdapterByDB(gdb)
+	adapter, err := gormadapter.NewAdapterByDBWithCustomTable(gdb, &managermodel.CasbinRule{})
 	if err != nil {
 		return nil, err
 	}
@@ -42,7 +43,7 @@ func NewEnforcer(gdb *gorm.DB) (*casbin.Enforcer, error) {
 	if err != nil {
 		return nil, err
 	}
-	enforcer, err := casbin.NewEnforcer(m, gormAdapter)
+	enforcer, err := casbin.NewEnforcer(m, adapter)
 	if err != nil {
 		return nil, err
 	}
diff --git a/manager/service/service_grpc.go b/manager/service/service_grpc.go
index e85470fa..6d0a664f 100644
--- a/manager/service/service_grpc.go
+++ b/manager/service/service_grpc.go
@@ -16,7 +16,6 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	emptypb "google.golang.org/protobuf/types/known/emptypb"
-	"gorm.io/datatypes"
 	"gorm.io/gorm"
 )
 
@@ -177,9 +176,16 @@ func (s *GRPC) AddCDNToCDNCluster(ctx context.Context, req *manager.AddCDNToCDNC
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
+	// It will automatically associate with the first cluster when cdn cluster id is empty.
 	cdnCluster := model.CDNCluster{}
-	if err := s.db.First(&cdnCluster, req.CdnClusterId).Error; err != nil {
-		return nil, status.Error(codes.Unknown, err.Error())
+	if req.CdnClusterId > 0 {
+		if err := s.db.First(&cdnCluster, req.CdnClusterId).Error; err != nil {
+			return nil, status.Error(codes.Unknown, err.Error())
+		}
+	} else {
+		if err := s.db.First(&cdnCluster).Error; err != nil {
+			return nil, status.Error(codes.Unknown, err.Error())
+		}
 	}
 
 	cdn := model.CDN{}
@@ -301,7 +307,7 @@ func (s *GRPC) createScheduler(ctx context.Context, req *manager.UpdateScheduler
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
-	var netConfig datatypes.JSONMap
+	var netConfig model.JSONMap
 	if len(req.NetConfig) > 0 {
 		if err := netConfig.UnmarshalJSON(req.NetConfig); err != nil {
 			return nil, status.Error(codes.InvalidArgument, err.Error())
@@ -348,7 +354,7 @@ func (s *GRPC) UpdateScheduler(ctx context.Context, req *manager.UpdateScheduler
 		return nil, status.Error(codes.Unknown, err.Error())
 	}
 
-	var netConfig datatypes.JSONMap
+	var netConfig model.JSONMap
 	if len(req.NetConfig) > 0 {
 		if err := netConfig.UnmarshalJSON(req.NetConfig); err != nil {
 			return nil, status.Error(codes.InvalidArgument, err.Error())
@@ -391,9 +397,16 @@ func (s *GRPC) AddSchedulerToSchedulerCluster(ctx context.Context, req *manager.
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
+	// It will automatically associate with the first cluster when scheduler cluster id is empty.
 	schedulerCluster := model.SchedulerCluster{}
-	if err := s.db.First(&schedulerCluster, req.SchedulerClusterId).Error; err != nil {
-		return nil, status.Error(codes.Unknown, err.Error())
+	if req.SchedulerClusterId > 0 {
+		if err := s.db.First(&schedulerCluster, req.SchedulerClusterId).Error; err != nil {
+			return nil, status.Error(codes.Unknown, err.Error())
+		}
+	} else {
+		if err := s.db.First(&schedulerCluster).Error; err != nil {
+			return nil, status.Error(codes.Unknown, err.Error())
+		}
 	}
 
 	scheduler := model.Scheduler{}
diff --git a/manager/types/scheduler_cluster.go b/manager/types/scheduler_cluster.go
index 0b837e9a..32fceb06 100644
--- a/manager/types/scheduler_cluster.go
+++ b/manager/types/scheduler_cluster.go
@@ -30,7 +30,7 @@ type UpdateSchedulerClusterRequest struct {
 }
 
 type GetSchedulerClustersQuery struct {
-	Name    string `form:"name" binding:"required"`
+	Name    string `form:"name" binding:"omitempty"`
 	Page    int    `form:"page" binding:"omitempty,gte=1"`
 	PerPage int    `form:"per_page" binding:"omitempty,gte=1,lte=50"`
 }
diff --git a/pkg/ratelimiter/limitreader/limit_reader.go b/pkg/ratelimiter/limitreader/limit_reader.go
index 44e9490a..4dabf362 100644
--- a/pkg/ratelimiter/limitreader/limit_reader.go
+++ b/pkg/ratelimiter/limitreader/limit_reader.go
@@ -17,10 +17,12 @@
 package limitreader
 
 import (
-	"crypto/md5"
+	"fmt"
 	"hash"
 	"io"
 
+	"github.com/opencontainers/go-digest"
+
 	"d7y.io/dragonfly/v2/pkg/ratelimiter/ratelimiter"
 	"d7y.io/dragonfly/v2/pkg/util/digestutils"
 )
@@ -28,40 +30,17 @@ import (
 // NewLimitReader creates a LimitReader.
 // src: reader
 // rate: bytes/second
-func NewLimitReader(src io.Reader, rate int64, calculateMd5 bool) *LimitReader {
-	return NewLimitReaderWithLimiter(newRateLimiterWithDefaultWindow(rate), src, calculateMd5)
+func NewLimitReader(src io.Reader, rate int64) *LimitReader {
+	return NewLimitReaderWithLimiter(newRateLimiterWithDefaultWindow(rate), src)
 }
 
 // NewLimitReaderWithLimiter creates LimitReader with a rateLimiter.
 // src: reader
 // rate: bytes/second
-func NewLimitReaderWithLimiter(rl *ratelimiter.RateLimiter, src io.Reader, calculateMd5 bool) *LimitReader {
-	var md5sum hash.Hash
-	if calculateMd5 {
-		md5sum = md5.New()
-	}
-	return &LimitReader{
-		Src:     src,
-		Limiter: rl,
-		md5sum:  md5sum,
-	}
-}
-
-// NewLimitReaderWithMD5Sum creates LimitReader with a md5 sum.
-// src: reader
-// rate: bytes/second
-func NewLimitReaderWithMD5Sum(src io.Reader, rate int64, md5sum hash.Hash) *LimitReader {
-	return NewLimitReaderWithLimiterAndMD5Sum(src, newRateLimiterWithDefaultWindow(rate), md5sum)
-}
-
-// NewLimitReaderWithLimiterAndMD5Sum creates LimitReader with rateLimiter and md5 sum.
-// src: reader
-// rate: bytes/second
-func NewLimitReaderWithLimiterAndMD5Sum(src io.Reader, rl *ratelimiter.RateLimiter, md5sum hash.Hash) *LimitReader {
+func NewLimitReaderWithLimiter(rl *ratelimiter.RateLimiter, src io.Reader) *LimitReader {
 	return &LimitReader{
 		Src:     src,
 		Limiter: rl,
-		md5sum:  md5sum,
 	}
 }
 
@@ -71,9 +50,10 @@ func newRateLimiterWithDefaultWindow(rate int64) *ratelimiter.RateLimiter {
 
 // LimitReader reads stream with RateLimiter.
 type LimitReader struct {
-	Src     io.Reader
-	Limiter *ratelimiter.RateLimiter
-	md5sum  hash.Hash
+	Src        io.Reader
+	Limiter    *ratelimiter.RateLimiter
+	digest     hash.Hash
+	digestType string
 }
 
 func (lr *LimitReader) Read(p []byte) (n int, err error) {
@@ -82,18 +62,30 @@ func (lr *LimitReader) Read(p []byte) (n int, err error) {
 		return n, e
 	}
 	if n > 0 {
-		if lr.md5sum != nil {
-			lr.md5sum.Write(p[:n])
+		if lr.digest != nil {
+			lr.digest.Write(p[:n])
 		}
 		lr.Limiter.AcquireBlocking(int64(n))
 	}
 	return n, e
 }
 
-// Md5 calculates the md5 of all contents read.
-func (lr *LimitReader) Md5() string {
-	if lr.md5sum != nil {
-		return digestutils.ToHashString(lr.md5sum)
+// NewLimitReaderWithLimiterAndDigest creates LimitReader with rateLimiter and digest.
+// src: reader
+// rate: bytes/second
+func NewLimitReaderWithLimiterAndDigest(src io.Reader, rl *ratelimiter.RateLimiter, digest hash.Hash, digestType digest.Algorithm) *LimitReader {
+	return &LimitReader{
+		Src:        src,
+		Limiter:    rl,
+		digest:     digest,
+		digestType: digestType.String(),
+	}
+}
+
+// Digest calculates the digest of all contents read, return value is like <algo>:<hex_value>
+func (lr *LimitReader) Digest() string {
+	if lr.digest != nil {
+		return fmt.Sprintf("%s:%s", lr.digestType, digestutils.ToHashString(lr.digest))
 	}
 	return ""
 }
diff --git a/pkg/rpc/client.go b/pkg/rpc/client.go
index 3456e6be..c8348567 100644
--- a/pkg/rpc/client.go
+++ b/pkg/rpc/client.go
@@ -338,7 +338,7 @@ func (conn *Connection) GetClientConn(hashKey string, stick bool) (*grpc.ClientC
 	if stick && !ok {
 		conn.rwMutex.RUnlock()
 		// if request is stateful, hash key must exist in key2NodeMap
-		return nil, fmt.Errorf("it is a stateful request， but cannot find hash key(%s) in key2NodeMap", hashKey)
+		return nil, fmt.Errorf("it is a stateful request, but cannot find hash key(%s) in key2NodeMap", hashKey)
 	}
 	if ok {
 		// if exist
diff --git a/pkg/rpc/manager/manager.pb.go b/pkg/rpc/manager/manager.pb.go
index 2c48a248..03914171 100644
--- a/pkg/rpc/manager/manager.pb.go
+++ b/pkg/rpc/manager/manager.pb.go
@@ -1245,167 +1245,168 @@ var file_pkg_rpc_manager_manager_proto_rawDesc = []byte{
 	0x0a, 0x0d, 0x64, 0x6f, 0x77, 0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18,
 	0x07, 0x20, 0x01, 0x28, 0x05, 0x42, 0x0c, 0xfa, 0x42, 0x09, 0x1a, 0x07, 0x10, 0xff, 0xff, 0x03,
 	0x28, 0x80, 0x08, 0x52, 0x0c, 0x64, 0x6f, 0x77, 0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x50, 0x6f, 0x72,
-	0x74, 0x22, 0xaa, 0x01, 0x0a, 0x19, 0x41, 0x64, 0x64, 0x43, 0x44, 0x4e, 0x54, 0x6f, 0x43, 0x44,
+	0x74, 0x22, 0xac, 0x01, 0x0a, 0x19, 0x41, 0x64, 0x64, 0x43, 0x44, 0x4e, 0x54, 0x6f, 0x43, 0x44,
 	0x4e, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12,
 	0x3e, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01,
 	0x20, 0x01, 0x28, 0x0e, 0x32, 0x13, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x53,
 	0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x42, 0x08, 0xfa, 0x42, 0x05, 0x82, 0x01,
 	0x02, 0x10, 0x01, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12,
 	0x1e, 0x0a, 0x06, 0x63, 0x64, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x04, 0x42,
-	0x07, 0xfa, 0x42, 0x04, 0x32, 0x02, 0x40, 0x01, 0x52, 0x05, 0x63, 0x64, 0x6e, 0x49, 0x64, 0x12,
-	0x2d, 0x0a, 0x0e, 0x63, 0x64, 0x6e, 0x5f, 0x63, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x5f, 0x69,
-	0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x04, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x32, 0x02, 0x40, 0x01,
-	0x52, 0x0c, 0x63, 0x64, 0x6e, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x49, 0x64, 0x22, 0xc4,
-	0x01, 0x0a, 0x10, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x43, 0x6c, 0x75, 0x73,
-	0x74, 0x65, 0x72, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52,
-	0x02, 0x69, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28,
-	0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x62, 0x69, 0x6f, 0x18, 0x03,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x62, 0x69, 0x6f, 0x12, 0x16, 0x0a, 0x06, 0x63, 0x6f, 0x6e,
-	0x66, 0x69, 0x67, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x63, 0x6f, 0x6e, 0x66, 0x69,
-	0x67, 0x12, 0x23, 0x0a, 0x0d, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x63, 0x6f, 0x6e, 0x66,
-	0x69, 0x67, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74,
-	0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x3d, 0x0a, 0x0e, 0x73, 0x65, 0x63, 0x75, 0x72, 0x69,
-	0x74, 0x79, 0x5f, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x16,
-	0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x53, 0x65, 0x63, 0x75, 0x72, 0x69, 0x74,
-	0x79, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x52, 0x0d, 0x73, 0x65, 0x63, 0x75, 0x72, 0x69, 0x74, 0x79,
-	0x47, 0x72, 0x6f, 0x75, 0x70, 0x22, 0xbf, 0x02, 0x0a, 0x09, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75,
-	0x6c, 0x65, 0x72, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52,
-	0x02, 0x69, 0x64, 0x12, 0x1b, 0x0a, 0x09, 0x68, 0x6f, 0x73, 0x74, 0x5f, 0x6e, 0x61, 0x6d, 0x65,
-	0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x68, 0x6f, 0x73, 0x74, 0x4e, 0x61, 0x6d, 0x65,
-	0x12, 0x12, 0x0a, 0x04, 0x76, 0x69, 0x70, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04,
-	0x76, 0x69, 0x70, 0x73, 0x12, 0x10, 0x0a, 0x03, 0x69, 0x64, 0x63, 0x18, 0x04, 0x20, 0x01, 0x28,
-	0x09, 0x52, 0x03, 0x69, 0x64, 0x63, 0x12, 0x1a, 0x0a, 0x08, 0x6c, 0x6f, 0x63, 0x61, 0x74, 0x69,
-	0x6f, 0x6e, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x6c, 0x6f, 0x63, 0x61, 0x74, 0x69,
-	0x6f, 0x6e, 0x12, 0x1d, 0x0a, 0x0a, 0x6e, 0x65, 0x74, 0x5f, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67,
-	0x18, 0x06, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x09, 0x6e, 0x65, 0x74, 0x43, 0x6f, 0x6e, 0x66, 0x69,
-	0x67, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69,
-	0x70, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x08, 0x20, 0x01, 0x28, 0x05, 0x52,
-	0x04, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x18,
-	0x09, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x46, 0x0a,
-	0x11, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x5f, 0x63, 0x6c, 0x75, 0x73, 0x74,
-	0x65, 0x72, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x19, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67,
-	0x65, 0x72, 0x2e, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x43, 0x6c, 0x75, 0x73,
-	0x74, 0x65, 0x72, 0x52, 0x10, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x43, 0x6c,
-	0x75, 0x73, 0x74, 0x65, 0x72, 0x12, 0x20, 0x0a, 0x04, 0x63, 0x64, 0x6e, 0x73, 0x18, 0x0c, 0x20,
-	0x03, 0x28, 0x0b, 0x32, 0x0c, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x43, 0x44,
-	0x4e, 0x52, 0x04, 0x63, 0x64, 0x6e, 0x73, 0x22, 0x7b, 0x0a, 0x13, 0x47, 0x65, 0x74, 0x53, 0x63,
-	0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x3e,
-	0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20,
-	0x01, 0x28, 0x0e, 0x32, 0x13, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x53, 0x6f,
-	0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x42, 0x08, 0xfa, 0x42, 0x05, 0x82, 0x01, 0x02,
-	0x10, 0x01, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12, 0x24,
-	0x0a, 0x09, 0x68, 0x6f, 0x73, 0x74, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28,
-	0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x68, 0x01, 0x52, 0x08, 0x68, 0x6f, 0x73, 0x74,
-	0x4e, 0x61, 0x6d, 0x65, 0x22, 0xd0, 0x02, 0x0a, 0x16, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x53,
-	0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12,
-	0x3e, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01,
-	0x20, 0x01, 0x28, 0x0e, 0x32, 0x13, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x53,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x42, 0x08, 0xfa, 0x42, 0x05, 0x82, 0x01,
-	0x02, 0x10, 0x01, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12,
-	0x24, 0x0a, 0x09, 0x68, 0x6f, 0x73, 0x74, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01,
-	0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x68, 0x01, 0x52, 0x08, 0x68, 0x6f, 0x73,
-	0x74, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x21, 0x0a, 0x04, 0x76, 0x69, 0x70, 0x73, 0x18, 0x04, 0x20,
-	0x01, 0x28, 0x09, 0x42, 0x0d, 0xfa, 0x42, 0x0a, 0x72, 0x08, 0x10, 0x01, 0x18, 0x80, 0x08, 0xd0,
-	0x01, 0x01, 0x52, 0x04, 0x76, 0x69, 0x70, 0x73, 0x12, 0x1f, 0x0a, 0x03, 0x69, 0x64, 0x63, 0x18,
-	0x05, 0x20, 0x01, 0x28, 0x09, 0x42, 0x0d, 0xfa, 0x42, 0x0a, 0x72, 0x08, 0x10, 0x01, 0x18, 0x80,
-	0x08, 0xd0, 0x01, 0x01, 0x52, 0x03, 0x69, 0x64, 0x63, 0x12, 0x27, 0x0a, 0x08, 0x6c, 0x6f, 0x63,
-	0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x42, 0x0b, 0xfa, 0x42, 0x08,
-	0x72, 0x06, 0x18, 0x80, 0x08, 0xd0, 0x01, 0x01, 0x52, 0x08, 0x6c, 0x6f, 0x63, 0x61, 0x74, 0x69,
-	0x6f, 0x6e, 0x12, 0x28, 0x0a, 0x0a, 0x6e, 0x65, 0x74, 0x5f, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67,
-	0x18, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x42, 0x09, 0xfa, 0x42, 0x06, 0x7a, 0x04, 0x10, 0x01, 0x70,
-	0x01, 0x52, 0x09, 0x6e, 0x65, 0x74, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x17, 0x0a, 0x02,
-	0x69, 0x70, 0x18, 0x08, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x70,
-	0x01, 0x52, 0x02, 0x69, 0x70, 0x12, 0x20, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x09, 0x20,
-	0x01, 0x28, 0x05, 0x42, 0x0c, 0xfa, 0x42, 0x09, 0x1a, 0x07, 0x10, 0xff, 0xff, 0x03, 0x28, 0x80,
-	0x08, 0x52, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x22, 0xce, 0x01, 0x0a, 0x25, 0x41, 0x64, 0x64, 0x53,
-	0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x54, 0x6f, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75,
-	0x6c, 0x65, 0x72, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x07, 0xfa, 0x42, 0x04, 0x32, 0x02, 0x28, 0x01, 0x52, 0x05, 0x63, 0x64, 0x6e, 0x49, 0x64, 0x12,
+	0x2f, 0x0a, 0x0e, 0x63, 0x64, 0x6e, 0x5f, 0x63, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x5f, 0x69,
+	0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x04, 0x42, 0x09, 0xfa, 0x42, 0x06, 0x32, 0x04, 0x28, 0x01,
+	0x40, 0x01, 0x52, 0x0c, 0x63, 0x64, 0x6e, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x49, 0x64,
+	0x22, 0xc4, 0x01, 0x0a, 0x10, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x43, 0x6c,
+	0x75, 0x73, 0x74, 0x65, 0x72, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x04, 0x52, 0x02, 0x69, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x62, 0x69, 0x6f,
+	0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x62, 0x69, 0x6f, 0x12, 0x16, 0x0a, 0x06, 0x63,
+	0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x63, 0x6f, 0x6e,
+	0x66, 0x69, 0x67, 0x12, 0x23, 0x0a, 0x0d, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x63, 0x6f,
+	0x6e, 0x66, 0x69, 0x67, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0c, 0x63, 0x6c, 0x69, 0x65,
+	0x6e, 0x74, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x3d, 0x0a, 0x0e, 0x73, 0x65, 0x63, 0x75,
+	0x72, 0x69, 0x74, 0x79, 0x5f, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b,
+	0x32, 0x16, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x53, 0x65, 0x63, 0x75, 0x72,
+	0x69, 0x74, 0x79, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x52, 0x0d, 0x73, 0x65, 0x63, 0x75, 0x72, 0x69,
+	0x74, 0x79, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x22, 0xbf, 0x02, 0x0a, 0x09, 0x53, 0x63, 0x68, 0x65,
+	0x64, 0x75, 0x6c, 0x65, 0x72, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x04, 0x52, 0x02, 0x69, 0x64, 0x12, 0x1b, 0x0a, 0x09, 0x68, 0x6f, 0x73, 0x74, 0x5f, 0x6e, 0x61,
+	0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x68, 0x6f, 0x73, 0x74, 0x4e, 0x61,
+	0x6d, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x76, 0x69, 0x70, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x04, 0x76, 0x69, 0x70, 0x73, 0x12, 0x10, 0x0a, 0x03, 0x69, 0x64, 0x63, 0x18, 0x04, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x03, 0x69, 0x64, 0x63, 0x12, 0x1a, 0x0a, 0x08, 0x6c, 0x6f, 0x63, 0x61,
+	0x74, 0x69, 0x6f, 0x6e, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x6c, 0x6f, 0x63, 0x61,
+	0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1d, 0x0a, 0x0a, 0x6e, 0x65, 0x74, 0x5f, 0x63, 0x6f, 0x6e, 0x66,
+	0x69, 0x67, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x09, 0x6e, 0x65, 0x74, 0x43, 0x6f, 0x6e,
+	0x66, 0x69, 0x67, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x02, 0x69, 0x70, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x08, 0x20, 0x01, 0x28,
+	0x05, 0x52, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75,
+	0x73, 0x18, 0x09, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12,
+	0x46, 0x0a, 0x11, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x5f, 0x63, 0x6c, 0x75,
+	0x73, 0x74, 0x65, 0x72, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x19, 0x2e, 0x6d, 0x61, 0x6e,
+	0x61, 0x67, 0x65, 0x72, 0x2e, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x43, 0x6c,
+	0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x10, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72,
+	0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x12, 0x20, 0x0a, 0x04, 0x63, 0x64, 0x6e, 0x73, 0x18,
+	0x0c, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0c, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e,
+	0x43, 0x44, 0x4e, 0x52, 0x04, 0x63, 0x64, 0x6e, 0x73, 0x22, 0x7b, 0x0a, 0x13, 0x47, 0x65, 0x74,
+	0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x12, 0x3e, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18,
+	0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x13, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e,
+	0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x42, 0x08, 0xfa, 0x42, 0x05, 0x82,
+	0x01, 0x02, 0x10, 0x01, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65,
+	0x12, 0x24, 0x0a, 0x09, 0x68, 0x6f, 0x73, 0x74, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20,
+	0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x68, 0x01, 0x52, 0x08, 0x68, 0x6f,
+	0x73, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x22, 0xd0, 0x02, 0x0a, 0x16, 0x55, 0x70, 0x64, 0x61, 0x74,
+	0x65, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
 	0x74, 0x12, 0x3e, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x74, 0x79, 0x70, 0x65,
 	0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x13, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72,
 	0x2e, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x42, 0x08, 0xfa, 0x42, 0x05,
 	0x82, 0x01, 0x02, 0x10, 0x01, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70,
-	0x65, 0x12, 0x2a, 0x0a, 0x0c, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x5f, 0x69,
-	0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x04, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x32, 0x02, 0x40, 0x01,
-	0x52, 0x0b, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x49, 0x64, 0x12, 0x39, 0x0a,
-	0x14, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x5f, 0x63, 0x6c, 0x75, 0x73, 0x74,
-	0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x04, 0x42, 0x07, 0xfa, 0x42, 0x04,
-	0x32, 0x02, 0x40, 0x01, 0x52, 0x12, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x43,
-	0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x49, 0x64, 0x22, 0xa8, 0x02, 0x0a, 0x15, 0x4c, 0x69, 0x73,
-	0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65,
-	0x73, 0x74, 0x12, 0x3e, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x74, 0x79, 0x70,
-	0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x13, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65,
-	0x72, 0x2e, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x42, 0x08, 0xfa, 0x42,
-	0x05, 0x82, 0x01, 0x02, 0x10, 0x01, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79,
-	0x70, 0x65, 0x12, 0x24, 0x0a, 0x09, 0x68, 0x6f, 0x73, 0x74, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18,
-	0x02, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x68, 0x01, 0x52, 0x08,
-	0x68, 0x6f, 0x73, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x17, 0x0a, 0x02, 0x69, 0x70, 0x18, 0x03,
-	0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x70, 0x01, 0x52, 0x02, 0x69,
-	0x70, 0x12, 0x53, 0x0a, 0x09, 0x68, 0x6f, 0x73, 0x74, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18, 0x04,
-	0x20, 0x03, 0x28, 0x0b, 0x32, 0x2c, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x4c,
-	0x69, 0x73, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x73, 0x52, 0x65, 0x71,
-	0x75, 0x65, 0x73, 0x74, 0x2e, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x45, 0x6e, 0x74,
-	0x72, 0x79, 0x42, 0x08, 0xfa, 0x42, 0x05, 0x9a, 0x01, 0x02, 0x30, 0x01, 0x52, 0x08, 0x68, 0x6f,
-	0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x1a, 0x3b, 0x0a, 0x0d, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e,
-	0x66, 0x6f, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c,
-	0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a,
-	0x02, 0x38, 0x01, 0x22, 0x4c, 0x0a, 0x16, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64,
-	0x75, 0x6c, 0x65, 0x72, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x32, 0x0a,
-	0x0a, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28,
-	0x0b, 0x32, 0x12, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x53, 0x63, 0x68, 0x65,
-	0x64, 0x75, 0x6c, 0x65, 0x72, 0x52, 0x0a, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72,
-	0x73, 0x22, 0x78, 0x0a, 0x10, 0x4b, 0x65, 0x65, 0x70, 0x41, 0x6c, 0x69, 0x76, 0x65, 0x52, 0x65,
-	0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x3e, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f,
-	0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x13, 0x2e, 0x6d, 0x61, 0x6e,
-	0x61, 0x67, 0x65, 0x72, 0x2e, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x42,
-	0x08, 0xfa, 0x42, 0x05, 0x82, 0x01, 0x02, 0x10, 0x01, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63,
-	0x65, 0x54, 0x79, 0x70, 0x65, 0x12, 0x24, 0x0a, 0x09, 0x68, 0x6f, 0x73, 0x74, 0x5f, 0x6e, 0x61,
-	0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x68,
-	0x01, 0x52, 0x08, 0x68, 0x6f, 0x73, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x2a, 0x45, 0x0a, 0x0a, 0x53,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12, 0x14, 0x0a, 0x10, 0x53, 0x43, 0x48,
-	0x45, 0x44, 0x55, 0x4c, 0x45, 0x52, 0x5f, 0x53, 0x4f, 0x55, 0x52, 0x43, 0x45, 0x10, 0x00, 0x12,
-	0x11, 0x0a, 0x0d, 0x43, 0x4c, 0x49, 0x45, 0x4e, 0x54, 0x5f, 0x53, 0x4f, 0x55, 0x52, 0x43, 0x45,
-	0x10, 0x01, 0x12, 0x0e, 0x0a, 0x0a, 0x43, 0x44, 0x4e, 0x5f, 0x53, 0x4f, 0x55, 0x52, 0x43, 0x45,
-	0x10, 0x02, 0x32, 0xca, 0x04, 0x0a, 0x07, 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x12, 0x2e,
-	0x0a, 0x06, 0x47, 0x65, 0x74, 0x43, 0x44, 0x4e, 0x12, 0x16, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67,
-	0x65, 0x72, 0x2e, 0x47, 0x65, 0x74, 0x43, 0x44, 0x4e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
-	0x1a, 0x0c, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x43, 0x44, 0x4e, 0x12, 0x34,
-	0x0a, 0x09, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x43, 0x44, 0x4e, 0x12, 0x19, 0x2e, 0x6d, 0x61,
-	0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x43, 0x44, 0x4e, 0x52,
-	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x0c, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72,
-	0x2e, 0x43, 0x44, 0x4e, 0x12, 0x50, 0x0a, 0x12, 0x41, 0x64, 0x64, 0x43, 0x44, 0x4e, 0x54, 0x6f,
-	0x43, 0x44, 0x4e, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x12, 0x22, 0x2e, 0x6d, 0x61, 0x6e,
-	0x61, 0x67, 0x65, 0x72, 0x2e, 0x41, 0x64, 0x64, 0x43, 0x44, 0x4e, 0x54, 0x6f, 0x43, 0x44, 0x4e,
-	0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16,
-	0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66,
-	0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x12, 0x40, 0x0a, 0x0c, 0x47, 0x65, 0x74, 0x53, 0x63, 0x68,
-	0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x12, 0x1c, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72,
-	0x2e, 0x47, 0x65, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x52, 0x65, 0x71,
-	0x75, 0x65, 0x73, 0x74, 0x1a, 0x12, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x53,
-	0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x12, 0x46, 0x0a, 0x0f, 0x55, 0x70, 0x64, 0x61,
-	0x74, 0x65, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x12, 0x1f, 0x2e, 0x6d, 0x61,
-	0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x53, 0x63, 0x68, 0x65,
-	0x64, 0x75, 0x6c, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x12, 0x2e, 0x6d,
-	0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72,
-	0x12, 0x68, 0x0a, 0x1e, 0x41, 0x64, 0x64, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72,
-	0x54, 0x6f, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x43, 0x6c, 0x75, 0x73, 0x74,
-	0x65, 0x72, 0x12, 0x2e, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x41, 0x64, 0x64,
-	0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x54, 0x6f, 0x53, 0x63, 0x68, 0x65, 0x64,
-	0x75, 0x6c, 0x65, 0x72, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65,
+	0x65, 0x12, 0x24, 0x0a, 0x09, 0x68, 0x6f, 0x73, 0x74, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02,
+	0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x68, 0x01, 0x52, 0x08, 0x68,
+	0x6f, 0x73, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x21, 0x0a, 0x04, 0x76, 0x69, 0x70, 0x73, 0x18,
+	0x04, 0x20, 0x01, 0x28, 0x09, 0x42, 0x0d, 0xfa, 0x42, 0x0a, 0x72, 0x08, 0x10, 0x01, 0x18, 0x80,
+	0x08, 0xd0, 0x01, 0x01, 0x52, 0x04, 0x76, 0x69, 0x70, 0x73, 0x12, 0x1f, 0x0a, 0x03, 0x69, 0x64,
+	0x63, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x42, 0x0d, 0xfa, 0x42, 0x0a, 0x72, 0x08, 0x10, 0x01,
+	0x18, 0x80, 0x08, 0xd0, 0x01, 0x01, 0x52, 0x03, 0x69, 0x64, 0x63, 0x12, 0x27, 0x0a, 0x08, 0x6c,
+	0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x42, 0x0b, 0xfa,
+	0x42, 0x08, 0x72, 0x06, 0x18, 0x80, 0x08, 0xd0, 0x01, 0x01, 0x52, 0x08, 0x6c, 0x6f, 0x63, 0x61,
+	0x74, 0x69, 0x6f, 0x6e, 0x12, 0x28, 0x0a, 0x0a, 0x6e, 0x65, 0x74, 0x5f, 0x63, 0x6f, 0x6e, 0x66,
+	0x69, 0x67, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x42, 0x09, 0xfa, 0x42, 0x06, 0x7a, 0x04, 0x10,
+	0x01, 0x70, 0x01, 0x52, 0x09, 0x6e, 0x65, 0x74, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x17,
+	0x0a, 0x02, 0x69, 0x70, 0x18, 0x08, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72,
+	0x02, 0x70, 0x01, 0x52, 0x02, 0x69, 0x70, 0x12, 0x20, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18,
+	0x09, 0x20, 0x01, 0x28, 0x05, 0x42, 0x0c, 0xfa, 0x42, 0x09, 0x1a, 0x07, 0x10, 0xff, 0xff, 0x03,
+	0x28, 0x80, 0x08, 0x52, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x22, 0xd0, 0x01, 0x0a, 0x25, 0x41, 0x64,
+	0x64, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x54, 0x6f, 0x53, 0x63, 0x68, 0x65,
+	0x64, 0x75, 0x6c, 0x65, 0x72, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75,
+	0x65, 0x73, 0x74, 0x12, 0x3e, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x74, 0x79,
+	0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x13, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67,
+	0x65, 0x72, 0x2e, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x42, 0x08, 0xfa,
+	0x42, 0x05, 0x82, 0x01, 0x02, 0x10, 0x01, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54,
+	0x79, 0x70, 0x65, 0x12, 0x2a, 0x0a, 0x0c, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72,
+	0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x04, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x32, 0x02,
+	0x28, 0x01, 0x52, 0x0b, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x49, 0x64, 0x12,
+	0x3b, 0x0a, 0x14, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x5f, 0x63, 0x6c, 0x75,
+	0x73, 0x74, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x04, 0x42, 0x09, 0xfa,
+	0x42, 0x06, 0x32, 0x04, 0x28, 0x01, 0x40, 0x01, 0x52, 0x12, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75,
+	0x6c, 0x65, 0x72, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x49, 0x64, 0x22, 0xa8, 0x02, 0x0a,
+	0x15, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x73, 0x52,
+	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x3e, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
+	0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x13, 0x2e, 0x6d, 0x61,
+	0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65,
+	0x42, 0x08, 0xfa, 0x42, 0x05, 0x82, 0x01, 0x02, 0x10, 0x01, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72,
+	0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12, 0x24, 0x0a, 0x09, 0x68, 0x6f, 0x73, 0x74, 0x5f, 0x6e,
+	0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02,
+	0x68, 0x01, 0x52, 0x08, 0x68, 0x6f, 0x73, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x17, 0x0a, 0x02,
+	0x69, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x70,
+	0x01, 0x52, 0x02, 0x69, 0x70, 0x12, 0x53, 0x0a, 0x09, 0x68, 0x6f, 0x73, 0x74, 0x5f, 0x69, 0x6e,
+	0x66, 0x6f, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x2c, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67,
+	0x65, 0x72, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72,
+	0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x2e, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66,
+	0x6f, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x42, 0x08, 0xfa, 0x42, 0x05, 0x9a, 0x01, 0x02, 0x30, 0x01,
+	0x52, 0x08, 0x68, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x1a, 0x3b, 0x0a, 0x0d, 0x48, 0x6f,
+	0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b,
+	0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a,
+	0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61,
+	0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0x4c, 0x0a, 0x16, 0x4c, 0x69, 0x73, 0x74, 0x53,
+	0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x12, 0x32, 0x0a, 0x0a, 0x73, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x73, 0x18,
+	0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e,
+	0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x52, 0x0a, 0x73, 0x63, 0x68, 0x65, 0x64,
+	0x75, 0x6c, 0x65, 0x72, 0x73, 0x22, 0x78, 0x0a, 0x10, 0x4b, 0x65, 0x65, 0x70, 0x41, 0x6c, 0x69,
+	0x76, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x3e, 0x0a, 0x0b, 0x73, 0x6f, 0x75,
+	0x72, 0x63, 0x65, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x13,
+	0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54,
+	0x79, 0x70, 0x65, 0x42, 0x08, 0xfa, 0x42, 0x05, 0x82, 0x01, 0x02, 0x10, 0x01, 0x52, 0x0a, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12, 0x24, 0x0a, 0x09, 0x68, 0x6f, 0x73,
+	0x74, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42,
+	0x04, 0x72, 0x02, 0x68, 0x01, 0x52, 0x08, 0x68, 0x6f, 0x73, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x2a,
+	0x45, 0x0a, 0x0a, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12, 0x14, 0x0a,
+	0x10, 0x53, 0x43, 0x48, 0x45, 0x44, 0x55, 0x4c, 0x45, 0x52, 0x5f, 0x53, 0x4f, 0x55, 0x52, 0x43,
+	0x45, 0x10, 0x00, 0x12, 0x11, 0x0a, 0x0d, 0x43, 0x4c, 0x49, 0x45, 0x4e, 0x54, 0x5f, 0x53, 0x4f,
+	0x55, 0x52, 0x43, 0x45, 0x10, 0x01, 0x12, 0x0e, 0x0a, 0x0a, 0x43, 0x44, 0x4e, 0x5f, 0x53, 0x4f,
+	0x55, 0x52, 0x43, 0x45, 0x10, 0x02, 0x32, 0xca, 0x04, 0x0a, 0x07, 0x4d, 0x61, 0x6e, 0x61, 0x67,
+	0x65, 0x72, 0x12, 0x2e, 0x0a, 0x06, 0x47, 0x65, 0x74, 0x43, 0x44, 0x4e, 0x12, 0x16, 0x2e, 0x6d,
+	0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x47, 0x65, 0x74, 0x43, 0x44, 0x4e, 0x52, 0x65, 0x71,
+	0x75, 0x65, 0x73, 0x74, 0x1a, 0x0c, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x43,
+	0x44, 0x4e, 0x12, 0x34, 0x0a, 0x09, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x43, 0x44, 0x4e, 0x12,
+	0x19, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65,
+	0x43, 0x44, 0x4e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x0c, 0x2e, 0x6d, 0x61, 0x6e,
+	0x61, 0x67, 0x65, 0x72, 0x2e, 0x43, 0x44, 0x4e, 0x12, 0x50, 0x0a, 0x12, 0x41, 0x64, 0x64, 0x43,
+	0x44, 0x4e, 0x54, 0x6f, 0x43, 0x44, 0x4e, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x12, 0x22,
+	0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x41, 0x64, 0x64, 0x43, 0x44, 0x4e, 0x54,
+	0x6f, 0x43, 0x44, 0x4e, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65,
 	0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74,
-	0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x12, 0x51, 0x0a, 0x0e, 0x4c, 0x69,
-	0x73, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x73, 0x12, 0x1e, 0x2e, 0x6d,
-	0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64,
-	0x75, 0x6c, 0x65, 0x72, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1f, 0x2e, 0x6d,
-	0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64,
-	0x75, 0x6c, 0x65, 0x72, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x40, 0x0a,
-	0x09, 0x4b, 0x65, 0x65, 0x70, 0x41, 0x6c, 0x69, 0x76, 0x65, 0x12, 0x19, 0x2e, 0x6d, 0x61, 0x6e,
-	0x61, 0x67, 0x65, 0x72, 0x2e, 0x4b, 0x65, 0x65, 0x70, 0x41, 0x6c, 0x69, 0x76, 0x65, 0x52, 0x65,
-	0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70,
-	0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x28, 0x01, 0x42,
-	0x25, 0x5a, 0x23, 0x64, 0x37, 0x79, 0x2e, 0x69, 0x6f, 0x2f, 0x64, 0x72, 0x61, 0x67, 0x6f, 0x6e,
-	0x66, 0x6c, 0x79, 0x2f, 0x76, 0x32, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x72, 0x70, 0x63, 0x2f, 0x6d,
-	0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x12, 0x40, 0x0a, 0x0c, 0x47, 0x65,
+	0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x12, 0x1c, 0x2e, 0x6d, 0x61, 0x6e,
+	0x61, 0x67, 0x65, 0x72, 0x2e, 0x47, 0x65, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65,
+	0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x12, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67,
+	0x65, 0x72, 0x2e, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x12, 0x46, 0x0a, 0x0f,
+	0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x12,
+	0x1f, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65,
+	0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x1a, 0x12, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x53, 0x63, 0x68, 0x65, 0x64,
+	0x75, 0x6c, 0x65, 0x72, 0x12, 0x68, 0x0a, 0x1e, 0x41, 0x64, 0x64, 0x53, 0x63, 0x68, 0x65, 0x64,
+	0x75, 0x6c, 0x65, 0x72, 0x54, 0x6f, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x43,
+	0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x12, 0x2e, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72,
+	0x2e, 0x41, 0x64, 0x64, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x54, 0x6f, 0x53,
+	0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52,
+	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x12, 0x51,
+	0x0a, 0x0e, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x73,
+	0x12, 0x1e, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x53,
+	0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x1a, 0x1f, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x4c, 0x69, 0x73, 0x74, 0x53,
+	0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x72, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x12, 0x40, 0x0a, 0x09, 0x4b, 0x65, 0x65, 0x70, 0x41, 0x6c, 0x69, 0x76, 0x65, 0x12, 0x19,
+	0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x4b, 0x65, 0x65, 0x70, 0x41, 0x6c, 0x69,
+	0x76, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67,
+	0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74,
+	0x79, 0x28, 0x01, 0x42, 0x25, 0x5a, 0x23, 0x64, 0x37, 0x79, 0x2e, 0x69, 0x6f, 0x2f, 0x64, 0x72,
+	0x61, 0x67, 0x6f, 0x6e, 0x66, 0x6c, 0x79, 0x2f, 0x76, 0x32, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x72,
+	0x70, 0x63, 0x2f, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74,
+	0x6f, 0x33,
 }
 
 var (
diff --git a/pkg/rpc/manager/manager.pb.validate.go b/pkg/rpc/manager/manager.pb.validate.go
index 5c7dbddc..cd0c8c62 100644
--- a/pkg/rpc/manager/manager.pb.validate.go
+++ b/pkg/rpc/manager/manager.pb.validate.go
@@ -558,12 +558,22 @@ func (m *AddCDNToCDNClusterRequest) Validate() error {
 		}
 	}
 
-	if m.GetCdnId() != 0 {
-
+	if m.GetCdnId() < 1 {
+		return AddCDNToCDNClusterRequestValidationError{
+			field:  "CdnId",
+			reason: "value must be greater than or equal to 1",
+		}
 	}
 
 	if m.GetCdnClusterId() != 0 {
 
+		if m.GetCdnClusterId() < 1 {
+			return AddCDNToCDNClusterRequestValidationError{
+				field:  "CdnClusterId",
+				reason: "value must be greater than or equal to 1",
+			}
+		}
+
 	}
 
 	return nil
@@ -1114,12 +1124,22 @@ func (m *AddSchedulerToSchedulerClusterRequest) Validate() error {
 		}
 	}
 
-	if m.GetSchedulerId() != 0 {
-
+	if m.GetSchedulerId() < 1 {
+		return AddSchedulerToSchedulerClusterRequestValidationError{
+			field:  "SchedulerId",
+			reason: "value must be greater than or equal to 1",
+		}
 	}
 
 	if m.GetSchedulerClusterId() != 0 {
 
+		if m.GetSchedulerClusterId() < 1 {
+			return AddSchedulerToSchedulerClusterRequestValidationError{
+				field:  "SchedulerClusterId",
+				reason: "value must be greater than or equal to 1",
+			}
+		}
+
 	}
 
 	return nil
diff --git a/pkg/rpc/manager/manager.proto b/pkg/rpc/manager/manager.proto
index b9fd873f..7c1afe1d 100644
--- a/pkg/rpc/manager/manager.proto
+++ b/pkg/rpc/manager/manager.proto
@@ -74,8 +74,8 @@ message UpdateCDNRequest {
 
 message AddCDNToCDNClusterRequest {
   SourceType source_type = 1 [(validate.rules).enum.defined_only = true];
-  uint64 cdn_id = 2 [(validate.rules).uint64 = {ignore_empty: true}];
-  uint64 cdn_cluster_id = 3 [(validate.rules).uint64 = {ignore_empty: true}];
+  uint64 cdn_id = 2 [(validate.rules).uint64.gte = 1];
+  uint64 cdn_cluster_id = 3 [(validate.rules).uint64 = {ignore_empty: true, gte: 1}];
 }
 
 message SchedulerCluster {
@@ -119,8 +119,8 @@ message UpdateSchedulerRequest {
 
 message AddSchedulerToSchedulerClusterRequest {
   SourceType source_type = 1 [(validate.rules).enum.defined_only = true];
-  uint64 scheduler_id = 2 [(validate.rules).uint64 = {ignore_empty: true}];
-  uint64 scheduler_cluster_id = 3 [(validate.rules).uint64 = {ignore_empty: true}];
+  uint64 scheduler_id = 2 [(validate.rules).uint64.gte = 1];
+  uint64 scheduler_cluster_id = 3 [(validate.rules).uint64 = {ignore_empty: true, gte: 1}];
 }
 
 message ListSchedulersRequest {
diff --git a/pkg/source/httpprotocol/http_source_client.go b/pkg/source/httpprotocol/http_source_client.go
index aec3bdaf..b6ee911f 100644
--- a/pkg/source/httpprotocol/http_source_client.go
+++ b/pkg/source/httpprotocol/http_source_client.go
@@ -22,38 +22,60 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"net/url"
+	"os"
 	"time"
 
 	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
 
+	"github.com/go-http-utils/headers"
+
 	"d7y.io/dragonfly/v2/cdnsystem/daemon/task"
 	"d7y.io/dragonfly/v2/pkg/source"
 	"d7y.io/dragonfly/v2/pkg/structure/maputils"
 	"d7y.io/dragonfly/v2/pkg/util/stringutils"
 	"d7y.io/dragonfly/v2/pkg/util/timeutils"
-	"github.com/go-http-utils/headers"
 )
 
 const (
 	HTTPClient  = "http"
 	HTTPSClient = "https"
+
+	ProxyEnv = "D7Y_SOURCE_PROXY"
 )
 
 var _defaultHTTPClient *http.Client
 var _ source.ResourceClient = (*httpSourceClient)(nil)
 
 func init() {
+	// TODO support customize source client
+	var (
+		proxy *url.URL
+		err   error
+	)
+	if proxyEnv := os.Getenv(ProxyEnv); len(proxyEnv) > 0 {
+		proxy, err = url.Parse(proxyEnv)
+		if err != nil {
+			fmt.Printf("Back source proxy parse error: %s\n", err)
+		}
+	}
+
 	transport := http.DefaultTransport.(*http.Transport).Clone()
 	transport.DialContext = (&net.Dialer{
 		Timeout:   3 * time.Second,
 		KeepAlive: 30 * time.Second,
 	}).DialContext
+
+	if proxy != nil {
+		transport.Proxy = http.ProxyURL(proxy)
+	}
+
 	_defaultHTTPClient = &http.Client{
 		Transport: transport,
 	}
-	httpSourceClient := NewHTTPSourceClient()
-	source.Register(HTTPClient, httpSourceClient)
-	source.Register(HTTPSClient, httpSourceClient)
+	sc := NewHTTPSourceClient()
+	source.Register(HTTPClient, sc)
+	source.Register(HTTPSClient, sc)
 }
 
 // httpSourceClient is an implementation of the interface of source.ResourceClient.
diff --git a/pkg/source/source_client.go b/pkg/source/source_client.go
index cd002df7..b607cf0c 100644
--- a/pkg/source/source_client.go
+++ b/pkg/source/source_client.go
@@ -25,16 +25,15 @@ import (
 	"sync"
 	"time"
 
-	"github.com/pkg/errors"
-
-	rangers "d7y.io/dragonfly/v2/pkg/util/rangeutils"
 	"github.com/go-http-utils/headers"
+	"github.com/pkg/errors"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
+	rangers "d7y.io/dragonfly/v2/pkg/util/rangeutils"
 )
 
-var _ ResourceClient = (*ClientManagerImpl)(nil)
-var _ ClientManager = (*ClientManagerImpl)(nil)
+var _ ResourceClient = (*clientManager)(nil)
+var _ ClientManager = (*clientManager)(nil)
 
 // ResourceClient supply apis that interact with the source.
 type ResourceClient interface {
@@ -50,13 +49,13 @@ type ResourceClient interface {
 	// IsExpired checks if a resource received or stored is the same.
 	IsExpired(ctx context.Context, url string, header RequestHeader, expireInfo map[string]string) (bool, error)
 
-	// Download download from source
+	// Download downloads from source
 	Download(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, error)
 
 	// DownloadWithResponseHeader download from source with responseHeader
 	DownloadWithResponseHeader(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, ResponseHeader, error)
 
-	// GetLastModified get lastModified timestamp milliseconds of resource
+	// GetLastModifiedMillis gets last modified timestamp milliseconds of resource
 	GetLastModifiedMillis(ctx context.Context, url string, header RequestHeader) (int64, error)
 }
 
@@ -66,17 +65,15 @@ type ClientManager interface {
 	UnRegister(schema string)
 }
 
-type ClientManagerImpl struct {
+type clientManager struct {
 	sync.RWMutex
 	clients map[string]ResourceClient
 }
 
-var _defaultMgr = &ClientManagerImpl{
-	clients: make(map[string]ResourceClient),
-}
+var _defaultManager = NewManager()
 
-func (clientMgr *ClientManagerImpl) GetContentLength(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (int64, error) {
-	sourceClient, err := clientMgr.getSourceClient(url)
+func (m *clientManager) GetContentLength(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (int64, error) {
+	sourceClient, err := m.getSourceClient(url)
 	if err != nil {
 		return -1, err
 	}
@@ -88,8 +85,8 @@ func (clientMgr *ClientManagerImpl) GetContentLength(ctx context.Context, url st
 	return sourceClient.GetContentLength(ctx, url, header, rang)
 }
 
-func (clientMgr *ClientManagerImpl) IsSupportRange(ctx context.Context, url string, header RequestHeader) (bool, error) {
-	sourceClient, err := clientMgr.getSourceClient(url)
+func (m *clientManager) IsSupportRange(ctx context.Context, url string, header RequestHeader) (bool, error) {
+	sourceClient, err := m.getSourceClient(url)
 	if err != nil {
 		return false, err
 	}
@@ -101,8 +98,8 @@ func (clientMgr *ClientManagerImpl) IsSupportRange(ctx context.Context, url stri
 	return sourceClient.IsSupportRange(ctx, url, header)
 }
 
-func (clientMgr *ClientManagerImpl) IsExpired(ctx context.Context, url string, header RequestHeader, expireInfo map[string]string) (bool, error) {
-	sourceClient, err := clientMgr.getSourceClient(url)
+func (m *clientManager) IsExpired(ctx context.Context, url string, header RequestHeader, expireInfo map[string]string) (bool, error) {
+	sourceClient, err := m.getSourceClient(url)
 	if err != nil {
 		return false, err
 	}
@@ -114,25 +111,25 @@ func (clientMgr *ClientManagerImpl) IsExpired(ctx context.Context, url string, h
 	return sourceClient.IsExpired(ctx, url, header, expireInfo)
 }
 
-func (clientMgr *ClientManagerImpl) Download(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
-	sourceClient, err := clientMgr.getSourceClient(url)
+func (m *clientManager) Download(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, error) {
+	sourceClient, err := m.getSourceClient(url)
 	if err != nil {
 		return nil, err
 	}
 	return sourceClient.Download(ctx, url, header, rang)
 }
 
-func (clientMgr *ClientManagerImpl) DownloadWithResponseHeader(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, ResponseHeader,
+func (m *clientManager) DownloadWithResponseHeader(ctx context.Context, url string, header RequestHeader, rang *rangers.Range) (io.ReadCloser, ResponseHeader,
 	error) {
-	sourceClient, err := clientMgr.getSourceClient(url)
+	sourceClient, err := m.getSourceClient(url)
 	if err != nil {
 		return nil, nil, err
 	}
 	return sourceClient.DownloadWithResponseHeader(ctx, url, header, rang)
 }
 
-func (clientMgr *ClientManagerImpl) GetLastModifiedMillis(ctx context.Context, url string, header RequestHeader) (int64, error) {
-	sourceClient, err := clientMgr.getSourceClient(url)
+func (m *clientManager) GetLastModifiedMillis(ctx context.Context, url string, header RequestHeader) (int64, error) {
+	sourceClient, err := m.getSourceClient(url)
 	if err != nil {
 		return -1, err
 	}
@@ -145,31 +142,31 @@ func (clientMgr *ClientManagerImpl) GetLastModifiedMillis(ctx context.Context, u
 }
 
 func NewManager() ClientManager {
-	return &ClientManagerImpl{
+	return &clientManager{
 		clients: make(map[string]ResourceClient),
 	}
 }
 
-func (clientMgr *ClientManagerImpl) Register(schema string, resourceClient ResourceClient) {
-	if client, ok := clientMgr.clients[strings.ToLower(schema)]; ok {
+func (m *clientManager) Register(schema string, resourceClient ResourceClient) {
+	if client, ok := m.clients[strings.ToLower(schema)]; ok {
 		logger.Infof("replace client %#v with %#v for schema %s", client, resourceClient, schema)
 	}
-	clientMgr.clients[strings.ToLower(schema)] = resourceClient
+	m.clients[strings.ToLower(schema)] = resourceClient
 }
 
 func Register(schema string, resourceClient ResourceClient) {
-	_defaultMgr.Register(schema, resourceClient)
+	_defaultManager.Register(schema, resourceClient)
 }
 
-func (clientMgr *ClientManagerImpl) UnRegister(schema string) {
-	if client, ok := clientMgr.clients[strings.ToLower(schema)]; ok {
+func (m *clientManager) UnRegister(schema string) {
+	if client, ok := m.clients[strings.ToLower(schema)]; ok {
 		logger.Infof("remove client %#v for schema %s", client, schema)
 	}
-	delete(clientMgr.clients, strings.ToLower(schema))
+	delete(m.clients, strings.ToLower(schema))
 }
 
 func UnRegister(schema string) {
-	_defaultMgr.UnRegister(schema)
+	_defaultManager.UnRegister(schema)
 }
 
 func GetContentLength(ctx context.Context, url string, header RequestHeader) (int64, error) {
@@ -181,15 +178,15 @@ func GetContentLength(ctx context.Context, url string, header RequestHeader) (in
 			return -1, err
 		}
 	}
-	return _defaultMgr.GetContentLength(ctx, url, header, httpRange)
+	return _defaultManager.GetContentLength(ctx, url, header, httpRange)
 }
 
 func IsSupportRange(ctx context.Context, url string, header RequestHeader) (bool, error) {
-	return _defaultMgr.IsSupportRange(ctx, url, header)
+	return _defaultManager.IsSupportRange(ctx, url, header)
 }
 
 func IsExpired(ctx context.Context, url string, header RequestHeader, expireInfo map[string]string) (bool, error) {
-	return _defaultMgr.IsExpired(ctx, url, header, expireInfo)
+	return _defaultManager.IsExpired(ctx, url, header, expireInfo)
 }
 
 func Download(ctx context.Context, url string, header RequestHeader) (io.ReadCloser, error) {
@@ -201,7 +198,7 @@ func Download(ctx context.Context, url string, header RequestHeader) (io.ReadClo
 			return nil, err
 		}
 	}
-	return _defaultMgr.Download(ctx, url, header, httpRange)
+	return _defaultManager.Download(ctx, url, header, httpRange)
 }
 
 func DownloadWithResponseHeader(ctx context.Context, url string, header RequestHeader) (io.ReadCloser, ResponseHeader, error) {
@@ -214,34 +211,34 @@ func DownloadWithResponseHeader(ctx context.Context, url string, header RequestH
 			return nil, nil, err
 		}
 	}
-	return _defaultMgr.DownloadWithResponseHeader(ctx, url, header, httpRange)
+	return _defaultManager.DownloadWithResponseHeader(ctx, url, header, httpRange)
 }
 
 func GetLastModifiedMillis(ctx context.Context, url string, header RequestHeader) (int64, error) {
-	return _defaultMgr.GetLastModifiedMillis(ctx, url, header)
+	return _defaultManager.GetLastModifiedMillis(ctx, url, header)
 }
 
 // getSourceClient get a source client from source manager with specified schema.
-func (clientMgr *ClientManagerImpl) getSourceClient(rawURL string) (ResourceClient, error) {
-	logger.Debugf("current clients: %#v", clientMgr.clients)
+func (m *clientManager) getSourceClient(rawURL string) (ResourceClient, error) {
+	logger.Debugf("current clients: %#v", m.clients)
 	parsedURL, err := url.Parse(rawURL)
 	if err != nil {
 		return nil, err
 	}
-	clientMgr.RLock()
-	client, ok := clientMgr.clients[strings.ToLower(parsedURL.Scheme)]
-	clientMgr.RUnlock()
+	m.RLock()
+	client, ok := m.clients[strings.ToLower(parsedURL.Scheme)]
+	m.RUnlock()
 	if !ok || client == nil {
-		return nil, errors.Errorf("can not find client for supporting url %s, clients:%v", rawURL, clientMgr.clients)
+		return nil, errors.Errorf("can not find client for supporting url %s, clients:%v", rawURL, m.clients)
 	}
 	return client, nil
 }
 
-func (clientMgr *ClientManagerImpl) loadSourcePlugin(schema string) (ResourceClient, error) {
-	clientMgr.Lock()
-	defer clientMgr.Unlock()
+func (m *clientManager) loadSourcePlugin(schema string) (ResourceClient, error) {
+	m.Lock()
+	defer m.Unlock()
 	// double check
-	client, ok := clientMgr.clients[schema]
+	client, ok := m.clients[schema]
 	if ok {
 		return client, nil
 	}
@@ -250,6 +247,6 @@ func (clientMgr *ClientManagerImpl) loadSourcePlugin(schema string) (ResourceCli
 	if err != nil {
 		return nil, err
 	}
-	clientMgr.clients[schema] = client
+	m.clients[schema] = client
 	return client, nil
 }
diff --git a/pkg/util/digestutils/digest.go b/pkg/util/digestutils/digest.go
index dd6b6f30..90cdd6da 100644
--- a/pkg/util/digestutils/digest.go
+++ b/pkg/util/digestutils/digest.go
@@ -26,11 +26,27 @@ import (
 	"os"
 	"strings"
 
-	"d7y.io/dragonfly/v2/internal/constants"
+	"github.com/opencontainers/go-digest"
+
 	"d7y.io/dragonfly/v2/pkg/unit"
 	"d7y.io/dragonfly/v2/pkg/util/fileutils"
 )
 
+const (
+	Sha256Hash digest.Algorithm = "sha256"
+	Md5Hash    digest.Algorithm = "md5"
+)
+
+var (
+	// Algorithms is used to check if an algorithm is supported.
+	// If algo is not supported, Algorithms[algo] will return empty string.
+	// Please don't use digest.Algorithm() to convert a string to digest.Algorithm.
+	Algorithms = map[string]digest.Algorithm{
+		Sha256Hash.String(): Sha256Hash,
+		Md5Hash.String():    Md5Hash,
+	}
+)
+
 func Sha256(values ...string) string {
 	if len(values) == 0 {
 		return ""
@@ -62,8 +78,8 @@ func Md5Bytes(bytes []byte) string {
 }
 
 // HashFile computes hash value corresponding to hashType,
-// hashType is from constants.Md5Hash and constants.Sha256Hash.
-func HashFile(file string, hashType string) string {
+// hashType is from digestutils.Md5Hash and digestutils.Sha256Hash.
+func HashFile(file string, hashType digest.Algorithm) string {
 	if !fileutils.IsRegular(file) {
 		return ""
 	}
@@ -76,9 +92,9 @@ func HashFile(file string, hashType string) string {
 	defer f.Close()
 
 	var h hash.Hash
-	if hashType == constants.Md5Hash {
+	if hashType == Md5Hash {
 		h = md5.New()
-	} else if hashType == constants.Sha256Hash {
+	} else if hashType == Sha256Hash {
 		h = sha256.New()
 	} else {
 		return ""
@@ -102,3 +118,15 @@ func Parse(digest string) []string {
 	digest = strings.Trim(digest, " ")
 	return strings.Split(digest, ":")
 }
+
+func CreateHash(hashType string) hash.Hash {
+	algo := Algorithms[hashType]
+	switch algo {
+	case Sha256Hash:
+		return sha256.New()
+	case Md5Hash:
+		return md5.New()
+	default:
+		return nil
+	}
+}
diff --git a/scheduler/core/events.go b/scheduler/core/events.go
index 6e9b24f4..164a2bfc 100644
--- a/scheduler/core/events.go
+++ b/scheduler/core/events.go
@@ -69,7 +69,7 @@ func (s *state) start() {
 		}
 		parent, candidates, hashParent := s.sched.ScheduleParent(peer)
 		if !hashParent && !peer.Host.CDN {
-			logger.WithTaskAndPeerID(peer.Task.TaskID, peer.PeerID).Warnf("waitScheduleParentPeerQueue: there is no available parent，reschedule it in one second")
+			logger.WithTaskAndPeerID(peer.Task.TaskID, peer.PeerID).Warnf("waitScheduleParentPeerQueue: there is no available parent, reschedule it in one second")
 			s.waitScheduleParentPeerQueue.Done(v)
 			s.waitScheduleParentPeerQueue.AddAfter(peer, time.Second)
 			continue
@@ -96,7 +96,7 @@ var _ event = startReportPieceResultEvent{}
 func (e startReportPieceResultEvent) apply(s *state) {
 	parent, candidates, hasParent := s.sched.ScheduleParent(e.peer)
 	if !hasParent {
-		logger.WithTaskAndPeerID(e.peer.Task.TaskID, e.peer.PeerID).Warnf("peerScheduleParentEvent: there is no available parent，reschedule it in one second")
+		logger.WithTaskAndPeerID(e.peer.Task.TaskID, e.peer.PeerID).Warnf("peerScheduleParentEvent: there is no available parent, reschedule it in one second")
 		s.waitScheduleParentPeerQueue.AddAfter(e.peer, time.Second)
 		return
 	}
@@ -123,7 +123,7 @@ func (e peerDownloadPieceSuccessEvent) apply(s *state) {
 		var hasParent bool
 		parentPeer, candidates, hasParent = s.sched.ScheduleParent(e.peer)
 		if !hasParent {
-			logger.WithTaskAndPeerID(e.peer.Task.TaskID, e.peer.PeerID).Warnf("peerDownloadPieceSuccessEvent: there is no available parent，reschedule it in one second")
+			logger.WithTaskAndPeerID(e.peer.Task.TaskID, e.peer.PeerID).Warnf("peerDownloadPieceSuccessEvent: there is no available parent, reschedule it in one second")
 			s.waitScheduleParentPeerQueue.AddAfter(e.peer, time.Second)
 			return
 		}
@@ -237,7 +237,7 @@ func (e peerDownloadFailEvent) apply(s *state) {
 		child := (value).(*types.Peer)
 		parent, candidates, hasParent := s.sched.ScheduleParent(child)
 		if !hasParent {
-			logger.WithTaskAndPeerID(child.Task.TaskID, child.PeerID).Warnf("peerDownloadFailEvent: there is no available parent，reschedule it in one second")
+			logger.WithTaskAndPeerID(child.Task.TaskID, child.PeerID).Warnf("peerDownloadFailEvent: there is no available parent, reschedule it in one second")
 			s.waitScheduleParentPeerQueue.AddAfter(e.peer, time.Second)
 			return true
 		}
@@ -306,7 +306,7 @@ func handlePeerLeave(peer *types.Peer, s *state) {
 		child := value.(*types.Peer)
 		parent, candidates, hasParent := s.sched.ScheduleParent(child)
 		if !hasParent {
-			logger.WithTaskAndPeerID(child.Task.TaskID, child.PeerID).Warnf("handlePeerLeave: there is no available parent，reschedule it in one second")
+			logger.WithTaskAndPeerID(child.Task.TaskID, child.PeerID).Warnf("handlePeerLeave: there is no available parent, reschedule it in one second")
 			s.waitScheduleParentPeerQueue.AddAfter(child, time.Second)
 			return true
 		}
diff --git a/scheduler/server/server.go b/scheduler/server/server.go
index 9a00cf7b..2c0c0257 100644
--- a/scheduler/server/server.go
+++ b/scheduler/server/server.go
@@ -78,7 +78,6 @@ func New(cfg *config.Config) (*Server, error) {
 		if err := s.register(context.Background()); err != nil {
 			return nil, err
 		}
-		logger.Info("scheduler register to manager")
 
 		if cfg.DynConfig.Type == dynconfig.ManagerSourceType {
 			options = append(options,
@@ -186,16 +185,14 @@ func (s *Server) register(ctx context.Context) error {
 	logger.Infof("update scheduler %s to manager successfully", scheduler.HostName)
 
 	schedulerClusterID := s.config.Manager.SchedulerClusterID
-	if schedulerClusterID != 0 {
-		if _, err := s.managerClient.AddSchedulerToSchedulerCluster(ctx, &manager.AddSchedulerToSchedulerClusterRequest{
-			SchedulerId:        scheduler.Id,
-			SchedulerClusterId: schedulerClusterID,
-		}); err != nil {
-			logger.Warnf("add scheduler %s to scheduler cluster %s failed %v", scheduler.HostName, schedulerClusterID, err)
-			return err
-		}
-		logger.Infof("add scheduler %s to scheduler cluster %s successfully", scheduler.HostName, schedulerClusterID)
+	if _, err := s.managerClient.AddSchedulerToSchedulerCluster(ctx, &manager.AddSchedulerToSchedulerClusterRequest{
+		SchedulerId:        scheduler.Id,
+		SchedulerClusterId: schedulerClusterID,
+	}); err != nil {
+		logger.Warnf("add scheduler %s to scheduler cluster %s failed %v", scheduler.HostName, schedulerClusterID, err)
+		return err
 	}
+	logger.Infof("add scheduler %s to scheduler cluster %s successfully", scheduler.HostName, schedulerClusterID)
 	return nil
 }
 
diff --git a/scheduler/tasks/tasks.go b/scheduler/tasks/tasks.go
index f7766f81..4182875f 100644
--- a/scheduler/tasks/tasks.go
+++ b/scheduler/tasks/tasks.go
@@ -2,7 +2,6 @@ package tasks
 
 import (
 	"context"
-	"strings"
 	"time"
 
 	"github.com/go-playground/validator/v10"
@@ -139,11 +138,7 @@ func (t *tasks) preheat(req string) error {
 		Header: request.Headers,
 		Tag:    request.Tag,
 		Filter: request.Filter,
-	}
-
-	//TODO(@zzy987) CDN don't support sha256
-	if strings.HasPrefix(request.Digest, "md5") {
-		meta.Digest = request.Digest
+		Digest: request.Digest,
 	}
 
 	// Generate range
diff --git a/scheduler/types/task.go b/scheduler/types/task.go
index 605388d0..9ee7a2b7 100644
--- a/scheduler/types/task.go
+++ b/scheduler/types/task.go
@@ -144,8 +144,6 @@ func (task *Task) GetLastAccessTime() time.Time {
 }
 
 func (task *Task) ListPeers() *sortedlist.SortedList {
-	task.lock.Lock()
-	defer task.lock.Unlock()
 	return task.peers
 }
 
diff --git a/version/version.go b/version/version.go
index 5fa651d7..227fa180 100644
--- a/version/version.go
+++ b/version/version.go
@@ -16,11 +16,14 @@
 
 package version
 
-const (
+var (
 	Major      = "2"
 	Minor      = "0"
 	GitVersion = "v2.0.0-rc.0"
-	GoVersion  = "go1.15.2"
+	GitCommit  = "unknown"
 	Platform   = osArch
-	BuildDay   = "2021-04-26"
+	BuildTime  = "unknown"
+	GoVersion  = "unknown"
+	Gotags     = "unknown"
+	Gogcflags  = "unknown"
 )


Test patch:
diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml
index 679ca4c4..948953c5 100644
--- a/.github/workflows/e2e.yml
+++ b/.github/workflows/e2e.yml
@@ -43,6 +43,7 @@ jobs:
       - name: Get dependencies
         run: |
           go mod vendor
+          mkdir -p /tmp/artifact
 
       - name: Setup Kind
         uses: engineerd/setup-kind@v0.5.0
@@ -59,3 +60,10 @@ jobs:
           token: ${{ secrets.CODECOV_TOKEN }}
           files: ./coverage.txt
           flags: e2etests
+
+      - name: Upload Logs
+        uses: actions/upload-artifact@v2
+        if: failure()
+        with:
+          name: dragonfly-logs
+          path: /tmp/artifact/
diff --git a/cdnsystem/daemon/cdn/cache_detector_test.go b/cdnsystem/daemon/cdn/cache_detector_test.go
index 24ef6bcf..0258930b 100644
--- a/cdnsystem/daemon/cdn/cache_detector_test.go
+++ b/cdnsystem/daemon/cdn/cache_detector_test.go
@@ -218,37 +218,37 @@ var fullPieceMetaRecords = append(partialPieceMetaRecords, &storage.PieceMetaRec
 
 func newCompletedFileMeta(taskID string, URL string, success bool) *storage.FileMetaData {
 	return &storage.FileMetaData{
-		TaskID:          taskID,
-		TaskURL:         URL,
-		PieceSize:       2000,
-		SourceFileLen:   9789,
-		AccessTime:      1624126443284,
-		Interval:        0,
-		CdnFileLength:   9789,
-		SourceRealMd5:   "",
-		PieceMd5Sign:    "98166bdfebb7b71dd5c6d47492d844f4421d90199641ca11fd8ce3111894115a",
-		ExpireInfo:      nil,
-		Finish:          true,
-		Success:         success,
-		TotalPieceCount: 5,
+		TaskID:           taskID,
+		TaskURL:          URL,
+		PieceSize:        2000,
+		SourceFileLen:    9789,
+		AccessTime:       1624126443284,
+		Interval:         0,
+		CdnFileLength:    9789,
+		SourceRealDigest: "",
+		PieceMd5Sign:     "98166bdfebb7b71dd5c6d47492d844f4421d90199641ca11fd8ce3111894115a",
+		ExpireInfo:       nil,
+		Finish:           true,
+		Success:          success,
+		TotalPieceCount:  5,
 	}
 }
 
 func newPartialFileMeta(taskID string, URL string) *storage.FileMetaData {
 	return &storage.FileMetaData{
-		TaskID:          taskID,
-		TaskURL:         URL,
-		PieceSize:       2000,
-		SourceFileLen:   9789,
-		AccessTime:      1624126443284,
-		Interval:        0,
-		CdnFileLength:   0,
-		SourceRealMd5:   "",
-		PieceMd5Sign:    "",
-		ExpireInfo:      nil,
-		Finish:          false,
-		Success:         false,
-		TotalPieceCount: 0,
+		TaskID:           taskID,
+		TaskURL:          URL,
+		PieceSize:        2000,
+		SourceFileLen:    9789,
+		AccessTime:       1624126443284,
+		Interval:         0,
+		CdnFileLength:    0,
+		SourceRealDigest: "",
+		PieceMd5Sign:     "",
+		ExpireInfo:       nil,
+		Finish:           false,
+		Success:          false,
+		TotalPieceCount:  0,
 	}
 }
 
diff --git a/cdnsystem/daemon/cdn/manager_test.go b/cdnsystem/daemon/cdn/manager_test.go
index 4bddb1f5..58e09a81 100644
--- a/cdnsystem/daemon/cdn/manager_test.go
+++ b/cdnsystem/daemon/cdn/manager_test.go
@@ -61,13 +61,16 @@ func (suite *CDNManagerTestSuite) SetupSuite() {
 	}
 	ctrl := gomock.NewController(suite.T())
 	progressMgr := mock.NewMockSeedProgressMgr(ctrl)
-	progressMgr.EXPECT().PublishPiece(taskID, gomock.Any()).Return(nil).Times(98 * 2)
+	progressMgr.EXPECT().PublishPiece(md5TaskID, gomock.Any()).Return(nil).Times(98 * 2)
+	progressMgr.EXPECT().PublishPiece(sha256TaskID, gomock.Any()).Return(nil).Times(98 * 2)
 	suite.cm, _ = newManager(config.New(), storeMgr, progressMgr)
 }
 
-var dragonflyURL = "http://dragonfly.io.com?a=a&b=b&c=c"
-
-var taskID = idgen.TaskID(dragonflyURL, &base.UrlMeta{Digest: "f1e2488bba4d1267948d9e2f7008571c", Tag: "dragonfly", Filter: "a&b"})
+var (
+	dragonflyURL = "http://dragonfly.io.com?a=a&b=b&c=c"
+	md5TaskID    = idgen.TaskID(dragonflyURL, &base.UrlMeta{Digest: "md5:f1e2488bba4d1267948d9e2f7008571c", Tag: "dragonfly", Filter: "a&b"})
+	sha256TaskID = idgen.TaskID(dragonflyURL, &base.UrlMeta{Digest: "sha256:b9907b9a5ba2b0223868c201b9addfe2ec1da1b90325d57c34f192966b0a68c5", Tag: "dragonfly", Filter: "a&b"})
+)
 
 func (suite *CDNManagerTestSuite) TearDownSuite() {
 	if suite.workHome != "" {
@@ -120,40 +123,86 @@ func (suite *CDNManagerTestSuite) TestTriggerCDN() {
 			return fileInfo.Size(), nil
 		},
 	).AnyTimes()
-	sourceTask := &types.SeedTask{
-		TaskID:           taskID,
-		URL:              dragonflyURL,
-		TaskURL:          urlutils.FilterURLParam(dragonflyURL, []string{"a", "b"}),
-		SourceFileLength: 9789,
-		CdnFileLength:    0,
-		PieceSize:        100,
-		Header:           map[string]string{"md5": "f1e2488bba4d1267948d9e2f7008571c"},
-		CdnStatus:        types.TaskInfoCdnStatusRunning,
-		PieceTotal:       0,
-		RequestMd5:       "f1e2488bba4d1267948d9e2f7008571c",
-		SourceRealMd5:    "",
-		PieceMd5Sign:     "",
+
+	tests := []struct {
+		name       string
+		sourceTask *types.SeedTask
+		targetTask *types.SeedTask
+	}{
+		{
+			name: "trigger_md5",
+			sourceTask: &types.SeedTask{
+				TaskID:           md5TaskID,
+				URL:              dragonflyURL,
+				TaskURL:          urlutils.FilterURLParam(dragonflyURL, []string{"a", "b"}),
+				SourceFileLength: 9789,
+				CdnFileLength:    0,
+				PieceSize:        100,
+				Header:           map[string]string{"md5": "f1e2488bba4d1267948d9e2f7008571c"},
+				CdnStatus:        types.TaskInfoCdnStatusRunning,
+				PieceTotal:       0,
+				RequestDigest:    "md5:f1e2488bba4d1267948d9e2f7008571c",
+				SourceRealDigest: "",
+				PieceMd5Sign:     "",
+			},
+			targetTask: &types.SeedTask{
+				TaskID:           md5TaskID,
+				URL:              dragonflyURL,
+				TaskURL:          urlutils.FilterURLParam(dragonflyURL, []string{"a", "b"}),
+				SourceFileLength: 9789,
+				CdnFileLength:    9789,
+				PieceSize:        100,
+				Header:           map[string]string{"md5": "f1e2488bba4d1267948d9e2f7008571c"},
+				CdnStatus:        types.TaskInfoCdnStatusSuccess,
+				PieceTotal:       0,
+				RequestDigest:    "md5:f1e2488bba4d1267948d9e2f7008571c",
+				SourceRealDigest: "md5:f1e2488bba4d1267948d9e2f7008571c",
+				PieceMd5Sign:     "bb138842f338fff90af737e4a6b2c6f8e2a7031ca9d5900bc9b646f6406d890f",
+			},
+		},
+		{
+			name: "trigger_sha256",
+			sourceTask: &types.SeedTask{
+				TaskID:           sha256TaskID,
+				URL:              dragonflyURL,
+				TaskURL:          urlutils.FilterURLParam(dragonflyURL, []string{"a", "b"}),
+				SourceFileLength: 9789,
+				CdnFileLength:    0,
+				PieceSize:        100,
+				Header:           map[string]string{"sha256": "b9907b9a5ba2b0223868c201b9addfe2ec1da1b90325d57c34f192966b0a68c5"},
+				CdnStatus:        types.TaskInfoCdnStatusRunning,
+				PieceTotal:       0,
+				RequestDigest:    "sha256:b9907b9a5ba2b0223868c201b9addfe2ec1da1b90325d57c34f192966b0a68c5",
+				SourceRealDigest: "",
+				PieceMd5Sign:     "",
+			},
+			targetTask: &types.SeedTask{
+				TaskID:           sha256TaskID,
+				URL:              dragonflyURL,
+				TaskURL:          urlutils.FilterURLParam(dragonflyURL, []string{"a", "b"}),
+				SourceFileLength: 9789,
+				CdnFileLength:    9789,
+				PieceSize:        100,
+				Header:           map[string]string{"sha256": "b9907b9a5ba2b0223868c201b9addfe2ec1da1b90325d57c34f192966b0a68c5"},
+				CdnStatus:        types.TaskInfoCdnStatusSuccess,
+				PieceTotal:       0,
+				RequestDigest:    "sha256:b9907b9a5ba2b0223868c201b9addfe2ec1da1b90325d57c34f192966b0a68c5",
+				SourceRealDigest: "sha256:b9907b9a5ba2b0223868c201b9addfe2ec1da1b90325d57c34f192966b0a68c5",
+				PieceMd5Sign:     "bb138842f338fff90af737e4a6b2c6f8e2a7031ca9d5900bc9b646f6406d890f",
+			},
+		},
 	}
 
-	targetTask := &types.SeedTask{
-		TaskID:           taskID,
-		URL:              dragonflyURL,
-		TaskURL:          urlutils.FilterURLParam(dragonflyURL, []string{"a", "b"}),
-		SourceFileLength: 9789,
-		CdnFileLength:    9789,
-		PieceSize:        100,
-		Header:           map[string]string{"md5": "f1e2488bba4d1267948d9e2f7008571c"},
-		CdnStatus:        types.TaskInfoCdnStatusSuccess,
-		PieceTotal:       0,
-		RequestMd5:       "f1e2488bba4d1267948d9e2f7008571c",
-		SourceRealMd5:    "f1e2488bba4d1267948d9e2f7008571c",
-		PieceMd5Sign:     "bb138842f338fff90af737e4a6b2c6f8e2a7031ca9d5900bc9b646f6406d890f",
+	for _, tt := range tests {
+		suite.Run(tt.name, func() {
+			gotSeedTask, err := suite.cm.TriggerCDN(context.Background(), tt.sourceTask)
+			suite.Nil(err)
+			suite.Equal(tt.targetTask, gotSeedTask)
+			cacheSeedTask, err := suite.cm.TriggerCDN(context.Background(), gotSeedTask)
+			suite.Nil(err)
+			suite.Equal(tt.targetTask, cacheSeedTask)
+		})
 	}
-	gotSeedTask, err := suite.cm.TriggerCDN(context.Background(), sourceTask)
-	suite.Nil(err)
-	suite.Equal(targetTask, gotSeedTask)
-	cacheSeedTask, err := suite.cm.TriggerCDN(context.Background(), gotSeedTask)
-	suite.Nil(err)
-	suite.Equal(targetTask, cacheSeedTask)
+
 	// TODO test range download
 }
diff --git a/cdnsystem/daemon/task/manager_test.go b/cdnsystem/daemon/task/manager_test.go
index 7530047f..7f19f20e 100644
--- a/cdnsystem/daemon/task/manager_test.go
+++ b/cdnsystem/daemon/task/manager_test.go
@@ -40,7 +40,6 @@ type TaskManagerTestSuite struct {
 
 func (suite *TaskManagerTestSuite) TestRegister() {
 	dragonflyURL := "http://dragonfly.io.com?a=a&b=b&c=c"
-	taskID := idgen.TaskID(dragonflyURL, &base.UrlMeta{Filter: "a&b", Tag: "dragonfly", Digest: "f1e2488bba4d1267948d9e2f7008571c"})
 	ctrl := gomock.NewController(suite.T())
 	cdnMgr := mock.NewMockCDNMgr(ctrl)
 	progressMgr := mock.NewMockSeedProgressMgr(ctrl)
@@ -59,13 +58,28 @@ func (suite *TaskManagerTestSuite) TestRegister() {
 		wantErr       bool
 	}{
 		{
-			name: "register",
+			name: "register_md5",
 			args: args{
 				ctx: context.Background(),
 				req: &types.TaskRegisterRequest{
 					URL:    dragonflyURL,
-					TaskID: taskID,
-					Md5:    "f1e2488bba4d1267948d9e2f7008571c",
+					TaskID: idgen.TaskID(dragonflyURL, &base.UrlMeta{Filter: "a&b", Tag: "dragonfly", Digest: "md5:f1e2488bba4d1267948d9e2f7008571c"}),
+					Digest: "md5:f1e2488bba4d1267948d9e2f7008571c",
+					Filter: []string{"a", "b"},
+					Header: nil,
+				},
+			},
+			wantPieceChan: nil,
+			wantErr:       false,
+		},
+		{
+			name: "register_sha256",
+			args: args{
+				ctx: context.Background(),
+				req: &types.TaskRegisterRequest{
+					URL:    dragonflyURL,
+					TaskID: idgen.TaskID(dragonflyURL, &base.UrlMeta{Filter: "a&b", Tag: "dragonfly", Digest: "sha256:b9907b9a5ba2b0223868c201b9addfe2ec1da1b90325d57c34f192966b0a68c5"}),
+					Digest: "sha256:b9907b9a5ba2b0223868c201b9addfe2ec1da1b90325d57c34f192966b0a68c5",
 					Filter: []string{"a", "b"},
 					Header: nil,
 				},
diff --git a/client/daemon/peer/peertask_file_test.go b/client/daemon/peer/peertask_file_test.go
index cb9ff1b2..9ad34bac 100644
--- a/client/daemon/peer/peertask_file_test.go
+++ b/client/daemon/peer/peertask_file_test.go
@@ -61,7 +61,14 @@ func TestFilePeerTask_BackSource_WithContentLength(t *testing.T) {
 	)
 	defer os.Remove(output)
 
-	schedulerClient, storageManager := setupPeerTaskManagerComponents(ctrl, taskID, int64(mockContentLength), int32(pieceSize), pieceParallelCount)
+	schedulerClient, storageManager := setupPeerTaskManagerComponents(
+		ctrl,
+		componentsOption{
+			taskID:             taskID,
+			contentLength:      int64(mockContentLength),
+			pieceSize:          int32(pieceSize),
+			pieceParallelCount: pieceParallelCount,
+		})
 	defer storageManager.CleanUp()
 
 	downloader := NewMockPieceDownloader(ctrl)
@@ -121,11 +128,12 @@ func TestFilePeerTask_BackSource_WithContentLength(t *testing.T) {
 		ptm.schedulerOption,
 		0)
 	assert.Nil(err, "new file peer task")
-	pt.(*filePeerTask).backSource = true
+	pt.needBackSource = true
 
 	pt.SetCallback(&filePeerTaskCallback{
 		ctx:   ctx,
 		ptm:   ptm,
+		pt:    pt,
 		req:   req,
 		start: time.Now(),
 	})
@@ -171,7 +179,14 @@ func TestFilePeerTask_BackSource_WithoutContentLength(t *testing.T) {
 	)
 	defer os.Remove(output)
 
-	schedulerClient, storageManager := setupPeerTaskManagerComponents(ctrl, taskID, int64(mockContentLength), int32(pieceSize), pieceParallelCount)
+	schedulerClient, storageManager := setupPeerTaskManagerComponents(
+		ctrl,
+		componentsOption{
+			taskID:             taskID,
+			contentLength:      int64(mockContentLength),
+			pieceSize:          int32(pieceSize),
+			pieceParallelCount: pieceParallelCount,
+		})
 	defer storageManager.CleanUp()
 
 	downloader := NewMockPieceDownloader(ctrl)
@@ -232,11 +247,12 @@ func TestFilePeerTask_BackSource_WithoutContentLength(t *testing.T) {
 		ptm.schedulerOption,
 		0)
 	assert.Nil(err, "new file peer task")
-	pt.(*filePeerTask).backSource = true
+	pt.needBackSource = true
 
 	pt.SetCallback(&filePeerTaskCallback{
 		ctx:   ctx,
 		ptm:   ptm,
+		pt:    pt,
 		req:   req,
 		start: time.Now(),
 	})
diff --git a/client/daemon/peer/peertask_manager_test.go b/client/daemon/peer/peertask_manager_test.go
index d105ad4d..92222f24 100644
--- a/client/daemon/peer/peertask_manager_test.go
+++ b/client/daemon/peer/peertask_manager_test.go
@@ -23,6 +23,8 @@ import (
 	"io"
 	"io/ioutil"
 	"math"
+	"net/http"
+	"net/http/httptest"
 	"os"
 	"sync"
 	"testing"
@@ -46,15 +48,21 @@ import (
 	daemonserver "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/server"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	schedulerclient "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
+	"d7y.io/dragonfly/v2/pkg/source"
+	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
 )
 
 var _ daemonserver.DaemonServer = mock_daemon.NewMockDaemonServer(nil)
 
-func setupPeerTaskManagerComponents(
-	ctrl *gomock.Controller,
-	taskID string,
-	contentLength int64,
-	pieceSize, pieceParallelCount int32) (
+type componentsOption struct {
+	taskID             string
+	contentLength      int64
+	pieceSize          int32
+	pieceParallelCount int32
+	peerPacketDelay    []time.Duration
+}
+
+func setupPeerTaskManagerComponents(ctrl *gomock.Controller, opt componentsOption) (
 	schedulerclient.SchedulerClient, storage.Manager) {
 	port := int32(freeport.GetPort())
 	// 1. setup a mock daemon server for uploading pieces info
@@ -62,13 +70,13 @@ func setupPeerTaskManagerComponents(
 	daemon.EXPECT().GetPieceTasks(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(ctx context.Context, request *base.PieceTaskRequest) (*base.PiecePacket, error) {
 		var tasks []*base.PieceInfo
 		for i := int32(0); i < request.Limit; i++ {
-			start := pieceSize * (request.StartNum + i)
-			if int64(start)+1 > contentLength {
+			start := opt.pieceSize * (request.StartNum + i)
+			if int64(start)+1 > opt.contentLength {
 				break
 			}
-			size := pieceSize
-			if int64(start+pieceSize) > contentLength {
-				size = int32(contentLength) - start
+			size := opt.pieceSize
+			if int64(start+opt.pieceSize) > opt.contentLength {
+				size = int32(opt.contentLength) - start
 			}
 			tasks = append(tasks,
 				&base.PieceInfo{
@@ -84,8 +92,8 @@ func setupPeerTaskManagerComponents(
 			TaskId:        request.TaskId,
 			DstPid:        "peer-x",
 			PieceInfos:    tasks,
-			ContentLength: contentLength,
-			TotalPiece:    int32(math.Ceil(float64(contentLength) / float64(pieceSize))),
+			ContentLength: opt.contentLength,
+			TotalPiece:    int32(math.Ceil(float64(opt.contentLength) / float64(opt.pieceSize))),
 		}, nil
 	})
 	ln, _ := rpc.Listen(dfnet.NetAddr{
@@ -101,32 +109,33 @@ func setupPeerTaskManagerComponents(
 		func(pr *scheduler.PieceResult) error {
 			return nil
 		})
-	var ppsent bool
+	var delayCount int
 	pps.EXPECT().Recv().AnyTimes().DoAndReturn(
 		func() (*scheduler.PeerPacket, error) {
-			if !ppsent {
-				ppsent = true
-				return &scheduler.PeerPacket{
-					Code:          dfcodes.Success,
-					TaskId:        taskID,
-					SrcPid:        "127.0.0.1",
-					ParallelCount: pieceParallelCount,
-					MainPeer: &scheduler.PeerPacket_DestPeer{
-						Ip:      "127.0.0.1",
-						RpcPort: port,
-						PeerId:  "peer-x",
-					},
-					StealPeers: nil,
-				}, nil
+			if len(opt.peerPacketDelay) > delayCount {
+				if delay := opt.peerPacketDelay[delayCount]; delay > 0 {
+					time.Sleep(delay)
+				}
+				delayCount++
 			}
-			time.Sleep(time.Hour)
-			return nil, nil
+			return &scheduler.PeerPacket{
+				Code:          dfcodes.Success,
+				TaskId:        opt.taskID,
+				SrcPid:        "127.0.0.1",
+				ParallelCount: opt.pieceParallelCount,
+				MainPeer: &scheduler.PeerPacket_DestPeer{
+					Ip:      "127.0.0.1",
+					RpcPort: port,
+					PeerId:  "peer-x",
+				},
+				StealPeers: nil,
+			}, nil
 		})
 	sched := mock_scheduler.NewMockSchedulerClient(ctrl)
 	sched.EXPECT().RegisterPeerTask(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(
 		func(ctx context.Context, ptr *scheduler.PeerTaskRequest, opts ...grpc.CallOption) (*scheduler.RegisterResult, error) {
 			return &scheduler.RegisterResult{
-				TaskId:      taskID,
+				TaskId:      opt.taskID,
 				SizeScope:   base.SizeScope_NORMAL,
 				DirectPiece: nil,
 			}, nil
@@ -172,7 +181,14 @@ func TestPeerTaskManager_StartFilePeerTask(t *testing.T) {
 	)
 	defer os.Remove(output)
 
-	schedulerClient, storageManager := setupPeerTaskManagerComponents(ctrl, taskID, int64(mockContentLength), int32(pieceSize), pieceParallelCount)
+	schedulerClient, storageManager := setupPeerTaskManagerComponents(
+		ctrl,
+		componentsOption{
+			taskID:             taskID,
+			contentLength:      int64(mockContentLength),
+			pieceSize:          int32(pieceSize),
+			pieceParallelCount: pieceParallelCount,
+		})
 	defer storageManager.CleanUp()
 
 	downloader := NewMockPieceDownloader(ctrl)
@@ -248,7 +264,14 @@ func TestPeerTaskManager_StartStreamPeerTask(t *testing.T) {
 		peerID = "peer-0"
 		taskID = "task-0"
 	)
-	sched, storageManager := setupPeerTaskManagerComponents(ctrl, taskID, int64(mockContentLength), int32(pieceSize), pieceParallelCount)
+	sched, storageManager := setupPeerTaskManagerComponents(
+		ctrl,
+		componentsOption{
+			taskID:             taskID,
+			contentLength:      int64(mockContentLength),
+			pieceSize:          int32(pieceSize),
+			pieceParallelCount: pieceParallelCount,
+		})
 	defer storageManager.CleanUp()
 
 	downloader := NewMockPieceDownloader(ctrl)
@@ -291,3 +314,74 @@ func TestPeerTaskManager_StartStreamPeerTask(t *testing.T) {
 	assert.Nil(err, "load read data")
 	assert.Equal(testBytes, outputBytes, "output and desired output must match")
 }
+
+func TestPeerTaskManager_StartStreamPeerTask_BackSource(t *testing.T) {
+	assert := testifyassert.New(t)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	testBytes, err := ioutil.ReadFile(test.File)
+	assert.Nil(err, "load test file")
+
+	var (
+		pieceParallelCount = int32(4)
+		pieceSize          = 1024
+
+		mockContentLength = len(testBytes)
+		//mockPieceCount    = int(math.Ceil(float64(mockContentLength) / float64(pieceSize)))
+
+		peerID = "peer-0"
+		taskID = "task-0"
+	)
+
+	source.Register("http", httpprotocol.NewHTTPSourceClient())
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		n, err := w.Write(testBytes)
+		assert.Nil(err)
+		assert.Equal(mockContentLength, n)
+	}))
+	defer ts.Close()
+
+	sched, storageManager := setupPeerTaskManagerComponents(
+		ctrl,
+		componentsOption{
+			taskID:             taskID,
+			contentLength:      int64(mockContentLength),
+			pieceSize:          int32(pieceSize),
+			pieceParallelCount: pieceParallelCount,
+			peerPacketDelay:    []time.Duration{time.Second},
+		})
+	defer storageManager.CleanUp()
+
+	ptm := &peerTaskManager{
+		host: &scheduler.PeerHost{
+			Ip: "127.0.0.1",
+		},
+		runningPeerTasks: sync.Map{},
+		pieceManager: &pieceManager{
+			storageManager:   storageManager,
+			pieceDownloader:  NewMockPieceDownloader(ctrl),
+			computePieceSize: computePieceSize,
+		},
+		storageManager:  storageManager,
+		schedulerClient: sched,
+		schedulerOption: config.SchedulerOption{
+			ScheduleTimeout: clientutil.Duration{Duration: time.Nanosecond},
+		},
+	}
+
+	r, _, err := ptm.StartStreamPeerTask(context.Background(), &scheduler.PeerTaskRequest{
+		Url: ts.URL,
+		UrlMeta: &base.UrlMeta{
+			Tag: "d7y-test",
+		},
+		PeerId:   peerID,
+		PeerHost: &scheduler.PeerHost{},
+	})
+	assert.Nil(err, "start stream peer task")
+
+	outputBytes, err := ioutil.ReadAll(r)
+	assert.Nil(err, "load read data")
+	assert.Equal(testBytes, outputBytes, "output and desired output must match")
+}
diff --git a/client/daemon/peer/peertask_stream_test.go b/client/daemon/peer/peertask_stream_test.go
index 1d483147..0ca48289 100644
--- a/client/daemon/peer/peertask_stream_test.go
+++ b/client/daemon/peer/peertask_stream_test.go
@@ -25,18 +25,17 @@ import (
 	"testing"
 	"time"
 
-	"d7y.io/dragonfly/v2/pkg/rpc/base"
-	rangers "d7y.io/dragonfly/v2/pkg/util/rangeutils"
-
-	"d7y.io/dragonfly/v2/pkg/source"
-	sourceMock "d7y.io/dragonfly/v2/pkg/source/mock"
 	"github.com/golang/mock/gomock"
 	testifyassert "github.com/stretchr/testify/assert"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/client/daemon/test"
+	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
+	"d7y.io/dragonfly/v2/pkg/source"
+	sourceMock "d7y.io/dragonfly/v2/pkg/source/mock"
+	rangers "d7y.io/dragonfly/v2/pkg/util/rangeutils"
 )
 
 func TestStreamPeerTask_BackSource_WithContentLength(t *testing.T) {
@@ -58,7 +57,14 @@ func TestStreamPeerTask_BackSource_WithContentLength(t *testing.T) {
 
 		url = "http://localhost/test/data"
 	)
-	schedulerClient, storageManager := setupPeerTaskManagerComponents(ctrl, taskID, int64(mockContentLength), int32(pieceSize), pieceParallelCount)
+	schedulerClient, storageManager := setupPeerTaskManagerComponents(
+		ctrl,
+		componentsOption{
+			taskID:             taskID,
+			contentLength:      int64(mockContentLength),
+			pieceSize:          int32(pieceSize),
+			pieceParallelCount: pieceParallelCount,
+		})
 	defer storageManager.CleanUp()
 
 	downloader := NewMockPieceDownloader(ctrl)
@@ -125,10 +131,11 @@ func TestStreamPeerTask_BackSource_WithContentLength(t *testing.T) {
 	pt.SetCallback(&streamPeerTaskCallback{
 		ctx:   ctx,
 		ptm:   ptm,
+		pt:    pt,
 		req:   req,
 		start: time.Now(),
 	})
-	pt.(*streamPeerTask).backSource = true
+	pt.needBackSource = true
 
 	rc, _, err := pt.Start(ctx)
 	assert.Nil(err, "start stream peer task")
@@ -157,7 +164,14 @@ func TestStreamPeerTask_BackSource_WithoutContentLength(t *testing.T) {
 
 		url = "http://localhost/test/data"
 	)
-	schedulerClient, storageManager := setupPeerTaskManagerComponents(ctrl, taskID, int64(mockContentLength), int32(pieceSize), pieceParallelCount)
+	schedulerClient, storageManager := setupPeerTaskManagerComponents(
+		ctrl,
+		componentsOption{
+			taskID:             taskID,
+			contentLength:      int64(mockContentLength),
+			pieceSize:          int32(pieceSize),
+			pieceParallelCount: pieceParallelCount,
+		})
 	defer storageManager.CleanUp()
 
 	downloader := NewMockPieceDownloader(ctrl)
@@ -224,10 +238,11 @@ func TestStreamPeerTask_BackSource_WithoutContentLength(t *testing.T) {
 	pt.SetCallback(&streamPeerTaskCallback{
 		ctx:   ctx,
 		ptm:   ptm,
+		pt:    pt,
 		req:   req,
 		start: time.Now(),
 	})
-	pt.(*streamPeerTask).backSource = true
+	pt.needBackSource = true
 
 	rc, _, err := pt.Start(ctx)
 	assert.Nil(err, "start stream peer task")
diff --git a/client/daemon/peer/piece_manager_test.go b/client/daemon/peer/piece_manager_test.go
index 63a1b4f3..462ea18d 100644
--- a/client/daemon/peer/piece_manager_test.go
+++ b/client/daemon/peer/piece_manager_test.go
@@ -30,7 +30,6 @@ import (
 	"testing"
 	"time"
 
-	"d7y.io/dragonfly/v2/pkg/source"
 	"github.com/golang/mock/gomock"
 	testifyassert "github.com/stretchr/testify/assert"
 
@@ -42,6 +41,7 @@ import (
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	_ "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/server"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
+	"d7y.io/dragonfly/v2/pkg/source"
 	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
 )
 
diff --git a/client/dfget/dfget_test.go b/client/dfget/dfget_test.go
index 326e1e0f..fd08fb37 100644
--- a/client/dfget/dfget_test.go
+++ b/client/dfget/dfget_test.go
@@ -25,7 +25,6 @@ import (
 	"testing"
 
 	"d7y.io/dragonfly/v2/client/config"
-	"d7y.io/dragonfly/v2/internal/constants"
 	"d7y.io/dragonfly/v2/internal/idgen"
 	"d7y.io/dragonfly/v2/pkg/source"
 	sourcemock "d7y.io/dragonfly/v2/pkg/source/mock"
@@ -49,7 +48,7 @@ func Test_downloadFromSource(t *testing.T) {
 	cfg := &config.DfgetConfig{
 		URL:    "http://a.b.c/xx",
 		Output: output,
-		Digest: strings.Join([]string{constants.Sha256Hash, digestutils.Sha256(content)}, ":"),
+		Digest: strings.Join([]string{digestutils.Sha256Hash.String(), digestutils.Sha256(content)}, ":"),
 	}
 
 	sourceClient.EXPECT().Download(context.Background(), cfg.URL, nil, nil).Return(ioutil.NopCloser(strings.NewReader(content)), nil)
diff --git a/pkg/util/digestutils/digest_test.go b/pkg/util/digestutils/digest_test.go
index aae00a1b..ae131ce9 100644
--- a/pkg/util/digestutils/digest_test.go
+++ b/pkg/util/digestutils/digest_test.go
@@ -22,7 +22,6 @@ import (
 	"syscall"
 	"testing"
 
-	"d7y.io/dragonfly/v2/internal/constants"
 	"d7y.io/dragonfly/v2/pkg/basic"
 	"d7y.io/dragonfly/v2/pkg/util/fileutils"
 	"github.com/google/uuid"
@@ -61,5 +60,5 @@ func TestHashFile(t *testing.T) {
 	f.Write([]byte("hello"))
 	f.Close()
 
-	assert.Equal(t, expected, HashFile(path, constants.Md5Hash))
+	assert.Equal(t, expected, HashFile(path, Md5Hash))
 }
diff --git a/test/e2e/dfget_test.go b/test/e2e/dfget_test.go
index c1add543..ba264a29 100644
--- a/test/e2e/dfget_test.go
+++ b/test/e2e/dfget_test.go
@@ -38,7 +38,7 @@ var _ = Describe("Download with dfget", func() {
 			Expect(err).NotTo(HaveOccurred())
 			fmt.Println(podName)
 			Expect(strings.HasPrefix(podName, "dragonfly-dfdaemon-")).Should(BeTrue())
-			pod := e2eutil.NewPodExec(dragonflyNamespace, podName)
+			pod := e2eutil.NewPodExec(dragonflyNamespace, podName, "dfdaemon")
 
 			for _, v := range e2eutil.GetFileList() {
 				url := e2eutil.GetFileURL(v)
diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go
index 59bd6ed2..2ae37dba 100644
--- a/test/e2e/e2e_test.go
+++ b/test/e2e/e2e_test.go
@@ -17,12 +17,30 @@
 package e2e
 
 import (
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-
+	"strings"
 	"testing"
+
+	"d7y.io/dragonfly/v2/test/e2e/e2eutil"
+	. "github.com/onsi/ginkgo" //nolint
+	. "github.com/onsi/gomega" //nolint
 )
 
+var _ = AfterSuite(func() {
+	out, err := e2eutil.KubeCtlCommand("-n", dragonflyNamespace, "get", "pod", "-l", "component=dfdaemon",
+		"-o", "jsonpath='{range .items[*]}{.metadata.name}{end}'").CombinedOutput()
+	podName := strings.Trim(string(out), "'")
+	Expect(err).NotTo(HaveOccurred())
+	Expect(strings.HasPrefix(podName, "dragonfly-dfdaemon-")).Should(BeTrue())
+
+	e2eutil.KubeCtlCopyCommand(dragonflyNamespace, podName, "/var/log/dragonfly/daemon/core.log", "/tmp/artifact/daemon.log").CombinedOutput()
+	e2eutil.KubeCtlCopyCommand(dragonflyNamespace, "dragonfly-cdn-0", "/var/log/dragonfly/cdn/core.log", "/tmp/artifact/cdn-0.log").CombinedOutput()
+	e2eutil.KubeCtlCopyCommand(dragonflyNamespace, "dragonfly-cdn-1", "/var/log/dragonfly/cdn/core.log", "/tmp/artifact/cdn-1.log").CombinedOutput()
+	e2eutil.KubeCtlCopyCommand(dragonflyNamespace, "dragonfly-cdn-2", "/var/log/dragonfly/cdn/core.log", "/tmp/artifact/cdn-2.log").CombinedOutput()
+	e2eutil.KubeCtlCopyCommand(dragonflyNamespace, "dragonfly-scheduler-0", "/var/log/dragonfly/scheduler/core.log", "/tmp/artifact/scheduler-0.log").CombinedOutput()
+	e2eutil.KubeCtlCopyCommand(dragonflyNamespace, "dragonfly-scheduler-1", "/var/log/dragonfly/scheduler/core.log", "/tmp/artifact/scheduler-1.log").CombinedOutput()
+	e2eutil.KubeCtlCopyCommand(dragonflyNamespace, "dragonfly-scheduler-2", "/var/log/dragonfly/scheduler/core.log", "/tmp/artifact/scheduler-2.log").CombinedOutput()
+})
+
 // TestE2E is the root of e2e test function
 func TestE2E(t *testing.T) {
 	RegisterFailHandler(Fail)
diff --git a/test/e2e/e2eutil/exec.go b/test/e2e/e2eutil/exec.go
index 51085bca..b916ada3 100644
--- a/test/e2e/e2eutil/exec.go
+++ b/test/e2e/e2eutil/exec.go
@@ -31,17 +31,26 @@ func ABCommand(arg ...string) *exec.Cmd {
 type PodExec struct {
 	namespace string
 	name      string
+	container string
 }
 
-func NewPodExec(namespace string, name string) *PodExec {
+func NewPodExec(namespace string, name string, container string) *PodExec {
 	return &PodExec{
 		namespace: namespace,
 		name:      name,
+		container: container,
 	}
 }
 
 func (p *PodExec) Command(arg ...string) *exec.Cmd {
 	extArgs := []string{"-n", p.namespace, "exec", p.name, "--"}
+	if p.container != "" {
+		extArgs = []string{"-n", p.namespace, "exec", "-c", p.container, p.name, "--"}
+	}
 	extArgs = append(extArgs, arg...)
 	return KubeCtlCommand(extArgs...)
 }
+
+func KubeCtlCopyCommand(ns, pod, source, target string) *exec.Cmd {
+	return KubeCtlCommand("-n", ns, "cp", pod+":"+source, target)
+}
diff --git a/test/testdata/charts/config.yaml b/test/testdata/charts/config.yaml
index 07c0645d..27a543e6 100644
--- a/test/testdata/charts/config.yaml
+++ b/test/testdata/charts/config.yaml
@@ -14,8 +14,8 @@ scheduler:
   nameOverride: ""
   fullnameOverride: ""
   replicas: 3
-  image: dragonflyoss/scheduler
-  tag: v0.1.0-beta
+  image: d7yio/scheduler
+  tag: latest
   pullPolicy: IfNotPresent
   resources:
     requests:
@@ -54,10 +54,10 @@ cdn:
   nameOverride: ""
   fullnameOverride: ""
   replicas: 3
-  image: dragonflyoss/cdn
-  tag: v0.1.0-beta
+  image: d7yio/cdn
+  tag: latest
   kubectlImage: bitnami/kubectl
-  pullPolicy: Always
+  pullPolicy: IfNotPresent
   resources:
     requests:
       cpu: "0"
@@ -110,8 +110,8 @@ dfdaemon:
   name: dfdaemon
   nameOverride: ""
   fullnameOverride: ""
-  image: dragonflyoss/dfdaemon
-  tag: v0.1.0-beta
+  image: d7yio/dfdaemon
+  tag: latest
   pullPolicy: IfNotPresent
   resources:
     requests: