Repository: dragonflyoss/Dragonfly2
Pull request #974 (2022-01-05 04:04:30)

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


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

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

```diff
@@            Coverage Diff             @@
##             main     #974      +/-   ##
==========================================
+ Coverage   42.31%   42.47%   +0.15%     
==========================================
  Files         116      116              
  Lines        8798     8798              
==========================================
+ Hits         3723     3737      +14     
+ Misses       4607     4598       -9     
+ Partials      468      463       -5     
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `42.47% <ø> (+0.15%)` | :arrow_up: |

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

| [Impacted Files](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/974?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [client/daemon/peer/peertask\_base.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/974/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2Jhc2UuZ28=) | `46.53% <0.00%> (+0.57%)` | :arrow_up: |
| [client/daemon/peer/piece\_manager.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/974/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BpZWNlX21hbmFnZXIuZ28=) | `63.46% <0.00%> (+0.64%)` | :arrow_up: |
| [client/daemon/peer/peertask\_stream.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/974/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3N0cmVhbS5nbw==) | `55.41% <0.00%> (+1.85%)` | :arrow_up: |
| [client/daemon/peer/peertask\_stream\_callback.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/974/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3N0cmVhbV9jYWxsYmFjay5nbw==) | `64.07% <0.00%> (+1.94%)` | :arrow_up: |
| [pkg/sync/krwmutex.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/974/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3N5bmMva3J3bXV0ZXguZ28=) | `86.36% <0.00%> (+4.54%)` | :arrow_up: |


Patch:
diff --git a/cdn/rpcserver/rpcserver.go b/cdn/rpcserver/rpcserver.go
index ca4eeaae..2a39b258 100644
--- a/cdn/rpcserver/rpcserver.go
+++ b/cdn/rpcserver/rpcserver.go
@@ -106,16 +106,19 @@ func (css *Server) ObtainSeeds(ctx context.Context, req *cdnsystem.SeedRequest,
 			PeerId:   peerID,
 			HostUuid: hostID,
 			PieceInfo: &base.PieceInfo{
-				PieceNum:    int32(piece.PieceNum),
-				RangeStart:  piece.PieceRange.StartIndex,
-				RangeSize:   piece.PieceLen,
-				PieceMd5:    piece.PieceMd5,
-				PieceOffset: piece.OriginRange.StartIndex,
-				PieceStyle:  piece.PieceStyle,
+				PieceNum:     int32(piece.PieceNum),
+				RangeStart:   piece.PieceRange.StartIndex,
+				RangeSize:    piece.PieceLen,
+				PieceMd5:     piece.PieceMd5,
+				PieceOffset:  piece.OriginRange.StartIndex,
+				PieceStyle:   piece.PieceStyle,
+				DownloadCost: piece.DownloadCost,
 			},
 			Done:            false,
 			ContentLength:   registeredTask.SourceFileLength,
 			TotalPieceCount: registeredTask.TotalPieceCount,
+			BeginTime:       piece.BeginDownloadTime,
+			EndTime:         piece.EndDownloadTime,
 		}
 		psc <- pieceSeed
 		jsonPiece, err := json.Marshal(pieceSeed)
@@ -163,7 +166,7 @@ func (css *Server) GetPieceTasks(ctx context.Context, req *base.PieceTaskRequest
 	defer span.End()
 	span.SetAttributes(constants.AttributeGetPieceTasksRequest.String(req.String()))
 	span.SetAttributes(constants.AttributeTaskID.String(req.TaskId))
-	logger.Infof("get piece tasks: %#v", req)
+	logger.Infof("get piece tasks: %s", req)
 	defer func() {
 		if r := recover(); r != nil {
 			err = dferrors.Newf(base.Code_UnknownError, "get task(%s) piece tasks encounter an panic: %v", req.TaskId, r)
@@ -177,7 +180,6 @@ func (css *Server) GetPieceTasks(ctx context.Context, req *base.PieceTaskRequest
 		}
 
 	}()
-	logger.Infof("get piece tasks: %#v", req)
 	seedTask, err := css.service.GetSeedTask(req.TaskId)
 	if err != nil {
 		if task.IsTaskNotFound(err) {
@@ -205,12 +207,13 @@ func (css *Server) GetPieceTasks(ctx context.Context, req *base.PieceTaskRequest
 	for _, piece := range pieces {
 		if piece.PieceNum >= req.StartNum && (count < req.Limit || req.Limit <= 0) {
 			p := &base.PieceInfo{
-				PieceNum:    int32(piece.PieceNum),
-				RangeStart:  piece.PieceRange.StartIndex,
-				RangeSize:   piece.PieceLen,
-				PieceMd5:    piece.PieceMd5,
-				PieceOffset: piece.OriginRange.StartIndex,
-				PieceStyle:  piece.PieceStyle,
+				PieceNum:     int32(piece.PieceNum),
+				RangeStart:   piece.PieceRange.StartIndex,
+				RangeSize:    piece.PieceLen,
+				PieceMd5:     piece.PieceMd5,
+				PieceOffset:  piece.OriginRange.StartIndex,
+				PieceStyle:   piece.PieceStyle,
+				DownloadCost: piece.DownloadCost,
 			}
 			pieceInfos = append(pieceInfos, p)
 			count++
diff --git a/cdn/supervisor/cdn/cache_writer.go b/cdn/supervisor/cdn/cache_writer.go
index 44857fba..aff5c5f0 100644
--- a/cdn/supervisor/cdn/cache_writer.go
+++ b/cdn/supervisor/cdn/cache_writer.go
@@ -23,6 +23,7 @@ import (
 	"encoding/json"
 	"io"
 	"sync"
+	"time"
 
 	"github.com/pkg/errors"
 	"go.opentelemetry.io/otel/trace"
@@ -39,10 +40,13 @@ import (
 )
 
 type piece struct {
-	taskID       string
-	pieceNum     uint32
-	pieceSize    uint32
-	pieceContent *bytes.Buffer
+	taskID            string
+	pieceNum          uint32
+	pieceSize         uint32
+	pieceContent      *bytes.Buffer
+	downloadCost      uint64
+	beginDownloadTime uint64
+	endDownloadTime   uint64
 }
 
 type downloadMetadata struct {
@@ -126,6 +130,7 @@ loop:
 		case <-writeCtx.Done():
 			break loop
 		default:
+			start := time.Now()
 			var bb = bufPool.Get().(*bytes.Buffer)
 			bb.Reset()
 			limitReader := io.LimitReader(reader, int64(seedTask.PieceSize))
@@ -139,12 +144,15 @@ loop:
 				break loop
 			}
 			backSourceLength += n
-
+			end := time.Now()
 			jobCh <- &piece{
-				taskID:       seedTask.ID,
-				pieceNum:     uint32(curPieceNum),
-				pieceSize:    uint32(seedTask.PieceSize),
-				pieceContent: bb,
+				taskID:            seedTask.ID,
+				pieceNum:          uint32(curPieceNum),
+				pieceSize:         uint32(seedTask.PieceSize),
+				pieceContent:      bb,
+				downloadCost:      uint64(end.Sub(start).Nanoseconds()),
+				beginDownloadTime: uint64(start.UnixNano()),
+				endDownloadTime:   uint64(end.UnixNano()),
 			}
 			curPieceNum++
 			if n < int64(seedTask.PieceSize) {
@@ -197,7 +205,10 @@ func (cw *cacheWriter) writerPool(ctx context.Context, g *errgroup.Group, routin
 							StartIndex: start,
 							EndIndex:   end,
 						},
-						PieceStyle: pieceStyle,
+						PieceStyle:        pieceStyle,
+						DownloadCost:      p.downloadCost,
+						BeginDownloadTime: p.beginDownloadTime,
+						EndDownloadTime:   p.endDownloadTime,
 					}
 					// write piece meta to storage
 					if err = cw.metadataManager.appendPieceMetadata(p.taskID, pieceRecord); err != nil {
diff --git a/cdn/supervisor/cdn/reporter.go b/cdn/supervisor/cdn/reporter.go
index 308996f6..89230564 100644
--- a/cdn/supervisor/cdn/reporter.go
+++ b/cdn/supervisor/cdn/reporter.go
@@ -74,11 +74,14 @@ func (re *reporter) reportPieceMetaRecord(ctx context.Context, taskID string, re
 */
 func convertPieceMeta2SeedPiece(record *storage.PieceMetaRecord) *task.PieceInfo {
 	return &task.PieceInfo{
-		PieceStyle:  base.PieceStyle(record.PieceStyle),
-		PieceNum:    record.PieceNum,
-		PieceMd5:    record.Md5,
-		PieceRange:  record.Range,
-		OriginRange: record.OriginRange,
-		PieceLen:    record.PieceLen,
+		PieceNum:          record.PieceNum,
+		PieceMd5:          record.Md5,
+		PieceRange:        record.Range,
+		OriginRange:       record.OriginRange,
+		PieceLen:          record.PieceLen,
+		PieceStyle:        base.PieceStyle(record.PieceStyle),
+		DownloadCost:      record.DownloadCost,
+		BeginDownloadTime: record.BeginDownloadTime,
+		EndDownloadTime:   record.EndDownloadTime,
 	}
 }
diff --git a/cdn/supervisor/cdn/storage/storage.go b/cdn/supervisor/cdn/storage/storage.go
index 4e9c3001..664326a0 100644
--- a/cdn/supervisor/cdn/storage/storage.go
+++ b/cdn/supervisor/cdn/storage/storage.go
@@ -111,13 +111,18 @@ type PieceMetaRecord struct {
 	OriginRange *rangeutils.Range `json:"originRange"`
 	// 0: PlainUnspecified
 	PieceStyle int32 `json:"pieceStyle"`
+	// total time(millisecond) consumed
+	DownloadCost      uint64 `json:"downloadCost"`
+	BeginDownloadTime uint64 `json:"beginDownloadTime"`
+	EndDownloadTime   uint64 `json:"endDownloadTime"`
 }
 
 const fieldSeparator = ":"
 
 func (record PieceMetaRecord) String() string {
 	return fmt.Sprint(record.PieceNum, fieldSeparator, record.PieceLen, fieldSeparator, record.Md5, fieldSeparator, record.Range, fieldSeparator,
-		record.OriginRange, fieldSeparator, record.PieceStyle)
+		record.OriginRange, fieldSeparator, record.PieceStyle, fieldSeparator, record.DownloadCost, fieldSeparator, record.BeginDownloadTime, fieldSeparator,
+		record.EndDownloadTime)
 }
 
 func ParsePieceMetaRecord(value string) (record *PieceMetaRecord, err error) {
@@ -148,13 +153,28 @@ func ParsePieceMetaRecord(value string) (record *PieceMetaRecord, err error) {
 	if err != nil {
 		return nil, errors.Wrapf(err, "invalid pieceStyle: %s", fields[5])
 	}
+	downloadCost, err := strconv.ParseUint(fields[6], 10, 64)
+	if err != nil {
+		return nil, errors.Wrapf(err, "invalid download cost: %s", fields[6])
+	}
+	beginDownloadTime, err := strconv.ParseUint(fields[7], 10, 64)
+	if err != nil {
+		return nil, errors.Wrapf(err, "invalid begin download time: %s", fields[7])
+	}
+	endDownloadTime, err := strconv.ParseUint(fields[8], 10, 64)
+	if err != nil {
+		return nil, errors.Wrapf(err, "invalid end download time: %s", fields[8])
+	}
 	return &PieceMetaRecord{
-		PieceNum:    uint32(pieceNum),
-		PieceLen:    uint32(pieceLen),
-		Md5:         md5,
-		Range:       pieceRange,
-		OriginRange: originRange,
-		PieceStyle:  int32(pieceStyle),
+		PieceNum:          uint32(pieceNum),
+		PieceLen:          uint32(pieceLen),
+		Md5:               md5,
+		Range:             pieceRange,
+		OriginRange:       originRange,
+		PieceStyle:        int32(pieceStyle),
+		DownloadCost:      downloadCost,
+		BeginDownloadTime: beginDownloadTime,
+		EndDownloadTime:   endDownloadTime,
 	}, nil
 }
 
diff --git a/cdn/supervisor/task/task.go b/cdn/supervisor/task/task.go
index 2f5f98f3..c60d4a9f 100644
--- a/cdn/supervisor/task/task.go
+++ b/cdn/supervisor/task/task.go
@@ -89,12 +89,15 @@ type SeedTask struct {
 }
 
 type PieceInfo struct {
-	PieceNum    uint32            `json:"piece_num"`
-	PieceMd5    string            `json:"piece_md5"`
-	PieceRange  *rangeutils.Range `json:"piece_range"`
-	OriginRange *rangeutils.Range `json:"origin_range"`
-	PieceLen    uint32            `json:"piece_len"`
-	PieceStyle  base.PieceStyle   `json:"piece_style"`
+	PieceNum          uint32            `json:"piece_num"`
+	PieceMd5          string            `json:"piece_md5"`
+	PieceRange        *rangeutils.Range `json:"piece_range"`
+	OriginRange       *rangeutils.Range `json:"origin_range"`
+	PieceLen          uint32            `json:"piece_len"`
+	PieceStyle        base.PieceStyle   `json:"piece_style"`
+	DownloadCost      uint64            `json:"download_cost"`
+	BeginDownloadTime uint64            `json:"begin_download_time"`
+	EndDownloadTime   uint64            `json:"end_download_time"`
 }
 
 const (
diff --git a/client/config/peerhost.go b/client/config/peerhost.go
index dba37d1a..be187a41 100644
--- a/client/config/peerhost.go
+++ b/client/config/peerhost.go
@@ -64,6 +64,9 @@ type DaemonOption struct {
 	Upload       UploadOption    `mapstructure:"upload" yaml:"upload"`
 	Storage      StorageOption   `mapstructure:"storage" yaml:"storage"`
 	ConfigServer string          `mapstructure:"configServer" yaml:"configServer"`
+	Health       *HealthOption   `mapstructure:"health" yaml:"health"`
+	// TODO WIP, did not use
+	Reload ReloadOption `mapstructure:"reloadOption" yaml:"reloadOption"`
 }
 
 func NewDaemonConfig() *DaemonOption {
@@ -196,7 +199,7 @@ type ProxyOption struct {
 	MaxConcurrency  int64           `mapstructure:"maxConcurrency" yaml:"maxConcurrency"`
 	RegistryMirror  *RegistryMirror `mapstructure:"registryMirror" yaml:"registryMirror"`
 	WhiteList       []*WhiteList    `mapstructure:"whiteList" yaml:"whiteList"`
-	Proxies         []*Proxy        `mapstructure:"proxies" yaml:"proxies"`
+	Proxies         []*ProxyRule    `mapstructure:"proxies" yaml:"proxies"`
 	HijackHTTPS     *HijackConfig   `mapstructure:"hijackHTTPS" yaml:"hijackHTTPS"`
 	DumpHTTPContent bool            `mapstructure:"dumpHTTPContent" yaml:"dumpHTTPContent"`
 	// ExtraRegistryMirrors add more mirror for different ports
@@ -283,7 +286,7 @@ func (p *ProxyOption) unmarshal(unmarshal func(in []byte, out interface{}) (err
 		MaxConcurrency  int64           `mapstructure:"maxConcurrency" yaml:"maxConcurrency"`
 		RegistryMirror  *RegistryMirror `mapstructure:"registryMirror" yaml:"registryMirror"`
 		WhiteList       []*WhiteList    `mapstructure:"whiteList" yaml:"whiteList"`
-		Proxies         []*Proxy        `mapstructure:"proxies" yaml:"proxies"`
+		Proxies         []*ProxyRule    `mapstructure:"proxies" yaml:"proxies"`
 		HijackHTTPS     *HijackConfig   `mapstructure:"hijackHTTPS" yaml:"hijackHTTPS"`
 		DumpHTTPContent bool            `mapstructure:"dumpHTTPContent" yaml:"dumpHTTPContent"`
 	}{}
@@ -444,6 +447,15 @@ type StorageOption struct {
 
 type StoreStrategy string
 
+type HealthOption struct {
+	ListenOption `yaml:",inline" mapstructure:",squash"`
+	Path         string `mapstructure:"path" yaml:"pash"`
+}
+
+type ReloadOption struct {
+	Interval clientutil.Duration
+}
+
 type FileString string
 
 func (f *FileString) UnmarshalJSON(b []byte) error {
@@ -650,8 +662,8 @@ func certPoolFromFiles(files ...string) (*x509.CertPool, error) {
 	return roots, nil
 }
 
-// Proxy describes a regular expression matching rule for how to proxy a request.
-type Proxy struct {
+// ProxyRule describes a regular expression matching rule for how to proxy a request.
+type ProxyRule struct {
 	Regx     *Regexp `yaml:"regx" mapstructure:"regx"`
 	UseHTTPS bool    `yaml:"useHTTPS" mapstructure:"useHTTPS"`
 	Direct   bool    `yaml:"direct" mapstructure:"direct"`
@@ -660,13 +672,13 @@ type Proxy struct {
 	Redirect string `yaml:"redirect" mapstructure:"redirect"`
 }
 
-func NewProxy(regx string, useHTTPS bool, direct bool, redirect string) (*Proxy, error) {
+func NewProxyRule(regx string, useHTTPS bool, direct bool, redirect string) (*ProxyRule, error) {
 	exp, err := NewRegexp(regx)
 	if err != nil {
 		return nil, errors.Wrap(err, "invalid regexp")
 	}
 
-	return &Proxy{
+	return &ProxyRule{
 		Regx:     exp,
 		UseHTTPS: useHTTPS,
 		Direct:   direct,
@@ -675,7 +687,7 @@ func NewProxy(regx string, useHTTPS bool, direct bool, redirect string) (*Proxy,
 }
 
 // Match checks if the given url matches the rule.
-func (r *Proxy) Match(url string) bool {
+func (r *ProxyRule) Match(url string) bool {
 	return r.Regx != nil && r.Regx.MatchString(url)
 }
 
diff --git a/client/config/peerhost_darwin.go b/client/config/peerhost_darwin.go
index 45f8e4aa..7a295faa 100644
--- a/client/config/peerhost_darwin.go
+++ b/client/config/peerhost_darwin.go
@@ -124,4 +124,9 @@ var peerHostConfig = DaemonOption{
 		Multiplex:              false,
 		DiskGCThresholdPercent: 95,
 	},
+	Reload: ReloadOption{
+		Interval: clientutil.Duration{
+			Duration: time.Minute,
+		},
+	},
 }
diff --git a/client/config/peerhost_linux.go b/client/config/peerhost_linux.go
index d0fc90a2..5d65e120 100644
--- a/client/config/peerhost_linux.go
+++ b/client/config/peerhost_linux.go
@@ -123,4 +123,9 @@ var peerHostConfig = DaemonOption{
 		Multiplex:              false,
 		DiskGCThresholdPercent: 95,
 	},
+	Reload: ReloadOption{
+		Interval: clientutil.Duration{
+			Duration: time.Minute,
+		},
+	},
 }
diff --git a/client/daemon/daemon.go b/client/daemon/daemon.go
index 10930548..bb47b24d 100644
--- a/client/daemon/daemon.go
+++ b/client/daemon/daemon.go
@@ -29,6 +29,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/gorilla/mux"
 	"github.com/pkg/errors"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"golang.org/x/sync/errgroup"
@@ -498,6 +499,30 @@ func (cd *clientDaemon) Serve() error {
 		}()
 	}
 
+	if cd.Option.Health != nil {
+		if cd.Option.Health.ListenOption.TCPListen == nil {
+			logger.Fatalf("health listen not found")
+		}
+		logger.Infof("serve http health at %#v", cd.Option.Health.ListenOption.TCPListen)
+		r := mux.NewRouter()
+		r.Path(cd.Option.Health.Path).HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+			writer.WriteHeader(http.StatusOK)
+			_, _ = writer.Write([]byte("success"))
+		})
+		listener, _, err := cd.prepareTCPListener(cd.Option.Health.ListenOption, false)
+		if err != nil {
+			logger.Fatalf("init health http server error: %v", err)
+		}
+		go func() {
+			if err = http.Serve(listener, r); err != nil {
+				if err == http.ErrServerClosed {
+					return
+				}
+				logger.Errorf("health http server error: %v", err)
+			}
+		}()
+	}
+
 	werr := g.Wait()
 	cd.Stop()
 	return werr
diff --git a/client/daemon/peer/peertask_file.go b/client/daemon/peer/peertask_file.go
index 01e13230..6f49fcc3 100644
--- a/client/daemon/peer/peertask_file.go
+++ b/client/daemon/peer/peertask_file.go
@@ -90,7 +90,7 @@ func (ptm *peerTaskManager) newFileTask(
 
 	// prefetch parent request
 	var parent *peerTaskConductor
-	if ptm.enablePrefetch && request.Range != nil {
+	if ptm.enabledPrefetch(request.Range) {
 		parent = ptm.prefetchParentTask(&request.PeerTaskRequest, request.Output)
 	}
 
diff --git a/client/daemon/peer/peertask_manager.go b/client/daemon/peer/peertask_manager.go
index 0f824b45..a896cd86 100644
--- a/client/daemon/peer/peertask_manager.go
+++ b/client/daemon/peer/peertask_manager.go
@@ -212,6 +212,10 @@ func (ptm *peerTaskManager) getOrCreatePeerTaskConductor(
 	return ptc, true, ptc.initStorage(desiredLocation)
 }
 
+func (ptm *peerTaskManager) enabledPrefetch(rg *clientutil.Range) bool {
+	return ptm.enablePrefetch && rg != nil
+}
+
 func (ptm *peerTaskManager) prefetchParentTask(request *scheduler.PeerTaskRequest, desiredLocation string) *peerTaskConductor {
 	req := &scheduler.PeerTaskRequest{
 		Url:         request.Url,
diff --git a/client/daemon/peer/peertask_reuse.go b/client/daemon/peer/peertask_reuse.go
index ef7bb995..330f77b9 100644
--- a/client/daemon/peer/peertask_reuse.go
+++ b/client/daemon/peer/peertask_reuse.go
@@ -37,31 +37,45 @@ import (
 
 var _ *logger.SugaredLoggerOnWith // pin this package for no log code generation
 
+// reuse task search logic:
+// A. prefetch feature enabled
+//    for ranged request, 1, find completed subtask, 2, find partial completed parent task
+//    for non-ranged request, just find completed task
+// B. prefetch feature disabled
+//    for ranged request, 1, find completed normal task, 2, find partial completed parent task
+//    for non-ranged request, just find completed task
+
 func (ptm *peerTaskManager) tryReuseFilePeerTask(ctx context.Context,
 	request *FileTaskRequest) (chan *FileTaskProgress, bool) {
 	taskID := idgen.TaskID(request.Url, request.UrlMeta)
-	var reuse *storage.ReusePeerTask
-	if ptm.enablePrefetch && request.Range != nil {
+	var (
+		reuse      *storage.ReusePeerTask
+		reuseRange *clientutil.Range // the range of parent peer task data to read
+		log        *logger.SugaredLoggerOnWith
+		length     int64
+		err        error
+	)
+
+	if ptm.enabledPrefetch(request.Range) {
 		reuse = ptm.storageManager.FindCompletedSubTask(taskID)
 	} else {
 		reuse = ptm.storageManager.FindCompletedTask(taskID)
 	}
-	var (
-		rg     *clientutil.Range // the range of parent peer task data to read
-		log    *logger.SugaredLoggerOnWith
-		length int64
-		err    error
-	)
+
 	if reuse == nil {
+		if request.Range == nil {
+			return nil, false
+		}
+		// for ranged request, check the parent task
+		reuseRange = request.Range
 		taskID = idgen.ParentTaskID(request.Url, request.UrlMeta)
-		reuse = ptm.storageManager.FindCompletedTask(taskID)
+		reuse = ptm.storageManager.FindPartialCompletedTask(taskID, reuseRange)
 		if reuse == nil {
 			return nil, false
 		}
-		rg = request.Range
 	}
 
-	if rg == nil {
+	if reuseRange == nil {
 		log = logger.With("peer", request.PeerId, "task", taskID, "component", "reuseFilePeerTask")
 		log.Infof("reuse from peer task: %s, total size: %d", reuse.PeerID, reuse.ContentLength)
 		length = reuse.ContentLength
@@ -70,7 +84,7 @@ func (ptm *peerTaskManager) tryReuseFilePeerTask(ctx context.Context,
 			"component", "reuseRangeFilePeerTask")
 		log.Infof("reuse partial data from peer task: %s, total size: %d, range: %s",
 			reuse.PeerID, reuse.ContentLength, request.UrlMeta.Range)
-		length = rg.Length
+		length = reuseRange.Length
 	}
 
 	_, span := tracer.Start(ctx, config.SpanReusePeerTask, trace.WithSpanKind(trace.SpanKindClient))
@@ -80,7 +94,7 @@ func (ptm *peerTaskManager) tryReuseFilePeerTask(ctx context.Context,
 	span.SetAttributes(config.AttributePeerID.String(request.PeerId))
 	span.SetAttributes(config.AttributeReusePeerID.String(reuse.PeerID))
 	span.SetAttributes(semconv.HTTPURLKey.String(request.Url))
-	if rg != nil {
+	if reuseRange != nil {
 		span.SetAttributes(config.AttributeReuseRange.String(request.UrlMeta.Range))
 	}
 	defer span.End()
@@ -89,7 +103,7 @@ func (ptm *peerTaskManager) tryReuseFilePeerTask(ctx context.Context,
 	span.AddEvent("reuse peer task", trace.WithAttributes(config.AttributePeerID.String(reuse.PeerID)))
 
 	start := time.Now()
-	if rg == nil || request.KeepOriginalOffset {
+	if reuseRange == nil || request.KeepOriginalOffset {
 		storeRequest := &storage.StoreRequest{
 			CommonTaskRequest: storage.CommonTaskRequest{
 				PeerID:      reuse.PeerID,
@@ -103,7 +117,7 @@ func (ptm *peerTaskManager) tryReuseFilePeerTask(ctx context.Context,
 		}
 		err = ptm.storageManager.Store(ctx, storeRequest)
 	} else {
-		err = ptm.storePartialFile(ctx, request, log, reuse, rg)
+		err = ptm.storePartialFile(ctx, request, log, reuse, reuseRange)
 	}
 
 	if err != nil {
@@ -168,36 +182,41 @@ func (ptm *peerTaskManager) storePartialFile(ctx context.Context, request *FileT
 func (ptm *peerTaskManager) tryReuseStreamPeerTask(ctx context.Context,
 	request *StreamTaskRequest) (io.ReadCloser, map[string]string, bool) {
 	taskID := idgen.TaskID(request.URL, request.URLMeta)
-	var reuse *storage.ReusePeerTask
-	if ptm.enablePrefetch && request.Range != nil {
+	var (
+		reuse      *storage.ReusePeerTask
+		reuseRange *clientutil.Range // the range of parent peer task data to read
+		log        *logger.SugaredLoggerOnWith
+		length     int64
+	)
+
+	if ptm.enabledPrefetch(request.Range) {
 		reuse = ptm.storageManager.FindCompletedSubTask(taskID)
 	} else {
 		reuse = ptm.storageManager.FindCompletedTask(taskID)
 	}
-	var (
-		rg  *clientutil.Range // the range of parent peer task data to read
-		log *logger.SugaredLoggerOnWith
-	)
+
 	if reuse == nil {
-		// for ranged request, check the parent task
 		if request.Range == nil {
 			return nil, nil, false
 		}
+		// for ranged request, check the parent task
+		reuseRange = request.Range
 		taskID = idgen.ParentTaskID(request.URL, request.URLMeta)
-		reuse = ptm.storageManager.FindCompletedTask(taskID)
+		reuse = ptm.storageManager.FindPartialCompletedTask(taskID, reuseRange)
 		if reuse == nil {
 			return nil, nil, false
 		}
-		rg = request.Range
 	}
 
-	if rg == nil {
+	if reuseRange == nil {
 		log = logger.With("peer", request.PeerID, "task", taskID, "component", "reuseStreamPeerTask")
 		log.Infof("reuse from peer task: %s, total size: %d", reuse.PeerID, reuse.ContentLength)
+		length = reuse.ContentLength
 	} else {
 		log = logger.With("peer", request.PeerID, "task", taskID, "component", "reuseRangeStreamPeerTask")
 		log.Infof("reuse partial data from peer task: %s, total size: %d, range: %s",
 			reuse.PeerID, reuse.ContentLength, request.URLMeta.Range)
+		length = reuseRange.Length
 	}
 
 	ctx, span := tracer.Start(ctx, config.SpanStreamTask, trace.WithSpanKind(trace.SpanKindClient))
@@ -207,13 +226,13 @@ func (ptm *peerTaskManager) tryReuseStreamPeerTask(ctx context.Context,
 	span.SetAttributes(config.AttributePeerID.String(request.PeerID))
 	span.SetAttributes(config.AttributeReusePeerID.String(reuse.PeerID))
 	span.SetAttributes(semconv.HTTPURLKey.String(request.URL))
-	if rg != nil {
+	if reuseRange != nil {
 		span.SetAttributes(config.AttributeReuseRange.String(request.URLMeta.Range))
 	}
 	defer span.End()
 
 	rc, err := ptm.storageManager.ReadAllPieces(ctx,
-		&storage.ReadAllPiecesRequest{PeerTaskMetadata: reuse.PeerTaskMetadata, Range: rg})
+		&storage.ReadAllPiecesRequest{PeerTaskMetadata: reuse.PeerTaskMetadata, Range: reuseRange})
 	if err != nil {
 		log.Errorf("read pieces error when reuse peer task: %s", err)
 		span.SetAttributes(config.AttributePeerTaskSuccess.Bool(false))
@@ -224,12 +243,20 @@ func (ptm *peerTaskManager) tryReuseStreamPeerTask(ctx context.Context,
 	attr := map[string]string{}
 	attr[config.HeaderDragonflyTask] = taskID
 	attr[config.HeaderDragonflyPeer] = request.PeerID
-	if rg != nil {
+	attr[headers.ContentLength] = fmt.Sprintf("%d", length)
+
+	if reuseRange != nil {
 		attr[config.HeaderDragonflyRange] = request.URLMeta.Range
-		attr[headers.ContentRange] = fmt.Sprintf("bytes %d-%d/%d", rg.Start, rg.Start+rg.Length-1, reuse.ContentLength)
-		attr[headers.ContentLength] = fmt.Sprintf("%d", rg.Length)
-	} else {
-		attr[headers.ContentLength] = fmt.Sprintf("%d", reuse.ContentLength)
+		attr[headers.ContentRange] = fmt.Sprintf("bytes %d-%d/%d", reuseRange.Start,
+			reuseRange.Start+reuseRange.Length-1, reuse.ContentLength)
+	} else if request.Range != nil {
+		// the length is from reuse task, ensure it equal with request
+		if length != request.Range.Length {
+			log.Errorf("target task length %d did not match range length %d", length, request.Range.Length)
+			return nil, nil, false
+		}
+		attr[headers.ContentRange] = fmt.Sprintf("bytes %d-%d/*", request.Range.Start,
+			request.Range.Start+request.Range.Length-1)
 	}
 
 	// TODO record time when file closed, need add a type to implement Close and WriteTo
diff --git a/client/daemon/peer/peertask_stream.go b/client/daemon/peer/peertask_stream.go
index 6ffc764a..099a76bf 100644
--- a/client/daemon/peer/peertask_stream.go
+++ b/client/daemon/peer/peertask_stream.go
@@ -74,7 +74,7 @@ func (ptm *peerTaskManager) newStreamTask(
 
 	// prefetch parent request
 	var parent *peerTaskConductor
-	if ptm.enablePrefetch && rg != nil {
+	if ptm.enabledPrefetch(rg) {
 		parent = ptm.prefetchParentTask(request, "")
 	}
 
diff --git a/client/daemon/proxy/proxy.go b/client/daemon/proxy/proxy.go
index e46a5c78..7d80467d 100644
--- a/client/daemon/proxy/proxy.go
+++ b/client/daemon/proxy/proxy.go
@@ -65,7 +65,7 @@ type Proxy struct {
 	registry *config.RegistryMirror
 
 	// proxy rules
-	rules []*config.Proxy
+	rules []*config.ProxyRule
 
 	// httpsHosts is the list of hosts whose https requests will be hijacked
 	httpsHosts []*config.HijackHost
@@ -166,7 +166,7 @@ func WithDirectHandler(h *http.ServeMux) Option {
 }
 
 // WithRules sets the proxy rules
-func WithRules(rules []*config.Proxy) Option {
+func WithRules(rules []*config.ProxyRule) Option {
 	return func(p *Proxy) *Proxy {
 		p.rules = rules
 		return p
@@ -511,7 +511,7 @@ func (proxy *Proxy) remoteConfig(host string) *tls.Config {
 }
 
 // setRules changes the rule lists of the proxy to the given rules.
-func (proxy *Proxy) setRules(rules []*config.Proxy) error {
+func (proxy *Proxy) setRules(rules []*config.ProxyRule) error {
 	proxy.rules = rules
 	return nil
 }
diff --git a/client/daemon/storage/local_storage.go b/client/daemon/storage/local_storage.go
index ee2f9907..58d13bf3 100644
--- a/client/daemon/storage/local_storage.go
+++ b/client/daemon/storage/local_storage.go
@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"math"
 	"os"
 	"path"
 	"sync"
@@ -29,7 +30,9 @@ import (
 
 	"go.uber.org/atomic"
 
+	"d7y.io/dragonfly/v2/client/clientutil"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
+	"d7y.io/dragonfly/v2/internal/util"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/util/digestutils"
 )
@@ -121,7 +124,7 @@ func (t *localTaskStore) WritePiece(ctx context.Context, req *WritePieceRequest)
 
 	n, err := io.Copy(file, io.LimitReader(req.Reader, req.Range.Length))
 	if err != nil {
-		return 0, err
+		return n, err
 	}
 
 	// when UnknownLength and size is align to piece num
@@ -543,6 +546,29 @@ func (t *localTaskStore) saveMetadata() error {
 	return err
 }
 
+func (t *localTaskStore) partialCompleted(rg *clientutil.Range) bool {
+	t.RLock()
+	defer t.RUnlock()
+
+	if t.ContentLength == -1 {
+		return false
+	}
+	start, end := computePiecePosition(t.ContentLength, rg, util.ComputePieceSize)
+	for i := start; i <= end; i++ {
+		if _, ok := t.Pieces[i]; !ok {
+			return false
+		}
+	}
+	return true
+}
+
+func computePiecePosition(total int64, rg *clientutil.Range, compute func(length int64) uint32) (start, end int32) {
+	pieceSize := compute(total)
+	start = int32(math.Floor(float64(rg.Start) / float64(pieceSize)))
+	end = int32(math.Floor(float64(rg.Start+rg.Length-1) / float64(pieceSize)))
+	return
+}
+
 // limitedReadFile implements io optimize for zero copy
 type limitedReadFile struct {
 	reader io.Reader
diff --git a/client/daemon/storage/storage_manager.go b/client/daemon/storage/storage_manager.go
index 29c7f7d1..e9f2212c 100644
--- a/client/daemon/storage/storage_manager.go
+++ b/client/daemon/storage/storage_manager.go
@@ -91,6 +91,8 @@ type Manager interface {
 	FindCompletedTask(taskID string) *ReusePeerTask
 	// FindCompletedSubTask try to find a completed subtask for fast path
 	FindCompletedSubTask(taskID string) *ReusePeerTask
+	// FindPartialCompletedTask try to find a partial completed task for fast path
+	FindPartialCompletedTask(taskID string, rg *clientutil.Range) *ReusePeerTask
 	// CleanUp cleans all storage data
 	CleanUp()
 }
@@ -456,16 +458,47 @@ func (s *storageManager) FindCompletedTask(taskID string) *ReusePeerTask {
 			continue
 		}
 
-		if !t.Done {
+		if t.Done {
+			return &ReusePeerTask{
+				PeerTaskMetadata: PeerTaskMetadata{
+					PeerID: t.PeerID,
+					TaskID: taskID,
+				},
+				ContentLength: t.ContentLength,
+				TotalPieces:   t.TotalPieces,
+			}
+		}
+	}
+	return nil
+}
+
+func (s *storageManager) FindPartialCompletedTask(taskID string, rg *clientutil.Range) *ReusePeerTask {
+	s.indexRWMutex.RLock()
+	defer s.indexRWMutex.RUnlock()
+	ts, ok := s.indexTask2PeerTask[taskID]
+	if !ok {
+		return nil
+	}
+	for _, t := range ts {
+		if t.invalid.Load() {
 			continue
 		}
-		return &ReusePeerTask{
-			PeerTaskMetadata: PeerTaskMetadata{
-				PeerID: t.PeerID,
-				TaskID: taskID,
-			},
-			ContentLength: t.ContentLength,
-			TotalPieces:   int32(t.TotalPieces),
+		// touch it before marking reclaim
+		t.touch()
+		// already marked, skip
+		if t.reclaimMarked.Load() {
+			continue
+		}
+
+		if t.Done || t.partialCompleted(rg) {
+			return &ReusePeerTask{
+				PeerTaskMetadata: PeerTaskMetadata{
+					PeerID: t.PeerID,
+					TaskID: taskID,
+				},
+				ContentLength: t.ContentLength,
+				TotalPieces:   t.TotalPieces,
+			}
 		}
 	}
 	return nil
diff --git a/deploy/docker-compose/template/scheduler.template.yaml b/deploy/docker-compose/template/scheduler.template.yaml
index c91fed70..1da64d69 100644
--- a/deploy/docker-compose/template/scheduler.template.yaml
+++ b/deploy/docker-compose/template/scheduler.template.yaml
@@ -6,8 +6,6 @@ server:
   # host: localhost
   # port is the ip and port scheduler server listens on.
   port: 8002
-  # limit the number of requests
-  listenLimit: 10000
   # cacheDir is dynconfig cache storage directory
   # in linux, default value is /var/cache/dragonfly
   # in macos(just for testing), default value is /Users/$USER/.dragonfly/cache
@@ -127,4 +125,4 @@ verbose: true
 pprof-port: -1
 
 # jaeger endpoint url, like: http://jaeger.dragonfly.svc:14268/api/traces
-jaeger: ""
\ No newline at end of file
+jaeger: ""
diff --git a/deploy/helm-charts b/deploy/helm-charts
index 5ee742a4..00a2ec43 160000
--- a/deploy/helm-charts
+++ b/deploy/helm-charts
@@ -1,1 +1,1 @@
-Subproject commit 5ee742a49f4d3da0990acdfe25c2aa1a0458f7e2
+Subproject commit 00a2ec437075a00dc752358064f4f6fe6f28c382
diff --git a/docs/en/deployment/configuration/scheduler.yaml b/docs/en/deployment/configuration/scheduler.yaml
index 85d62acc..61ebdf47 100644
--- a/docs/en/deployment/configuration/scheduler.yaml
+++ b/docs/en/deployment/configuration/scheduler.yaml
@@ -6,8 +6,6 @@ server:
   # host: localhost
   # port is the ip and port scheduler server listens on.
   port: 8002
-  # limit the number of requests
-  listenLimit: 10000
   # cacheDir is dynconfig cache storage directory
   # in linux, default value is /var/cache/dragonfly
   # in macos(just for testing), default value is /Users/$USER/.dragonfly/cache
diff --git a/docs/zh-CN/deployment/configuration/scheduler.yaml b/docs/zh-CN/deployment/configuration/scheduler.yaml
index 224d0233..a3301888 100644
--- a/docs/zh-CN/deployment/configuration/scheduler.yaml
+++ b/docs/zh-CN/deployment/configuration/scheduler.yaml
@@ -6,8 +6,6 @@ server:
   # host: localhost
   # 服务监听端口
   port:
-  # 限制请求并发数
-  listenLimit: 10000
   # daemon 动态配置缓存目录
   # linux 上默认目录 /var/cache/dragonfly
   # macos(仅开发、测试), 默认目录是 /Users/$USER/.dragonfly/cache
diff --git a/go.mod b/go.mod
index 7c296ba1..777ff633 100644
--- a/go.mod
+++ b/go.mod
@@ -48,7 +48,6 @@ require (
 	github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.11.0
-	github.com/prometheus/common v0.28.0
 	github.com/schollz/progressbar/v3 v3.8.2
 	github.com/serialx/hashring v0.0.0-20200727003509-22c0c7ab6b1b
 	github.com/shirou/gopsutil/v3 v3.21.11
@@ -68,7 +67,6 @@ require (
 	go.uber.org/atomic v1.9.0
 	go.uber.org/zap v1.19.0
 	golang.org/x/crypto v0.0.0-20211117183948-ae814b36b871
-	golang.org/x/net v0.0.0-20211209124913-491a49abca63
 	golang.org/x/oauth2 v0.0.0-20210819190943-2bc19b11175f
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e
@@ -164,6 +162,7 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
+	github.com/prometheus/common v0.28.0 // indirect
 	github.com/prometheus/procfs v0.6.0 // indirect
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/robfig/cron/v3 v3.0.1 // indirect
@@ -193,6 +192,7 @@ require (
 	golang.org/x/exp v0.0.0-20201221025956-e89b829e73ea // indirect
 	golang.org/x/lint v0.0.0-20210508222113-6edffad5e616 // indirect
 	golang.org/x/mod v0.4.2 // indirect
+	golang.org/x/net v0.0.0-20211209124913-491a49abca63 // indirect
 	golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/tools v0.1.6-0.20210820212750-d4cc65f0b2ff // indirect
diff --git a/internal/dflog/logger.go b/internal/dflog/logger.go
index 18c665f4..39f97752 100644
--- a/internal/dflog/logger.go
+++ b/internal/dflog/logger.go
@@ -18,6 +18,8 @@ package logger
 
 import (
 	"fmt"
+	"os"
+	"strconv"
 
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
@@ -87,7 +89,12 @@ func SetDownloadLogger(log *zap.Logger) {
 
 func SetGrpcLogger(log *zap.SugaredLogger) {
 	GrpcLogger = log
-	grpclog.SetLoggerV2(&zapGrpc{GrpcLogger})
+	var v int
+	vLevel := os.Getenv("GRPC_GO_LOG_VERBOSITY_LEVEL")
+	if vl, err := strconv.Atoi(vLevel); err == nil {
+		v = vl
+	}
+	grpclog.SetLoggerV2(&zapGrpc{GrpcLogger, v})
 }
 
 func SetJobLogger(log *zap.SugaredLogger) {
@@ -204,6 +211,7 @@ func Fatal(args ...interface{}) {
 
 type zapGrpc struct {
 	*zap.SugaredLogger
+	verbose int
 }
 
 func (z *zapGrpc) Infoln(args ...interface{}) {
@@ -231,5 +239,5 @@ func (z *zapGrpc) Fatalln(args ...interface{}) {
 }
 
 func (z *zapGrpc) V(level int) bool {
-	return level > 0
+	return level <= z.verbose
 }
diff --git a/manager/database/mysql.go b/manager/database/mysql.go
index 8fb770e6..b849e77f 100644
--- a/manager/database/mysql.go
+++ b/manager/database/mysql.go
@@ -32,8 +32,11 @@ import (
 )
 
 const (
-	defaultCDNLoadLimit    = 300
-	defaultClientLoadLimit = 100
+	// Default number of cdn load limit
+	defaultCDNLoadLimit = 300
+
+	// Default number of client load limit
+	defaultClientLoadLimit = 50
 )
 
 func newMyqsl(cfg *config.MysqlConfig) (*gorm.DB, error) {
diff --git a/manager/types/scheduler_cluster.go b/manager/types/scheduler_cluster.go
index 11408751..37b6b2c5 100644
--- a/manager/types/scheduler_cluster.go
+++ b/manager/types/scheduler_cluster.go
@@ -52,7 +52,7 @@ type GetSchedulerClustersQuery struct {
 }
 
 type SchedulerClusterConfig struct {
-	FilterParentCount uint32 `yaml:"filterParentCount" mapstructure:"filterParentCount" json:"filter_parent_count" binding:"omitempty,gte=1,lte=100"`
+	FilterParentLimit uint32 `yaml:"filterParentLimit" mapstructure:"filterParentLimit" json:"filter_parent_limit" binding:"omitempty,gte=1,lte=100"`
 }
 
 type SchedulerClusterClientConfig struct {
diff --git a/pkg/rpc/base/base.pb.go b/pkg/rpc/base/base.pb.go
index 640eb354..53e26826 100644
--- a/pkg/rpc/base/base.pb.go
+++ b/pkg/rpc/base/base.pb.go
@@ -562,6 +562,8 @@ type PieceInfo struct {
 	PieceMd5    string     `protobuf:"bytes,4,opt,name=piece_md5,json=pieceMd5,proto3" json:"piece_md5,omitempty"`
 	PieceOffset uint64     `protobuf:"varint,5,opt,name=piece_offset,json=pieceOffset,proto3" json:"piece_offset,omitempty"`
 	PieceStyle  PieceStyle `protobuf:"varint,6,opt,name=piece_style,json=pieceStyle,proto3,enum=base.PieceStyle" json:"piece_style,omitempty"`
+	// total time(millisecond) consumed
+	DownloadCost uint64 `protobuf:"varint,7,opt,name=download_cost,json=downloadCost,proto3" json:"download_cost,omitempty"`
 }
 
 func (x *PieceInfo) Reset() {
@@ -638,6 +640,13 @@ func (x *PieceInfo) GetPieceStyle() PieceStyle {
 	return PieceStyle_PLAIN
 }
 
+func (x *PieceInfo) GetDownloadCost() uint64 {
+	if x != nil {
+		return x.DownloadCost
+	}
+	return 0
+}
+
 type PiecePacket struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -787,7 +796,7 @@ var file_pkg_rpc_base_base_proto_rawDesc = []byte{
 	0x75, 0x6d, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0d, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x2a, 0x02, 0x28,
 	0x00, 0x52, 0x08, 0x73, 0x74, 0x61, 0x72, 0x74, 0x4e, 0x75, 0x6d, 0x12, 0x1d, 0x0a, 0x05, 0x6c,
 	0x69, 0x6d, 0x69, 0x74, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0d, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x2a,
-	0x02, 0x28, 0x00, 0x52, 0x05, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x22, 0xb3, 0x02, 0x0a, 0x09, 0x50,
+	0x02, 0x28, 0x00, 0x52, 0x05, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x22, 0xe1, 0x02, 0x0a, 0x09, 0x50,
 	0x69, 0x65, 0x63, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x69, 0x65, 0x63,
 	0x65, 0x5f, 0x6e, 0x75, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x08, 0x70, 0x69, 0x65,
 	0x63, 0x65, 0x4e, 0x75, 0x6d, 0x12, 0x28, 0x0a, 0x0b, 0x72, 0x61, 0x6e, 0x67, 0x65, 0x5f, 0x73,
@@ -807,73 +816,76 @@ var file_pkg_rpc_base_base_proto_rawDesc = []byte{
 	0x0b, 0x70, 0x69, 0x65, 0x63, 0x65, 0x5f, 0x73, 0x74, 0x79, 0x6c, 0x65, 0x18, 0x06, 0x20, 0x01,
 	0x28, 0x0e, 0x32, 0x10, 0x2e, 0x62, 0x61, 0x73, 0x65, 0x2e, 0x50, 0x69, 0x65, 0x63, 0x65, 0x53,
 	0x74, 0x79, 0x6c, 0x65, 0x52, 0x0a, 0x70, 0x69, 0x65, 0x63, 0x65, 0x53, 0x74, 0x79, 0x6c, 0x65,
-	0x22, 0x95, 0x02, 0x0a, 0x0b, 0x50, 0x69, 0x65, 0x63, 0x65, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74,
-	0x12, 0x20, 0x0a, 0x07, 0x74, 0x61, 0x73, 0x6b, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28,
-	0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x10, 0x01, 0x52, 0x06, 0x74, 0x61, 0x73, 0x6b,
-	0x49, 0x64, 0x12, 0x20, 0x0a, 0x07, 0x64, 0x73, 0x74, 0x5f, 0x70, 0x69, 0x64, 0x18, 0x03, 0x20,
-	0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x10, 0x01, 0x52, 0x06, 0x64, 0x73,
-	0x74, 0x50, 0x69, 0x64, 0x12, 0x22, 0x0a, 0x08, 0x64, 0x73, 0x74, 0x5f, 0x61, 0x64, 0x64, 0x72,
-	0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x10, 0x01, 0x52,
-	0x07, 0x64, 0x73, 0x74, 0x41, 0x64, 0x64, 0x72, 0x12, 0x30, 0x0a, 0x0b, 0x70, 0x69, 0x65, 0x63,
-	0x65, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x73, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0f, 0x2e,
-	0x62, 0x61, 0x73, 0x65, 0x2e, 0x50, 0x69, 0x65, 0x63, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x0a,
-	0x70, 0x69, 0x65, 0x63, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x73, 0x12, 0x1f, 0x0a, 0x0b, 0x74, 0x6f,
-	0x74, 0x61, 0x6c, 0x5f, 0x70, 0x69, 0x65, 0x63, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x05, 0x52,
-	0x0a, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x50, 0x69, 0x65, 0x63, 0x65, 0x12, 0x25, 0x0a, 0x0e, 0x63,
-	0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x5f, 0x6c, 0x65, 0x6e, 0x67, 0x74, 0x68, 0x18, 0x07, 0x20,
-	0x01, 0x28, 0x03, 0x52, 0x0d, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4c, 0x65, 0x6e, 0x67,
-	0x74, 0x68, 0x12, 0x24, 0x0a, 0x0e, 0x70, 0x69, 0x65, 0x63, 0x65, 0x5f, 0x6d, 0x64, 0x35, 0x5f,
-	0x73, 0x69, 0x67, 0x6e, 0x18, 0x08, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x70, 0x69, 0x65, 0x63,
-	0x65, 0x4d, 0x64, 0x35, 0x53, 0x69, 0x67, 0x6e, 0x2a, 0xa1, 0x05, 0x0a, 0x04, 0x43, 0x6f, 0x64,
-	0x65, 0x12, 0x11, 0x0a, 0x0d, 0x58, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49,
-	0x45, 0x44, 0x10, 0x00, 0x12, 0x0c, 0x0a, 0x07, 0x53, 0x75, 0x63, 0x63, 0x65, 0x73, 0x73, 0x10,
-	0xc8, 0x01, 0x12, 0x16, 0x0a, 0x11, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x55, 0x6e, 0x61, 0x76,
-	0x61, 0x69, 0x6c, 0x61, 0x62, 0x6c, 0x65, 0x10, 0xf4, 0x03, 0x12, 0x13, 0x0a, 0x0e, 0x52, 0x65,
-	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x4c, 0x61, 0x63, 0x6b, 0x65, 0x64, 0x10, 0xe8, 0x07, 0x12,
-	0x0f, 0x0a, 0x0a, 0x42, 0x61, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x10, 0xf8, 0x0a,
-	0x12, 0x15, 0x0a, 0x10, 0x50, 0x65, 0x65, 0x72, 0x54, 0x61, 0x73, 0x6b, 0x4e, 0x6f, 0x74, 0x46,
-	0x6f, 0x75, 0x6e, 0x64, 0x10, 0xfc, 0x0a, 0x12, 0x11, 0x0a, 0x0c, 0x55, 0x6e, 0x6b, 0x6e, 0x6f,
-	0x77, 0x6e, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0xdc, 0x0b, 0x12, 0x13, 0x0a, 0x0e, 0x52, 0x65,
-	0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x4f, 0x75, 0x74, 0x10, 0xe0, 0x0b, 0x12,
-	0x10, 0x0a, 0x0b, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0xa0,
-	0x1f, 0x12, 0x1b, 0x0a, 0x16, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x50, 0x69, 0x65, 0x63, 0x65,
-	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xa1, 0x1f, 0x12, 0x1a,
-	0x0a, 0x15, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65,
-	0x54, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x10, 0xa2, 0x1f, 0x12, 0x1a, 0x0a, 0x15, 0x43, 0x6c,
-	0x69, 0x65, 0x6e, 0x74, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x43, 0x61, 0x6e, 0x63, 0x65,
-	0x6c, 0x65, 0x64, 0x10, 0xa3, 0x1f, 0x12, 0x19, 0x0a, 0x14, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74,
-	0x57, 0x61, 0x69, 0x74, 0x50, 0x69, 0x65, 0x63, 0x65, 0x52, 0x65, 0x61, 0x64, 0x79, 0x10, 0xa4,
-	0x1f, 0x12, 0x1c, 0x0a, 0x17, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x50, 0x69, 0x65, 0x63, 0x65,
-	0x44, 0x6f, 0x77, 0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xa5, 0x1f, 0x12,
-	0x1b, 0x0a, 0x16, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
-	0x4c, 0x69, 0x6d, 0x69, 0x74, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xa6, 0x1f, 0x12, 0x1a, 0x0a, 0x15,
-	0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e,
-	0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0xa7, 0x1f, 0x12, 0x18, 0x0a, 0x13, 0x43, 0x6c, 0x69, 0x65,
-	0x6e, 0x74, 0x50, 0x69, 0x65, 0x63, 0x65, 0x4e, 0x6f, 0x74, 0x46, 0x6f, 0x75, 0x6e, 0x64, 0x10,
-	0xb4, 0x22, 0x12, 0x0f, 0x0a, 0x0a, 0x53, 0x63, 0x68, 0x65, 0x64, 0x45, 0x72, 0x72, 0x6f, 0x72,
-	0x10, 0x88, 0x27, 0x12, 0x18, 0x0a, 0x13, 0x53, 0x63, 0x68, 0x65, 0x64, 0x4e, 0x65, 0x65, 0x64,
-	0x42, 0x61, 0x63, 0x6b, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x10, 0x89, 0x27, 0x12, 0x12, 0x0a,
-	0x0d, 0x53, 0x63, 0x68, 0x65, 0x64, 0x50, 0x65, 0x65, 0x72, 0x47, 0x6f, 0x6e, 0x65, 0x10, 0x8a,
-	0x27, 0x12, 0x16, 0x0a, 0x11, 0x53, 0x63, 0x68, 0x65, 0x64, 0x50, 0x65, 0x65, 0x72, 0x4e, 0x6f,
-	0x74, 0x46, 0x6f, 0x75, 0x6e, 0x64, 0x10, 0x8c, 0x27, 0x12, 0x23, 0x0a, 0x1e, 0x53, 0x63, 0x68,
-	0x65, 0x64, 0x50, 0x65, 0x65, 0x72, 0x50, 0x69, 0x65, 0x63, 0x65, 0x52, 0x65, 0x73, 0x75, 0x6c,
-	0x74, 0x52, 0x65, 0x70, 0x6f, 0x72, 0x74, 0x46, 0x61, 0x69, 0x6c, 0x10, 0x8d, 0x27, 0x12, 0x19,
-	0x0a, 0x14, 0x53, 0x63, 0x68, 0x65, 0x64, 0x54, 0x61, 0x73, 0x6b, 0x53, 0x74, 0x61, 0x74, 0x75,
-	0x73, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0x8e, 0x27, 0x12, 0x0d, 0x0a, 0x08, 0x43, 0x44, 0x4e,
-	0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0xf0, 0x2e, 0x12, 0x18, 0x0a, 0x13, 0x43, 0x44, 0x4e, 0x54,
-	0x61, 0x73, 0x6b, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x72, 0x79, 0x46, 0x61, 0x69, 0x6c, 0x10,
-	0xf1, 0x2e, 0x12, 0x18, 0x0a, 0x13, 0x43, 0x44, 0x4e, 0x54, 0x61, 0x73, 0x6b, 0x44, 0x6f, 0x77,
-	0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xf2, 0x2e, 0x12, 0x14, 0x0a, 0x0f,
-	0x43, 0x44, 0x4e, 0x54, 0x61, 0x73, 0x6b, 0x4e, 0x6f, 0x74, 0x46, 0x6f, 0x75, 0x6e, 0x64, 0x10,
-	0x84, 0x32, 0x12, 0x18, 0x0a, 0x13, 0x49, 0x6e, 0x76, 0x61, 0x6c, 0x69, 0x64, 0x52, 0x65, 0x73,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xd9, 0x36, 0x2a, 0x17, 0x0a, 0x0a,
-	0x50, 0x69, 0x65, 0x63, 0x65, 0x53, 0x74, 0x79, 0x6c, 0x65, 0x12, 0x09, 0x0a, 0x05, 0x50, 0x4c,
-	0x41, 0x49, 0x4e, 0x10, 0x00, 0x2a, 0x2c, 0x0a, 0x09, 0x53, 0x69, 0x7a, 0x65, 0x53, 0x63, 0x6f,
-	0x70, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x4e, 0x4f, 0x52, 0x4d, 0x41, 0x4c, 0x10, 0x00, 0x12, 0x09,
-	0x0a, 0x05, 0x53, 0x4d, 0x41, 0x4c, 0x4c, 0x10, 0x01, 0x12, 0x08, 0x0a, 0x04, 0x54, 0x49, 0x4e,
-	0x59, 0x10, 0x02, 0x42, 0x22, 0x5a, 0x20, 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, 0x62, 0x61, 0x73, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x12, 0x2c, 0x0a, 0x0d, 0x64, 0x6f, 0x77, 0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x5f, 0x63, 0x6f, 0x73,
+	0x74, 0x18, 0x07, 0x20, 0x01, 0x28, 0x04, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x32, 0x02, 0x28, 0x00,
+	0x52, 0x0c, 0x64, 0x6f, 0x77, 0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x43, 0x6f, 0x73, 0x74, 0x22, 0x95,
+	0x02, 0x0a, 0x0b, 0x50, 0x69, 0x65, 0x63, 0x65, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x12, 0x20,
+	0x0a, 0x07, 0x74, 0x61, 0x73, 0x6b, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x42,
+	0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x10, 0x01, 0x52, 0x06, 0x74, 0x61, 0x73, 0x6b, 0x49, 0x64,
+	0x12, 0x20, 0x0a, 0x07, 0x64, 0x73, 0x74, 0x5f, 0x70, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28,
+	0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x10, 0x01, 0x52, 0x06, 0x64, 0x73, 0x74, 0x50,
+	0x69, 0x64, 0x12, 0x22, 0x0a, 0x08, 0x64, 0x73, 0x74, 0x5f, 0x61, 0x64, 0x64, 0x72, 0x18, 0x04,
+	0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x10, 0x01, 0x52, 0x07, 0x64,
+	0x73, 0x74, 0x41, 0x64, 0x64, 0x72, 0x12, 0x30, 0x0a, 0x0b, 0x70, 0x69, 0x65, 0x63, 0x65, 0x5f,
+	0x69, 0x6e, 0x66, 0x6f, 0x73, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0f, 0x2e, 0x62, 0x61,
+	0x73, 0x65, 0x2e, 0x50, 0x69, 0x65, 0x63, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x0a, 0x70, 0x69,
+	0x65, 0x63, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x73, 0x12, 0x1f, 0x0a, 0x0b, 0x74, 0x6f, 0x74, 0x61,
+	0x6c, 0x5f, 0x70, 0x69, 0x65, 0x63, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0a, 0x74,
+	0x6f, 0x74, 0x61, 0x6c, 0x50, 0x69, 0x65, 0x63, 0x65, 0x12, 0x25, 0x0a, 0x0e, 0x63, 0x6f, 0x6e,
+	0x74, 0x65, 0x6e, 0x74, 0x5f, 0x6c, 0x65, 0x6e, 0x67, 0x74, 0x68, 0x18, 0x07, 0x20, 0x01, 0x28,
+	0x03, 0x52, 0x0d, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4c, 0x65, 0x6e, 0x67, 0x74, 0x68,
+	0x12, 0x24, 0x0a, 0x0e, 0x70, 0x69, 0x65, 0x63, 0x65, 0x5f, 0x6d, 0x64, 0x35, 0x5f, 0x73, 0x69,
+	0x67, 0x6e, 0x18, 0x08, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x70, 0x69, 0x65, 0x63, 0x65, 0x4d,
+	0x64, 0x35, 0x53, 0x69, 0x67, 0x6e, 0x2a, 0xa1, 0x05, 0x0a, 0x04, 0x43, 0x6f, 0x64, 0x65, 0x12,
+	0x11, 0x0a, 0x0d, 0x58, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44,
+	0x10, 0x00, 0x12, 0x0c, 0x0a, 0x07, 0x53, 0x75, 0x63, 0x63, 0x65, 0x73, 0x73, 0x10, 0xc8, 0x01,
+	0x12, 0x16, 0x0a, 0x11, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x55, 0x6e, 0x61, 0x76, 0x61, 0x69,
+	0x6c, 0x61, 0x62, 0x6c, 0x65, 0x10, 0xf4, 0x03, 0x12, 0x13, 0x0a, 0x0e, 0x52, 0x65, 0x73, 0x6f,
+	0x75, 0x72, 0x63, 0x65, 0x4c, 0x61, 0x63, 0x6b, 0x65, 0x64, 0x10, 0xe8, 0x07, 0x12, 0x0f, 0x0a,
+	0x0a, 0x42, 0x61, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x10, 0xf8, 0x0a, 0x12, 0x15,
+	0x0a, 0x10, 0x50, 0x65, 0x65, 0x72, 0x54, 0x61, 0x73, 0x6b, 0x4e, 0x6f, 0x74, 0x46, 0x6f, 0x75,
+	0x6e, 0x64, 0x10, 0xfc, 0x0a, 0x12, 0x11, 0x0a, 0x0c, 0x55, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e,
+	0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0xdc, 0x0b, 0x12, 0x13, 0x0a, 0x0e, 0x52, 0x65, 0x71, 0x75,
+	0x65, 0x73, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x4f, 0x75, 0x74, 0x10, 0xe0, 0x0b, 0x12, 0x10, 0x0a,
+	0x0b, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0xa0, 0x1f, 0x12,
+	0x1b, 0x0a, 0x16, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x50, 0x69, 0x65, 0x63, 0x65, 0x52, 0x65,
+	0x71, 0x75, 0x65, 0x73, 0x74, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xa1, 0x1f, 0x12, 0x1a, 0x0a, 0x15,
+	0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x53, 0x63, 0x68, 0x65, 0x64, 0x75, 0x6c, 0x65, 0x54, 0x69,
+	0x6d, 0x65, 0x6f, 0x75, 0x74, 0x10, 0xa2, 0x1f, 0x12, 0x1a, 0x0a, 0x15, 0x43, 0x6c, 0x69, 0x65,
+	0x6e, 0x74, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x78, 0x74, 0x43, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x65,
+	0x64, 0x10, 0xa3, 0x1f, 0x12, 0x19, 0x0a, 0x14, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x57, 0x61,
+	0x69, 0x74, 0x50, 0x69, 0x65, 0x63, 0x65, 0x52, 0x65, 0x61, 0x64, 0x79, 0x10, 0xa4, 0x1f, 0x12,
+	0x1c, 0x0a, 0x17, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x50, 0x69, 0x65, 0x63, 0x65, 0x44, 0x6f,
+	0x77, 0x6e, 0x6c, 0x6f, 0x61, 0x64, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xa5, 0x1f, 0x12, 0x1b, 0x0a,
+	0x16, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x4c, 0x69,
+	0x6d, 0x69, 0x74, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xa6, 0x1f, 0x12, 0x1a, 0x0a, 0x15, 0x43, 0x6c,
+	0x69, 0x65, 0x6e, 0x74, 0x43, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x45, 0x72,
+	0x72, 0x6f, 0x72, 0x10, 0xa7, 0x1f, 0x12, 0x18, 0x0a, 0x13, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74,
+	0x50, 0x69, 0x65, 0x63, 0x65, 0x4e, 0x6f, 0x74, 0x46, 0x6f, 0x75, 0x6e, 0x64, 0x10, 0xb4, 0x22,
+	0x12, 0x0f, 0x0a, 0x0a, 0x53, 0x63, 0x68, 0x65, 0x64, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x10, 0x88,
+	0x27, 0x12, 0x18, 0x0a, 0x13, 0x53, 0x63, 0x68, 0x65, 0x64, 0x4e, 0x65, 0x65, 0x64, 0x42, 0x61,
+	0x63, 0x6b, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x10, 0x89, 0x27, 0x12, 0x12, 0x0a, 0x0d, 0x53,
+	0x63, 0x68, 0x65, 0x64, 0x50, 0x65, 0x65, 0x72, 0x47, 0x6f, 0x6e, 0x65, 0x10, 0x8a, 0x27, 0x12,
+	0x16, 0x0a, 0x11, 0x53, 0x63, 0x68, 0x65, 0x64, 0x50, 0x65, 0x65, 0x72, 0x4e, 0x6f, 0x74, 0x46,
+	0x6f, 0x75, 0x6e, 0x64, 0x10, 0x8c, 0x27, 0x12, 0x23, 0x0a, 0x1e, 0x53, 0x63, 0x68, 0x65, 0x64,
+	0x50, 0x65, 0x65, 0x72, 0x50, 0x69, 0x65, 0x63, 0x65, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x52,
+	0x65, 0x70, 0x6f, 0x72, 0x74, 0x46, 0x61, 0x69, 0x6c, 0x10, 0x8d, 0x27, 0x12, 0x19, 0x0a, 0x14,
+	0x53, 0x63, 0x68, 0x65, 0x64, 0x54, 0x61, 0x73, 0x6b, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x45,
+	0x72, 0x72, 0x6f, 0x72, 0x10, 0x8e, 0x27, 0x12, 0x0d, 0x0a, 0x08, 0x43, 0x44, 0x4e, 0x45, 0x72,
+	0x72, 0x6f, 0x72, 0x10, 0xf0, 0x2e, 0x12, 0x18, 0x0a, 0x13, 0x43, 0x44, 0x4e, 0x54, 0x61, 0x73,
+	0x6b, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x72, 0x79, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xf1, 0x2e,
+	0x12, 0x18, 0x0a, 0x13, 0x43, 0x44, 0x4e, 0x54, 0x61, 0x73, 0x6b, 0x44, 0x6f, 0x77, 0x6e, 0x6c,
+	0x6f, 0x61, 0x64, 0x46, 0x61, 0x69, 0x6c, 0x10, 0xf2, 0x2e, 0x12, 0x14, 0x0a, 0x0f, 0x43, 0x44,
+	0x4e, 0x54, 0x61, 0x73, 0x6b, 0x4e, 0x6f, 0x74, 0x46, 0x6f, 0x75, 0x6e, 0x64, 0x10, 0x84, 0x32,
+	0x12, 0x18, 0x0a, 0x13, 0x49, 0x6e, 0x76, 0x61, 0x6c, 0x69, 0x64, 0x52, 0x65, 0x73, 0x6f, 0x75,
+	0x72, 0x63, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xd9, 0x36, 0x2a, 0x17, 0x0a, 0x0a, 0x50, 0x69,
+	0x65, 0x63, 0x65, 0x53, 0x74, 0x79, 0x6c, 0x65, 0x12, 0x09, 0x0a, 0x05, 0x50, 0x4c, 0x41, 0x49,
+	0x4e, 0x10, 0x00, 0x2a, 0x2c, 0x0a, 0x09, 0x53, 0x69, 0x7a, 0x65, 0x53, 0x63, 0x6f, 0x70, 0x65,
+	0x12, 0x0a, 0x0a, 0x06, 0x4e, 0x4f, 0x52, 0x4d, 0x41, 0x4c, 0x10, 0x00, 0x12, 0x09, 0x0a, 0x05,
+	0x53, 0x4d, 0x41, 0x4c, 0x4c, 0x10, 0x01, 0x12, 0x08, 0x0a, 0x04, 0x54, 0x49, 0x4e, 0x59, 0x10,
+	0x02, 0x42, 0x22, 0x5a, 0x20, 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, 0x62, 0x61, 0x73, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
diff --git a/pkg/rpc/base/base.pb.validate.go b/pkg/rpc/base/base.pb.validate.go
index 84e75adb..c0e95cca 100644
--- a/pkg/rpc/base/base.pb.validate.go
+++ b/pkg/rpc/base/base.pb.validate.go
@@ -431,6 +431,13 @@ func (m *PieceInfo) Validate() error {
 
 	// no validation rules for PieceStyle
 
+	if m.GetDownloadCost() < 0 {
+		return PieceInfoValidationError{
+			field:  "DownloadCost",
+			reason: "value must be greater than or equal to 0",
+		}
+	}
+
 	return nil
 }
 
diff --git a/pkg/rpc/base/base.proto b/pkg/rpc/base/base.proto
index 07f12c1b..bcc8a379 100644
--- a/pkg/rpc/base/base.proto
+++ b/pkg/rpc/base/base.proto
@@ -124,6 +124,8 @@ message PieceInfo{
   string piece_md5 = 4 [(validate.rules).string = {pattern:"([a-f\\d]{32}|[A-F\\d]{32}|[a-f\\d]{16}|[A-F\\d]{16})", ignore_empty:true}];
   uint64 piece_offset = 5 [(validate.rules).uint64.gte = 0];
   base.PieceStyle piece_style = 6;
+  // total time(millisecond) consumed
+  uint64 download_cost = 7 [(validate.rules).uint64.gte = 0];
 }
 
 message PiecePacket{
diff --git a/pkg/rpc/cdnsystem/cdnsystem.pb.go b/pkg/rpc/cdnsystem/cdnsystem.pb.go
index 5d1b32b9..24732c2e 100644
--- a/pkg/rpc/cdnsystem/cdnsystem.pb.go
+++ b/pkg/rpc/cdnsystem/cdnsystem.pb.go
@@ -118,6 +118,10 @@ type PieceSeed struct {
 	ContentLength int64 `protobuf:"varint,6,opt,name=content_length,json=contentLength,proto3" json:"content_length,omitempty"`
 	// total piece count, -1 represents task is downloading or failed
 	TotalPieceCount int32 `protobuf:"varint,7,opt,name=total_piece_count,json=totalPieceCount,proto3" json:"total_piece_count,omitempty"`
+	// begin time for the piece downloading
+	BeginTime uint64 `protobuf:"varint,8,opt,name=begin_time,json=beginTime,proto3" json:"begin_time,omitempty"`
+	// end time for the piece downloading
+	EndTime uint64 `protobuf:"varint,9,opt,name=end_time,json=endTime,proto3" json:"end_time,omitempty"`
 }
 
 func (x *PieceSeed) Reset() {
@@ -194,6 +198,20 @@ func (x *PieceSeed) GetTotalPieceCount() int32 {
 	return 0
 }
 
+func (x *PieceSeed) GetBeginTime() uint64 {
+	if x != nil {
+		return x.BeginTime
+	}
+	return 0
+}
+
+func (x *PieceSeed) GetEndTime() uint64 {
+	if x != nil {
+		return x.EndTime
+	}
+	return 0
+}
+
 var File_pkg_rpc_cdnsystem_cdnsystem_proto protoreflect.FileDescriptor
 
 var file_pkg_rpc_cdnsystem_cdnsystem_proto_rawDesc = []byte{
@@ -210,7 +228,7 @@ var file_pkg_rpc_cdnsystem_cdnsystem_proto_rawDesc = []byte{
 	0xfa, 0x42, 0x05, 0x72, 0x03, 0x88, 0x01, 0x01, 0x52, 0x03, 0x75, 0x72, 0x6c, 0x12, 0x28, 0x0a,
 	0x08, 0x75, 0x72, 0x6c, 0x5f, 0x6d, 0x65, 0x74, 0x61, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32,
 	0x0d, 0x2e, 0x62, 0x61, 0x73, 0x65, 0x2e, 0x55, 0x72, 0x6c, 0x4d, 0x65, 0x74, 0x61, 0x52, 0x07,
-	0x75, 0x72, 0x6c, 0x4d, 0x65, 0x74, 0x61, 0x22, 0xea, 0x01, 0x0a, 0x09, 0x50, 0x69, 0x65, 0x63,
+	0x75, 0x72, 0x6c, 0x4d, 0x65, 0x74, 0x61, 0x22, 0xa4, 0x02, 0x0a, 0x09, 0x50, 0x69, 0x65, 0x63,
 	0x65, 0x53, 0x65, 0x65, 0x64, 0x12, 0x20, 0x0a, 0x07, 0x70, 0x65, 0x65, 0x72, 0x5f, 0x69, 0x64,
 	0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x42, 0x07, 0xfa, 0x42, 0x04, 0x72, 0x02, 0x10, 0x01, 0x52,
 	0x06, 0x70, 0x65, 0x65, 0x72, 0x49, 0x64, 0x12, 0x24, 0x0a, 0x09, 0x68, 0x6f, 0x73, 0x74, 0x5f,
@@ -225,18 +243,22 @@ var file_pkg_rpc_cdnsystem_cdnsystem_proto_rawDesc = []byte{
 	0x6e, 0x74, 0x4c, 0x65, 0x6e, 0x67, 0x74, 0x68, 0x12, 0x2a, 0x0a, 0x11, 0x74, 0x6f, 0x74, 0x61,
 	0x6c, 0x5f, 0x70, 0x69, 0x65, 0x63, 0x65, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x07, 0x20,
 	0x01, 0x28, 0x05, 0x52, 0x0f, 0x74, 0x6f, 0x74, 0x61, 0x6c, 0x50, 0x69, 0x65, 0x63, 0x65, 0x43,
-	0x6f, 0x75, 0x6e, 0x74, 0x32, 0x83, 0x01, 0x0a, 0x06, 0x53, 0x65, 0x65, 0x64, 0x65, 0x72, 0x12,
-	0x3d, 0x0a, 0x0b, 0x4f, 0x62, 0x74, 0x61, 0x69, 0x6e, 0x53, 0x65, 0x65, 0x64, 0x73, 0x12, 0x16,
-	0x2e, 0x63, 0x64, 0x6e, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x2e, 0x53, 0x65, 0x65, 0x64, 0x52,
-	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x14, 0x2e, 0x63, 0x64, 0x6e, 0x73, 0x79, 0x73, 0x74,
-	0x65, 0x6d, 0x2e, 0x50, 0x69, 0x65, 0x63, 0x65, 0x53, 0x65, 0x65, 0x64, 0x30, 0x01, 0x12, 0x3a,
-	0x0a, 0x0d, 0x47, 0x65, 0x74, 0x50, 0x69, 0x65, 0x63, 0x65, 0x54, 0x61, 0x73, 0x6b, 0x73, 0x12,
-	0x16, 0x2e, 0x62, 0x61, 0x73, 0x65, 0x2e, 0x50, 0x69, 0x65, 0x63, 0x65, 0x54, 0x61, 0x73, 0x6b,
-	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x11, 0x2e, 0x62, 0x61, 0x73, 0x65, 0x2e, 0x50,
-	0x69, 0x65, 0x63, 0x65, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x42, 0x27, 0x5a, 0x25, 0x64, 0x37,
-	0x79, 0x2e, 0x69, 0x6f, 0x2f, 0x64, 0x72, 0x61, 0x67, 0x6f, 0x6e, 0x66, 0x6c, 0x79, 0x2f, 0x76,
-	0x32, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x72, 0x70, 0x63, 0x2f, 0x63, 0x64, 0x6e, 0x73, 0x79, 0x73,
-	0x74, 0x65, 0x6d, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x6f, 0x75, 0x6e, 0x74, 0x12, 0x1d, 0x0a, 0x0a, 0x62, 0x65, 0x67, 0x69, 0x6e, 0x5f, 0x74, 0x69,
+	0x6d, 0x65, 0x18, 0x08, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x62, 0x65, 0x67, 0x69, 0x6e, 0x54,
+	0x69, 0x6d, 0x65, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x6e, 0x64, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18,
+	0x09, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07, 0x65, 0x6e, 0x64, 0x54, 0x69, 0x6d, 0x65, 0x32, 0x83,
+	0x01, 0x0a, 0x06, 0x53, 0x65, 0x65, 0x64, 0x65, 0x72, 0x12, 0x3d, 0x0a, 0x0b, 0x4f, 0x62, 0x74,
+	0x61, 0x69, 0x6e, 0x53, 0x65, 0x65, 0x64, 0x73, 0x12, 0x16, 0x2e, 0x63, 0x64, 0x6e, 0x73, 0x79,
+	0x73, 0x74, 0x65, 0x6d, 0x2e, 0x53, 0x65, 0x65, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x1a, 0x14, 0x2e, 0x63, 0x64, 0x6e, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x2e, 0x50, 0x69, 0x65,
+	0x63, 0x65, 0x53, 0x65, 0x65, 0x64, 0x30, 0x01, 0x12, 0x3a, 0x0a, 0x0d, 0x47, 0x65, 0x74, 0x50,
+	0x69, 0x65, 0x63, 0x65, 0x54, 0x61, 0x73, 0x6b, 0x73, 0x12, 0x16, 0x2e, 0x62, 0x61, 0x73, 0x65,
+	0x2e, 0x50, 0x69, 0x65, 0x63, 0x65, 0x54, 0x61, 0x73, 0x6b, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x74, 0x1a, 0x11, 0x2e, 0x62, 0x61, 0x73, 0x65, 0x2e, 0x50, 0x69, 0x65, 0x63, 0x65, 0x50, 0x61,
+	0x63, 0x6b, 0x65, 0x74, 0x42, 0x27, 0x5a, 0x25, 0x64, 0x37, 0x79, 0x2e, 0x69, 0x6f, 0x2f, 0x64,
+	0x72, 0x61, 0x67, 0x6f, 0x6e, 0x66, 0x6c, 0x79, 0x2f, 0x76, 0x32, 0x2f, 0x70, 0x6b, 0x67, 0x2f,
+	0x72, 0x70, 0x63, 0x2f, 0x63, 0x64, 0x6e, 0x73, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x62, 0x06, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
diff --git a/pkg/rpc/cdnsystem/cdnsystem.pb.validate.go b/pkg/rpc/cdnsystem/cdnsystem.pb.validate.go
index 7dddaf82..ebbe4330 100644
--- a/pkg/rpc/cdnsystem/cdnsystem.pb.validate.go
+++ b/pkg/rpc/cdnsystem/cdnsystem.pb.validate.go
@@ -165,6 +165,10 @@ func (m *PieceSeed) Validate() error {
 
 	// no validation rules for TotalPieceCount
 
+	// no validation rules for BeginTime
+
+	// no validation rules for EndTime
+
 	return nil
 }
 
diff --git a/pkg/rpc/cdnsystem/cdnsystem.proto b/pkg/rpc/cdnsystem/cdnsystem.proto
index 973afbd2..5247c432 100644
--- a/pkg/rpc/cdnsystem/cdnsystem.proto
+++ b/pkg/rpc/cdnsystem/cdnsystem.proto
@@ -44,6 +44,10 @@ message PieceSeed{
   int64 content_length = 6;
   // total piece count, -1 represents task is downloading or failed
   int32 total_piece_count = 7;
+  // begin time for the piece downloading
+  uint64 begin_time = 8;
+  // end time for the piece downloading
+  uint64 end_time = 9;
 }
 
 // CDN System RPC Service
diff --git a/pkg/rpc/manager/server/mocks/server_mock.go b/pkg/rpc/manager/server/mocks/server_mock.go
deleted file mode 100644
index 013fd619..00000000
--- a/pkg/rpc/manager/server/mocks/server_mock.go
+++ /dev/null
@@ -1,125 +0,0 @@
-// Code generated by MockGen. DO NOT EDIT.
-// Source: pkg/rpc/manager/server/server.go
-
-// Package mocks is a generated GoMock package.
-package mocks
-
-import (
-	context "context"
-	reflect "reflect"
-
-	manager "d7y.io/dragonfly/v2/pkg/rpc/manager"
-	gomock "github.com/golang/mock/gomock"
-)
-
-// MockManagerServer is a mock of ManagerServer interface.
-type MockManagerServer struct {
-	ctrl     *gomock.Controller
-	recorder *MockManagerServerMockRecorder
-}
-
-// MockManagerServerMockRecorder is the mock recorder for MockManagerServer.
-type MockManagerServerMockRecorder struct {
-	mock *MockManagerServer
-}
-
-// NewMockManagerServer creates a new mock instance.
-func NewMockManagerServer(ctrl *gomock.Controller) *MockManagerServer {
-	mock := &MockManagerServer{ctrl: ctrl}
-	mock.recorder = &MockManagerServerMockRecorder{mock}
-	return mock
-}
-
-// EXPECT returns an object that allows the caller to indicate expected use.
-func (m *MockManagerServer) EXPECT() *MockManagerServerMockRecorder {
-	return m.recorder
-}
-
-// GetCDN mocks base method.
-func (m *MockManagerServer) GetCDN(arg0 context.Context, arg1 *manager.GetCDNRequest) (*manager.CDN, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetCDN", arg0, arg1)
-	ret0, _ := ret[0].(*manager.CDN)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// GetCDN indicates an expected call of GetCDN.
-func (mr *MockManagerServerMockRecorder) GetCDN(arg0, arg1 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCDN", reflect.TypeOf((*MockManagerServer)(nil).GetCDN), arg0, arg1)
-}
-
-// GetScheduler mocks base method.
-func (m *MockManagerServer) GetScheduler(arg0 context.Context, arg1 *manager.GetSchedulerRequest) (*manager.Scheduler, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetScheduler", arg0, arg1)
-	ret0, _ := ret[0].(*manager.Scheduler)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// GetScheduler indicates an expected call of GetScheduler.
-func (mr *MockManagerServerMockRecorder) GetScheduler(arg0, arg1 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetScheduler", reflect.TypeOf((*MockManagerServer)(nil).GetScheduler), arg0, arg1)
-}
-
-// KeepAlive mocks base method.
-func (m *MockManagerServer) KeepAlive(arg0 manager.Manager_KeepAliveServer) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "KeepAlive", arg0)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// KeepAlive indicates an expected call of KeepAlive.
-func (mr *MockManagerServerMockRecorder) KeepAlive(arg0 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "KeepAlive", reflect.TypeOf((*MockManagerServer)(nil).KeepAlive), arg0)
-}
-
-// ListSchedulers mocks base method.
-func (m *MockManagerServer) ListSchedulers(arg0 context.Context, arg1 *manager.ListSchedulersRequest) (*manager.ListSchedulersResponse, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "ListSchedulers", arg0, arg1)
-	ret0, _ := ret[0].(*manager.ListSchedulersResponse)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// ListSchedulers indicates an expected call of ListSchedulers.
-func (mr *MockManagerServerMockRecorder) ListSchedulers(arg0, arg1 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListSchedulers", reflect.TypeOf((*MockManagerServer)(nil).ListSchedulers), arg0, arg1)
-}
-
-// UpdateCDN mocks base method.
-func (m *MockManagerServer) UpdateCDN(arg0 context.Context, arg1 *manager.UpdateCDNRequest) (*manager.CDN, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "UpdateCDN", arg0, arg1)
-	ret0, _ := ret[0].(*manager.CDN)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// UpdateCDN indicates an expected call of UpdateCDN.
-func (mr *MockManagerServerMockRecorder) UpdateCDN(arg0, arg1 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateCDN", reflect.TypeOf((*MockManagerServer)(nil).UpdateCDN), arg0, arg1)
-}
-
-// UpdateScheduler mocks base method.
-func (m *MockManagerServer) UpdateScheduler(arg0 context.Context, arg1 *manager.UpdateSchedulerRequest) (*manager.Scheduler, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "UpdateScheduler", arg0, arg1)
-	ret0, _ := ret[0].(*manager.Scheduler)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// UpdateScheduler indicates an expected call of UpdateScheduler.
-func (mr *MockManagerServerMockRecorder) UpdateScheduler(arg0, arg1 interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateScheduler", reflect.TypeOf((*MockManagerServer)(nil).UpdateScheduler), arg0, arg1)
-}
diff --git a/pkg/rpc/manager/server/server.go b/pkg/rpc/manager/server/server.go
deleted file mode 100644
index 0dd15137..00000000
--- a/pkg/rpc/manager/server/server.go
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- *     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 server
-
-import (
-	"context"
-
-	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
-	grpc_zap "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap"
-	grpc_recovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
-	grpc_validator "github.com/grpc-ecosystem/go-grpc-middleware/validator"
-	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
-	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
-	"google.golang.org/grpc"
-
-	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/pkg/rpc/manager"
-)
-
-var defaultStreamMiddleWares = []grpc.StreamServerInterceptor{
-	grpc_validator.StreamServerInterceptor(),
-	grpc_recovery.StreamServerInterceptor(),
-	grpc_prometheus.StreamServerInterceptor,
-	grpc_zap.StreamServerInterceptor(logger.GrpcLogger.Desugar()),
-}
-
-var defaultUnaryMiddleWares = []grpc.UnaryServerInterceptor{
-	grpc_validator.UnaryServerInterceptor(),
-	grpc_recovery.UnaryServerInterceptor(),
-	grpc_prometheus.UnaryServerInterceptor,
-	grpc_zap.UnaryServerInterceptor(logger.GrpcLogger.Desugar()),
-}
-
-// ManagerServer is the server API for Manager service.
-type ManagerServer interface {
-	// Get CDN and CDN cluster configuration
-	GetCDN(context.Context, *manager.GetCDNRequest) (*manager.CDN, error)
-	// Update CDN configuration
-	UpdateCDN(context.Context, *manager.UpdateCDNRequest) (*manager.CDN, error)
-	// Get Scheduler and Scheduler cluster configuration
-	GetScheduler(context.Context, *manager.GetSchedulerRequest) (*manager.Scheduler, error)
-	// Update scheduler configuration
-	UpdateScheduler(context.Context, *manager.UpdateSchedulerRequest) (*manager.Scheduler, error)
-	// List acitve schedulers configuration
-	ListSchedulers(context.Context, *manager.ListSchedulersRequest) (*manager.ListSchedulersResponse, error)
-	// KeepAlive with manager
-	KeepAlive(manager.Manager_KeepAliveServer) error
-}
-
-type proxy struct {
-	server ManagerServer
-	manager.UnimplementedManagerServer
-}
-
-func New(managerServer ManagerServer, jaeger bool, opts ...grpc.ServerOption) *grpc.Server {
-	if jaeger {
-		defaultStreamMiddleWares = append(defaultStreamMiddleWares, otelgrpc.StreamServerInterceptor())
-		defaultUnaryMiddleWares = append(defaultUnaryMiddleWares, otelgrpc.UnaryServerInterceptor())
-	}
-
-	grpcServer := grpc.NewServer(append([]grpc.ServerOption{
-		grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(defaultStreamMiddleWares...)),
-		grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(defaultUnaryMiddleWares...)),
-	}, opts...)...)
-
-	manager.RegisterManagerServer(grpcServer, &proxy{server: managerServer})
-
-	return grpcServer
-}
-
-func (p *proxy) GetCDN(ctx context.Context, req *manager.GetCDNRequest) (*manager.CDN, error) {
-	return p.server.GetCDN(ctx, req)
-}
-
-func (p *proxy) UpdateCDN(ctx context.Context, req *manager.UpdateCDNRequest) (*manager.CDN, error) {
-	return p.server.UpdateCDN(ctx, req)
-}
-
-func (p *proxy) GetScheduler(ctx context.Context, req *manager.GetSchedulerRequest) (*manager.Scheduler, error) {
-	return p.server.GetScheduler(ctx, req)
-}
-
-func (p *proxy) UpdateScheduler(ctx context.Context, req *manager.UpdateSchedulerRequest) (*manager.Scheduler, error) {
-	return p.server.UpdateScheduler(ctx, req)
-}
-
-func (p *proxy) ListSchedulers(ctx context.Context, req *manager.ListSchedulersRequest) (*manager.ListSchedulersResponse, error) {
-	return p.server.ListSchedulers(ctx, req)
-}
-
-func (p *proxy) KeepAlive(req manager.Manager_KeepAliveServer) error {
-	return p.server.KeepAlive(req)
-}
diff --git a/pkg/source/httpprotocol/http_source_client.go b/pkg/source/httpprotocol/http_source_client.go
index 51d380a8..2a6cb3c4 100644
--- a/pkg/source/httpprotocol/http_source_client.go
+++ b/pkg/source/httpprotocol/http_source_client.go
@@ -17,6 +17,7 @@
 package httpprotocol
 
 import (
+	"crypto/tls"
 	"fmt"
 	"net"
 	"net/http"
@@ -53,12 +54,19 @@ func init() {
 		}
 	}
 
-	transport := http.DefaultTransport.(*http.Transport).Clone()
-	transport.TLSClientConfig.InsecureSkipVerify = true
-	transport.DialContext = (&net.Dialer{
-		Timeout:   3 * time.Second,
-		KeepAlive: 30 * time.Second,
-	}).DialContext
+	transport := &http.Transport{
+		DialContext: (&net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+			DualStack: true,
+		}).DialContext,
+		IdleConnTimeout:       90 * time.Second,
+		ResponseHeaderTimeout: 5 * time.Second,
+		ExpectContinueTimeout: 2 * time.Second,
+		TLSClientConfig: &tls.Config{
+			InsecureSkipVerify: true,
+		},
+	}
 
 	if proxy != nil {
 		transport.Proxy = http.ProxyURL(proxy)
diff --git a/pkg/source/source_client.go b/pkg/source/source_client.go
index a2277eb1..49d6e8bd 100644
--- a/pkg/source/source_client.go
+++ b/pkg/source/source_client.go
@@ -31,6 +31,11 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 )
 
+var (
+	// Clinet back-to-source download timeout
+	contextTimeout = 1 * time.Minute
+)
+
 var (
 	// ErrResourceNotReachable represents the url resource is a not reachable.
 	ErrResourceNotReachable = errors.New("resource is not reachable")
@@ -260,7 +265,7 @@ func GetContentLength(request *Request) (int64, error) {
 		return UnknownSourceFileLen, errors.Wrapf(ErrNoClientFound, "scheme: %s", request.URL.Scheme)
 	}
 	if _, ok := request.Context().Deadline(); !ok {
-		ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+		ctx, cancel := context.WithTimeout(context.Background(), contextTimeout)
 		request = request.WithContext(ctx)
 		defer cancel()
 	}
@@ -273,7 +278,7 @@ func IsSupportRange(request *Request) (bool, error) {
 		return false, errors.Wrapf(ErrNoClientFound, "scheme: %s", request.URL.Scheme)
 	}
 	if _, ok := request.Context().Deadline(); !ok {
-		ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+		ctx, cancel := context.WithTimeout(context.Background(), contextTimeout)
 		request = request.WithContext(ctx)
 		defer cancel()
 	}
@@ -289,7 +294,7 @@ func IsExpired(request *Request, info *ExpireInfo) (bool, error) {
 		return false, errors.Wrapf(ErrNoClientFound, "scheme: %s", request.URL.Scheme)
 	}
 	if _, ok := request.Context().Deadline(); !ok {
-		ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+		ctx, cancel := context.WithTimeout(context.Background(), contextTimeout)
 		request = request.WithContext(ctx)
 		defer cancel()
 	}
@@ -302,7 +307,7 @@ func GetLastModified(request *Request) (int64, error) {
 		return -1, errors.Wrapf(ErrNoClientFound, "scheme: %s", request.URL.Scheme)
 	}
 	if _, ok := request.Context().Deadline(); !ok {
-		ctx, cancel := context.WithTimeout(context.Background(), 4*time.Second)
+		ctx, cancel := context.WithTimeout(context.Background(), contextTimeout)
 		request = request.WithContext(ctx)
 		defer cancel()
 	}
diff --git a/pkg/util/fileutils/file_utils.go b/pkg/util/fileutils/file_utils.go
index 7e132606..ae108662 100644
--- a/pkg/util/fileutils/file_utils.go
+++ b/pkg/util/fileutils/file_utils.go
@@ -25,7 +25,6 @@ import (
 
 	"github.com/pkg/errors"
 
-	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/unit"
 )
 
@@ -122,7 +121,7 @@ func PathExist(path string) bool {
 }
 
 func IsDir(path string) bool {
-	f, err := stat(path)
+	f, err := os.Stat(path)
 	if err != nil {
 		return false
 	}
@@ -131,7 +130,7 @@ func IsDir(path string) bool {
 }
 
 func IsRegular(path string) bool {
-	f, err := stat(path)
+	f, err := os.Stat(path)
 	if err != nil {
 		return false
 	}
@@ -153,16 +152,6 @@ func IsEmptyDir(path string) (bool, error) {
 	return false, err
 }
 
-func stat(path string) (os.FileInfo, error) {
-	f, err := os.Stat(path)
-
-	if err != nil {
-		logger.Warnf("stat file %s: %v", path, err)
-	}
-
-	return f, err
-}
-
 // GetFreeSpace gets the free disk space of the path.
 func GetFreeSpace(path string) (unit.Bytes, error) {
 	fs := syscall.Statfs_t{}
diff --git a/pkg/util/timeutils/time_utils.go b/pkg/util/timeutils/time_utils.go
index 9bcc6faa..a06ca0e2 100644
--- a/pkg/util/timeutils/time_utils.go
+++ b/pkg/util/timeutils/time_utils.go
@@ -65,3 +65,13 @@ func UnixSeconds(timeString string) int64 {
 func SecondsUnixTime(seconds int64) time.Time {
 	return time.Unix(seconds, 0)
 }
+
+// NanoToTime converts an int64 nanoseconds to a time
+func NanoToTime(nsec int64) time.Time {
+	return time.Unix(0, nsec)
+}
+
+// SubNano returns the difference between two nanoseconds
+func SubNano(x int64, y int64) time.Duration {
+	return NanoToTime(x).Sub(NanoToTime(y))
+}
diff --git a/scheduler/config/config.go b/scheduler/config/config.go
index c97f7e57..644ab254 100644
--- a/scheduler/config/config.go
+++ b/scheduler/config/config.go
@@ -59,10 +59,9 @@ type Config struct {
 func New() *Config {
 	return &Config{
 		Server: &ServerConfig{
-			IP:          iputils.IPv4,
-			Host:        hostutils.FQDNHostname,
-			Port:        8002,
-			ListenLimit: 1000,
+			IP:   iputils.IPv4,
+			Host: hostutils.FQDNHostname,
+			Port: 8002,
 		},
 		Scheduler: &SchedulerConfig{
 			Algorithm:            "default",
@@ -124,10 +123,6 @@ func (c *Config) Validate() error {
 		return errors.New("server requires parameter port")
 	}
 
-	if c.Server.ListenLimit <= 0 {
-		return errors.New("server requires parameter listenLimit")
-	}
-
 	if c.Scheduler.Algorithm == "" {
 		return errors.New("scheduler requires parameter algorithm")
 	}
@@ -221,9 +216,6 @@ type ServerConfig struct {
 	// Server port
 	Port int `yaml:"port" mapstructure:"port"`
 
-	// Limit the number of requests
-	ListenLimit int `yaml:"listenLimit" mapstructure:"listenLimit"`
-
 	// Server dynamic config cache directory
 	CacheDir string `yaml:"cacheDir" mapstructure:"cacheDir"`
 
diff --git a/scheduler/resource/cdn.go b/scheduler/resource/cdn.go
index 2f367a01..fc0f0778 100644
--- a/scheduler/resource/cdn.go
+++ b/scheduler/resource/cdn.go
@@ -33,6 +33,7 @@ import (
 	"d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
 	cdnclient "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem/client"
 	rpcscheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
+	"d7y.io/dragonfly/v2/pkg/util/timeutils"
 	"d7y.io/dragonfly/v2/scheduler/config"
 )
 
@@ -102,10 +103,9 @@ func (c *cdn) TriggerTask(ctx context.Context, task *Task) (*Peer, *rpcscheduler
 			}
 		}
 
-		peer.Log.Infof("receive piece: %#v %#v", piece, piece.PieceInfo)
-
 		// Handle begin of piece
 		if piece.PieceInfo != nil && piece.PieceInfo.PieceNum == common.BeginOfPiece {
+			peer.Log.Infof("receive begin of piece from cdn: %#v %#v", piece, piece.PieceInfo)
 			if err := peer.FSM.Event(PeerEventDownload); err != nil {
 				return nil, nil, err
 			}
@@ -114,6 +114,7 @@ func (c *cdn) TriggerTask(ctx context.Context, task *Task) (*Peer, *rpcscheduler
 
 		// Handle end of piece
 		if piece.Done {
+			peer.Log.Infof("receive end of from cdn: %#v %#v", piece, piece.PieceInfo)
 			return peer, &rpcscheduler.PeerResult{
 				TotalPieceCount: piece.TotalPieceCount,
 				ContentLength:   piece.ContentLength,
@@ -121,9 +122,9 @@ func (c *cdn) TriggerTask(ctx context.Context, task *Task) (*Peer, *rpcscheduler
 		}
 
 		// Handle piece download successfully
+		peer.Log.Infof("receive piece from cdn: %#v %#v", piece, piece.PieceInfo)
 		peer.Pieces.Set(uint(piece.PieceInfo.PieceNum))
-		// TODO(244372610) CDN should set piece cost
-		peer.AppendPieceCost(0)
+		peer.AppendPieceCost(timeutils.SubNano(int64(piece.EndTime), int64(piece.BeginTime)).Milliseconds())
 		task.StorePiece(piece.PieceInfo)
 	}
 }
diff --git a/scheduler/resource/peer.go b/scheduler/resource/peer.go
index 9544be0a..e414b195 100644
--- a/scheduler/resource/peer.go
+++ b/scheduler/resource/peer.go
@@ -90,9 +90,6 @@ const (
 	// Peer downloaded failed
 	PeerEventDownloadFailed = "DownloadFailed"
 
-	// Peer back to initial pending state
-	PeerEventRestart = "Restart"
-
 	// Peer leaves
 	PeerEventLeave = "Leave"
 )
@@ -173,7 +170,6 @@ func NewPeer(id string, task *Task, host *Host) *Peer {
 				PeerStatePending, PeerStateReceivedTiny, PeerStateReceivedSmall, PeerStateReceivedNormal,
 				PeerStateRunning, PeerStateBackToSource, PeerStateSucceeded,
 			}, Dst: PeerStateFailed},
-			{Name: PeerEventRestart, Src: []string{PeerStateSucceeded}, Dst: PeerStatePending},
 			{Name: PeerEventLeave, Src: []string{PeerStateFailed, PeerStateSucceeded}, Dst: PeerEventLeave},
 		},
 		fsm.Callbacks{
@@ -195,6 +191,7 @@ func NewPeer(id string, task *Task, host *Host) *Peer {
 			},
 			PeerEventDownloadFromBackToSource: func(e *fsm.Event) {
 				p.Task.BackToSourcePeers.Add(p)
+				p.DeleteParent()
 				p.UpdateAt.Store(time.Now())
 				p.Log.Infof("peer state is %s", e.FSM.Current())
 			},
@@ -212,14 +209,12 @@ func NewPeer(id string, task *Task, host *Host) *Peer {
 					p.Task.BackToSourcePeers.Delete(p)
 				}
 
-				p.UpdateAt.Store(time.Now())
-				p.Log.Infof("peer state is %s", e.FSM.Current())
-			},
-			PeerEventRestart: func(e *fsm.Event) {
+				p.DeleteParent()
 				p.UpdateAt.Store(time.Now())
 				p.Log.Infof("peer state is %s", e.FSM.Current())
 			},
 			PeerEventLeave: func(e *fsm.Event) {
+				p.DeleteParent()
 				p.Log.Infof("peer state is %s", e.FSM.Current())
 			},
 		},
@@ -258,10 +253,10 @@ func (p *Peer) DeleteChild(key string) {
 	if !ok {
 		return
 	}
-
 	if _, loaded := p.Children.LoadAndDelete(child.ID); loaded {
 		p.ChildCount.Dec()
 	}
+
 	child.Parent = &atomic.Value{}
 }
 
@@ -282,7 +277,7 @@ func (p *Peer) StoreParent(parent *Peer) {
 
 	p.Parent.Store(parent)
 	if _, loaded := parent.Children.LoadOrStore(p.ID, p); !loaded {
-		p.ChildCount.Inc()
+		parent.ChildCount.Inc()
 	}
 }
 
@@ -295,10 +290,10 @@ func (p *Peer) DeleteParent() {
 	if !ok {
 		return
 	}
-
 	p.Parent = &atomic.Value{}
+
 	if _, loaded := parent.Children.LoadAndDelete(p.ID); loaded {
-		p.ChildCount.Dec()
+		parent.ChildCount.Dec()
 	}
 }
 
@@ -310,9 +305,6 @@ func (p *Peer) ReplaceParent(parent *Peer) {
 
 // Depth represents depth of tree
 func (p *Peer) Depth() int {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-
 	node := p
 	var depth int
 	for node != nil {
@@ -328,7 +320,7 @@ func (p *Peer) Depth() int {
 		}
 
 		// Prevent traversal tree from infinite loop
-		if p == parent {
+		if p.ID == parent.ID {
 			p.Log.Info("tree structure produces an infinite loop")
 			break
 		}
@@ -357,9 +349,11 @@ func isDescendant(ancestor, descendant *Peer) bool {
 		if !ok {
 			return false
 		}
+
 		if parent.ID == ancestor.ID {
 			return true
 		}
+
 		node = parent
 	}
 
@@ -368,17 +362,11 @@ func isDescendant(ancestor, descendant *Peer) bool {
 
 // AppendPieceCost append piece cost to costs slice
 func (p *Peer) AppendPieceCost(cost int64) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
 	p.pieceCosts = append(p.pieceCosts, cost)
 }
 
 // PieceCosts return piece costs slice
 func (p *Peer) PieceCosts() []int64 {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-
 	return p.pieceCosts
 }
 
diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go
index 491a777f..b2a570fd 100644
--- a/scheduler/scheduler.go
+++ b/scheduler/scheduler.go
@@ -24,7 +24,6 @@ import (
 	"time"
 
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
-	"golang.org/x/net/netutil"
 	"google.golang.org/grpc"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
@@ -203,11 +202,10 @@ func (s *Server) Serve() error {
 		logger.Fatalf("net listener failed to start: %v", err)
 	}
 	defer listener.Close()
-	limitListener := netutil.LimitListener(listener, s.config.Server.ListenLimit)
 
 	// Started GRPC server
-	logger.Infof("started grpc server at %s://%s with max connection %d", limitListener.Addr().Network(), limitListener.Addr().String(), s.config.Server.ListenLimit)
-	if err := s.grpcServer.Serve(limitListener); err != nil {
+	logger.Infof("started grpc server at %s://%s", listener.Addr().Network(), listener.Addr().String())
+	if err := s.grpcServer.Serve(listener); err != nil {
 		logger.Errorf("stoped grpc server: %v", err)
 		return err
 	}
diff --git a/scheduler/scheduler/evaluator/evaluator_base.go b/scheduler/scheduler/evaluator/evaluator_base.go
index 35d47de4..4909b22a 100644
--- a/scheduler/scheduler/evaluator/evaluator_base.go
+++ b/scheduler/scheduler/evaluator/evaluator_base.go
@@ -56,9 +56,9 @@ const (
 )
 
 const (
-	// If the number of samples is greater than or equal to 10,
+	// If the number of samples is greater than or equal to 30,
 	// it is close to the normal distribution
-	normalDistributionLen = 10
+	normalDistributionLen = 30
 
 	// When costs len is greater than or equal to 2,
 	// the last cost can be compared and calculated
@@ -110,9 +110,13 @@ func calculatePieceScore(parent *resource.Peer, child *resource.Peer, totalPiece
 
 // calculateFreeLoadScore 0.0~1.0 larger and better
 func calculateFreeLoadScore(host *resource.Host) float64 {
-	load := host.PeerCount.Load()
-	totalLoad := host.UploadLoadLimit.Load()
-	return float64(totalLoad-int32(load)) / float64(totalLoad)
+	uploadLoadLimit := host.UploadLoadLimit.Load()
+	freeUploadLoad := host.FreeUploadLoad()
+	if uploadLoadLimit > 0 && freeUploadLoad > 0 {
+		return float64(freeUploadLoad) / float64(uploadLoadLimit)
+	}
+
+	return minScore
 }
 
 // calculateHostTypeAffinityScore 0.0~1.0 larger and better
@@ -174,7 +178,7 @@ func calculateMultiElementAffinityScore(dst, src string) float64 {
 
 func (eb *evaluatorBase) IsBadNode(peer *resource.Peer) bool {
 	if peer.FSM.Is(resource.PeerStateFailed) || peer.FSM.Is(resource.PeerStateLeave) || peer.FSM.Is(resource.PeerStatePending) {
-		peer.Log.Infof("peer is bad node because peer status is %s", peer.FSM.Current())
+		peer.Log.Debugf("peer is bad node because peer status is %s", peer.FSM.Current())
 		return true
 	}
 
@@ -183,7 +187,7 @@ func (eb *evaluatorBase) IsBadNode(peer *resource.Peer) bool {
 	len := len(costs)
 	// Peer has not finished downloading enough piece
 	if len < minAvailableCostLen {
-		logger.Infof("peer %s has not finished downloading enough piece, it can't be bad node", peer.ID)
+		logger.Debugf("peer %s has not finished downloading enough piece, it can't be bad node", peer.ID)
 		return false
 	}
 
@@ -194,7 +198,7 @@ func (eb *evaluatorBase) IsBadNode(peer *resource.Peer) bool {
 	// if the last cost is twenty times more than mean, it is bad node.
 	if len < normalDistributionLen {
 		isBadNode := big.NewFloat(lastCost).Cmp(big.NewFloat(mean*20)) > 0
-		logger.Infof("peer %s mean is %.2f and it is bad node: %t", peer.ID, mean, isBadNode)
+		logger.Debugf("peer %s mean is %.2f and it is bad node: %t", peer.ID, mean, isBadNode)
 		return isBadNode
 	}
 
@@ -203,7 +207,7 @@ func (eb *evaluatorBase) IsBadNode(peer *resource.Peer) bool {
 	// refer to https://en.wikipedia.org/wiki/68%E2%80%9395%E2%80%9399.7_rule
 	stdev, _ := stats.StandardDeviation(costs[:len-1]) // nolint: errcheck
 	isBadNode := big.NewFloat(lastCost).Cmp(big.NewFloat(mean+3*stdev)) > 0
-	logger.Infof("peer %s meet the normal distribution, costs mean is %.2f and standard deviation is %.2f, peer is bad node: %t",
+	logger.Debugf("peer %s meet the normal distribution, costs mean is %.2f and standard deviation is %.2f, peer is bad node: %t",
 		peer.ID, mean, stdev, isBadNode)
 	return isBadNode
 }
diff --git a/scheduler/scheduler/scheduler.go b/scheduler/scheduler/scheduler.go
index 5914033c..c320f0f5 100644
--- a/scheduler/scheduler/scheduler.go
+++ b/scheduler/scheduler/scheduler.go
@@ -35,11 +35,14 @@ const (
 	// Default number of pieces downloaded in parallel
 	defaultParallelCount = 4
 
-	// Default number of available parents after filtering
-	defaultFilterParentCount = 5
+	// Default limit the number of filter traversals
+	defaultFilterParentLimit = 10
+
+	// Default tree available depth
+	defaultAvailableDepth = 2
 
 	// Default tree depth limit
-	defaultDepthLimit = 10
+	defaultDepthLimit = 4
 )
 
 type Scheduler interface {
@@ -100,8 +103,8 @@ func (s *scheduler) ScheduleParent(ctx context.Context, peer *resource.Peer, blo
 				peer.Log.Errorf("send packet failed: %v", err)
 				return
 			}
-			peer.Log.Infof("peer scheduling %d times and back-to-source limit %d times, cdn peer is %#v, return code %d",
-				n, s.config.RetryBackSourceLimit, cdnPeer, base.Code_SchedNeedBackSource)
+			peer.Log.Infof("peer scheduling %d times and cdn peer is %#v, peer downloads back-to-source %d",
+				n, cdnPeer, base.Code_SchedNeedBackSource)
 
 			if err := peer.FSM.Event(resource.PeerEventDownloadFromBackToSource); err != nil {
 				peer.Log.Errorf("peer fsm event failed: %v", err)
@@ -119,7 +122,6 @@ func (s *scheduler) ScheduleParent(ctx context.Context, peer *resource.Peer, blo
 
 			// If the peer downloads back-to-source, its parent needs to be deleted
 			peer.DeleteParent()
-			peer.Task.Log.Info("peer back to source successfully")
 			return
 		}
 
@@ -136,7 +138,7 @@ func (s *scheduler) ScheduleParent(ctx context.Context, peer *resource.Peer, blo
 				peer.Log.Errorf("send packet failed: %v", err)
 				return
 			}
-			peer.Log.Infof("peer scheduling exceeds the limit %d times and return code %d", s.config.RetryLimit, base.Code_SchedTaskStatusError)
+			peer.Log.Errorf("peer scheduling exceeds the limit %d times and return code %d", s.config.RetryLimit, base.Code_SchedTaskStatusError)
 			return
 		}
 
@@ -221,17 +223,19 @@ func (s *scheduler) FindParent(ctx context.Context, peer *resource.Peer, blockli
 
 // Filter the parent that can be scheduled
 func (s *scheduler) filterParents(peer *resource.Peer, blocklist set.SafeSet) []*resource.Peer {
-	filterParentCount := defaultFilterParentCount
-	if config, ok := s.dynconfig.GetSchedulerClusterConfig(); ok && config.FilterParentCount > 0 {
-		filterParentCount = int(config.FilterParentCount)
+	filterParentLimit := defaultFilterParentLimit
+	if config, ok := s.dynconfig.GetSchedulerClusterConfig(); ok && filterParentLimit > 0 {
+		filterParentLimit = int(config.FilterParentLimit)
 	}
 
 	var parents []*resource.Peer
 	var parentIDs []string
+	var n int
 	peer.Task.Peers.Range(func(_, value interface{}) bool {
-		if len(parents) >= filterParentCount {
+		if n > filterParentLimit {
 			return false
 		}
+		n++
 
 		parent, ok := value.(*resource.Peer)
 		if !ok {
@@ -239,37 +243,45 @@ func (s *scheduler) filterParents(peer *resource.Peer, blocklist set.SafeSet) []
 		}
 
 		if blocklist.Contains(parent.ID) {
-			peer.Log.Infof("parent %s is not selected because it is in blocklist", parent.ID)
+			peer.Log.Debugf("parent %s is not selected because it is in blocklist", parent.ID)
 			return true
 		}
 
-		if parent == peer {
-			peer.Log.Info("parent is not selected because it is same")
+		if parent.ID == peer.ID {
+			peer.Log.Debug("parent is not selected because it is same")
 			return true
 		}
 
 		if s.evaluator.IsBadNode(parent) {
-			peer.Log.Infof("parent %s is not selected because it is bad node", parent.ID)
+			peer.Log.Debugf("parent %s is not selected because it is bad node", parent.ID)
+			return true
+		}
+
+		peerChildCount := peer.ChildCount.Load()
+		parentDepth := parent.Depth()
+		if peerChildCount > 0 && parentDepth > defaultAvailableDepth {
+			peer.Log.Debugf("peer has %d children and parent %s depth is %d", peerChildCount, parent.ID, parentDepth)
 			return true
 		}
 
-		if parent.Depth() > defaultDepthLimit {
-			peer.Log.Infof("exceeds the %d depth limit of the tree", defaultDepthLimit)
+		peerDepth := peer.Depth()
+		if parentDepth+peerDepth > defaultDepthLimit {
+			peer.Log.Debugf("exceeds the %d depth limit of the tree, peer depth is %d, parent %s is %d", defaultDepthLimit, peerDepth, parent.ID, parentDepth)
 			return true
 		}
 
 		if parent.IsDescendant(peer) {
-			peer.Log.Infof("parent %s is not selected because it is descendant", parent.ID)
+			peer.Log.Debugf("parent %s is not selected because it is descendant", parent.ID)
 			return true
 		}
 
 		if parent.IsAncestor(peer) {
-			peer.Log.Infof("parent %s is not selected because it is ancestor", parent.ID)
+			peer.Log.Debugf("parent %s is not selected because it is ancestor", parent.ID)
 			return true
 		}
 
 		if parent.Host.FreeUploadLoad() <= 0 {
-			peer.Log.Infof("parent %s is not selected because its free upload is empty", parent.ID)
+			peer.Log.Debugf("parent %s is not selected because its free upload is empty", parent.ID)
 			return true
 		}
 
diff --git a/scheduler/service/service.go b/scheduler/service/service.go
index 0a159067..a0bb2bc6 100644
--- a/scheduler/service/service.go
+++ b/scheduler/service/service.go
@@ -20,7 +20,6 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"time"
 
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
@@ -29,6 +28,7 @@ import (
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/base/common"
 	rpcscheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
+	"d7y.io/dragonfly/v2/pkg/util/timeutils"
 	"d7y.io/dragonfly/v2/scheduler/config"
 	"d7y.io/dragonfly/v2/scheduler/metrics"
 	"d7y.io/dragonfly/v2/scheduler/resource"
@@ -80,7 +80,7 @@ func (s *Service) RegisterPeerTask(ctx context.Context, req *rpcscheduler.PeerTa
 
 	// Task has been successful
 	if task.FSM.Is(resource.TaskStateSucceeded) {
-		peer.Log.Info("task has been successful")
+		peer.Log.Info("tasks can be reused")
 		sizeScope := task.SizeScope()
 		switch sizeScope {
 		case base.SizeScope_TINY:
@@ -107,8 +107,7 @@ func (s *Service) RegisterPeerTask(ctx context.Context, req *rpcscheduler.PeerTa
 			fallthrough
 		case base.SizeScope_SMALL:
 			peer.Log.Info("task size scope is small")
-			// If the file is registered as a small type,
-			// there is no need to build a tree, just find the parent and return
+			// There is no need to build a tree, just find the parent and return
 			parent, ok := s.scheduler.FindParent(ctx, peer, set.NewSafeSet())
 			if !ok {
 				peer.Log.Warn("task size scope is small and it can not select parent")
@@ -124,6 +123,22 @@ func (s *Service) RegisterPeerTask(ctx context.Context, req *rpcscheduler.PeerTa
 				}, nil
 			}
 
+			// When task size scope is small, parent must be downloaded successfully
+			// before returning to the parent directly
+			if !parent.FSM.Is(resource.PeerStateSucceeded) {
+				peer.Log.Infof("task size scope is small and download state %s is not PeerStateSucceeded", parent.FSM.Current())
+				if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
+					dferr := dferrors.New(base.Code_SchedError, err.Error())
+					peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
+					return nil, dferr
+				}
+
+				return &rpcscheduler.RegisterResult{
+					TaskId:    task.ID,
+					SizeScope: base.SizeScope_NORMAL,
+				}, nil
+			}
+
 			firstPiece, ok := task.LoadPiece(0)
 			if !ok {
 				peer.Log.Warn("task size scope is small and it can not get first piece")
@@ -183,7 +198,7 @@ func (s *Service) RegisterPeerTask(ctx context.Context, req *rpcscheduler.PeerTa
 	}
 
 	// Task is unsuccessful
-	peer.Log.Info("task is unsuccessful and needs to be register")
+	peer.Log.Infof("task state is %s and needs to be register", task.FSM.Current())
 	if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
 		dferr := dferrors.New(base.Code_SchedError, err.Error())
 		peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
@@ -238,17 +253,17 @@ func (s *Service) ReportPieceResult(stream rpcscheduler.Scheduler_ReportPieceRes
 			defer peer.DeleteStream()
 		}
 
-		peer.Log.Infof("receive piece: %#v %#v", piece, piece.PieceInfo)
-
 		if piece.PieceInfo != nil {
 			// Handle begin of piece
 			if piece.PieceInfo.PieceNum == common.BeginOfPiece {
+				peer.Log.Infof("receive begin of piece: %#v %#v", piece, piece.PieceInfo)
 				s.handleBeginOfPiece(ctx, peer)
 				continue
 			}
 
 			// Handle end of piece
 			if piece.PieceInfo.PieceNum == common.EndOfPiece {
+				peer.Log.Infof("receive end of piece: %#v %#v", piece, piece.PieceInfo)
 				s.handleEndOfPiece(ctx, peer)
 				continue
 			}
@@ -256,6 +271,7 @@ func (s *Service) ReportPieceResult(stream rpcscheduler.Scheduler_ReportPieceRes
 
 		// Handle piece download successfully
 		if piece.Success {
+			peer.Log.Infof("receive piece: %#v %#v", piece, piece.PieceInfo)
 			s.handlePieceSuccess(ctx, peer, piece)
 
 			// Collect peer host traffic metrics
@@ -272,16 +288,18 @@ func (s *Service) ReportPieceResult(stream rpcscheduler.Scheduler_ReportPieceRes
 
 		// Handle piece download code
 		if piece.Code != base.Code_Success {
-			// FIXME(244372610) When dfdaemon download peer return empty, retry later.
 			if piece.Code == base.Code_ClientWaitPieceReady {
-				peer.Log.Infof("receive piece code %d and wait for dfdaemon piece ready", piece.Code)
+				peer.Log.Debugf("receive piece code %d and wait for dfdaemon piece ready", piece.Code)
 				continue
 			}
 
 			// Handle piece download failed
-			peer.Log.Errorf("receive failed piece: %#v %#v", piece, piece.PieceInfo)
+			peer.Log.Errorf("receive failed piece: %#v", piece)
 			s.handlePieceFail(ctx, peer, piece)
+			continue
 		}
+
+		peer.Log.Warnf("receive unknow piece: %#v %#v", piece, piece.PieceInfo)
 	}
 }
 
@@ -289,12 +307,12 @@ func (s *Service) ReportPieceResult(stream rpcscheduler.Scheduler_ReportPieceRes
 func (s *Service) ReportPeerResult(ctx context.Context, req *rpcscheduler.PeerResult) error {
 	peer, ok := s.resource.PeerManager().Load(req.PeerId)
 	if !ok {
-		logger.Errorf("report peer result: peer %s is not exists", req.PeerId)
+		logger.Errorf("report peer result and peer %s is not exists", req.PeerId)
 		return dferrors.Newf(base.Code_SchedPeerNotFound, "peer %s not found", req.PeerId)
 	}
 
-	peer.Log.Infof("report peer result request: %#v", req)
 	if !req.Success {
+		peer.Log.Errorf("report peer failed result: %s %#v", req.Code, req)
 		if peer.FSM.Is(resource.PeerStateBackToSource) {
 			s.handleTaskFail(ctx, peer.Task)
 		}
@@ -302,6 +320,7 @@ func (s *Service) ReportPeerResult(ctx context.Context, req *rpcscheduler.PeerRe
 		return nil
 	}
 
+	peer.Log.Infof("report peer result: %#v", req)
 	if peer.FSM.Is(resource.PeerStateBackToSource) {
 		s.handleTaskSuccess(ctx, peer.Task, req)
 	}
@@ -313,11 +332,11 @@ func (s *Service) ReportPeerResult(ctx context.Context, req *rpcscheduler.PeerRe
 func (s *Service) LeaveTask(ctx context.Context, req *rpcscheduler.PeerTarget) error {
 	peer, ok := s.resource.PeerManager().Load(req.PeerId)
 	if !ok {
-		logger.Errorf("leave task: peer %s is not exists", req.PeerId)
+		logger.Errorf("leave task and peer %s is not exists", req.PeerId)
 		return dferrors.Newf(base.Code_SchedPeerNotFound, "peer %s not found", req.PeerId)
 	}
 
-	peer.Log.Infof("leave task request: %#v", req)
+	peer.Log.Infof("leave task: %#v", req)
 	if err := peer.FSM.Event(resource.PeerEventLeave); err != nil {
 		peer.Log.Errorf("peer fsm event failed: %v", err)
 		return dferrors.Newf(base.Code_SchedTaskStatusError, err.Error())
@@ -333,6 +352,7 @@ func (s *Service) LeaveTask(ctx context.Context, req *rpcscheduler.PeerTarget) e
 		blocklist := set.NewSafeSet()
 		blocklist.Add(peer.ID)
 
+		child.Log.Infof("schedule parent because of parent peer %s is leaving", peer.ID)
 		s.scheduler.ScheduleParent(ctx, child, blocklist)
 		return true
 	})
@@ -347,9 +367,6 @@ func (s *Service) registerTask(ctx context.Context, req *rpcscheduler.PeerTaskRe
 	task := resource.NewTask(idgen.TaskID(req.Url, req.UrlMeta), req.Url, s.config.Scheduler.BackSourceCount, req.UrlMeta)
 	task, loaded := s.resource.TaskManager().LoadOrStore(task)
 	if loaded && task.HasAvailablePeer() && (task.FSM.Is(resource.TaskStateRunning) || task.FSM.Is(resource.TaskStateSucceeded)) {
-		// Task is healthy and can be reused
-		task.UpdateAt.Store(time.Now())
-		task.Log.Infof("reuse task and status is %s", task.FSM.Current())
 		return task, nil
 	}
 
@@ -392,10 +409,10 @@ func (s *Service) registerPeer(ctx context.Context, req *rpcscheduler.PeerTaskRe
 	peer, loaded := s.resource.PeerManager().LoadOrStore(resource.NewPeer(req.PeerId, task, host))
 	if !loaded {
 		peer.Log.Info("create new peer")
-	} else {
-		peer.Log.Info("peer already exists")
+		return peer
 	}
 
+	peer.Log.Info("peer already exists")
 	return peer
 }
 
@@ -410,6 +427,7 @@ func (s *Service) triggerCDNTask(ctx context.Context, task *resource.Task) {
 	}
 
 	// Update the task status first to help peer scheduling evaluation and scoring
+	peer.Log.Info("trigger cdn download task successfully")
 	s.handleTaskSuccess(ctx, task, endOfPiece)
 	s.handlePeerSuccess(ctx, peer)
 }
@@ -419,7 +437,7 @@ func (s *Service) handleBeginOfPiece(ctx context.Context, peer *resource.Peer) {
 	switch peer.FSM.Current() {
 	case resource.PeerStateBackToSource:
 		// Back to the source download process, peer directly returns
-		peer.Log.Info("peer back to source")
+		peer.Log.Info("peer downloads back-to-source when receive the begin of piece")
 		return
 	case resource.PeerStateReceivedTiny:
 		// When the task is tiny,
@@ -447,6 +465,8 @@ func (s *Service) handleBeginOfPiece(ctx context.Context, peer *resource.Peer) {
 		// to help peer to schedule the parent node
 		blocklist := set.NewSafeSet()
 		blocklist.Add(peer.ID)
+
+		peer.Log.Infof("schedule parent because of peer receive begin of piece")
 		s.scheduler.ScheduleParent(ctx, peer, blocklist)
 	default:
 		peer.Log.Warnf("peer state is %s when receive the begin of piece", peer.FSM.Current())
@@ -460,7 +480,7 @@ func (s *Service) handleEndOfPiece(ctx context.Context, peer *resource.Peer) {}
 func (s *Service) handlePieceSuccess(ctx context.Context, peer *resource.Peer, piece *rpcscheduler.PieceResult) {
 	// Update peer piece info
 	peer.Pieces.Set(uint(piece.PieceInfo.PieceNum))
-	peer.AppendPieceCost(int64(piece.EndTime - piece.BeginTime))
+	peer.AppendPieceCost(timeutils.SubNano(int64(piece.EndTime), int64(piece.BeginTime)).Milliseconds())
 
 	// When the peer downloads back-to-source,
 	// piece downloads successfully updates the task piece info
@@ -473,14 +493,13 @@ func (s *Service) handlePieceSuccess(ctx context.Context, peer *resource.Peer, p
 func (s *Service) handlePieceFail(ctx context.Context, peer *resource.Peer, piece *rpcscheduler.PieceResult) {
 	// Failed to download piece back-to-source
 	if peer.FSM.Is(resource.PeerStateBackToSource) {
-		peer.Log.Error("peer back to source finished with fail piece")
 		return
 	}
 
 	// If parent can not found, reschedule parent.
 	parent, ok := s.resource.PeerManager().Load(piece.DstPid)
 	if !ok {
-		peer.Log.Errorf("can not found parent %s and reschedule", piece.DstPid)
+		peer.Log.Errorf("schedule parent because of peer can not found parent %s", piece.DstPid)
 		s.scheduler.ScheduleParent(ctx, peer, set.NewSafeSet())
 		return
 	}
@@ -488,7 +507,7 @@ func (s *Service) handlePieceFail(ctx context.Context, peer *resource.Peer, piec
 	// It’s not a case of back-to-source downloading failed,
 	// to help peer to reschedule the parent node
 	switch piece.Code {
-	case base.Code_ClientPieceDownloadFail, base.Code_PeerTaskNotFound, base.Code_CDNError, base.Code_CDNTaskDownloadFail:
+	case base.Code_PeerTaskNotFound, base.Code_CDNError, base.Code_CDNTaskDownloadFail:
 		if err := parent.FSM.Event(resource.PeerEventDownloadFailed); err != nil {
 			peer.Log.Errorf("peer fsm event failed: %v", err)
 			break
@@ -523,6 +542,7 @@ func (s *Service) handlePieceFail(ctx context.Context, peer *resource.Peer, piec
 	blocklist := set.NewSafeSet()
 	blocklist.Add(parent.ID)
 
+	peer.Log.Infof("schedule parent because of peer receive failed piece")
 	s.scheduler.ScheduleParent(ctx, peer, blocklist)
 }
 
@@ -536,7 +556,7 @@ func (s *Service) handlePeerSuccess(ctx context.Context, peer *resource.Peer) {
 			// Tiny file downloaded successfully
 			peer.Task.DirectPiece = data
 		} else {
-			peer.Log.Warnf("download tiny file length is %d, task content length is %d, download is failed: %v", len(data), peer.Task.ContentLength.Load(), err)
+			peer.Log.Warnf("download tiny file length is %d, task content length is %d, downloading is failed: %v", len(data), peer.Task.ContentLength.Load(), err)
 		}
 	}
 
@@ -563,6 +583,7 @@ func (s *Service) handlePeerFail(ctx context.Context, peer *resource.Peer) {
 			return true
 		}
 
+		child.Log.Infof("schedule parent because of parent peer %s is failed", peer.ID)
 		s.scheduler.ScheduleParent(ctx, child, blocklist)
 		return true
 	})


Test patch:
diff --git a/client/config/peerhost_test.go b/client/config/peerhost_test.go
index 6be66729..7a9c2840 100644
--- a/client/config/peerhost_test.go
+++ b/client/config/peerhost_test.go
@@ -347,7 +347,7 @@ func TestPeerHostOption_Load(t *testing.T) {
 				Insecure: true,
 				Direct:   false,
 			},
-			Proxies: []*Proxy{
+			Proxies: []*ProxyRule{
 				{
 					Regx:     proxyExp,
 					UseHTTPS: false,
diff --git a/client/daemon/peer/peertask_reuse_test.go b/client/daemon/peer/peertask_reuse_test.go
new file mode 100644
index 00000000..2003f9fc
--- /dev/null
+++ b/client/daemon/peer/peertask_reuse_test.go
@@ -0,0 +1,501 @@
+/*
+ *     Copyright 2022 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package peer
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"os"
+	"path"
+	"testing"
+
+	"github.com/go-http-utils/headers"
+	"github.com/golang/mock/gomock"
+	testifyassert "github.com/stretchr/testify/assert"
+
+	"d7y.io/dragonfly/v2/client/clientutil"
+	"d7y.io/dragonfly/v2/client/daemon/storage"
+	"d7y.io/dragonfly/v2/client/daemon/test"
+	ms "d7y.io/dragonfly/v2/client/daemon/test/mock/storage"
+	"d7y.io/dragonfly/v2/pkg/rpc/base"
+	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
+)
+
+func TestReuseFilePeerTask(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	assert := testifyassert.New(t)
+
+	testBytes, err := os.ReadFile(test.File)
+	assert.Nil(err)
+	testOutput := path.Join(os.TempDir(), "d7y-reuse-output.data")
+	defer os.Remove(testOutput)
+
+	var testCases = []struct {
+		name           string
+		request        *FileTaskRequest
+		enablePrefetch bool
+		storageManager func(sm *ms.MockManager)
+		verify         func(pg chan *FileTaskProgress, ok bool)
+	}{
+		{
+			name: "normal completed task found",
+			request: &FileTaskRequest{
+				PeerTaskRequest: scheduler.PeerTaskRequest{
+					PeerId: "",
+					Url:    "http://example.com/1",
+					UrlMeta: &base.UrlMeta{
+						Digest: "",
+						Tag:    "",
+						Range:  "",
+						Filter: "",
+						Header: nil,
+					},
+				},
+				Output: testOutput,
+				Range:  nil,
+			},
+			enablePrefetch: false,
+			storageManager: func(sm *ms.MockManager) {
+				var taskID string
+				sm.EXPECT().FindCompletedTask(gomock.Any()).DoAndReturn(
+					func(id string) *storage.ReusePeerTask {
+						taskID = id
+						return &storage.ReusePeerTask{
+							PeerTaskMetadata: storage.PeerTaskMetadata{
+								TaskID: taskID,
+							},
+							ContentLength: 10,
+							TotalPieces:   0,
+							PieceMd5Sign:  "",
+						}
+					})
+				sm.EXPECT().Store(gomock.Any(), gomock.Any()).DoAndReturn(
+					func(ctx context.Context, req *storage.StoreRequest) error {
+						return os.WriteFile(req.Destination, testBytes[0:10], 0644)
+					})
+			},
+			verify: func(pg chan *FileTaskProgress, ok bool) {
+				assert.True(ok)
+				data, err := os.ReadFile(testOutput)
+				assert.Nil(err)
+				assert.Equal(testBytes[0:10], data)
+			},
+		},
+		{
+			name: "normal completed task not found",
+			request: &FileTaskRequest{
+				PeerTaskRequest: scheduler.PeerTaskRequest{
+					PeerId: "",
+					Url:    "http://example.com/1",
+					UrlMeta: &base.UrlMeta{
+						Digest: "",
+						Tag:    "",
+						Range:  "",
+						Filter: "",
+						Header: nil,
+					},
+				},
+				Output: testOutput,
+				Range:  nil,
+			},
+			enablePrefetch: false,
+			storageManager: func(sm *ms.MockManager) {
+				//sm.EXPECT().FindPartialCompletedTask(gomock.Any(), gomock.Any()).DoAndReturn(
+				//	func(taskID string, rg *clientutil.Range) *storage.ReusePeerTask {
+				//		return nil
+				//	})
+				//sm.EXPECT().FindCompletedSubTask(gomock.Any()).DoAndReturn(
+				//	func(taskID string) *storage.ReusePeerTask {
+				//		return nil
+				//	})
+				sm.EXPECT().FindCompletedTask(gomock.Any()).DoAndReturn(
+					func(taskID string) *storage.ReusePeerTask {
+						return nil
+					})
+			},
+			verify: func(pg chan *FileTaskProgress, ok bool) {
+				assert.False(ok)
+				assert.Nil(pg)
+			},
+		},
+		{
+			name: "normal completed subtask found",
+			request: &FileTaskRequest{
+				PeerTaskRequest: scheduler.PeerTaskRequest{
+					PeerId: "",
+					Url:    "http://example.com/1",
+					UrlMeta: &base.UrlMeta{
+						Digest: "",
+						Tag:    "",
+						Range:  "",
+						Filter: "",
+						Header: nil,
+					},
+				},
+				Output: testOutput,
+				Range:  &clientutil.Range{Start: 200, Length: 100},
+			},
+			enablePrefetch: true,
+			storageManager: func(sm *ms.MockManager) {
+				var taskID string
+				sm.EXPECT().FindCompletedSubTask(gomock.Any()).DoAndReturn(
+					func(id string) *storage.ReusePeerTask {
+						taskID = id
+						return &storage.ReusePeerTask{
+							PeerTaskMetadata: storage.PeerTaskMetadata{
+								TaskID: taskID,
+							},
+							ContentLength: 10,
+							TotalPieces:   0,
+							PieceMd5Sign:  "",
+						}
+					})
+				sm.EXPECT().Store(gomock.Any(), gomock.Any()).DoAndReturn(
+					func(ctx context.Context, req *storage.StoreRequest) error {
+						return os.WriteFile(req.Destination, testBytes[200:300], 0644)
+					})
+			},
+			verify: func(pg chan *FileTaskProgress, ok bool) {
+				assert.True(ok)
+				data, err := os.ReadFile(testOutput)
+				assert.Nil(err)
+				assert.Equal(testBytes[200:300], data)
+			},
+		},
+		{
+			name: "normal completed subtask not found",
+			request: &FileTaskRequest{
+				PeerTaskRequest: scheduler.PeerTaskRequest{
+					PeerId: "",
+					Url:    "http://example.com/1",
+					UrlMeta: &base.UrlMeta{
+						Digest: "",
+						Tag:    "",
+						Range:  "",
+						Filter: "",
+						Header: nil,
+					},
+				},
+				Output: testOutput,
+				Range:  &clientutil.Range{Start: 0, Length: 10},
+			},
+			enablePrefetch: true,
+			storageManager: func(sm *ms.MockManager) {
+				sm.EXPECT().FindPartialCompletedTask(gomock.Any(), gomock.Any()).DoAndReturn(
+					func(taskID string, rg *clientutil.Range) *storage.ReusePeerTask {
+						return nil
+					})
+				sm.EXPECT().FindCompletedSubTask(gomock.Any()).DoAndReturn(
+					func(taskID string) *storage.ReusePeerTask {
+						return nil
+					})
+			},
+			verify: func(pg chan *FileTaskProgress, ok bool) {
+				assert.False(ok)
+				assert.Nil(pg)
+			},
+		},
+		{
+			name: "partial task found",
+			request: &FileTaskRequest{
+				PeerTaskRequest: scheduler.PeerTaskRequest{
+					PeerId: "",
+					Url:    "http://example.com/1",
+					UrlMeta: &base.UrlMeta{
+						Digest: "",
+						Tag:    "",
+						Range:  "",
+						Filter: "",
+						Header: nil,
+					},
+				},
+				Output: testOutput,
+				Range:  &clientutil.Range{Start: 300, Length: 100},
+			},
+			enablePrefetch: true,
+			storageManager: func(sm *ms.MockManager) {
+				var taskID string
+				sm.EXPECT().FindCompletedSubTask(gomock.Any()).DoAndReturn(
+					func(id string) *storage.ReusePeerTask {
+						return nil
+					})
+				sm.EXPECT().FindPartialCompletedTask(gomock.Any(), gomock.Any()).DoAndReturn(
+					func(id string, rg *clientutil.Range) *storage.ReusePeerTask {
+						taskID = id
+						return &storage.ReusePeerTask{
+							PeerTaskMetadata: storage.PeerTaskMetadata{
+								TaskID: taskID,
+							},
+							ContentLength: 100,
+							TotalPieces:   0,
+							PieceMd5Sign:  "",
+						}
+					})
+				sm.EXPECT().ReadAllPieces(gomock.Any(), gomock.Any()).DoAndReturn(
+					func(ctx context.Context, req *storage.ReadAllPiecesRequest) (io.ReadCloser, error) {
+						assert.Equal(taskID, req.TaskID)
+						return io.NopCloser(bytes.NewBuffer(testBytes[300:400])), nil
+					})
+			},
+			verify: func(pg chan *FileTaskProgress, ok bool) {
+				assert.True(ok)
+				data, err := os.ReadFile(testOutput)
+				assert.Nil(err)
+				assert.Equal(testBytes[300:400], data)
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			defer os.Remove(testOutput)
+			sm := ms.NewMockManager(ctrl)
+			tc.storageManager(sm)
+			ptm := &peerTaskManager{
+				host:           &scheduler.PeerHost{},
+				enablePrefetch: tc.enablePrefetch,
+				storageManager: sm,
+			}
+			tc.verify(ptm.tryReuseFilePeerTask(context.Background(), tc.request))
+		})
+	}
+}
+
+func TestReuseStreamPeerTask(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	assert := testifyassert.New(t)
+
+	var testCases = []struct {
+		name           string
+		request        *StreamTaskRequest
+		enablePrefetch bool
+		storageManager func(sm *ms.MockManager)
+		verify         func(rc io.ReadCloser, attr map[string]string, ok bool)
+	}{
+		{
+			name: "normal completed task found",
+			request: &StreamTaskRequest{
+				URL: "http://example.com/1",
+				URLMeta: &base.UrlMeta{
+					Digest: "",
+					Tag:    "",
+					Range:  "",
+					Filter: "",
+					Header: nil,
+				},
+				Range:  nil,
+				PeerID: "",
+			},
+			enablePrefetch: false,
+			storageManager: func(sm *ms.MockManager) {
+				var taskID string
+				sm.EXPECT().FindCompletedTask(gomock.Any()).DoAndReturn(
+					func(id string) *storage.ReusePeerTask {
+						taskID = id
+						return &storage.ReusePeerTask{
+							PeerTaskMetadata: storage.PeerTaskMetadata{
+								TaskID: taskID,
+							},
+							ContentLength: 10,
+							TotalPieces:   0,
+							PieceMd5Sign:  "",
+						}
+					})
+				sm.EXPECT().ReadAllPieces(gomock.Any(), gomock.Any()).DoAndReturn(
+					func(ctx context.Context, req *storage.ReadAllPiecesRequest) (io.ReadCloser, error) {
+						assert.Equal(taskID, req.TaskID)
+						return io.NopCloser(bytes.NewBuffer([]byte("1111111111"))), nil
+					})
+			},
+			verify: func(rc io.ReadCloser, attr map[string]string, ok bool) {
+				assert.True(ok)
+				assert.NotNil(rc)
+				assert.Equal("10", attr[headers.ContentLength])
+				_, exist := attr[headers.ContentRange]
+				assert.False(exist)
+			},
+		},
+		{
+			name: "normal completed task not found",
+			request: &StreamTaskRequest{
+				URL: "http://example.com/1",
+				URLMeta: &base.UrlMeta{
+					Digest: "",
+					Tag:    "",
+					Range:  "",
+					Filter: "",
+					Header: nil,
+				},
+				Range:  nil,
+				PeerID: "",
+			},
+			enablePrefetch: false,
+			storageManager: func(sm *ms.MockManager) {
+				//sm.EXPECT().FindPartialCompletedTask(gomock.Any(), gomock.Any()).DoAndReturn(
+				//	func(taskID string, rg *clientutil.Range) *storage.ReusePeerTask {
+				//		return nil
+				//	})
+				//sm.EXPECT().FindCompletedSubTask(gomock.Any()).DoAndReturn(
+				//	func(taskID string) *storage.ReusePeerTask {
+				//		return nil
+				//	})
+				sm.EXPECT().FindCompletedTask(gomock.Any()).DoAndReturn(
+					func(taskID string) *storage.ReusePeerTask {
+						return nil
+					})
+			},
+			verify: func(rc io.ReadCloser, attr map[string]string, ok bool) {
+				assert.False(ok)
+				assert.Nil(rc)
+				assert.Nil(attr)
+			},
+		},
+		{
+			name: "normal completed subtask found",
+			request: &StreamTaskRequest{
+				URL: "http://example.com/1",
+				URLMeta: &base.UrlMeta{
+					Digest: "",
+					Tag:    "",
+					Range:  "",
+					Filter: "",
+					Header: nil,
+				},
+				Range:  &clientutil.Range{Start: 0, Length: 10},
+				PeerID: "",
+			},
+			enablePrefetch: true,
+			storageManager: func(sm *ms.MockManager) {
+				var taskID string
+				sm.EXPECT().FindCompletedSubTask(gomock.Any()).DoAndReturn(
+					func(id string) *storage.ReusePeerTask {
+						taskID = id
+						return &storage.ReusePeerTask{
+							PeerTaskMetadata: storage.PeerTaskMetadata{
+								TaskID: taskID,
+							},
+							ContentLength: 10,
+							TotalPieces:   0,
+							PieceMd5Sign:  "",
+						}
+					})
+				sm.EXPECT().ReadAllPieces(gomock.Any(), gomock.Any()).DoAndReturn(
+					func(ctx context.Context, req *storage.ReadAllPiecesRequest) (io.ReadCloser, error) {
+						assert.Equal(taskID, req.TaskID)
+						return io.NopCloser(bytes.NewBuffer([]byte("1111111111"))), nil
+					})
+			},
+			verify: func(rc io.ReadCloser, attr map[string]string, ok bool) {
+				assert.True(ok)
+				assert.NotNil(rc)
+				assert.Equal("10", attr[headers.ContentLength])
+				assert.Equal("bytes 0-9/*", attr[headers.ContentRange])
+			},
+		},
+		{
+			name: "normal completed subtask not found",
+			request: &StreamTaskRequest{
+				URL: "http://example.com/1",
+				URLMeta: &base.UrlMeta{
+					Digest: "",
+					Tag:    "",
+					Range:  "",
+					Filter: "",
+					Header: nil,
+				},
+				Range:  &clientutil.Range{Start: 0, Length: 10},
+				PeerID: "",
+			},
+			enablePrefetch: true,
+			storageManager: func(sm *ms.MockManager) {
+				sm.EXPECT().FindPartialCompletedTask(gomock.Any(), gomock.Any()).DoAndReturn(
+					func(taskID string, rg *clientutil.Range) *storage.ReusePeerTask {
+						return nil
+					})
+				sm.EXPECT().FindCompletedSubTask(gomock.Any()).DoAndReturn(
+					func(taskID string) *storage.ReusePeerTask {
+						return nil
+					})
+			},
+			verify: func(rc io.ReadCloser, attr map[string]string, ok bool) {
+				assert.False(ok)
+				assert.Nil(rc)
+				assert.Nil(attr)
+			},
+		},
+		{
+			name: "partial task found",
+			request: &StreamTaskRequest{
+				URL: "http://example.com/1",
+				URLMeta: &base.UrlMeta{
+					Digest: "",
+					Tag:    "",
+					Range:  "",
+					Filter: "",
+					Header: nil,
+				},
+				Range:  &clientutil.Range{Start: 0, Length: 10},
+				PeerID: "",
+			},
+			enablePrefetch: true,
+			storageManager: func(sm *ms.MockManager) {
+				var taskID string
+				sm.EXPECT().FindCompletedSubTask(gomock.Any()).DoAndReturn(
+					func(id string) *storage.ReusePeerTask {
+						return nil
+					})
+				sm.EXPECT().FindPartialCompletedTask(gomock.Any(), gomock.Any()).DoAndReturn(
+					func(id string, rg *clientutil.Range) *storage.ReusePeerTask {
+						taskID = id
+						return &storage.ReusePeerTask{
+							PeerTaskMetadata: storage.PeerTaskMetadata{
+								TaskID: taskID,
+							},
+							ContentLength: 100,
+							TotalPieces:   0,
+							PieceMd5Sign:  "",
+						}
+					})
+				sm.EXPECT().ReadAllPieces(gomock.Any(), gomock.Any()).DoAndReturn(
+					func(ctx context.Context, req *storage.ReadAllPiecesRequest) (io.ReadCloser, error) {
+						assert.Equal(taskID, req.TaskID)
+						return io.NopCloser(bytes.NewBuffer([]byte("1111111111"))), nil
+					})
+			},
+			verify: func(rc io.ReadCloser, attr map[string]string, ok bool) {
+				assert.True(ok)
+				assert.NotNil(rc)
+				assert.Equal("10", attr[headers.ContentLength])
+				assert.Equal("bytes 0-9/100", attr[headers.ContentRange])
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			sm := ms.NewMockManager(ctrl)
+			tc.storageManager(sm)
+			ptm := &peerTaskManager{
+				host:           &scheduler.PeerHost{},
+				enablePrefetch: tc.enablePrefetch,
+				storageManager: sm,
+			}
+			tc.verify(ptm.tryReuseStreamPeerTask(context.Background(), tc.request))
+		})
+	}
+}
diff --git a/client/daemon/proxy/proxy_test.go b/client/daemon/proxy/proxy_test.go
index 1218c099..3a8f4869 100644
--- a/client/daemon/proxy/proxy_test.go
+++ b/client/daemon/proxy/proxy_test.go
@@ -36,7 +36,7 @@ type testItem struct {
 
 type testCase struct {
 	Error          error
-	Rules          []*config.Proxy
+	Rules          []*config.ProxyRule
 	RegistryMirror *config.RegistryMirror
 	Items          []testItem
 }
@@ -50,8 +50,8 @@ func (tc *testCase) WithRule(regx string, direct bool, useHTTPS bool, redirect s
 		return tc
 	}
 
-	var r *config.Proxy
-	r, tc.Error = config.NewProxy(regx, useHTTPS, direct, redirect)
+	var r *config.ProxyRule
+	r, tc.Error = config.NewProxyRule(regx, useHTTPS, direct, redirect)
 	tc.Rules = append(tc.Rules, r)
 	return tc
 }
diff --git a/client/daemon/storage/local_storage_test.go b/client/daemon/storage/local_storage_test.go
index 0e46c129..1f98fea9 100644
--- a/client/daemon/storage/local_storage_test.go
+++ b/client/daemon/storage/local_storage_test.go
@@ -391,3 +391,92 @@ func calcPieceMd5(data []byte) string {
 	hash.Write(data)
 	return hex.EncodeToString(hash.Sum(nil)[:16])
 }
+
+func Test_computePiecePosition(t *testing.T) {
+	var testCases = []struct {
+		name  string
+		total int64
+		rg    *clientutil.Range
+		start int32
+		end   int32
+		piece uint32
+	}{
+		{
+			name:  "0",
+			total: 500,
+			rg: &clientutil.Range{
+				Start:  0,
+				Length: 10,
+			},
+			start: 0,
+			end:   0,
+			piece: 100,
+		},
+		{
+			name:  "1",
+			total: 500,
+			rg: &clientutil.Range{
+				Start:  30,
+				Length: 60,
+			},
+			start: 0,
+			end:   0,
+			piece: 100,
+		},
+		{
+			name:  "2",
+			total: 500,
+			rg: &clientutil.Range{
+				Start:  30,
+				Length: 130,
+			},
+			start: 0,
+			end:   1,
+			piece: 100,
+		},
+		{
+			name:  "3",
+			total: 500,
+			rg: &clientutil.Range{
+				Start:  350,
+				Length: 100,
+			},
+			start: 3,
+			end:   4,
+			piece: 100,
+		},
+		{
+			name:  "4",
+			total: 500,
+			rg: &clientutil.Range{
+				Start:  400,
+				Length: 100,
+			},
+			start: 4,
+			end:   4,
+			piece: 100,
+		},
+		{
+			name:  "5",
+			total: 500,
+			rg: &clientutil.Range{
+				Start:  0,
+				Length: 500,
+			},
+			start: 0,
+			end:   4,
+			piece: 100,
+		},
+	}
+
+	assert := testifyassert.New(t)
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			start, end := computePiecePosition(tc.total, tc.rg, func(length int64) uint32 {
+				return tc.piece
+			})
+			assert.Equal(tc.start, start)
+			assert.Equal(tc.end, end)
+		})
+	}
+}
diff --git a/client/daemon/test/mock/storage/manager.go b/client/daemon/test/mock/storage/manager.go
index ecf6f540..2af3834b 100644
--- a/client/daemon/test/mock/storage/manager.go
+++ b/client/daemon/test/mock/storage/manager.go
@@ -10,6 +10,7 @@ import (
 	reflect "reflect"
 	time "time"
 
+	clientutil "d7y.io/dragonfly/v2/client/clientutil"
 	storage "d7y.io/dragonfly/v2/client/daemon/storage"
 	base "d7y.io/dragonfly/v2/pkg/rpc/base"
 	gomock "github.com/golang/mock/gomock"
@@ -296,6 +297,20 @@ func (mr *MockManagerMockRecorder) FindCompletedTask(taskID interface{}) *gomock
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "FindCompletedTask", reflect.TypeOf((*MockManager)(nil).FindCompletedTask), taskID)
 }
 
+// FindPartialCompletedTask mocks base method.
+func (m *MockManager) FindPartialCompletedTask(taskID string, rg *clientutil.Range) *storage.ReusePeerTask {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "FindPartialCompletedTask", taskID, rg)
+	ret0, _ := ret[0].(*storage.ReusePeerTask)
+	return ret0
+}
+
+// FindPartialCompletedTask indicates an expected call of FindPartialCompletedTask.
+func (mr *MockManagerMockRecorder) FindPartialCompletedTask(taskID, rg interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "FindPartialCompletedTask", reflect.TypeOf((*MockManager)(nil).FindPartialCompletedTask), taskID, rg)
+}
+
 // GetPieces mocks base method.
 func (m *MockManager) GetPieces(ctx context.Context, req *base.PieceTaskRequest) (*base.PiecePacket, error) {
 	m.ctrl.T.Helper()
diff --git a/pkg/util/timeutils/time_utils_test.go b/pkg/util/timeutils/time_utils_test.go
index 99c4c689..56ab4b36 100644
--- a/pkg/util/timeutils/time_utils_test.go
+++ b/pkg/util/timeutils/time_utils_test.go
@@ -139,7 +139,63 @@ func TestSecondsUnixTime(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			if got := SecondsUnixTime(tt.args); got != tt.want {
-				t.Errorf("MillisUnixTime() = %v, want %v", got, tt.want)
+				t.Errorf("SecondsUnixTime() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestNanoToTime(t *testing.T) {
+	const Layout = "2006-01-02 15:04:05"
+	sample, _ := time.ParseInLocation(Layout, "2021-01-02 12:04:05", time.Local)
+	now := time.Now()
+	tests := []struct {
+		name string
+		args int64
+		want time.Time
+	}{
+		{
+			name: "convert an int64 nanosecond to a time",
+			args: sample.UnixNano(),
+			want: sample.Local(),
+		},
+		{
+			name: "convert now",
+			args: now.UnixNano(),
+			want: time.Unix(0, now.UnixNano()),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := NanoToTime(tt.args); got != tt.want {
+				t.Errorf("NanoToTime() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestSubNano(t *testing.T) {
+	now := time.Now()
+	tests := []struct {
+		name string
+		args []int64
+		want int64
+	}{
+		{
+			name: "nanoseconds are not equal",
+			args: []int64{now.Add(1 * time.Nanosecond).UnixNano(), now.UnixNano()},
+			want: 1,
+		},
+		{
+			name: "nanoseconds are equal",
+			args: []int64{now.UnixNano(), now.UnixNano()},
+			want: 0,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := SubNano(tt.args[0], tt.args[1]); got.Nanoseconds() != tt.want {
+				t.Errorf("SubNano() = %v, want %v", got, tt.want)
 			}
 		})
 	}
diff --git a/scheduler/config/config_test.go b/scheduler/config/config_test.go
index 1ae28e0d..7b7e1991 100644
--- a/scheduler/config/config_test.go
+++ b/scheduler/config/config_test.go
@@ -31,12 +31,11 @@ func TestConfig_Load(t *testing.T) {
 
 	config := &Config{
 		Server: &ServerConfig{
-			IP:          "127.0.0.1",
-			Host:        "foo",
-			Port:        8002,
-			ListenLimit: 1000,
-			CacheDir:    "foo",
-			LogDir:      "bar",
+			IP:       "127.0.0.1",
+			Host:     "foo",
+			Port:     8002,
+			CacheDir: "foo",
+			LogDir:   "bar",
 		},
 		Scheduler: &SchedulerConfig{
 			Algorithm:            "default",
diff --git a/scheduler/config/testdata/scheduler.yaml b/scheduler/config/testdata/scheduler.yaml
index a27b84f7..3a88b515 100644
--- a/scheduler/config/testdata/scheduler.yaml
+++ b/scheduler/config/testdata/scheduler.yaml
@@ -2,7 +2,6 @@ server:
   ip: 127.0.0.1
   host: foo
   port: 8002
-  listenLimit: 1000
   cacheDir: foo
   logDir: bar
 
diff --git a/scheduler/scheduler/evaluator/evaluator_base_test.go b/scheduler/scheduler/evaluator/evaluator_base_test.go
index c0b357f4..22e98f0a 100644
--- a/scheduler/scheduler/evaluator/evaluator_base_test.go
+++ b/scheduler/scheduler/evaluator/evaluator_base_test.go
@@ -627,17 +627,10 @@ func TestEvaluatorBase_IsBadNode(t *testing.T) {
 			totalPieceCount: 1,
 			mock: func(peer *resource.Peer) {
 				peer.FSM.SetState(resource.PeerStateRunning)
-				peer.AppendPieceCost(10)
-				peer.AppendPieceCost(11)
-				peer.AppendPieceCost(12)
-				peer.AppendPieceCost(13)
-				peer.AppendPieceCost(14)
-				peer.AppendPieceCost(9)
-				peer.AppendPieceCost(8)
-				peer.AppendPieceCost(7)
-				peer.AppendPieceCost(6)
-				peer.AppendPieceCost(5)
-				peer.AppendPieceCost(19)
+				for i := 0; i < 30; i++ {
+					peer.AppendPieceCost(int64(i))
+				}
+				peer.AppendPieceCost(50)
 			},
 			expect: func(t *testing.T, isBadNode bool) {
 				assert := assert.New(t)
@@ -650,16 +643,9 @@ func TestEvaluatorBase_IsBadNode(t *testing.T) {
 			totalPieceCount: 1,
 			mock: func(peer *resource.Peer) {
 				peer.FSM.SetState(resource.PeerStateRunning)
-				peer.AppendPieceCost(10)
-				peer.AppendPieceCost(11)
-				peer.AppendPieceCost(12)
-				peer.AppendPieceCost(13)
-				peer.AppendPieceCost(14)
-				peer.AppendPieceCost(9)
-				peer.AppendPieceCost(8)
-				peer.AppendPieceCost(7)
-				peer.AppendPieceCost(6)
-				peer.AppendPieceCost(5)
+				for i := 0; i < 30; i++ {
+					peer.AppendPieceCost(int64(i))
+				}
 				peer.AppendPieceCost(18)
 			},
 			expect: func(t *testing.T, isBadNode bool) {
@@ -673,16 +659,9 @@ func TestEvaluatorBase_IsBadNode(t *testing.T) {
 			totalPieceCount: 1,
 			mock: func(peer *resource.Peer) {
 				peer.FSM.SetState(resource.PeerStateRunning)
-				peer.AppendPieceCost(10)
-				peer.AppendPieceCost(11)
-				peer.AppendPieceCost(12)
-				peer.AppendPieceCost(13)
-				peer.AppendPieceCost(14)
-				peer.AppendPieceCost(9)
-				peer.AppendPieceCost(8)
-				peer.AppendPieceCost(7)
-				peer.AppendPieceCost(6)
-				peer.AppendPieceCost(5)
+				for i := 20; i < 50; i++ {
+					peer.AppendPieceCost(int64(i))
+				}
 				peer.AppendPieceCost(0)
 			},
 			expect: func(t *testing.T, isBadNode bool) {
diff --git a/scheduler/scheduler/scheduler_test.go b/scheduler/scheduler/scheduler_test.go
index 9b413653..c404a05d 100644
--- a/scheduler/scheduler/scheduler_test.go
+++ b/scheduler/scheduler/scheduler_test.go
@@ -808,7 +808,7 @@ func TestScheduler_FindParent(t *testing.T) {
 			},
 		},
 		{
-			name: "find parent and fetch filterParentCount from manager dynconfig",
+			name: "find parent and fetch filterParentLimit from manager dynconfig",
 			mock: func(peer *resource.Peer, mockPeers []*resource.Peer, blocklist set.SafeSet, md *configmocks.MockDynconfigInterfaceMockRecorder) {
 				peer.FSM.SetState(resource.PeerStateRunning)
 				mockPeers[0].FSM.SetState(resource.PeerStateRunning)
@@ -821,7 +821,7 @@ func TestScheduler_FindParent(t *testing.T) {
 				mockPeers[1].Pieces.Set(2)
 
 				md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{
-					FilterParentCount: 1,
+					FilterParentLimit: 1,
 				}, true).Times(1)
 			},
 			expect: func(t *testing.T, mockPeers []*resource.Peer, parent *resource.Peer, ok bool) {
diff --git a/scheduler/service/service_test.go b/scheduler/service/service_test.go
index 0959b939..b9c34db3 100644
--- a/scheduler/service/service_test.go
+++ b/scheduler/service/service_test.go
@@ -346,6 +346,46 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				assert.True(peer.FSM.Is(resource.PeerStateReceivedNormal))
 			},
 		},
+		{
+			name: "task scope size is SizeScope_SMALL and load piece error, parent state is PeerStateRunning",
+			req: &rpcscheduler.PeerTaskRequest{
+				PeerHost: &rpcscheduler.PeerHost{
+					Uuid: mockRawHost.Uuid,
+				},
+			},
+			mock: func(
+				req *rpcscheduler.PeerTaskRequest, mockPeer *resource.Peer, mockCDNPeer *resource.Peer,
+				scheduler scheduler.Scheduler, res resource.Resource, hostManager resource.HostManager, taskManager resource.TaskManager, peerManager resource.PeerManager,
+				ms *mocks.MockSchedulerMockRecorder, mr *resource.MockResourceMockRecorder, mh *resource.MockHostManagerMockRecorder, mt *resource.MockTaskManagerMockRecorder, mp *resource.MockPeerManagerMockRecorder,
+			) {
+				mockPeer.Task.FSM.SetState(resource.TaskStateSucceeded)
+				mockPeer.Task.StorePeer(mockCDNPeer)
+				mockPeer.Task.ContentLength.Store(129)
+				mockPeer.Task.StorePiece(&base.PieceInfo{
+					PieceNum: 0,
+				})
+				mockPeer.Task.TotalPieceCount.Store(1)
+				mockPeer.FSM.SetState(resource.PeerStatePending)
+				mockCDNPeer.FSM.SetState(resource.PeerStateRunning)
+
+				gomock.InOrder(
+					mr.TaskManager().Return(taskManager).Times(1),
+					mt.LoadOrStore(gomock.Any()).Return(mockPeer.Task, true).Times(1),
+					mr.HostManager().Return(hostManager).Times(1),
+					mh.Load(gomock.Eq(mockPeer.Host.ID)).Return(mockPeer.Host, true).Times(1),
+					mr.PeerManager().Return(peerManager).Times(1),
+					mp.LoadOrStore(gomock.Any()).Return(mockPeer, true).Times(1),
+					ms.FindParent(gomock.Any(), gomock.Any(), gomock.Any()).Return(mockCDNPeer, true).Times(1),
+				)
+			},
+			expect: func(t *testing.T, peer *resource.Peer, result *rpcscheduler.RegisterResult, err error) {
+				assert := assert.New(t)
+				assert.NoError(err)
+				assert.Equal(result.TaskId, peer.Task.ID)
+				assert.Equal(result.SizeScope, base.SizeScope_NORMAL)
+				assert.True(peer.FSM.Is(resource.PeerStateReceivedNormal))
+			},
+		},
 		{
 			name: "task scope size is SizeScope_SMALL and load piece error, peer state is PeerStateFailed",
 			req: &rpcscheduler.PeerTaskRequest{
@@ -366,6 +406,8 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				})
 				mockPeer.Task.TotalPieceCount.Store(1)
 				mockPeer.FSM.SetState(resource.PeerStateFailed)
+				mockCDNPeer.FSM.SetState(resource.PeerStateSucceeded)
+
 				gomock.InOrder(
 					mr.TaskManager().Return(taskManager).Times(1),
 					mt.LoadOrStore(gomock.Any()).Return(mockPeer.Task, true).Times(1),
@@ -403,6 +445,8 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				})
 				mockPeer.Task.TotalPieceCount.Store(1)
 				mockPeer.FSM.SetState(resource.PeerStateFailed)
+				mockCDNPeer.FSM.SetState(resource.PeerStateSucceeded)
+
 				gomock.InOrder(
 					mr.TaskManager().Return(taskManager).Times(1),
 					mt.LoadOrStore(gomock.Any()).Return(mockPeer.Task, true).Times(1),
@@ -439,6 +483,8 @@ func TestService_RegisterPeerTask(t *testing.T) {
 					PieceNum: 0,
 				})
 				mockPeer.Task.TotalPieceCount.Store(1)
+				mockCDNPeer.FSM.SetState(resource.PeerStateSucceeded)
+
 				gomock.InOrder(
 					mr.TaskManager().Return(taskManager).Times(1),
 					mt.LoadOrStore(gomock.Any()).Return(mockPeer.Task, true).Times(1),
@@ -756,7 +802,7 @@ func TestService_ReportPieceResult(t *testing.T) {
 			},
 		},
 		{
-			name: "revice Code_ClientPieceDownloadFail code",
+			name: "revice Code_PeerTaskNotFound code",
 			mock: func(
 				mockPeer *resource.Peer,
 				res resource.Resource, peerManager resource.PeerManager,
@@ -768,7 +814,7 @@ func TestService_ReportPieceResult(t *testing.T) {
 					ms.Context().Return(context.Background()).Times(1),
 					ms.Recv().Return(&rpcscheduler.PieceResult{
 						SrcPid: mockPeerID,
-						Code:   base.Code_ClientPieceDownloadFail,
+						Code:   base.Code_PeerTaskNotFound,
 					}, nil).Times(1),
 					mr.PeerManager().Return(peerManager).Times(1),
 					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
@@ -1762,6 +1808,7 @@ func TestService_registerPeer(t *testing.T) {
 func TestService_handlePieceSuccess(t *testing.T) {
 	mockHost := resource.NewHost(mockRawHost)
 	mockTask := resource.NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
+	now := time.Now()
 
 	tests := []struct {
 		name   string
@@ -1777,8 +1824,8 @@ func TestService_handlePieceSuccess(t *testing.T) {
 					PieceNum: 0,
 					PieceMd5: "ac32345ef819f03710e2105c81106fdd",
 				},
-				BeginTime: uint64(time.Now().Unix()),
-				EndTime:   uint64(time.Now().Add(1 * time.Second).Unix()),
+				BeginTime: uint64(now.UnixNano()),
+				EndTime:   uint64(now.Add(1 * time.Millisecond).UnixNano()),
 			},
 			peer: resource.NewPeer(mockPeerID, mockTask, mockHost),
 			mock: func(peer *resource.Peer) {
@@ -1797,8 +1844,8 @@ func TestService_handlePieceSuccess(t *testing.T) {
 					PieceNum: 0,
 					PieceMd5: "ac32345ef819f03710e2105c81106fdd",
 				},
-				BeginTime: uint64(time.Now().Unix()),
-				EndTime:   uint64(time.Now().Add(1 * time.Second).Unix()),
+				BeginTime: uint64(now.UnixNano()),
+				EndTime:   uint64(now.Add(1 * time.Millisecond).UnixNano()),
 			},
 			peer: resource.NewPeer(mockPeerID, mockTask, mockHost),
 			mock: func(peer *resource.Peer) {
@@ -1890,36 +1937,6 @@ func TestService_handlePieceFail(t *testing.T) {
 				assert.True(peer.FSM.Is(resource.PeerStateRunning))
 			},
 		},
-		{
-			name: "piece result code is Code_ClientPieceDownloadFail and parent state set PeerEventDownloadFailed",
-			config: &config.Config{
-				Scheduler: mockSchedulerConfig,
-				CDN:       &config.CDNConfig{Enable: true},
-				Metrics:   &config.MetricsConfig{EnablePeerHost: true},
-			},
-			piece: &rpcscheduler.PieceResult{
-				Code:   base.Code_ClientPieceDownloadFail,
-				DstPid: mockCDNPeerID,
-			},
-			peer:   resource.NewPeer(mockPeerID, mockTask, mockHost),
-			parent: resource.NewPeer(mockCDNPeerID, mockTask, mockHost),
-			run: func(t *testing.T, svc *Service, peer *resource.Peer, parent *resource.Peer, piece *rpcscheduler.PieceResult, peerManager resource.PeerManager, cdn resource.CDN, ms *mocks.MockSchedulerMockRecorder, mr *resource.MockResourceMockRecorder, mp *resource.MockPeerManagerMockRecorder, mc *resource.MockCDNMockRecorder) {
-				peer.FSM.SetState(resource.PeerStateRunning)
-				parent.FSM.SetState(resource.PeerStateRunning)
-				blocklist := set.NewSafeSet()
-				blocklist.Add(parent.ID)
-				gomock.InOrder(
-					mr.PeerManager().Return(peerManager).Times(1),
-					mp.Load(gomock.Eq(parent.ID)).Return(parent, true).Times(1),
-					ms.ScheduleParent(gomock.Any(), gomock.Eq(peer), gomock.Eq(blocklist)).Return().Times(1),
-				)
-
-				svc.handlePieceFail(context.Background(), peer, piece)
-				assert := assert.New(t)
-				assert.True(peer.FSM.Is(resource.PeerStateRunning))
-				assert.True(parent.FSM.Is(resource.PeerStateFailed))
-			},
-		},
 		{
 			name: "piece result code is Code_PeerTaskNotFound and parent state set PeerEventDownloadFailed",
 			config: &config.Config{
@@ -2094,8 +2111,8 @@ func TestService_handlePieceFail(t *testing.T) {
 				gomock.InOrder(
 					mr.PeerManager().Return(peerManager).Times(1),
 					mp.Load(gomock.Eq(parent.ID)).Return(parent, true).Times(1),
-					ms.ScheduleParent(gomock.Any(), gomock.Eq(peer), gomock.Eq(blocklist)).Return().Times(1),
 				)
+				ms.ScheduleParent(gomock.Any(), gomock.Eq(peer), gomock.Eq(blocklist)).Return().Times(1)
 				mr.CDN().Do(func() { wg.Done() }).Return(cdn).Times(1)
 				mc.TriggerTask(gomock.Any(), gomock.Any()).Do(func(ctx context.Context, task *resource.Task) { wg.Done() }).Return(peer, &rpcscheduler.PeerResult{}, nil).Times(1)