Repository: dragonflyoss/Dragonfly2
Pull request #1258 (2022-04-17 06:36:37)

Feature: use a golang native file server to replace nginx
<!--- 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! -->

- [ ] 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.


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

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

```diff
@@            Coverage Diff             @@
##             main    #1258      +/-   ##
==========================================
+ Coverage   46.26%   46.30%   +0.03%     
==========================================
  Files         135      135              
  Lines       11202    11202              
==========================================
+ Hits         5183     5187       +4     
+ Misses       5451     5447       -4     
  Partials      568      568              
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `46.30% <ø> (+0.03%)` | :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/1258?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [...ent/daemon/peer/peertask\_piecetask\_synchronizer.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19zeW5jaHJvbml6ZXIuZ28=) | `57.22% <0.00%> (-1.81%)` | :arrow_down: |
| [scheduler/service/service.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3NlcnZpY2Uvc2VydmljZS5nbw==) | `86.75% <0.00%> (+0.66%)` | :arrow_up: |
| [cdn/supervisor/progress/progress.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1258/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2RuL3N1cGVydmlzb3IvcHJvZ3Jlc3MvcHJvZ3Jlc3MuZ28=) | `96.87% <0.00%> (+6.25%)` | :arrow_up: |

anjia0532: 用golang写静态文件服务器 替换nginx我是认可的，只是有几点担心
1. 需要压测来评估 nginx 和 golang 静态文件的吞吐（QPS/TPS）和性能（比如内存），可以用 k6,gatling,wrk等工具来实现，如果集成到 e2e 更好，需要综合测试 kb,mb,gb 三种量级的文件对于 nginx/golang 的影响，假设（我对这块不熟悉），golang是整个读取 load 到内存，那么测kb/mb时没问题，但是到gb时可能会oom
2. 需要评估代码会不会逃逸/越权访问未授权目录， 比如 `http.FileServer(http.Dir("/path/to/static-dir"))` ,url访问的是 `http://example.com/../xxxx.tar.gz`  会不会访问到 `/path/to/` 目录
3. nginx 会根据后缀给resp返回 content-type , golang 如果不返回或者默认写死一个，会不会影响下游使用？
4. golang 静态文件服务器支不支持 断点续传、下载

Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 0fc067eb..fcd688e5 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -7,6 +7,8 @@ on:
   pull_request:
     branches: [ main, release-* ]
     paths-ignore: [ '**.md', '**.png', '**.jpg', '**.svg', '**/docs/**' ]
+  schedule:
+    - cron: '0 4 * * *'
 
 env:
   GO_VERSION: 1.17
diff --git a/.goreleaser.yml b/.goreleaser.yml
index a62204d1..84bf28d3 100644
--- a/.goreleaser.yml
+++ b/.goreleaser.yml
@@ -21,6 +21,23 @@ builds:
       - -X "d7y.io/dragonfly/v2/version.Gotags=none"
       - -X "d7y.io/dragonfly/v2/version.Gogcflags=none"
 
+  - main: ./cmd/dfcache
+    id: dfcache
+    binary: dfcache
+    goos:
+      - linux
+      - darwin
+    goarch:
+      - amd64
+    ldflags:
+      - -X d7y.io/dragonfly/v2/version.Major={{ .Major }}
+      - -X d7y.io/dragonfly/v2/version.Minor={{ .Minor }}
+      - -X d7y.io/dragonfly/v2/version.GitVersion={{ .Tag }}
+      - -X d7y.io/dragonfly/v2/version.GitCommit={{ .ShortCommit }}
+      - -X d7y.io/dragonfly/v2/version.BuildTime={{ .Date }}
+      - -X "d7y.io/dragonfly/v2/version.Gotags=none"
+      - -X "d7y.io/dragonfly/v2/version.Gogcflags=none"
+
   - main: ./cmd/scheduler
     id: scheduler
     binary: scheduler
@@ -138,3 +155,45 @@ nfpms:
       postinstall: build/package/nfpm/dfget-postinstall.sh
       preremove: build/package/nfpm/dfget-preremove.sh
       postremove: build/package/nfpm/dfget-postremove.sh
+  - id: dfcache
+    file_name_template: "dfcache-{{ .Version }}-{{ .Os }}-{{ .Arch }}"
+    package_name: dfcache
+    description: dfcache is the cache client to of dragonfly that communicates with dfdaemon and operates on files in P2P network
+    license: "Apache 2.0"
+    homepage: https://d7y.io
+    bindir: /usr/bin
+    builds:
+      - dfcache
+    formats:
+      - rpm
+      - deb
+    contents:
+      - src: build/package/docs/dfcache/dfcache.1
+        dst: /usr/share/man/man1/dfcache.1
+
+      - src: build/package/docs/dfcache/dfcache-delete.1
+        dst: /usr/share/man/man1/dfcache-delete.1
+
+      - src: build/package/docs/dfcache/dfcache-doc.1
+        dst: /usr/share/man/man1/dfcache-doc.1
+
+      - src: build/package/docs/dfcache/dfcache-export.1
+        dst: /usr/share/man/man1/dfcache-export.1
+
+      - src: build/package/docs/dfcache/dfcache-import.1
+        dst: /usr/share/man/man1/dfcache-import.1
+
+      - src: build/package/docs/dfcache/dfcache-plugin.1
+        dst: /usr/share/man/man1/dfcache-plugin.1
+
+      - src: build/package/docs/dfcache/dfcache-stat.1
+        dst: /usr/share/man/man1/dfcache-stat.1
+
+      - src: build/package/docs/dfcache/dfcache-version.1
+        dst: /usr/share/man/man1/dfcache-version.1
+
+      - src: LICENSE
+        dst: /usr/share/doc/dfcache/License
+
+      - src: CHANGELOG.md
+        dst: /usr/share/doc/dfcache/ChangeLog
diff --git a/Makefile b/Makefile
index 00490f05..e2f8de1f 100644
--- a/Makefile
+++ b/Makefile
@@ -14,6 +14,7 @@
 
 PROJECT_NAME := "d7y.io/dragonfly/v2"
 DFGET_NAME := "dfget"
+DFCACHE_NAME := "dfcache"
 SEMVER := "2.0.1"
 VERSION_RELEASE := "1"
 PKG := "$(PROJECT_NAME)"
@@ -21,6 +22,7 @@ PKG_LIST := $(shell go list ${PKG}/... | grep -v /vendor/ | grep -v '\(/test/\)'
 GIT_COMMIT := $(shell git rev-parse --verify HEAD --short=7)
 GIT_COMMIT_LONG := $(shell git rev-parse --verify HEAD)
 DFGET_ARCHIVE_PREFIX := "$(DFGET_NAME)_$(SEMVER)-$(VERSION_RELEASE)_$(GIT_COMMIT)"
+DFCACHE_ARCHIVE_PREFIX := "$(DFCACHE_NAME)_$(SEMVER)-$(VERSION_RELEASE)_$(GIT_COMMIT)"
 
 all: help
 
@@ -186,6 +188,23 @@ build-rpm-dfget: build-linux-dfget
 	@echo "Build package output: ./bin/$(DFGET_ARCHIVE_PREFIX)_linux_amd64.rpm"
 .PHONY: build-rpm-dfget
 
+# Build rpm dfcache
+build-rpm-dfcache: build-linux-dfcache build-dfcache-man-page
+	@echo "Begin to build rpm dfcache"
+	@docker run --rm \
+	-v "$(PWD)/build:/root/build" \
+	-v "$(PWD)/build/package/docs:/root/docs" \
+	-v "$(PWD)/LICENSE:/root/License" \
+	-v "$(PWD)/CHANGELOG.md:/root/CHANGELOG.md" \
+	-v "$(PWD)/bin:/root/bin" \
+	-e "SEMVER=$(SEMVER)" \
+	-e "VERSION_RELEASE=$(VERSION_RELEASE)" \
+	goreleaser/nfpm pkg \
+		--config /root/build/package/nfpm/dfcache.yaml \
+		--target /root/bin/$(DFCACHE_ARCHIVE_PREFIX)_linux_amd64.rpm
+	@echo "Build package output: ./bin/$(DFCACHE_ARCHIVE_PREFIX)_linux_amd64.rpm"
+.PHONY: build-rpm-dfcache
+
 # Build deb dfget
 build-deb-dfget: build-linux-dfget
 	@echo "Begin to build deb dfget"
@@ -203,11 +222,40 @@ build-deb-dfget: build-linux-dfget
 	@echo "Build package output: ./bin/$(DFGET_ARCHIVE_PREFIX)_linux_amd64.deb"
 .PHONY: build-deb-dfget
 
+# Build deb dfcache
+build-deb-dfcache: build-linux-dfcache build-dfcache-man-page
+	@echo "Begin to build deb dfcache"
+	@docker run --rm \
+	-v "$(PWD)/build:/root/build" \
+	-v "$(PWD)/build/package/docs:/root/docs" \
+	-v "$(PWD)/LICENSE:/root/License" \
+	-v "$(PWD)/CHANGELOG.md:/root/CHANGELOG.md" \
+	-v "$(PWD)/bin:/root/bin" \
+	-e "SEMVER=$(SEMVER)" \
+	-e "VERSION_RELEASE=$(VERSION_RELEASE)" \
+	goreleaser/nfpm pkg \
+		--config /root/build/package/nfpm/dfcache.yaml \
+		--target /root/bin/$(DFCACHE_ARCHIVE_PREFIX)_linux_amd64.deb
+	@echo "Build package output: ./bin/$(DFCACHE_ARCHIVE_PREFIX)_linux_amd64.deb"
+.PHONY: build-deb-dfcache
+
 # Generate dfget man page
 build-dfget-man-page:
 	@pandoc -s -t man ./build/package/docs/dfget.1.md -o ./build/package/docs/dfget.1
 .PHONY: build-dfget-man-page
 
+# Genrate dfcache man pages
+build-dfcache-man-page:
+	@pandoc -s -t man ./build/package/docs/dfcache/dfcache.md -o ./build/package/docs/dfcache/dfcache.1
+	@pandoc -s -t man ./build/package/docs/dfcache/dfcache_delete.md -o ./build/package/docs/dfcache/dfcache-delete.1
+	@pandoc -s -t man ./build/package/docs/dfcache/dfcache_doc.md -o ./build/package/docs/dfcache/dfcache-doc.1
+	@pandoc -s -t man ./build/package/docs/dfcache/dfcache_export.md -o ./build/package/docs/dfcache/dfcache-export.1
+	@pandoc -s -t man ./build/package/docs/dfcache/dfcache_import.md -o ./build/package/docs/dfcache/dfcache-import.1
+	@pandoc -s -t man ./build/package/docs/dfcache/dfcache_plugin.md -o ./build/package/docs/dfcache/dfcache-plugin.1
+	@pandoc -s -t man ./build/package/docs/dfcache/dfcache_stat.md -o ./build/package/docs/dfcache/dfcache-stat.1
+	@pandoc -s -t man ./build/package/docs/dfcache/dfcache_version.md -o ./build/package/docs/dfcache/dfcache-version.1
+.PHONY: build-dfcache-man-page
+
 # Generate e2e sha256sum
 build-e2e-sha256sum:
 	@GOOS=linux GOARCH=amd64 go build -o /tmp/sha256sum-offset test/tools/sha256sum-offset/main.go
@@ -345,8 +393,11 @@ help:
 	@echo "make install-scheduler              install scheduler"
 	@echo "make install-manager                install manager"
 	@echo "make build-rpm-dfget                build rpm dfget"
+	@echo "make build-rpm-dfcache              build rpm dfcache"
 	@echo "make build-deb-dfget                build deb dfget"
+	@echo "make build-deb-dfcache              build deb dfcache"
 	@echo "make build-dfget-man-page           generate dfget man page"
+	@echo "make build-dfcache-man-page         generate dfcache man page"
 	@echo "make test                           run unit tests"
 	@echo "make test-coverage                  run tests with coverage"
 	@echo "make actions-e2e-test-coverage      run github actons E2E tests with coverage"
diff --git a/build/images/cdn/Dockerfile b/build/images/cdn/Dockerfile
index cf0dd654..63c24e9d 100644
--- a/build/images/cdn/Dockerfile
+++ b/build/images/cdn/Dockerfile
@@ -1,3 +1,5 @@
+ARG BASE_IMAGE=alpine:3.14
+
 FROM golang:1.17.3-alpine3.14 as builder
 
 ARG GOPROXY
@@ -18,14 +20,12 @@ RUN GRPC_HEALTH_PROBE_VERSION=v0.4.8 && \
     wget -qO/bin/grpc_health_probe https://github.com/grpc-ecosystem/grpc-health-probe/releases/download/${GRPC_HEALTH_PROBE_VERSION}/grpc_health_probe-linux-amd64 && \
     chmod +x /bin/grpc_health_probe
 
-# TODO support BASE_IMAGE
-FROM nginx:1.21-alpine
+FROM ${BASE_IMAGE}
 
 ENV PATH=/opt/dragonfly/bin:$PATH
 RUN echo "hosts: files dns" > /etc/nsswitch.conf
 
 COPY --from=builder /go/src/d7y.io/dragonfly/v2/hack/start-cdn.sh /root/start.sh
-COPY --from=builder /go/src/d7y.io/dragonfly/v2/hack/cdn-nginx.conf /etc/nginx/nginx.conf
 COPY --from=builder /opt/dragonfly/bin/cdn /opt/dragonfly/bin/cdn
 COPY --from=health /bin/grpc_health_probe /bin/grpc_health_probe
 
diff --git a/build/package/docs/dfcache/dfcache-delete.1 b/build/package/docs/dfcache/dfcache-delete.1
new file mode 100644
index 00000000..03bf6da0
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache-delete.1
@@ -0,0 +1,44 @@
+.\" Automatically generated by Pandoc 2.14.0.3
+.\"
+.TH "DFCACHE" "1" "" "Version v2.0.2" "Frivolous \[lq]Dfcache\[rq] Documentation"
+.hy
+.SS dfcache delete
+.PP
+delete file from P2P cache system
+.SS Synopsis
+.PP
+delete file from P2P cache system
+.IP
+.nf
+\f[C]
+dfcache delete <-i cid> [flags]
+\f[R]
+.fi
+.SS Options
+.IP
+.nf
+\f[C]
+  -h, --help   help for delete
+\f[R]
+.fi
+.SS Options inherited from parent commands
+.IP
+.nf
+\f[C]
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default \[dq]dragonfly-dfcache\[dq])
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+\f[R]
+.fi
+.SS SEE ALSO
+.IP \[bu] 2
+dfcache - the P2P cache client of dragonfly
diff --git a/build/package/docs/dfcache/dfcache-doc.1 b/build/package/docs/dfcache/dfcache-doc.1
new file mode 100644
index 00000000..b179493e
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache-doc.1
@@ -0,0 +1,45 @@
+.\" Automatically generated by Pandoc 2.14.0.3
+.\"
+.TH "DFCACHE" "1" "" "Version v2.0.2" "Frivolous \[lq]Dfcache\[rq] Documentation"
+.hy
+.SS dfcache doc
+.PP
+generate documents
+.SS Synopsis
+.PP
+generate markdown documents for cmd: dfcache .
+.IP
+.nf
+\f[C]
+dfcache doc [flags]
+\f[R]
+.fi
+.SS Options
+.IP
+.nf
+\f[C]
+  -h, --help          help for doc
+      --path string   destination dir of generated markdown documents (default \[dq]./\[dq])
+\f[R]
+.fi
+.SS Options inherited from parent commands
+.IP
+.nf
+\f[C]
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default \[dq]dragonfly-dfcache\[dq])
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+\f[R]
+.fi
+.SS SEE ALSO
+.IP \[bu] 2
+dfcache - the P2P cache client of dragonfly
diff --git a/build/package/docs/dfcache/dfcache-export.1 b/build/package/docs/dfcache/dfcache-export.1
new file mode 100644
index 00000000..668bec35
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache-export.1
@@ -0,0 +1,46 @@
+.\" Automatically generated by Pandoc 2.14.0.3
+.\"
+.TH "DFCACHE" "1" "" "Version v2.0.2" "Frivolous \[lq]Dfcache\[rq] Documentation"
+.hy
+.SS dfcache export
+.PP
+export file from P2P cache system
+.SS Synopsis
+.PP
+export file from P2P cache system
+.IP
+.nf
+\f[C]
+dfcache export <-i cid> <output>|<-O output> [flags]
+\f[R]
+.fi
+.SS Options
+.IP
+.nf
+\f[C]
+  -h, --help            help for export
+  -l, --local           only export file from local cache
+  -O, --output string   export file path
+\f[R]
+.fi
+.SS Options inherited from parent commands
+.IP
+.nf
+\f[C]
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default \[dq]dragonfly-dfcache\[dq])
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+\f[R]
+.fi
+.SS SEE ALSO
+.IP \[bu] 2
+dfcache - the P2P cache client of dragonfly
diff --git a/build/package/docs/dfcache/dfcache-import.1 b/build/package/docs/dfcache/dfcache-import.1
new file mode 100644
index 00000000..58f2c77f
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache-import.1
@@ -0,0 +1,45 @@
+.\" Automatically generated by Pandoc 2.14.0.3
+.\"
+.TH "DFCACHE" "1" "" "Version v2.0.2" "Frivolous \[lq]Dfcache\[rq] Documentation"
+.hy
+.SS dfcache import
+.PP
+import file into P2P cache system
+.SS Synopsis
+.PP
+import file into P2P cache system
+.IP
+.nf
+\f[C]
+dfcache import <-i cid> <file>|<-I file> [flags]
+\f[R]
+.fi
+.SS Options
+.IP
+.nf
+\f[C]
+  -h, --help           help for import
+  -I, --input string   import the given file into P2P network
+\f[R]
+.fi
+.SS Options inherited from parent commands
+.IP
+.nf
+\f[C]
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default \[dq]dragonfly-dfcache\[dq])
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+\f[R]
+.fi
+.SS SEE ALSO
+.IP \[bu] 2
+dfcache - the P2P cache client of dragonfly
diff --git a/build/package/docs/dfcache/dfcache-plugin.1 b/build/package/docs/dfcache/dfcache-plugin.1
new file mode 100644
index 00000000..edea895f
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache-plugin.1
@@ -0,0 +1,44 @@
+.\" Automatically generated by Pandoc 2.14.0.3
+.\"
+.TH "DFCACHE" "1" "" "Version v2.0.2" "Frivolous \[lq]Dfcache\[rq] Documentation"
+.hy
+.SS dfcache plugin
+.PP
+show plugin
+.SS Synopsis
+.PP
+show the plugin details of dragonfly.
+.IP
+.nf
+\f[C]
+dfcache plugin [flags]
+\f[R]
+.fi
+.SS Options
+.IP
+.nf
+\f[C]
+  -h, --help   help for plugin
+\f[R]
+.fi
+.SS Options inherited from parent commands
+.IP
+.nf
+\f[C]
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default \[dq]dragonfly-dfcache\[dq])
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+\f[R]
+.fi
+.SS SEE ALSO
+.IP \[bu] 2
+dfcache - the P2P cache client of dragonfly
diff --git a/build/package/docs/dfcache/dfcache-stat.1 b/build/package/docs/dfcache/dfcache-stat.1
new file mode 100644
index 00000000..39da84c6
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache-stat.1
@@ -0,0 +1,45 @@
+.\" Automatically generated by Pandoc 2.14.0.3
+.\"
+.TH "DFCACHE" "1" "" "Version v2.0.2" "Frivolous \[lq]Dfcache\[rq] Documentation"
+.hy
+.SS dfcache stat
+.PP
+stat checks if a file exists in P2P cache system
+.SS Synopsis
+.PP
+stat checks if a file exists in P2P cache system
+.IP
+.nf
+\f[C]
+dfcache stat <-i cid> [flags]
+\f[R]
+.fi
+.SS Options
+.IP
+.nf
+\f[C]
+  -h, --help    help for stat
+  -l, --local   only check task exists locally, and don\[aq]t check other peers in P2P network
+\f[R]
+.fi
+.SS Options inherited from parent commands
+.IP
+.nf
+\f[C]
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default \[dq]dragonfly-dfcache\[dq])
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+\f[R]
+.fi
+.SS SEE ALSO
+.IP \[bu] 2
+dfcache - the P2P cache client of dragonfly
diff --git a/build/package/docs/dfcache/dfcache-version.1 b/build/package/docs/dfcache/dfcache-version.1
new file mode 100644
index 00000000..300cd0ae
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache-version.1
@@ -0,0 +1,44 @@
+.\" Automatically generated by Pandoc 2.14.0.3
+.\"
+.TH "DFCACHE" "1" "" "Version v2.0.2" "Frivolous \[lq]Dfcache\[rq] Documentation"
+.hy
+.SS dfcache version
+.PP
+show version
+.SS Synopsis
+.PP
+show the version details of dragonfly.
+.IP
+.nf
+\f[C]
+dfcache version [flags]
+\f[R]
+.fi
+.SS Options
+.IP
+.nf
+\f[C]
+  -h, --help   help for version
+\f[R]
+.fi
+.SS Options inherited from parent commands
+.IP
+.nf
+\f[C]
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default \[dq]dragonfly-dfcache\[dq])
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+\f[R]
+.fi
+.SS SEE ALSO
+.IP \[bu] 2
+dfcache - the P2P cache client of dragonfly
diff --git a/build/package/docs/dfcache/dfcache.1 b/build/package/docs/dfcache/dfcache.1
new file mode 100644
index 00000000..bf020d82
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache.1
@@ -0,0 +1,57 @@
+.\" Automatically generated by Pandoc 2.14.0.3
+.\"
+.TH "DFCACHE" "1" "" "Version v2.0.2" "Frivolous \[lq]Dfcache\[rq] Documentation"
+.hy
+.SS dfcache
+.PP
+the P2P cache client of dragonfly
+.SS Synopsis
+.PP
+dfcache is the cache client to of dragonfly that communicates with
+dfdaemon and operates on files in P2P network, where the P2P network
+acts as a cache system.
+.PP
+The difference between dfcache and dfget is that, dfget downloads file
+from a given URL, the file might be on other peers in P2P network or a
+CDN peer, it\[cq]s the P2P network\[cq]s responsibility to download file
+from source; but dfcache could only export or download a file that has
+been imported or added into P2P network by other peer, it\[cq]s the
+user\[cq]s responsibility to go back to source and add file into P2P
+network.
+.SS Options
+.IP
+.nf
+\f[C]
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+  -h, --help                  help for dfcache
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default \[dq]dragonfly-dfcache\[dq])
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+\f[R]
+.fi
+.SS SEE ALSO
+.IP \[bu] 2
+dfcache completion - generate the autocompletion script for the
+specified shell
+.IP \[bu] 2
+dfcache delete - delete file from P2P cache system
+.IP \[bu] 2
+dfcache doc - generate documents
+.IP \[bu] 2
+dfcache export - export file from P2P cache system
+.IP \[bu] 2
+dfcache import - import file into P2P cache system
+.IP \[bu] 2
+dfcache plugin - show plugin
+.IP \[bu] 2
+dfcache stat - stat checks if a file exists in P2P cache system
+.IP \[bu] 2
+dfcache version - show version
diff --git a/build/package/docs/dfcache/dfcache.md b/build/package/docs/dfcache/dfcache.md
new file mode 100644
index 00000000..ecfb225e
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache.md
@@ -0,0 +1,48 @@
+% DFCACHE(1) Version v2.0.2 | Frivolous "Dfcache" Documentation
+
+## dfcache
+
+the P2P cache client of dragonfly
+
+### Synopsis
+
+
+dfcache is the cache client to of dragonfly that communicates with dfdaemon and operates
+on files in P2P network, where the P2P network acts as a cache system.
+
+The difference between dfcache and dfget is that, dfget downloads file from a given URL,
+the file might be on other peers in P2P network or a CDN peer, it's the P2P network's
+responsibility to download file from source; but dfcache could only export or download a
+file that has been imported or added into P2P network by other peer, it's the user's
+responsibility to go back to source and add file into P2P network.
+
+
+### Options
+
+```
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+  -h, --help                  help for dfcache
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default "dragonfly-dfcache")
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+```
+
+### SEE ALSO
+
+* [dfcache completion](dfcache_completion.md)	 - generate the autocompletion script for the specified shell
+* [dfcache delete](dfcache_delete.md)	 - delete file from P2P cache system
+* [dfcache doc](dfcache_doc.md)	 - generate documents
+* [dfcache export](dfcache_export.md)	 - export file from P2P cache system
+* [dfcache import](dfcache_import.md)	 - import file into P2P cache system
+* [dfcache plugin](dfcache_plugin.md)	 - show plugin
+* [dfcache stat](dfcache_stat.md)	 - stat checks if a file exists in P2P cache system
+* [dfcache version](dfcache_version.md)	 - show version
+
diff --git a/build/package/docs/dfcache/dfcache_delete.md b/build/package/docs/dfcache/dfcache_delete.md
new file mode 100644
index 00000000..d9f611b8
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache_delete.md
@@ -0,0 +1,41 @@
+% DFCACHE(1) Version v2.0.2 | Frivolous "Dfcache" Documentation
+
+## dfcache delete
+
+delete file from P2P cache system
+
+### Synopsis
+
+delete file from P2P cache system
+
+```
+dfcache delete <-i cid> [flags]
+```
+
+### Options
+
+```
+  -h, --help   help for delete
+```
+
+### Options inherited from parent commands
+
+```
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default "dragonfly-dfcache")
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+```
+
+### SEE ALSO
+
+* [dfcache](dfcache.md)	 - the P2P cache client of dragonfly
+
diff --git a/build/package/docs/dfcache/dfcache_doc.md b/build/package/docs/dfcache/dfcache_doc.md
new file mode 100644
index 00000000..d5b51597
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache_doc.md
@@ -0,0 +1,42 @@
+% DFCACHE(1) Version v2.0.2 | Frivolous "Dfcache" Documentation
+
+## dfcache doc
+
+generate documents
+
+### Synopsis
+
+generate markdown documents for cmd: dfcache .
+
+```
+dfcache doc [flags]
+```
+
+### Options
+
+```
+  -h, --help          help for doc
+      --path string   destination dir of generated markdown documents (default "./")
+```
+
+### Options inherited from parent commands
+
+```
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default "dragonfly-dfcache")
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+```
+
+### SEE ALSO
+
+* [dfcache](dfcache.md)	 - the P2P cache client of dragonfly
+
diff --git a/build/package/docs/dfcache/dfcache_export.md b/build/package/docs/dfcache/dfcache_export.md
new file mode 100644
index 00000000..5568e87d
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache_export.md
@@ -0,0 +1,43 @@
+% DFCACHE(1) Version v2.0.2 | Frivolous "Dfcache" Documentation
+
+## dfcache export
+
+export file from P2P cache system
+
+### Synopsis
+
+export file from P2P cache system
+
+```
+dfcache export <-i cid> <output>|<-O output> [flags]
+```
+
+### Options
+
+```
+  -h, --help            help for export
+  -l, --local           only export file from local cache
+  -O, --output string   export file path
+```
+
+### Options inherited from parent commands
+
+```
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default "dragonfly-dfcache")
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+```
+
+### SEE ALSO
+
+* [dfcache](dfcache.md)	 - the P2P cache client of dragonfly
+
diff --git a/build/package/docs/dfcache/dfcache_import.md b/build/package/docs/dfcache/dfcache_import.md
new file mode 100644
index 00000000..8ac7412d
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache_import.md
@@ -0,0 +1,42 @@
+% DFCACHE(1) Version v2.0.2 | Frivolous "Dfcache" Documentation
+
+## dfcache import
+
+import file into P2P cache system
+
+### Synopsis
+
+import file into P2P cache system
+
+```
+dfcache import <-i cid> <file>|<-I file> [flags]
+```
+
+### Options
+
+```
+  -h, --help           help for import
+  -I, --input string   import the given file into P2P network
+```
+
+### Options inherited from parent commands
+
+```
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default "dragonfly-dfcache")
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+```
+
+### SEE ALSO
+
+* [dfcache](dfcache.md)	 - the P2P cache client of dragonfly
+
diff --git a/build/package/docs/dfcache/dfcache_plugin.md b/build/package/docs/dfcache/dfcache_plugin.md
new file mode 100644
index 00000000..85344f96
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache_plugin.md
@@ -0,0 +1,41 @@
+% DFCACHE(1) Version v2.0.2 | Frivolous "Dfcache" Documentation
+
+## dfcache plugin
+
+show plugin
+
+### Synopsis
+
+show the plugin details of dragonfly.
+
+```
+dfcache plugin [flags]
+```
+
+### Options
+
+```
+  -h, --help   help for plugin
+```
+
+### Options inherited from parent commands
+
+```
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default "dragonfly-dfcache")
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+```
+
+### SEE ALSO
+
+* [dfcache](dfcache.md)	 - the P2P cache client of dragonfly
+
diff --git a/build/package/docs/dfcache/dfcache_stat.md b/build/package/docs/dfcache/dfcache_stat.md
new file mode 100644
index 00000000..ab7662ae
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache_stat.md
@@ -0,0 +1,42 @@
+% DFCACHE(1) Version v2.0.2 | Frivolous "Dfcache" Documentation
+
+## dfcache stat
+
+stat checks if a file exists in P2P cache system
+
+### Synopsis
+
+stat checks if a file exists in P2P cache system
+
+```
+dfcache stat <-i cid> [flags]
+```
+
+### Options
+
+```
+  -h, --help    help for stat
+  -l, --local   only check task exists locally, and don't check other peers in P2P network
+```
+
+### Options inherited from parent commands
+
+```
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default "dragonfly-dfcache")
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+```
+
+### SEE ALSO
+
+* [dfcache](dfcache.md)	 - the P2P cache client of dragonfly
+
diff --git a/build/package/docs/dfcache/dfcache_version.md b/build/package/docs/dfcache/dfcache_version.md
new file mode 100644
index 00000000..450872a1
--- /dev/null
+++ b/build/package/docs/dfcache/dfcache_version.md
@@ -0,0 +1,41 @@
+% DFCACHE(1) Version v2.0.2 | Frivolous "Dfcache" Documentation
+
+## dfcache version
+
+show version
+
+### Synopsis
+
+show the version details of dragonfly.
+
+```
+dfcache version [flags]
+```
+
+### Options
+
+```
+  -h, --help   help for version
+```
+
+### Options inherited from parent commands
+
+```
+      --callsystem string     The caller name which is mainly used for statistics and access control
+  -i, --cid string            content or cache ID, e.g. sha256 digest of the content
+      --config string         the path of configuration file with yaml extension name, default is /etc/dragonfly/dfcache.yaml, it can also be set by env var: DFCACHE_CONFIG
+      --console               whether logger output records to the stdout
+      --jaeger string         jaeger endpoint url, like: http://localhost:14250/api/traces
+      --logdir string         Dfcache log directory
+      --pprof-port int        listen port for pprof, 0 represents random port (default -1)
+      --service-name string   name of the service for tracer (default "dragonfly-dfcache")
+  -t, --tag string            different tags for the same cid will be recognized as different  files in P2P network
+      --timeout duration      Timeout for this cache operation, 0 is infinite
+      --verbose               whether logger use debug level
+      --workhome string       Dfcache working directory
+```
+
+### SEE ALSO
+
+* [dfcache](dfcache.md)	 - the P2P cache client of dragonfly
+
diff --git a/build/package/nfpm/dfcache.yaml b/build/package/nfpm/dfcache.yaml
new file mode 100644
index 00000000..59d60f6a
--- /dev/null
+++ b/build/package/nfpm/dfcache.yaml
@@ -0,0 +1,50 @@
+name: dfcache
+arch: amd64
+platform: linux
+epoch: 2
+version: v${SEMVER}
+version_schema: semver
+release: ${VERSION_RELEASE}
+section: default
+priority: extra
+description: dfcache is the cache client to of dragonfly that communicates with dfdaemon and operates on files in P2P network
+license: "Apache 2.0"
+homepage: https://d7y.io
+maintainer: Dragonfly Maintainers
+
+provides:
+  - dfcache
+
+contents:
+  - src: /root/bin/linux_amd64/dfcache
+    dst: /usr/bin/dfcache
+
+  - src: /root/docs/dfcache/dfcache.1
+    dst: /usr/share/man/man1/dfcache.1
+
+  - src: /root/docs/dfcache/dfcache-delete.1
+    dst: /usr/share/man/man1/dfcache-delete.1
+
+  - src: /root/docs/dfcache/dfcache-doc.1
+    dst: /usr/share/man/man1/dfcache-doc.1
+
+  - src: /root/docs/dfcache/dfcache-export.1
+    dst: /usr/share/man/man1/dfcache-export.1
+
+  - src: /root/docs/dfcache/dfcache-import.1
+    dst: /usr/share/man/man1/dfcache-import.1
+
+  - src: /root/docs/dfcache/dfcache-plugin.1
+    dst: /usr/share/man/man1/dfcache-plugin.1
+
+  - src: /root/docs/dfcache/dfcache-stat.1
+    dst: /usr/share/man/man1/dfcache-stat.1
+
+  - src: /root/docs/dfcache/dfcache-version.1
+    dst: /usr/share/man/man1/dfcache-version.1
+
+  - src: /root/License
+    dst: /usr/share/doc/dfcache/License
+
+  - src: /root/CHANGELOG.md
+    dst: /usr/share/doc/dfcache/ChangeLog
diff --git a/build/package/nfpm/systemd/dfget-daemon.service b/build/package/nfpm/systemd/dfget-daemon.service
index fd0660e5..eef18068 100644
--- a/build/package/nfpm/systemd/dfget-daemon.service
+++ b/build/package/nfpm/systemd/dfget-daemon.service
@@ -10,7 +10,7 @@ After=network.target
 
 [Service]
 Type=simple
-ExecStartPre=/bin/bash /opt/dragonfly/fix.dfget-daemon.cpuset.sh
+ExecStartPre=-/bin/sh /opt/dragonfly/fix.dfget-daemon.cpuset.sh
 ExecStart=/usr/bin/dfget daemon
 ExecReload=/bin/kill -HUP $MAINPID
 Restart=always
diff --git a/build/package/nfpm/systemd/fix.dfget-daemon.cpuset.sh b/build/package/nfpm/systemd/fix.dfget-daemon.cpuset.sh
index f292b285..7592696a 100644
--- a/build/package/nfpm/systemd/fix.dfget-daemon.cpuset.sh
+++ b/build/package/nfpm/systemd/fix.dfget-daemon.cpuset.sh
@@ -1,5 +1,10 @@
 #!/bin/sh
 
+# skip unsupported and newly kernel
+if [ ! -d /sys/fs/cgroup/cpuset,cpu,cpuacct ]; then
+  exit 0
+fi
+
 # fix cpu set not settled in old kernel
 mkdir -p /sys/fs/cgroup/cpuset,cpu,cpuacct/dragonfly.slice/dfget-daemon.service
 cat /sys/fs/cgroup/cpuset,cpu,cpuacct/cpuset.cpus > \
diff --git a/cdn/cdn.go b/cdn/cdn.go
index 44854220..dbe77178 100644
--- a/cdn/cdn.go
+++ b/cdn/cdn.go
@@ -18,6 +18,7 @@ package cdn
 
 import (
 	"context"
+	"net/http"
 
 	"github.com/pkg/errors"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
@@ -25,6 +26,7 @@ import (
 	"google.golang.org/grpc"
 
 	"d7y.io/dragonfly/v2/cdn/config"
+	"d7y.io/dragonfly/v2/cdn/fileserver"
 	"d7y.io/dragonfly/v2/cdn/gc"
 	"d7y.io/dragonfly/v2/cdn/metrics"
 	"d7y.io/dragonfly/v2/cdn/rpcserver"
@@ -33,6 +35,7 @@ import (
 	"d7y.io/dragonfly/v2/cdn/supervisor/cdn/storage"
 	"d7y.io/dragonfly/v2/cdn/supervisor/progress"
 	"d7y.io/dragonfly/v2/cdn/supervisor/task"
+	"d7y.io/dragonfly/v2/client/daemon/upload"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/rpc/manager"
 	managerClient "d7y.io/dragonfly/v2/pkg/rpc/manager/client"
@@ -54,6 +57,9 @@ type Server struct {
 
 	// gc Server
 	gcServer *gc.Server
+
+	// fileServer
+	fileServer *fileserver.Server
 }
 
 // New creates a brand-new server instance.
@@ -97,6 +103,8 @@ func New(config *config.Config) (*Server, error) {
 		return nil, errors.Wrap(err, "create rpcServer")
 	}
 
+	fileServer := fileserver.New(config.RPCServer.DownloadPort, upload.PeerDownloadHTTPPathPrefix, storageManager.GetUploadPath())
+
 	// Initialize gc server
 	gcServer, err := gc.New()
 	if err != nil {
@@ -126,6 +134,7 @@ func New(config *config.Config) (*Server, error) {
 		metricsServer: metricsServer,
 		configServer:  configServer,
 		gcServer:      gcServer,
+		fileServer:    fileServer,
 	}, nil
 }
 
@@ -172,6 +181,16 @@ func (s *Server) Serve() error {
 		}
 	}()
 
+	go func() {
+		// Start file server
+		if err := s.fileServer.ListenAndServe(); err != nil {
+			if err == http.ErrServerClosed {
+				return
+			}
+			logger.Fatalf("start cdn file server failed: %v", err)
+		}
+	}()
+
 	// Start grpc server
 	return s.grpcServer.ListenAndServe()
 }
@@ -198,5 +217,10 @@ func (s *Server) Stop() error {
 		// Stop grpc server
 		return s.grpcServer.Shutdown()
 	})
+
+	g.Go(func() error {
+		// Stop file server
+		return s.fileServer.Shutdown(ctx)
+	})
 	return g.Wait()
 }
diff --git a/cdn/fileserver/fileserver.go b/cdn/fileserver/fileserver.go
new file mode 100644
index 00000000..513c1f17
--- /dev/null
+++ b/cdn/fileserver/fileserver.go
@@ -0,0 +1,111 @@
+/*
+ *     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 fileserver
+
+import (
+	"context"
+	"fmt"
+	"io/fs"
+	"net/http"
+	"os"
+	"path"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/pkg/errors"
+
+	logger "d7y.io/dragonfly/v2/internal/dflog"
+)
+
+const (
+	maxRetryAttempts = 5
+)
+
+// d7yDir is modified from http.Dir, for http.Dir does not treat syscall.EMFILE as we want.
+type d7yDir string
+
+func mapDirOpenError(originalErr error, name string) error {
+	if os.IsNotExist(originalErr) || os.IsPermission(originalErr) {
+		return originalErr
+	}
+
+	parts := strings.Split(name, string(filepath.Separator))
+	for i := range parts {
+		if parts[i] == "" {
+			continue
+		}
+		fi, err := os.Stat(strings.Join(parts[:i+1], string(filepath.Separator)))
+		if err != nil {
+			return originalErr
+		}
+		if !fi.IsDir() {
+			pathError, ok := originalErr.(*fs.PathError)
+			if ok && pathError.Err == syscall.EMFILE {
+				return syscall.EMFILE
+			}
+			return fs.ErrNotExist
+		}
+	}
+	return originalErr
+}
+
+func (d d7yDir) Open(name string) (http.File, error) {
+	if filepath.Separator != '/' && strings.ContainsRune(name, filepath.Separator) {
+		return nil, errors.New("http: invalid character in file path")
+	}
+	dir := string(d)
+	if dir == "" {
+		dir = "."
+	}
+	fullName := filepath.Join(dir, filepath.FromSlash(path.Clean("/"+name)))
+
+	f, err := os.Open(fullName)
+	if err != nil {
+		mappedErr := mapDirOpenError(err, fullName)
+		// retry when syscall.EMFILE
+		for i := 0; mappedErr == syscall.EMFILE && i < maxRetryAttempts; i++ {
+			time.Sleep(100 * time.Millisecond)
+			f, err = os.Open(fullName)
+			if err == nil {
+				return f, nil
+			}
+			mappedErr = mapDirOpenError(err, fullName)
+		}
+		return nil, mappedErr
+	}
+	return f, nil
+}
+
+type Server struct {
+	*http.Server
+}
+
+func New(port int, prefix, uploadPath string) *Server {
+	return &Server{
+		&http.Server{
+			Addr:    fmt.Sprintf(":%d", port),
+			Handler: http.StripPrefix(prefix, http.FileServer(d7yDir(uploadPath))),
+		},
+	}
+}
+
+func (server *Server) Shutdown(ctx context.Context) error {
+	defer logger.Infof("====stopped file server====")
+	return server.Server.Shutdown(ctx)
+}
diff --git a/cdn/supervisor/cdn/storage/disk/disk.go b/cdn/supervisor/cdn/storage/disk/disk.go
index 12565dc2..2284d796 100644
--- a/cdn/supervisor/cdn/storage/disk/disk.go
+++ b/cdn/supervisor/cdn/storage/disk/disk.go
@@ -281,6 +281,10 @@ func (s *diskStorageManager) TryFreeSpace(fileLength int64) (bool, error) {
 	return true, nil
 }
 
+func (s *diskStorageManager) GetUploadPath() string {
+	return s.diskDriver.GetPath(storage.GetUploadHomeRaw())
+}
+
 func (s *diskStorageManager) GC() error {
 	logger.StorageGCLogger.With("type", "disk").Debug("start the disk storage gc job")
 	gcTaskIDs, err := s.diskCleaner.GC("disk", false)
diff --git a/cdn/supervisor/cdn/storage/hybrid/hybrid.go b/cdn/supervisor/cdn/storage/hybrid/hybrid.go
index 2e6cf0eb..f3d7957a 100644
--- a/cdn/supervisor/cdn/storage/hybrid/hybrid.go
+++ b/cdn/supervisor/cdn/storage/hybrid/hybrid.go
@@ -381,6 +381,10 @@ func (h *hybridStorageManager) tryShmSpace(url, taskID string, fileLength int64)
 	return "", fmt.Errorf("shared memory is not allowed")
 }
 
+func (h *hybridStorageManager) GetUploadPath() string {
+	return h.diskDriver.GetPath(storage.GetUploadHomeRaw())
+}
+
 func (h *hybridStorageManager) GC() error {
 	logger.StorageGCLogger.With("type", "hybrid").Debug("start the hybrid storage gc job")
 	var wg sync.WaitGroup
diff --git a/cdn/supervisor/cdn/storage/mock/mock_storage_manager.go b/cdn/supervisor/cdn/storage/mock/mock_storage_manager.go
index c56d659b..442cc1b3 100644
--- a/cdn/supervisor/cdn/storage/mock/mock_storage_manager.go
+++ b/cdn/supervisor/cdn/storage/mock/mock_storage_manager.go
@@ -65,6 +65,20 @@ func (mr *MockManagerMockRecorder) DeleteTask(arg0 interface{}) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteTask", reflect.TypeOf((*MockManager)(nil).DeleteTask), arg0)
 }
 
+// GetUploadPath mocks base method.
+func (m *MockManager) GetUploadPath() string {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetUploadPath")
+	ret0, _ := ret[0].(string)
+	return ret0
+}
+
+// GetUploadPath indicates an expected call of GetUploadPath.
+func (mr *MockManagerMockRecorder) GetUploadPath() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetUploadPath", reflect.TypeOf((*MockManager)(nil).GetUploadPath))
+}
+
 // ReadDownloadFile mocks base method.
 func (m *MockManager) ReadDownloadFile(arg0 string) (io.ReadCloser, error) {
 	m.ctrl.T.Helper()
diff --git a/cdn/supervisor/cdn/storage/storage.go b/cdn/supervisor/cdn/storage/storage.go
index 664326a0..6a30efe1 100644
--- a/cdn/supervisor/cdn/storage/storage.go
+++ b/cdn/supervisor/cdn/storage/storage.go
@@ -74,6 +74,9 @@ type Manager interface {
 
 	// TryFreeSpace checks if there is enough space for the file, return true while we are sure that there is enough space.
 	TryFreeSpace(fileLength int64) (bool, error)
+
+	// GetUploadPath get path of upload file
+	GetUploadPath() string
 }
 
 // FileMetadata meta data of task
diff --git a/client/config/constants_otel.go b/client/config/constants_otel.go
index 97758e76..b5803f54 100644
--- a/client/config/constants_otel.go
+++ b/client/config/constants_otel.go
@@ -55,6 +55,7 @@ const (
 	SpanBackSource        = "client-back-source"
 	SpanFirstSchedule     = "schedule-#1"
 	SpanGetPieceTasks     = "get-piece-tasks"
+	SpanSyncPieceTasks    = "sync-piece-tasks"
 	SpanDownloadPiece     = "download-piece-#%d"
 	SpanProxy             = "proxy"
 	SpanWritePiece        = "write-piece"
diff --git a/client/config/flags.go b/client/config/flags.go
index d8718fc9..c1c24659 100644
--- a/client/config/flags.go
+++ b/client/config/flags.go
@@ -27,7 +27,7 @@ import (
 	"golang.org/x/time/rate"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 )
 
 // SchedulersValue implements the pflag.Value interface.
diff --git a/client/config/peerhost.go b/client/config/peerhost.go
index 29e79879..b8cc10aa 100644
--- a/client/config/peerhost.go
+++ b/client/config/peerhost.go
@@ -35,7 +35,7 @@ import (
 
 	"d7y.io/dragonfly/v2/client/clientutil"
 	"d7y.io/dragonfly/v2/cmd/dependency/base"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/unit"
 	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
 )
diff --git a/client/config/peerhost_darwin.go b/client/config/peerhost_darwin.go
index 7a295faa..68d05a2a 100644
--- a/client/config/peerhost_darwin.go
+++ b/client/config/peerhost_darwin.go
@@ -26,7 +26,7 @@ import (
 	"golang.org/x/time/rate"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/util/hostutils"
 	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
 )
diff --git a/client/config/peerhost_linux.go b/client/config/peerhost_linux.go
index 5d65e120..da98d305 100644
--- a/client/config/peerhost_linux.go
+++ b/client/config/peerhost_linux.go
@@ -26,7 +26,7 @@ import (
 	"golang.org/x/time/rate"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/util/hostutils"
 	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
 )
diff --git a/client/daemon/daemon.go b/client/daemon/daemon.go
index c0789ed3..fbcaaa6b 100644
--- a/client/daemon/daemon.go
+++ b/client/daemon/daemon.go
@@ -47,7 +47,7 @@ import (
 	"d7y.io/dragonfly/v2/client/daemon/storage"
 	"d7y.io/dragonfly/v2/client/daemon/upload"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/reachable"
diff --git a/client/daemon/peer/peertask_conductor.go b/client/daemon/peer/peertask_conductor.go
index 89851420..081c0ed8 100644
--- a/client/daemon/peer/peertask_conductor.go
+++ b/client/daemon/peer/peertask_conductor.go
@@ -65,6 +65,10 @@ type peerTaskConductor struct {
 	// ctx is with span info for tracing
 	// we use successCh and failCh mark task success or fail
 	ctx context.Context
+	// piece download uses this context
+	pieceDownloadCtx context.Context
+	// when back source, cancel all piece download action
+	pieceDownloadCancel context.CancelFunc
 
 	// host info about current host
 	host *scheduler.PeerHost
@@ -103,7 +107,9 @@ type peerTaskConductor struct {
 	// peerPacketStream stands schedulerclient.PeerPacketStream from scheduler
 	peerPacketStream schedulerclient.PeerPacketStream
 	// peerPacket is the latest available peers from peerPacketCh
-	peerPacket atomic.Value // *scheduler.PeerPacket
+	// Deprecated: remove in future release
+	peerPacket      atomic.Value // *scheduler.PeerPacket
+	legacyPeerCount *atomic.Int64
 	// peerPacketReady will receive a ready signal for peerPacket ready
 	peerPacketReady chan bool
 	// pieceTaskPoller pulls piece task from other peers
@@ -114,7 +120,6 @@ type peerTaskConductor struct {
 
 	// same actions must be done only once, like close done channel and so on
 	statusOnce sync.Once
-	cancelOnce sync.Once
 	// done channel will be closed when peer task success
 	successCh chan struct{}
 	// fail channel will be closed after peer task fail
@@ -142,7 +147,7 @@ type peerTaskConductor struct {
 	// requestedPieces stands all pieces requested from peers
 	requestedPieces *Bitmap
 	// lock used by piece download worker
-	requestedPiecesLock sync.Mutex
+	requestedPiecesLock sync.RWMutex
 	// lock used by send piece result
 	sendPieceResultLock sync.Mutex
 	// limiter will be used when enable per peer task rate limit
@@ -205,6 +210,7 @@ func (ptm *peerTaskManager) newPeerTaskConductor(
 		taskID:              taskID,
 		successCh:           make(chan struct{}),
 		failCh:              make(chan struct{}),
+		legacyPeerCount:     atomic.NewInt64(0),
 		span:                span,
 		readyPieces:         NewBitmap(),
 		runningPieces:       NewBitmap(),
@@ -230,6 +236,8 @@ func (ptm *peerTaskManager) newPeerTaskConductor(
 		peerTaskConductor: ptc,
 	}
 
+	ptc.pieceDownloadCtx, ptc.pieceDownloadCancel = context.WithCancel(ptc.ctx)
+
 	return ptc
 }
 
@@ -384,14 +392,49 @@ func (pt *peerTaskConductor) Log() *logger.SugaredLoggerOnWith {
 }
 
 func (pt *peerTaskConductor) cancel(code base.Code, reason string) {
-	pt.cancelOnce.Do(func() {
+	pt.statusOnce.Do(func() {
 		pt.failedCode = code
 		pt.failedReason = reason
-		pt.Fail()
+		pt.fail()
+	})
+}
+
+// only use when receive back source code from scheduler
+func (pt *peerTaskConductor) markBackSource() {
+	pt.needBackSource.Store(true)
+	// when close peerPacketReady, pullPiecesFromPeers will invoke backSource
+	close(pt.peerPacketReady)
+	// let legacy mode exit
+	pt.peerPacket.Store(&scheduler.PeerPacket{
+		TaskId:        pt.taskID,
+		SrcPid:        pt.peerID,
+		ParallelCount: 1,
+		MainPeer:      nil,
+		StealPeers: []*scheduler.PeerPacket_DestPeer{
+			{
+				Ip:      pt.host.Ip,
+				RpcPort: pt.host.RpcPort,
+				PeerId:  pt.peerID,
+			},
+		},
+		Code: base.Code_SchedNeedBackSource,
 	})
 }
 
+// only use when legacy get piece from peers schedule timeout
+func (pt *peerTaskConductor) forceBackSource() {
+	pt.needBackSource.Store(true)
+	pt.backSource()
+}
+
 func (pt *peerTaskConductor) backSource() {
+	// cancel all piece download
+	pt.pieceDownloadCancel()
+	// cancel all sync pieces
+	if pt.pieceTaskSyncManager != nil {
+		pt.pieceTaskSyncManager.cancel()
+	}
+
 	ctx, span := tracer.Start(pt.ctx, config.SpanBackSource)
 	pt.contentLength.Store(-1)
 	err := pt.pieceManager.DownloadSource(ctx, pt, pt.request)
@@ -433,7 +476,7 @@ func (pt *peerTaskConductor) pullPieces() {
 
 func (pt *peerTaskConductor) pullPiecesWithP2P() {
 	var (
-		// keep same size with pt.failedPieceCh for avoiding dead lock
+		// keep same size with pt.failedPieceCh for avoiding deadlock
 		pieceBufferSize = uint32(config.DefaultPieceChanSize)
 		pieceRequestCh  = make(chan *DownloadPieceRequest, pieceBufferSize)
 	)
@@ -519,7 +562,7 @@ func (pt *peerTaskConductor) storeTinyPeerTask() {
 
 func (pt *peerTaskConductor) receivePeerPacket(pieceRequestCh chan *DownloadPieceRequest) {
 	var (
-		lastPieceNum        int32 = 0
+		lastNotReadyPiece   int32 = 0
 		peerPacket          *scheduler.PeerPacket
 		err                 error
 		firstPacketReceived bool
@@ -570,9 +613,7 @@ loop:
 		pt.Debugf("receive peerPacket %v", peerPacket)
 		if peerPacket.Code != base.Code_Success {
 			if peerPacket.Code == base.Code_SchedNeedBackSource {
-				pt.needBackSource.Store(true)
-				pt.pieceTaskSyncManager.cancel()
-				close(pt.peerPacketReady)
+				pt.markBackSource()
 				pt.Infof("receive back source code")
 				return
 			}
@@ -609,18 +650,22 @@ loop:
 			firstPeerSpan.End()
 		}
 		// updateSynchronizer will update legacy peers to peerPacket.StealPeers only
-		pt.updateSynchronizer(lastPieceNum, peerPacket)
+		lastNotReadyPiece = pt.updateSynchronizer(lastNotReadyPiece, peerPacket)
 		if !firstPacketReceived {
 			// trigger legacy get piece once to avoid first schedule timeout
 			firstPacketReceived = true
 		} else if len(peerPacket.StealPeers) == 0 {
+			pt.Debugf("no legacy peers, skip to send peerPacketReady")
+			pt.legacyPeerCount.Store(0)
 			continue
 		}
 
-		pt.Debugf("connect to %d legacy peers", len(peerPacket.StealPeers))
-		pt.peerPacket.Store(peerPacket)
+		legacyPeerCount := int64(len(peerPacket.StealPeers))
+		pt.Debugf("connect to %d legacy peers", legacyPeerCount)
+		pt.legacyPeerCount.Store(legacyPeerCount)
 
-		// legacy mode: send peerPacketReady
+		// legacy mode: update peer packet, then send peerPacketReady
+		pt.peerPacket.Store(peerPacket)
 		select {
 		case pt.peerPacketReady <- true:
 		case <-pt.successCh:
@@ -635,19 +680,22 @@ loop:
 }
 
 // updateSynchronizer will convert peers to synchronizer, if failed, will update failed peers to scheduler.PeerPacket
-func (pt *peerTaskConductor) updateSynchronizer(lastNum int32, p *scheduler.PeerPacket) {
-	num, ok := pt.getNextPieceNum(lastNum)
+func (pt *peerTaskConductor) updateSynchronizer(lastNum int32, p *scheduler.PeerPacket) int32 {
+	desiredPiece, ok := pt.getNextNotReadyPieceNum(lastNum)
 	if !ok {
-		pt.Infof("peer task completed")
-		return
+		pt.Infof("all pieces is ready, peer task completed, skip to synchronize")
+		p.MainPeer = nil
+		p.StealPeers = nil
+		return desiredPiece
 	}
 	var peers = []*scheduler.PeerPacket_DestPeer{p.MainPeer}
 	peers = append(peers, p.StealPeers...)
 
-	legacyPeers := pt.pieceTaskSyncManager.newMultiPieceTaskSynchronizer(peers, num)
+	legacyPeers := pt.pieceTaskSyncManager.newMultiPieceTaskSynchronizer(peers, desiredPiece)
 
 	p.MainPeer = nil
 	p.StealPeers = legacyPeers
+	return desiredPiece
 }
 
 func (pt *peerTaskConductor) confirmReceivePeerPacketError(err error) {
@@ -664,8 +712,7 @@ func (pt *peerTaskConductor) confirmReceivePeerPacketError(err error) {
 	)
 	de, ok := err.(*dferrors.DfError)
 	if ok && de.Code == base.Code_SchedNeedBackSource {
-		pt.needBackSource.Store(true)
-		close(pt.peerPacketReady)
+		pt.markBackSource()
 		pt.Infof("receive back source code")
 		return
 	} else if ok && de.Code != base.Code_SchedNeedBackSource {
@@ -814,7 +861,7 @@ loop:
 		// 3. dispatch piece request to all workers
 		pt.dispatchPieceRequest(pieceRequestCh, piecePacket)
 
-		// 4. get next piece
+		// 4. get next not request piece
 		if num, ok = pt.getNextPieceNum(num); ok {
 			// get next piece success
 			limit = config.DefaultPieceChanSize
@@ -851,7 +898,7 @@ func (pt *peerTaskConductor) updateMetadata(piecePacket *base.PiecePacket) {
 	}
 
 	// update content length
-	if piecePacket.ContentLength > -1 {
+	if piecePacket.ContentLength > -1 && pt.GetContentLength() == -1 {
 		metadataChanged = true
 		pt.SetContentLength(piecePacket.ContentLength)
 		pt.span.SetAttributes(config.AttributeTaskContentLength.Int64(piecePacket.ContentLength))
@@ -881,8 +928,8 @@ func (pt *peerTaskConductor) waitFirstPeerPacket() (done bool, backSource bool)
 	case _, ok := <-pt.peerPacketReady:
 		if ok {
 			// preparePieceTasksByPeer func already send piece result with error
-			pt.Infof("new peer client ready, scheduler time cost: %dus, main peer: %s",
-				time.Since(pt.startTime).Microseconds(), pt.peerPacket.Load().(*scheduler.PeerPacket).MainPeer)
+			pt.Infof("new peer client ready, scheduler time cost: %dus, peer count: %d",
+				time.Since(pt.startTime).Microseconds(), len(pt.peerPacket.Load().(*scheduler.PeerPacket).StealPeers))
 			return true, false
 		}
 		// when scheduler says base.Code_SchedNeedBackSource, receivePeerPacket will close pt.peerPacketReady
@@ -900,8 +947,7 @@ func (pt *peerTaskConductor) waitFirstPeerPacket() (done bool, backSource bool)
 		}
 		pt.Warnf("start download from source due to %s", reasonScheduleTimeout)
 		pt.span.AddEvent("back source due to schedule timeout")
-		pt.needBackSource.Store(true)
-		pt.backSource()
+		pt.forceBackSource()
 		return false, true
 	}
 }
@@ -918,7 +964,7 @@ func (pt *peerTaskConductor) waitAvailablePeerPacket() (int32, bool) {
 	case _, ok := <-pt.peerPacketReady:
 		if ok {
 			// preparePieceTasksByPeer func already send piece result with error
-			pt.Infof("new peer client ready, main peer: %s", pt.peerPacket.Load().(*scheduler.PeerPacket).MainPeer)
+			pt.Infof("new peer client ready, peer count: %d", len(pt.peerPacket.Load().(*scheduler.PeerPacket).StealPeers))
 			// research from piece 0
 			return 0, true
 		}
@@ -1012,6 +1058,9 @@ func (pt *peerTaskConductor) downloadPieceWorker(id int32, requests chan *Downlo
 			}
 			pt.readyPiecesLock.RUnlock()
 			pt.downloadPiece(id, request)
+		case <-pt.pieceDownloadCtx.Done():
+			pt.Infof("piece download cancelled, peer download worker #%d exit", id)
+			return
 		case <-pt.successCh:
 			pt.Infof("peer task success, peer download worker #%d exit", id)
 			return
@@ -1028,6 +1077,7 @@ func (pt *peerTaskConductor) downloadPiece(workerID int32, request *DownloadPiec
 	if pt.runningPieces.IsSet(request.piece.PieceNum) {
 		pt.runningPiecesLock.Unlock()
 		pt.Log().Debugf("piece %d is downloading, skip", request.piece.PieceNum)
+		// TODO save to queue for failed pieces
 		return
 	}
 	pt.runningPieces.Set(request.piece.PieceNum)
@@ -1039,7 +1089,7 @@ func (pt *peerTaskConductor) downloadPiece(workerID int32, request *DownloadPiec
 		pt.runningPiecesLock.Unlock()
 	}()
 
-	ctx, span := tracer.Start(pt.ctx, fmt.Sprintf(config.SpanDownloadPiece, request.piece.PieceNum))
+	ctx, span := tracer.Start(pt.pieceDownloadCtx, fmt.Sprintf(config.SpanDownloadPiece, request.piece.PieceNum))
 	span.SetAttributes(config.AttributePiece.Int(int(request.piece.PieceNum)))
 	span.SetAttributes(config.AttributePieceWorker.Int(int(workerID)))
 
@@ -1057,26 +1107,41 @@ func (pt *peerTaskConductor) downloadPiece(workerID int32, request *DownloadPiec
 	// result is always not nil, pieceManager will report begin and end time
 	result, err := pt.pieceManager.DownloadPiece(ctx, request)
 	if err != nil {
-		pt.pieceTaskSyncManager.acquire(
+		pt.ReportPieceResult(request, result, err)
+		span.SetAttributes(config.AttributePieceSuccess.Bool(false))
+		span.End()
+		if pt.needBackSource.Load() {
+			pt.Infof("switch to back source, skip send failed piece")
+			return
+		}
+		attempt, success := pt.pieceTaskSyncManager.acquire(
 			&base.PieceTaskRequest{
+				Limit:    1,
 				TaskId:   pt.taskID,
 				SrcPid:   pt.peerID,
 				StartNum: uint32(request.piece.PieceNum),
-				Limit:    1,
 			})
+		pt.Infof("send failed piece %d to remote, attempt: %d, success: %d",
+			request.piece.PieceNum, attempt, success)
+
+		// when there is no legacy peers, skip send to failedPieceCh for legacy peers in background
+		if pt.legacyPeerCount.Load() == 0 {
+			pt.Infof("there is no legacy peers, skip send to failedPieceCh for legacy peers")
+			return
+		}
 		// Deprecated
 		// send to fail chan and retry
 		// try to send directly first, if failed channel is busy, create a new goroutine to do this
 		select {
 		case pt.failedPieceCh <- request.piece.PieceNum:
+			pt.Infof("success to send failed piece %d to failedPieceCh", request.piece.PieceNum)
 		default:
+			pt.Infof("start to send failed piece %d to failedPieceCh in background", request.piece.PieceNum)
 			go func() {
 				pt.failedPieceCh <- request.piece.PieceNum
+				pt.Infof("success to send failed piece %d to failedPieceCh in background", request.piece.PieceNum)
 			}()
 		}
-		pt.ReportPieceResult(request, result, err)
-		span.SetAttributes(config.AttributePieceSuccess.Bool(false))
-		span.End()
 		return
 	}
 	// broadcast success piece
@@ -1122,12 +1187,16 @@ func (pt *peerTaskConductor) isCompleted() bool {
 	return pt.completedLength.Load() == pt.contentLength.Load()
 }
 
+// for legacy peers only
 func (pt *peerTaskConductor) getNextPieceNum(cur int32) (int32, bool) {
 	if pt.isCompleted() {
 		return -1, false
 	}
 	i := cur
 	// try to find next not requested piece
+	pt.requestedPiecesLock.RLock()
+	defer pt.requestedPiecesLock.RUnlock()
+
 	for ; pt.requestedPieces.IsSet(i); i++ {
 	}
 	totalPiece := pt.GetTotalPieces()
@@ -1142,6 +1211,29 @@ func (pt *peerTaskConductor) getNextPieceNum(cur int32) (int32, bool) {
 	return i, true
 }
 
+func (pt *peerTaskConductor) getNextNotReadyPieceNum(cur int32) (int32, bool) {
+	if pt.isCompleted() {
+		return 0, false
+	}
+	i := cur
+	// try to find next not ready piece
+	pt.readyPiecesLock.RLock()
+	defer pt.readyPiecesLock.RUnlock()
+
+	for ; pt.readyPieces.IsSet(i); i++ {
+	}
+	totalPiece := pt.GetTotalPieces()
+	if totalPiece > 0 && i >= totalPiece {
+		// double check, re-search
+		for i = int32(0); pt.readyPieces.IsSet(i); i++ {
+		}
+		if totalPiece > 0 && i >= totalPiece {
+			return 0, false
+		}
+	}
+	return i, true
+}
+
 func (pt *peerTaskConductor) recoverFromPanic() {
 	if r := recover(); r != nil {
 		pt.Errorf("recovered from panic %q. Call stack:\n%v", r, string(debug.Stack()))
@@ -1276,6 +1368,7 @@ func (pt *peerTaskConductor) done() {
 	defer func() {
 		pt.broker.Stop()
 		pt.span.End()
+		pt.pieceDownloadCancel()
 		if pt.pieceTaskSyncManager != nil {
 			pt.pieceTaskSyncManager.cancel()
 		}
@@ -1367,6 +1460,7 @@ func (pt *peerTaskConductor) fail() {
 		close(pt.failCh)
 		pt.broker.Stop()
 		pt.span.End()
+		pt.pieceDownloadCancel()
 		if pt.pieceTaskSyncManager != nil {
 			pt.pieceTaskSyncManager.cancel()
 		}
diff --git a/client/daemon/peer/peertask_dummy.go b/client/daemon/peer/peertask_dummy.go
index 14df5bff..c684cbdc 100644
--- a/client/daemon/peer/peertask_dummy.go
+++ b/client/daemon/peer/peertask_dummy.go
@@ -22,7 +22,7 @@ import (
 	"google.golang.org/grpc"
 
 	"d7y.io/dragonfly/v2/internal/dferrors"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	schedulerclient "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
diff --git a/client/daemon/peer/peertask_piecetask_poller.go b/client/daemon/peer/peertask_piecetask_poller.go
index 8d1cf877..77f6261c 100644
--- a/client/daemon/peer/peertask_piecetask_poller.go
+++ b/client/daemon/peer/peertask_piecetask_poller.go
@@ -88,6 +88,7 @@ func (poller *pieceTaskPoller) preparePieceTasksByPeer(
 	span.SetAttributes(config.AttributeGetPieceLimit.Int(int(request.Limit)))
 	defer span.End()
 
+	var maxRetries = 60
 	// when cdn returns base.Code_CDNTaskNotFound, report it to scheduler and wait cdn download it.
 retry:
 	ptc.Debugf("try get piece task from peer %s, piece num: %d, limit: %d\"", peer.PeerId, request.StartNum, request.Limit)
@@ -140,8 +141,10 @@ retry:
 	}
 
 	// currently, before cdn gc tasks, it did not notify scheduler, when cdn complains Code_CDNTaskNotFound, retry
-	if code == base.Code_CDNTaskNotFound && curPeerPacket == ptc.peerPacket.Load().(*scheduler.PeerPacket) {
+	if maxRetries > 0 && code == base.Code_CDNTaskNotFound && curPeerPacket == ptc.peerPacket.Load().(*scheduler.PeerPacket) {
 		span.AddEvent("retry for CdnTaskNotFound")
+		time.Sleep(time.Second)
+		maxRetries--
 		goto retry
 	}
 	return nil, err
diff --git a/client/daemon/peer/peertask_piecetask_synchronizer.go b/client/daemon/peer/peertask_piecetask_synchronizer.go
index f1132cf0..48eec280 100644
--- a/client/daemon/peer/peertask_piecetask_synchronizer.go
+++ b/client/daemon/peer/peertask_piecetask_synchronizer.go
@@ -18,13 +18,17 @@ package peer
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"sync"
 
+	"github.com/pkg/errors"
+	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/atomic"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	"d7y.io/dragonfly/v2/client/config"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
@@ -43,6 +47,7 @@ type pieceTaskSyncManager struct {
 
 type pieceTaskSynchronizer struct {
 	*logger.SugaredLoggerOnWith
+	span              trace.Span
 	client            dfdaemon.Daemon_SyncPieceTasksClient
 	dstPeer           *scheduler.PeerPacket_DestPeer
 	error             atomic.Value
@@ -50,45 +55,54 @@ type pieceTaskSynchronizer struct {
 	pieceRequestCh    chan *DownloadPieceRequest
 }
 
+type pieceTaskSynchronizerError struct {
+	err error
+}
+
 // FIXME for compatibility, sync will be called after the dfclient.GetPieceTasks deprecated and the pieceTaskPoller removed
-func (s *pieceTaskSyncManager) sync(pp *scheduler.PeerPacket, request *base.PieceTaskRequest) error {
+func (s *pieceTaskSyncManager) sync(pp *scheduler.PeerPacket, desiredPiece int32) error {
 	var (
-		peers  = map[string]bool{}
-		errors []error
+		peers = map[string]bool{}
+		errs  []error
 	)
 	peers[pp.MainPeer.PeerId] = true
 	// TODO if the worker failed, reconnect and retry
 	s.Lock()
 	defer s.Unlock()
 	if _, ok := s.workers[pp.MainPeer.PeerId]; !ok {
-		err := s.newPieceTaskSynchronizer(s.ctx, pp.MainPeer, request)
+		err := s.newPieceTaskSynchronizer(s.ctx, pp.MainPeer, desiredPiece)
 		if err != nil {
 			s.peerTaskConductor.Errorf("main peer SyncPieceTasks error: %s", err)
-			errors = append(errors, err)
+			errs = append(errs, err)
 		}
 	}
 	for _, p := range pp.StealPeers {
 		peers[p.PeerId] = true
 		if _, ok := s.workers[p.PeerId]; !ok {
-			err := s.newPieceTaskSynchronizer(s.ctx, p, request)
+			err := s.newPieceTaskSynchronizer(s.ctx, p, desiredPiece)
 			if err != nil {
 				s.peerTaskConductor.Errorf("steal peer SyncPieceTasks error: %s", err)
-				errors = append(errors, err)
+				errs = append(errs, err)
 			}
 		}
 	}
 
 	// cancel old workers
 	if len(s.workers) != len(peers) {
+		var peersToRemove []string
 		for p, worker := range s.workers {
 			if !peers[p] {
 				worker.close()
+				peersToRemove = append(peersToRemove, p)
 			}
 		}
+		for _, p := range peersToRemove {
+			delete(s.workers, p)
+		}
 	}
 
-	if len(errors) > 0 {
-		return errors[0]
+	if len(errs) > 0 {
+		return errs[0]
 	}
 	return nil
 }
@@ -103,29 +117,46 @@ func (s *pieceTaskSyncManager) cleanStaleWorker(destPeers []*scheduler.PeerPacke
 
 	// cancel old workers
 	if len(s.workers) != len(peers) {
+		var peersToRemove []string
 		for p, worker := range s.workers {
 			if !peers[p] {
 				worker.close()
+				peersToRemove = append(peersToRemove, p)
 			}
 		}
+		for _, p := range peersToRemove {
+			delete(s.workers, p)
+		}
 	}
 }
 
 func (s *pieceTaskSyncManager) newPieceTaskSynchronizer(
 	ctx context.Context,
 	dstPeer *scheduler.PeerPacket_DestPeer,
-	request *base.PieceTaskRequest) error {
-
-	request.DstPid = dstPeer.PeerId
+	lastNum int32) error {
+	request := &base.PieceTaskRequest{
+		TaskId:   s.peerTaskConductor.taskID,
+		SrcPid:   s.peerTaskConductor.peerID,
+		DstPid:   dstPeer.PeerId,
+		StartNum: uint32(lastNum),
+		Limit:    16,
+	}
 	if worker, ok := s.workers[dstPeer.PeerId]; ok {
-		// clean error worker
-		if worker.error.Load() != nil {
-			delete(s.workers, dstPeer.PeerId)
+		// worker is okay, keep it go on
+		if worker.error.Load() == nil {
+			s.peerTaskConductor.Infof("reuse PieceTaskSynchronizer %s", dstPeer.PeerId)
+			return nil
 		}
+		// clean error worker
+		delete(s.workers, dstPeer.PeerId)
 	}
 
-	request.DstPid = dstPeer.PeerId
 	client, err := dfclient.SyncPieceTasks(ctx, dstPeer, request)
+	// Refer: https://github.com/grpc/grpc-go/blob/v1.44.0/stream.go#L104
+	// When receive io.EOF, the real error should be discovered using RecvMsg, here is client.Recv() here
+	if err == io.EOF && client != nil {
+		_, err = client.Recv()
+	}
 	if err != nil {
 		s.peerTaskConductor.Errorf("call SyncPieceTasks error: %s, dest peer: %s", err, dstPeer.PeerId)
 		return err
@@ -140,7 +171,10 @@ func (s *pieceTaskSyncManager) newPieceTaskSynchronizer(
 		return err
 	}
 
+	_, span := tracer.Start(s.ctx, config.SpanSyncPieceTasks)
+	span.SetAttributes(config.AttributeTargetPeerID.String(dstPeer.PeerId))
 	synchronizer := &pieceTaskSynchronizer{
+		span:                span,
 		peerTaskConductor:   s.peerTaskConductor,
 		pieceRequestCh:      s.pieceRequestCh,
 		client:              client,
@@ -155,29 +189,32 @@ func (s *pieceTaskSyncManager) newPieceTaskSynchronizer(
 
 func (s *pieceTaskSyncManager) newMultiPieceTaskSynchronizer(
 	destPeers []*scheduler.PeerPacket_DestPeer,
-	lastNum int32) (legacyPeers []*scheduler.PeerPacket_DestPeer) {
+	desiredPiece int32) (legacyPeers []*scheduler.PeerPacket_DestPeer) {
 	s.Lock()
 	defer s.Unlock()
 	for _, peer := range destPeers {
-		request := &base.PieceTaskRequest{
-			TaskId:   s.peerTaskConductor.taskID,
-			SrcPid:   s.peerTaskConductor.peerID,
-			DstPid:   "",
-			StartNum: uint32(lastNum),
-			Limit:    16,
-		}
-		err := s.newPieceTaskSynchronizer(s.ctx, peer, request)
+		err := s.newPieceTaskSynchronizer(s.ctx, peer, desiredPiece)
 		if err == nil {
 			s.peerTaskConductor.Infof("connected to peer: %s", peer.PeerId)
 			continue
 		}
-		legacyPeers = append(legacyPeers, peer)
 		// when err is codes.Unimplemented, fallback to legacy get piece grpc
 		stat, ok := status.FromError(err)
 		if ok && stat.Code() == codes.Unimplemented {
+			// for legacy peers, when get pieces error, will report the error
 			s.peerTaskConductor.Warnf("connect peer %s error: %s, fallback to legacy get piece grpc", peer.PeerId, err)
+			legacyPeers = append(legacyPeers, peer)
+			continue
+		}
+
+		// other errors, report to scheduler
+		if errors.Is(err, context.DeadlineExceeded) {
+			// connect timeout error, report to scheduler to get more available peers
+			s.reportInvalidPeer(peer, base.Code_ClientConnectionError)
+			s.peerTaskConductor.Infof("connect to peer %s with error: %s, peer is invalid, skip legacy grpc", peer.PeerId, err)
 		} else {
-			s.reportError(peer)
+			// other errors, report to scheduler to get more available peers
+			s.reportInvalidPeer(peer, base.Code_ClientPieceRequestFail)
 			s.peerTaskConductor.Errorf("connect peer %s error: %s, not codes.Unimplemented", peer.PeerId, err)
 		}
 	}
@@ -185,50 +222,59 @@ func (s *pieceTaskSyncManager) newMultiPieceTaskSynchronizer(
 	return legacyPeers
 }
 
-func compositePieceResult(peerTaskConductor *peerTaskConductor, destPeer *scheduler.PeerPacket_DestPeer) *scheduler.PieceResult {
+func compositePieceResult(peerTaskConductor *peerTaskConductor, destPeer *scheduler.PeerPacket_DestPeer, code base.Code) *scheduler.PieceResult {
 	return &scheduler.PieceResult{
 		TaskId:        peerTaskConductor.taskID,
 		SrcPid:        peerTaskConductor.peerID,
 		DstPid:        destPeer.PeerId,
 		PieceInfo:     &base.PieceInfo{},
 		Success:       false,
-		Code:          base.Code_ClientPieceRequestFail,
+		Code:          code,
 		HostLoad:      nil,
 		FinishedCount: peerTaskConductor.readyPieces.Settled(),
 	}
 }
 
-func (s *pieceTaskSyncManager) reportError(destPeer *scheduler.PeerPacket_DestPeer) {
-	sendError := s.peerTaskConductor.sendPieceResult(compositePieceResult(s.peerTaskConductor, destPeer))
+func (s *pieceTaskSyncManager) reportInvalidPeer(destPeer *scheduler.PeerPacket_DestPeer, code base.Code) {
+	sendError := s.peerTaskConductor.sendPieceResult(compositePieceResult(s.peerTaskConductor, destPeer, code))
 	if sendError != nil {
-		s.peerTaskConductor.cancel(base.Code_SchedError, sendError.Error())
 		s.peerTaskConductor.Errorf("connect peer %s failed and send piece result with error: %s", destPeer.PeerId, sendError)
+		go s.peerTaskConductor.cancel(base.Code_SchedError, sendError.Error())
+	} else {
+		s.peerTaskConductor.Debugf("report invalid peer %s/%d to scheduler", destPeer.PeerId, code)
 	}
 }
 
 // acquire send the target piece to other peers
-func (s *pieceTaskSyncManager) acquire(request *base.PieceTaskRequest) {
+func (s *pieceTaskSyncManager) acquire(request *base.PieceTaskRequest) (attempt int, success int) {
 	s.RLock()
 	for _, p := range s.workers {
-		p.acquire(request)
+		attempt++
+		if p.acquire(request) == nil {
+			success++
+		}
 	}
 	s.RUnlock()
+	return
 }
 
 func (s *pieceTaskSyncManager) cancel() {
-	s.RLock()
+	s.ctxCancel()
+	s.Lock()
 	for _, p := range s.workers {
 		p.close()
 	}
-	s.RUnlock()
-	s.ctxCancel()
+	s.workers = map[string]*pieceTaskSynchronizer{}
+	s.Unlock()
 }
 
 func (s *pieceTaskSynchronizer) close() {
 	if err := s.client.CloseSend(); err != nil {
-		s.error.Store(err)
+		s.error.Store(&pieceTaskSynchronizerError{err})
 		s.Debugf("close send error: %s, dest peer: %s", err, s.dstPeer.PeerId)
+		s.span.RecordError(err)
 	}
+	s.span.End()
 }
 
 func (s *pieceTaskSynchronizer) dispatchPieceRequest(piecePacket *base.PiecePacket) {
@@ -242,10 +288,11 @@ func (s *pieceTaskSynchronizer) dispatchPieceRequest(piecePacket *base.PiecePack
 		if finished {
 			s.peerTaskConductor.Done()
 		}
+		return
 	}
 	for _, piece := range piecePacket.PieceInfos {
-		s.Infof("got piece %d from %s/%s, digest: %s, start: %d, size: %d, dest peer: %s",
-			piece.PieceNum, piecePacket.DstAddr, piecePacket.DstPid, piece.PieceMd5, piece.RangeStart, piece.RangeSize, s.dstPeer.PeerId)
+		s.Infof("got piece %d from %s/%s, digest: %s, start: %d, size: %d",
+			piece.PieceNum, piecePacket.DstAddr, piecePacket.DstPid, piece.PieceMd5, piece.RangeStart, piece.RangeSize)
 		// FIXME when set total piece but no total digest, fetch again
 		s.peerTaskConductor.requestedPiecesLock.Lock()
 		if !s.peerTaskConductor.requestedPieces.IsSet(piece.PieceNum) {
@@ -263,6 +310,7 @@ func (s *pieceTaskSynchronizer) dispatchPieceRequest(piecePacket *base.PiecePack
 		}
 		select {
 		case s.pieceRequestCh <- req:
+			s.span.AddEvent(fmt.Sprintf("send piece #%d request to piece download queue", piece.PieceNum))
 		case <-s.peerTaskConductor.successCh:
 			s.Infof("peer task success, stop dispatch piece request, dest peer: %s", s.dstPeer.PeerId)
 		case <-s.peerTaskConductor.failCh:
@@ -272,50 +320,54 @@ func (s *pieceTaskSynchronizer) dispatchPieceRequest(piecePacket *base.PiecePack
 }
 
 func (s *pieceTaskSynchronizer) receive(piecePacket *base.PiecePacket) {
-	var (
-		err error
-	)
-	s.dispatchPieceRequest(piecePacket)
+	var err error
 	for {
+		s.dispatchPieceRequest(piecePacket)
 		piecePacket, err = s.client.Recv()
-		if err == io.EOF {
-			s.Debugf("synchronizer receives io.EOF")
-			return
-		}
 		if err != nil {
-			if s.canceled(err) {
-				s.Debugf("synchronizer receives canceled")
-				return
-			}
-			s.error.Store(err)
-			s.reportError()
-			s.Errorf("synchronizer receives with error: %s", err)
-			return
+			break
 		}
+	}
 
-		s.dispatchPieceRequest(piecePacket)
+	if err == io.EOF {
+		s.Debugf("synchronizer receives io.EOF")
+	} else if s.canceled(err) {
+		s.Debugf("synchronizer receives canceled")
+		s.error.Store(&pieceTaskSynchronizerError{err})
+	} else {
+		s.Errorf("synchronizer receives with error: %s", err)
+		s.error.Store(&pieceTaskSynchronizerError{err})
+		s.reportError(err)
+		s.Errorf("synchronizer receives with error: %s", err)
 	}
 }
 
-func (s *pieceTaskSynchronizer) acquire(request *base.PieceTaskRequest) {
+func (s *pieceTaskSynchronizer) acquire(request *base.PieceTaskRequest) error {
+	if s.error.Load() != nil {
+		err := s.error.Load().(*pieceTaskSynchronizerError).err
+		s.Debugf("synchronizer already error %s, skip acquire more pieces", err)
+		return err
+	}
+	request.DstPid = s.dstPeer.PeerId
 	err := s.client.Send(request)
+	s.span.AddEvent(fmt.Sprintf("send piece #%d request", request.StartNum))
 	if err != nil {
-		if s.canceled(err) {
-			s.Debugf("synchronizer sends canceled")
-			return
-		}
+		// send should always ok
+		s.error.Store(&pieceTaskSynchronizerError{err})
 		s.Errorf("synchronizer sends with error: %s", err)
-		s.error.Store(err)
-		s.reportError()
-		return
+		s.reportError(err)
 	}
+	return err
 }
 
-func (s *pieceTaskSynchronizer) reportError() {
-	sendError := s.peerTaskConductor.sendPieceResult(compositePieceResult(s.peerTaskConductor, s.dstPeer))
+func (s *pieceTaskSynchronizer) reportError(err error) {
+	s.span.RecordError(err)
+	sendError := s.peerTaskConductor.sendPieceResult(compositePieceResult(s.peerTaskConductor, s.dstPeer, base.Code_ClientPieceRequestFail))
 	if sendError != nil {
-		s.peerTaskConductor.cancel(base.Code_SchedError, sendError.Error())
 		s.Errorf("sync piece info failed and send piece result with error: %s", sendError)
+		go s.peerTaskConductor.cancel(base.Code_SchedError, sendError.Error())
+	} else {
+		s.Debugf("report sync piece error to scheduler")
 	}
 }
 
diff --git a/client/daemon/rpcserver/rpcserver.go b/client/daemon/rpcserver/rpcserver.go
index 4f998711..089426f0 100644
--- a/client/daemon/rpcserver/rpcserver.go
+++ b/client/daemon/rpcserver/rpcserver.go
@@ -19,6 +19,7 @@ package rpcserver
 import (
 	"context"
 	"fmt"
+	"io"
 	"math"
 	"net"
 	"os"
@@ -139,58 +140,85 @@ func (s *server) GetPieceTasks(ctx context.Context, request *base.PieceTaskReque
 }
 
 // sendExistPieces will send as much as possible pieces
-func (s *server) sendExistPieces(request *base.PieceTaskRequest, sync dfdaemongrpc.Daemon_SyncPieceTasksServer, sentMap map[int32]struct{}) (total int32, sent int, err error) {
-	return sendExistPieces(sync.Context(), s.GetPieceTasks, request, sync, sentMap, true)
+func (s *server) sendExistPieces(log *logger.SugaredLoggerOnWith, request *base.PieceTaskRequest, sync dfdaemongrpc.Daemon_SyncPieceTasksServer, sentMap map[int32]struct{}) (total int32, err error) {
+	return sendExistPieces(sync.Context(), log, s.GetPieceTasks, request, sync, sentMap, true)
 }
 
 // sendFirstPieceTasks will send as much as possible pieces, even if no available pieces
-func (s *server) sendFirstPieceTasks(request *base.PieceTaskRequest, sync dfdaemongrpc.Daemon_SyncPieceTasksServer, sentMap map[int32]struct{}) (total int32, sent int, err error) {
-	return sendExistPieces(sync.Context(), s.GetPieceTasks, request, sync, sentMap, false)
+func (s *server) sendFirstPieceTasks(log *logger.SugaredLoggerOnWith, request *base.PieceTaskRequest, sync dfdaemongrpc.Daemon_SyncPieceTasksServer, sentMap map[int32]struct{}) (total int32, err error) {
+	return sendExistPieces(sync.Context(), log, s.GetPieceTasks, request, sync, sentMap, false)
 }
 
 func (s *server) SyncPieceTasks(sync dfdaemongrpc.Daemon_SyncPieceTasksServer) error {
 	request, err := sync.Recv()
 	if err != nil {
+		logger.Errorf("receive first sync piece tasks request error: %s", err.Error())
 		return err
 	}
+	log := logger.With("taskID", request.TaskId,
+		"localPeerID", request.DstPid, "remotePeerID", request.SrcPid)
+
 	skipPieceCount := request.StartNum
 	var sentMap = make(map[int32]struct{})
-	total, sent, err := s.sendFirstPieceTasks(request, sync, sentMap)
+
+	// TODO if not found, try to send to peer task conductor, then download it first
+	total, err := s.sendFirstPieceTasks(log, request, sync, sentMap)
 	if err != nil {
+		log.Errorf("send first piece tasks error: %s", err)
 		return err
 	}
 
-	// task is done, just return
-	if int(total) == sent {
-		return nil
+	recvReminding := func() error {
+		for {
+			request, err = sync.Recv()
+			if err == io.EOF {
+				return nil
+			}
+			if err != nil {
+				logger.Errorf("receive reminding piece tasks request error: %s", err)
+				return err
+			}
+			total, err = s.sendExistPieces(log, request, sync, sentMap)
+			if err != nil {
+				logger.Errorf("send reminding piece tasks error: %s", err)
+				return err
+			}
+		}
+	}
+
+	// task is done, just receive new piece tasks requests only
+	if int(total) == len(sentMap)+int(skipPieceCount) {
+		log.Infof("all piece tasks sent, receive new piece tasks requests only")
+		return recvReminding()
 	}
 
 	// subscribe peer task message for remaining pieces
 	result, ok := s.peerTaskManager.Subscribe(request)
 	if !ok {
-		// task not found, double check for done task
-		total, sent, err = s.sendExistPieces(request, sync, sentMap)
+		// running task not found, double check for done task
+		request.StartNum = searchNextPieceNum(sentMap, skipPieceCount)
+		total, err = s.sendExistPieces(log, request, sync, sentMap)
 		if err != nil {
+			log.Errorf("send exist piece tasks error: %s", err)
 			return err
 		}
 
-		if int(total) > sent {
+		if int(total) > len(sentMap)+int(skipPieceCount) {
 			return status.Errorf(codes.Unavailable, "peer task not finish, but no running task found")
 		}
-		return nil
+		return recvReminding()
 	}
 
 	var sub = &subscriber{
-		SubscribeResult: result,
-		sync:            sync,
-		request:         request,
-		skipPieceCount:  skipPieceCount,
-		totalPieces:     total,
-		sentMap:         sentMap,
-		done:            make(chan struct{}),
-		uploadAddr:      s.uploadAddr,
-		SugaredLoggerOnWith: logger.With("taskID", request.TaskId,
-			"localPeerID", request.DstPid, "remotePeerID", request.SrcPid),
+		SubscribeResult:     result,
+		sync:                sync,
+		request:             request,
+		skipPieceCount:      skipPieceCount,
+		totalPieces:         total,
+		sentMap:             sentMap,
+		done:                make(chan struct{}),
+		uploadAddr:          s.uploadAddr,
+		SugaredLoggerOnWith: log,
 	}
 
 	go sub.receiveRemainingPieceTaskRequests()
diff --git a/client/daemon/rpcserver/subscriber.go b/client/daemon/rpcserver/subscriber.go
index a2fbbf21..fcd10231 100644
--- a/client/daemon/rpcserver/subscriber.go
+++ b/client/daemon/rpcserver/subscriber.go
@@ -52,41 +52,56 @@ func (s *subscriber) getPieces(ctx context.Context, request *base.PieceTaskReque
 
 func sendExistPieces(
 	ctx context.Context,
+	log *logger.SugaredLoggerOnWith,
 	get func(ctx context.Context, request *base.PieceTaskRequest) (*base.PiecePacket, error),
 	request *base.PieceTaskRequest,
 	sync dfdaemon.Daemon_SyncPieceTasksServer,
-	sendMap map[int32]struct{},
-	skipSendZeroPiece bool) (total int32, sent int, err error) {
+	sentMap map[int32]struct{},
+	skipSendZeroPiece bool) (total int32, err error) {
 	if request.Limit <= 0 {
 		request.Limit = 16
 	}
+	var pp *base.PiecePacket
 	for {
-		pp, err := get(ctx, request)
+		pp, err = get(ctx, request)
 		if err != nil {
-			return -1, -1, err
+			log.Errorf("get piece error: %s", err)
+			return -1, err
 		}
 		if len(pp.PieceInfos) == 0 && skipSendZeroPiece {
-			return pp.TotalPiece, sent, nil
+			return pp.TotalPiece, nil
 		}
 		if err = sync.Send(pp); err != nil {
-			return pp.TotalPiece, sent, err
+			log.Errorf("send pieces error: %s", err)
+			return pp.TotalPiece, err
 		}
 		for _, p := range pp.PieceInfos {
-			sendMap[p.PieceNum] = struct{}{}
+			log.Infof("send ready piece %d", p.PieceNum)
+			sentMap[p.PieceNum] = struct{}{}
 		}
-		sent += len(pp.PieceInfos)
 		if uint32(len(pp.PieceInfos)) < request.Limit {
-			return pp.TotalPiece, sent, nil
+			log.Infof("sent %d pieces, total: %d", len(pp.PieceInfos), pp.TotalPiece)
+			return pp.TotalPiece, nil
 		}
 		// the get piece func always return sorted pieces, use last piece num + 1 to get more pieces
 		request.StartNum = uint32(pp.PieceInfos[request.Limit-1].PieceNum + 1)
 	}
 }
 
+func searchNextPieceNum(sentMap map[int32]struct{}, cur uint32) (nextPieceNum uint32) {
+	for i := int32(cur); ; i++ {
+		if _, ok := sentMap[i]; !ok {
+			nextPieceNum = uint32(i)
+			break
+		}
+	}
+	return nextPieceNum
+}
+
 // sendExistPieces will send as much as possible pieces
-func (s *subscriber) sendExistPieces(startNum uint32) (total int32, sent int, err error) {
+func (s *subscriber) sendExistPieces(startNum uint32) (total int32, err error) {
 	s.request.StartNum = startNum
-	return sendExistPieces(s.sync.Context(), s.getPieces, s.request, s.sync, s.sentMap, true)
+	return sendExistPieces(s.sync.Context(), s.SugaredLoggerOnWith, s.getPieces, s.request, s.sync, s.sentMap, true)
 }
 
 func (s *subscriber) receiveRemainingPieceTaskRequests() {
@@ -94,7 +109,7 @@ func (s *subscriber) receiveRemainingPieceTaskRequests() {
 	for {
 		request, err := s.sync.Recv()
 		if err == io.EOF {
-			s.Infof("SyncPieceTasks done, exit receiving")
+			s.Infof("remote SyncPieceTasks done, exit receiving")
 			return
 		}
 		if err != nil {
@@ -123,6 +138,7 @@ func (s *subscriber) receiveRemainingPieceTaskRequests() {
 			return
 		}
 		for _, p := range pp.PieceInfos {
+			s.Infof("send ready piece %d", p.PieceNum)
 			s.sentMap[p.PieceNum] = struct{}{}
 		}
 		s.Unlock()
@@ -145,27 +161,19 @@ loop:
 	for {
 		select {
 		case <-s.done:
-			s.Infof("SyncPieceTasks done, exit sending, local task is running")
+			s.Infof("remote SyncPieceTasks done, exit sending, local task is running")
 			return nil
 		case info := <-s.PieceInfoChannel:
+			s.Infof("receive piece info, num: %d, %v", info.Num, info.Finished)
 			// not desired piece
-			s.Debugf("receive piece info, num: %d, %v", info.Num, info.Finished)
 			if s.totalPieces > -1 && uint32(info.Num) < nextPieceNum {
 				continue
 			}
-			s.Lock()
-			total, _, err := s.sendExistPieces(uint32(info.Num))
 
+			s.Lock()
+			total, err := s.sendExistPieces(uint32(info.Num))
 			if err != nil {
-				if stat, ok := status.FromError(err); !ok {
-					// not grpc error
-					s.Errorf("sent exist pieces error: %s", err)
-				} else if stat.Code() == codes.Canceled {
-					err = nil
-					s.Debugf("SyncPieceTasks canceled, exit sending")
-				} else {
-					s.Warnf("SyncPieceTasks send error code %d/%s", stat.Code(), stat.Message())
-				}
+				err = s.saveError(err)
 				s.Unlock()
 				return err
 			}
@@ -183,23 +191,16 @@ loop:
 			nextPieceNum = s.searchNextPieceNum(nextPieceNum)
 			s.Unlock()
 		case <-s.Success:
+			s.Infof("peer task is success, send remaining pieces")
 			s.Lock()
 			// all pieces already sent
 			if s.totalPieces > -1 && nextPieceNum == uint32(s.totalPieces) {
 				s.Unlock()
 				break loop
 			}
-			total, _, err := s.sendExistPieces(nextPieceNum)
+			total, err := s.sendExistPieces(nextPieceNum)
 			if err != nil {
-				if stat, ok := status.FromError(err); !ok {
-					// not grpc error
-					s.Errorf("sent exist pieces error: %s", err)
-				} else if stat.Code() == codes.Canceled {
-					err = nil
-					s.Debugf("SyncPieceTasks canceled, exit sending")
-				} else {
-					s.Warnf("SyncPieceTasks send error code %d/%s", stat.Code(), stat.Message())
-				}
+				err = s.saveError(err)
 				s.Unlock()
 				return err
 			}
@@ -208,11 +209,14 @@ loop:
 			}
 			if s.totalPieces > -1 && len(s.sentMap)+int(s.skipPieceCount) != int(s.totalPieces) {
 				s.Unlock()
-				return dferrors.Newf(base.Code_ClientError, "peer task success, but can not send all pieces")
+				msg := "peer task success, but can not send all pieces"
+				s.Errorf(msg)
+				return dferrors.Newf(base.Code_ClientError, msg)
 			}
 			s.Unlock()
 			break loop
 		case <-s.Fail:
+			s.Errorf("peer task failed")
 			return dferrors.Newf(base.Code_ClientError, "peer task failed")
 		}
 	}
@@ -223,12 +227,19 @@ loop:
 	}
 }
 
-func (s *subscriber) searchNextPieceNum(cur uint32) (nextPieceNum uint32) {
-	for i := int32(cur); ; i++ {
-		if _, ok := s.sentMap[i]; !ok {
-			nextPieceNum = uint32(i)
-			break
-		}
+func (s *subscriber) saveError(err error) error {
+	if stat, ok := status.FromError(err); !ok {
+		// not grpc error
+		s.Errorf("sent exist pieces error: %s", err)
+	} else if stat.Code() == codes.Canceled {
+		err = nil
+		s.Debugf("SyncPieceTasks canceled, exit sending")
+	} else {
+		s.Warnf("SyncPieceTasks send error code %d/%s", stat.Code(), stat.Message())
 	}
-	return nextPieceNum
+	return err
+}
+
+func (s *subscriber) searchNextPieceNum(cur uint32) (nextPieceNum uint32) {
+	return searchNextPieceNum(s.sentMap, cur)
 }
diff --git a/cmd/cdn/main.go b/cmd/cdn/main.go
index b9086ccd..deee0fee 100644
--- a/cmd/cdn/main.go
+++ b/cmd/cdn/main.go
@@ -19,8 +19,7 @@ package main
 import (
 	_ "d7y.io/dragonfly/v2/cdn/supervisor/cdn/storage/disk"   //nolint:gci    // Register disk storage manager
 	_ "d7y.io/dragonfly/v2/cdn/supervisor/cdn/storage/hybrid" // Register hybrid storage manager
-	_ "d7y.io/dragonfly/v2/pkg/source/httpprotocol"           // Register http client
-	_ "d7y.io/dragonfly/v2/pkg/source/ossprotocol"            // Register oss client
+	_ "d7y.io/dragonfly/v2/pkg/source/loader"                 // Register all source clients
 
 	"d7y.io/dragonfly/v2/cmd/cdn/cmd" //nolint:gci
 )
diff --git a/cmd/dependency/dependency.go b/cmd/dependency/dependency.go
index ac872f57..526d6b03 100644
--- a/cmd/dependency/dependency.go
+++ b/cmd/dependency/dependency.go
@@ -46,7 +46,7 @@ import (
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/cmd/dependency/base"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/unit"
 	"d7y.io/dragonfly/v2/pkg/util/hostutils"
diff --git a/cmd/dependency/plugin_cmd.go b/cmd/dependency/plugin_cmd.go
index b0c5469a..3beb732b 100644
--- a/cmd/dependency/plugin_cmd.go
+++ b/cmd/dependency/plugin_cmd.go
@@ -25,6 +25,7 @@ import (
 
 	"d7y.io/dragonfly/v2/internal/dfplugin"
 	"d7y.io/dragonfly/v2/pkg/dfpath"
+	"d7y.io/dragonfly/v2/pkg/source"
 )
 
 var PluginCmd = &cobra.Command{
@@ -35,11 +36,19 @@ var PluginCmd = &cobra.Command{
 	DisableAutoGenTag: true,
 	SilenceUsage:      true,
 	Run: func(cmd *cobra.Command, args []string) {
-		ListAvailablePlugins()
+		ListAvailableInTreePlugins()
+		ListAvailableOutOfTreePlugins()
 	},
 }
 
-func ListAvailablePlugins() {
+func ListAvailableInTreePlugins() {
+	clients := source.ListClients()
+	for _, scheme := range clients {
+		fmt.Printf("source plugin: %s, location: in-tree\n", scheme)
+	}
+}
+
+func ListAvailableOutOfTreePlugins() {
 	d, err := dfpath.New()
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "failed to get plugin path: %q\n", err)
@@ -56,6 +65,10 @@ func ListAvailablePlugins() {
 		fmt.Fprintf(os.Stderr, "read plugin dir %s error: %s\n", d.PluginDir(), err)
 		return
 	}
+	if len(files) == 0 {
+		fmt.Fprintf(os.Stderr, "no out of tree plugin found\n")
+		return
+	}
 	for _, file := range files {
 		var attr []byte
 		fileName := file.Name()
@@ -86,6 +99,6 @@ func ListAvailablePlugins() {
 			continue
 		}
 
-		fmt.Printf("%s plugin %s, location: %s, attribute: %s\n", typ, name, fileName, string(attr))
+		fmt.Printf("%s plugin: %s, location: %s, attribute: %s\n", typ, name, fileName, string(attr))
 	}
 }
diff --git a/cmd/dfcache/cmd/root.go b/cmd/dfcache/cmd/root.go
index 6ea13ac6..92a374c3 100644
--- a/cmd/dfcache/cmd/root.go
+++ b/cmd/dfcache/cmd/root.go
@@ -29,7 +29,7 @@ import (
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/cmd/dependency"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/client"
 	"d7y.io/dragonfly/v2/version"
diff --git a/cmd/dfget/cmd/daemon.go b/cmd/dfget/cmd/daemon.go
index c577c9bd..c7c04abd 100644
--- a/cmd/dfget/cmd/daemon.go
+++ b/cmd/dfget/cmd/daemon.go
@@ -20,7 +20,6 @@ import (
 	"context"
 	"os"
 	"path"
-	"syscall"
 	"time"
 
 	"github.com/gofrs/flock"
@@ -33,7 +32,7 @@ import (
 	server "d7y.io/dragonfly/v2/client/daemon"
 	"d7y.io/dragonfly/v2/cmd/dependency"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/client"
 	"d7y.io/dragonfly/v2/version"
@@ -85,19 +84,20 @@ it supports container engine, wget and other downloading tools through proxy fun
 // daemon will be launched by dfget command
 // redirect stdout and stderr to file for debugging
 func redirectStdoutAndStderr(console bool, logDir string) {
-	// when console log is enabled, skip redirect stdout
-	if !console {
-		stdoutPath := path.Join(logDir, "daemon", "stdout.log")
-		if stdout, err := os.OpenFile(stdoutPath, os.O_WRONLY|os.O_CREATE|os.O_APPEND|os.O_SYNC, 0644); err != nil {
-			logger.Warnf("open %s error: %s", stdoutPath, err)
-		} else if err := syscall.Dup2(int(stdout.Fd()), 1); err != nil {
-			logger.Warnf("redirect stdout error: %s", err)
-		}
+	// when console log is enabled, skip redirect
+	if console {
+		return
+	}
+	stdoutPath := path.Join(logDir, "daemon", "stdout.log")
+	if stdout, err := os.OpenFile(stdoutPath, os.O_WRONLY|os.O_CREATE|os.O_APPEND|os.O_SYNC, 0644); err != nil {
+		logger.Warnf("open %s error: %s", stdoutPath, err)
+	} else if err := dup2(int(stdout.Fd()), 1); err != nil {
+		logger.Warnf("redirect stdout error: %s", err)
 	}
 	stderrPath := path.Join(logDir, "daemon", "stderr.log")
 	if stderr, err := os.OpenFile(stderrPath, os.O_WRONLY|os.O_CREATE|os.O_APPEND|os.O_SYNC, 0644); err != nil {
 		logger.Warnf("open %s error: %s", stderrPath, err)
-	} else if err := syscall.Dup2(int(stderr.Fd()), 2); err != nil {
+	} else if err := dup2(int(stderr.Fd()), 2); err != nil {
 		logger.Warnf("redirect stderr error: %s", err)
 	}
 }
diff --git a/cmd/dfget/cmd/daemon_dup2.go b/cmd/dfget/cmd/daemon_dup2.go
new file mode 100644
index 00000000..22124703
--- /dev/null
+++ b/cmd/dfget/cmd/daemon_dup2.go
@@ -0,0 +1,26 @@
+//go:build !windows && !arm && !arm64
+// +build !windows,!arm,!arm64
+
+/*
+ *     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 cmd
+
+import "syscall"
+
+func dup2(from int, to int) (err error) {
+	return syscall.Dup2(from, to)
+}
diff --git a/cmd/dfget/cmd/daemon_dup3.go b/cmd/dfget/cmd/daemon_dup3.go
new file mode 100644
index 00000000..ed261643
--- /dev/null
+++ b/cmd/dfget/cmd/daemon_dup3.go
@@ -0,0 +1,29 @@
+//go:build !windows && (arm || arm64)
+// +build !windows
+// +build arm arm64
+
+/*
+ *     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 cmd
+
+import "syscall"
+
+func dup2(from int, to int) (err error) {
+	// linux_arm64 platform doesn't have syscall.Dup2
+	// so use the nearly identical syscall.Dup3 instead.
+	return syscall.Dup3(from, to, 0)
+}
diff --git a/cmd/dfget/cmd/root.go b/cmd/dfget/cmd/root.go
index 202ed94c..eccdd61e 100644
--- a/cmd/dfget/cmd/root.go
+++ b/cmd/dfget/cmd/root.go
@@ -36,8 +36,8 @@ import (
 	"d7y.io/dragonfly/v2/cmd/dependency"
 	"d7y.io/dragonfly/v2/internal/constants"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
 	"d7y.io/dragonfly/v2/pkg/basic"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/dfpath"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/client"
 	"d7y.io/dragonfly/v2/pkg/source"
diff --git a/cmd/dfget/main.go b/cmd/dfget/main.go
index 06915819..43244862 100644
--- a/cmd/dfget/main.go
+++ b/cmd/dfget/main.go
@@ -18,10 +18,8 @@ package main
 
 import (
 	"d7y.io/dragonfly/v2/cmd/dfget/cmd"
-	// Register http client
-	_ "d7y.io/dragonfly/v2/pkg/source/httpprotocol"
-	// Register oss client
-	_ "d7y.io/dragonfly/v2/pkg/source/ossprotocol"
+	// register all source clients
+	_ "d7y.io/dragonfly/v2/pkg/source/loader"
 )
 
 func main() {
diff --git a/deploy/docker-compose/run.sh b/deploy/docker-compose/run.sh
index 285726af..8483b66c 100755
--- a/deploy/docker-compose/run.sh
+++ b/deploy/docker-compose/run.sh
@@ -11,17 +11,12 @@ cd $DIR
 prepare(){
     mkdir -p config
 
-    cat template/cdn.template.yaml > config/cdn.yaml
-    cat template/dfget.template.yaml > config/dfget.yaml
-    cat template/scheduler.template.yaml > config/scheduler.yaml
-    cat template/manager.template.yaml > config/manager.yaml
-
     ip=${IP:-$(hostname -i)}
 
-    sed -i "s,__IP__,$ip," config/dfget.yaml
-    sed -i "s,__IP__,$ip," config/cdn.yaml
-    sed -i "s,__IP__,$ip," config/scheduler.yaml
-    sed -i "s,__IP__,$ip," config/manager.yaml
+    sed "s,__IP__,$ip," template/dfget.template.yaml > config/dfget.yaml
+    sed "s,__IP__,$ip," template/cdn.template.yaml > config/cdn.yaml
+    sed "s,__IP__,$ip," template/scheduler.template.yaml > config/scheduler.yaml
+    sed "s,__IP__,$ip," template/manager.template.yaml > config/manager.yaml
 }
 
 run_container(){
diff --git a/hack/start-cdn.sh b/hack/start-cdn.sh
index ff4c5893..12a23607 100755
--- a/hack/start-cdn.sh
+++ b/hack/start-cdn.sh
@@ -4,6 +4,4 @@ set -o nounset
 set -o errexit
 set -o pipefail
 
-nginx
-
 /opt/dragonfly/bin/cdn "$@"
diff --git a/internal/dfnet/dfnet.go b/pkg/dfnet/dfnet.go
similarity index 100%
rename from internal/dfnet/dfnet.go
rename to pkg/dfnet/dfnet.go
diff --git a/pkg/rpc/cdnsystem/client/client.go b/pkg/rpc/cdnsystem/client/client.go
index 1fc3236a..4ce71ff1 100644
--- a/pkg/rpc/cdnsystem/client/client.go
+++ b/pkg/rpc/cdnsystem/client/client.go
@@ -25,7 +25,7 @@ import (
 	"google.golang.org/grpc"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
@@ -103,32 +103,22 @@ func (cc *cdnClient) ObtainSeeds(ctx context.Context, sr *cdnsystem.SeedRequest,
 }
 
 func (cc *cdnClient) GetPieceTasks(ctx context.Context, addr dfnet.NetAddr, req *base.PieceTaskRequest, opts ...grpc.CallOption) (*base.PiecePacket, error) {
-	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, err := cc.getSeederClientWithTarget(addr.GetEndpoint())
-		if err != nil {
-			return nil, err
-		}
-		return client.GetPieceTasks(ctx, req, opts...)
-	}, 0.2, 2.0, 3, nil)
+	client, err := cc.getSeederClientWithTarget(addr.GetEndpoint())
 	if err != nil {
-		logger.WithTaskID(req.TaskId).Infof("GetPieceTasks: invoke cdn node %s GetPieceTasks failed: %v", addr.GetEndpoint(), err)
 		return nil, err
 	}
-	return res.(*base.PiecePacket), nil
+	return client.GetPieceTasks(ctx, req, opts...)
 }
 
 func (cc *cdnClient) SyncPieceTasks(ctx context.Context, addr dfnet.NetAddr, req *base.PieceTaskRequest, opts ...grpc.CallOption) (cdnsystem.Seeder_SyncPieceTasksClient, error) {
-	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, err := cc.getSeederClientWithTarget(addr.GetEndpoint())
-		if err != nil {
-			return nil, err
-		}
-		return client.SyncPieceTasks(ctx, opts...)
-	}, 0.2, 2.0, 3, nil)
+	client, err := cc.getSeederClientWithTarget(addr.GetEndpoint())
+	if err != nil {
+		return nil, err
+	}
+	syncClient, err := client.SyncPieceTasks(ctx, opts...)
 	if err != nil {
 		logger.WithTaskID(req.TaskId).Infof("SyncPieceTasks: invoke cdn node %s SyncPieceTasks failed: %v", addr.GetEndpoint(), err)
 		return nil, err
 	}
-	client := res.(cdnsystem.Seeder_SyncPieceTasksClient)
-	return client, client.Send(req)
+	return syncClient, syncClient.Send(req)
 }
diff --git a/pkg/rpc/client.go b/pkg/rpc/client.go
index dff07941..cea09795 100644
--- a/pkg/rpc/client.go
+++ b/pkg/rpc/client.go
@@ -33,7 +33,7 @@ import (
 
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 )
 
@@ -269,7 +269,6 @@ type candidateClient struct {
 }
 
 func (conn *Connection) createClient(target string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
-	// should not retry
 	ctx, cancel := context.WithTimeout(context.Background(), conn.dialTimeout)
 	defer cancel()
 	return grpc.DialContext(ctx, target, opts...)
diff --git a/pkg/rpc/dfdaemon/cdnsystem/mocks/client_mock.go b/pkg/rpc/dfdaemon/cdnsystem/mocks/client_mock.go
index 654f6f35..22794b47 100644
--- a/pkg/rpc/dfdaemon/cdnsystem/mocks/client_mock.go
+++ b/pkg/rpc/dfdaemon/cdnsystem/mocks/client_mock.go
@@ -8,7 +8,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	dfnet "d7y.io/dragonfly/v2/internal/dfnet"
+	dfnet "d7y.io/dragonfly/v2/pkg/dfnet"
 	base "d7y.io/dragonfly/v2/pkg/rpc/base"
 	cdnsystem "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
 	client "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem/client"
diff --git a/pkg/rpc/dfdaemon/client/client.go b/pkg/rpc/dfdaemon/client/client.go
index 4131cc35..560f7d35 100644
--- a/pkg/rpc/dfdaemon/client/client.go
+++ b/pkg/rpc/dfdaemon/client/client.go
@@ -28,7 +28,7 @@ import (
 	"google.golang.org/protobuf/types/known/emptypb"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
@@ -119,35 +119,25 @@ func (dc *daemonClient) Download(ctx context.Context, req *dfdaemon.DownRequest,
 
 func (dc *daemonClient) GetPieceTasks(ctx context.Context, target dfnet.NetAddr, ptr *base.PieceTaskRequest, opts ...grpc.CallOption) (*base.PiecePacket,
 	error) {
-	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, err := dc.getDaemonClientWithTarget(target.GetEndpoint())
-		if err != nil {
-			return nil, err
-		}
-		return client.GetPieceTasks(ctx, ptr, opts...)
-	}, 0.2, 2.0, 3, nil)
+	client, err := dc.getDaemonClientWithTarget(target.GetEndpoint())
 	if err != nil {
-		logger.WithTaskID(ptr.TaskId).Infof("GetPieceTasks: invoke daemon node %s GetPieceTasks failed: %v", target, err)
 		return nil, err
 	}
-	return res.(*base.PiecePacket), nil
+	return client.GetPieceTasks(ctx, ptr, opts...)
 }
 
 func (dc *daemonClient) SyncPieceTasks(ctx context.Context, target dfnet.NetAddr, ptr *base.PieceTaskRequest, opts ...grpc.CallOption) (dfdaemon.Daemon_SyncPieceTasksClient, error) {
-	res, err := rpc.ExecuteWithRetry(func() (interface{}, error) {
-		client, err := dc.getDaemonClientWithTarget(target.GetEndpoint())
-		if err != nil {
-			return nil, err
-		}
-		return client.SyncPieceTasks(ctx, opts...)
-	}, 0.2, 2.0, 3, nil)
+	client, err := dc.getDaemonClientWithTarget(target.GetEndpoint())
+	if err != nil {
+		return nil, err
+	}
+	syncClient, err := client.SyncPieceTasks(ctx, opts...)
 	if err != nil {
 		logger.WithTaskID(ptr.TaskId).Infof("SyncPieceTasks: invoke daemon node %s SyncPieceTasks failed: %v", target, err)
 		return nil, err
 	}
 
-	client := res.(dfdaemon.Daemon_SyncPieceTasksClient)
-	return client, client.Send(ptr)
+	return syncClient, syncClient.Send(ptr)
 }
 
 func (dc *daemonClient) CheckHealth(ctx context.Context, target dfnet.NetAddr, opts ...grpc.CallOption) (err error) {
diff --git a/pkg/rpc/dfdaemon/client/mocks/client_mock.go b/pkg/rpc/dfdaemon/client/mocks/client_mock.go
index c100e9c0..14b76f9e 100644
--- a/pkg/rpc/dfdaemon/client/mocks/client_mock.go
+++ b/pkg/rpc/dfdaemon/client/mocks/client_mock.go
@@ -8,7 +8,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	dfnet "d7y.io/dragonfly/v2/internal/dfnet"
+	dfnet "d7y.io/dragonfly/v2/pkg/dfnet"
 	base "d7y.io/dragonfly/v2/pkg/rpc/base"
 	dfdaemon "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
 	client "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/client"
diff --git a/pkg/rpc/dfdaemon/client/peer.go b/pkg/rpc/dfdaemon/client/peer.go
index c2b3fb13..77463e76 100644
--- a/pkg/rpc/dfdaemon/client/peer.go
+++ b/pkg/rpc/dfdaemon/client/peer.go
@@ -23,7 +23,7 @@ import (
 
 	"google.golang.org/grpc"
 
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/base/common"
diff --git a/pkg/rpc/manager/client/client.go b/pkg/rpc/manager/client/client.go
index e2e04167..4daf9a85 100644
--- a/pkg/rpc/manager/client/client.go
+++ b/pkg/rpc/manager/client/client.go
@@ -29,7 +29,7 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/reachable"
 	"d7y.io/dragonfly/v2/pkg/rpc/manager"
 )
@@ -102,10 +102,10 @@ func NewWithAddrs(netAddrs []dfnet.NetAddr) (Client, error) {
 			logger.Infof("use %s address for manager grpc client", netAddr.Addr)
 			return New(netAddr.Addr)
 		}
-		logger.Warnf("%s address can not reachable", netAddr.Addr)
+		logger.Warnf("%s manager address can not reachable", netAddr.Addr)
 	}
 
-	return nil, errors.New("can not find available addresses")
+	return nil, errors.New("can not find available manager addresses")
 }
 
 func (c *client) GetScheduler(req *manager.GetSchedulerRequest) (*manager.Scheduler, error) {
diff --git a/pkg/rpc/scheduler/client/client.go b/pkg/rpc/scheduler/client/client.go
index 4ce5b3f9..7968d14e 100644
--- a/pkg/rpc/scheduler/client/client.go
+++ b/pkg/rpc/scheduler/client/client.go
@@ -27,7 +27,7 @@ import (
 
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
diff --git a/pkg/rpc/scheduler/client/mocks/client_mock.go b/pkg/rpc/scheduler/client/mocks/client_mock.go
index 7807c4f6..529f55d6 100644
--- a/pkg/rpc/scheduler/client/mocks/client_mock.go
+++ b/pkg/rpc/scheduler/client/mocks/client_mock.go
@@ -8,7 +8,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	dfnet "d7y.io/dragonfly/v2/internal/dfnet"
+	dfnet "d7y.io/dragonfly/v2/pkg/dfnet"
 	scheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	client "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
 	gomock "github.com/golang/mock/gomock"
diff --git a/pkg/rpc/server_listen.go b/pkg/rpc/server_listen.go
index 98d4152b..0cce05b2 100644
--- a/pkg/rpc/server_listen.go
+++ b/pkg/rpc/server_listen.go
@@ -23,7 +23,7 @@ import (
 	"syscall"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 )
 
 // Listen wraps net.Listen with dfnet.NetAddr
diff --git a/pkg/source/clients/example/dfs.go b/pkg/source/clients/example/dfs.go
new file mode 100644
index 00000000..64f5e142
--- /dev/null
+++ b/pkg/source/clients/example/dfs.go
@@ -0,0 +1,61 @@
+/*
+ *     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 example
+
+import (
+	"bytes"
+	"io"
+
+	"d7y.io/dragonfly/v2/pkg/source"
+)
+
+const scheme = "dfs"
+
+var data = "hello world"
+
+type client struct {
+}
+
+func init() {
+	if err := source.Register(scheme, New(), nil); err != nil {
+		panic(err)
+	}
+}
+
+func New() source.ResourceClient {
+	return &client{}
+}
+
+func (c *client) GetContentLength(request *source.Request) (int64, error) {
+	return int64(len(data)), nil
+}
+
+func (c *client) IsSupportRange(request *source.Request) (bool, error) {
+	return false, nil
+}
+
+func (c *client) IsExpired(request *source.Request, info *source.ExpireInfo) (bool, error) {
+	panic("implement me")
+}
+
+func (c *client) Download(request *source.Request) (*source.Response, error) {
+	return source.NewResponse(io.NopCloser(bytes.NewBufferString(data))), nil
+}
+
+func (c *client) GetLastModified(request *source.Request) (int64, error) {
+	panic("implement me")
+}
diff --git a/pkg/source/hdfsprotocol/hdfs_source_client.go b/pkg/source/clients/hdfsprotocol/hdfs_source_client.go
similarity index 100%
rename from pkg/source/hdfsprotocol/hdfs_source_client.go
rename to pkg/source/clients/hdfsprotocol/hdfs_source_client.go
diff --git a/pkg/source/hdfsprotocol/hdfs_source_client_test.go b/pkg/source/clients/hdfsprotocol/hdfs_source_client_test.go
similarity index 100%
rename from pkg/source/hdfsprotocol/hdfs_source_client_test.go
rename to pkg/source/clients/hdfsprotocol/hdfs_source_client_test.go
diff --git a/pkg/source/httpprotocol/http_source_client.go b/pkg/source/clients/httpprotocol/http_source_client.go
similarity index 100%
rename from pkg/source/httpprotocol/http_source_client.go
rename to pkg/source/clients/httpprotocol/http_source_client.go
diff --git a/pkg/source/httpprotocol/http_source_client_test.go b/pkg/source/clients/httpprotocol/http_source_client_test.go
similarity index 100%
rename from pkg/source/httpprotocol/http_source_client_test.go
rename to pkg/source/clients/httpprotocol/http_source_client_test.go
diff --git a/pkg/source/ossprotocol/oss_source_client.go b/pkg/source/clients/ossprotocol/oss_source_client.go
similarity index 100%
rename from pkg/source/ossprotocol/oss_source_client.go
rename to pkg/source/clients/ossprotocol/oss_source_client.go
diff --git a/pkg/source/loader/dfs.go.example b/pkg/source/loader/dfs.go.example
new file mode 100644
index 00000000..49b092f2
--- /dev/null
+++ b/pkg/source/loader/dfs.go.example
@@ -0,0 +1,21 @@
+/*
+ *     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 loader
+
+import (
+	_ "d7y.io/dragonfly/v2/pkg/source/clients/example" // Register dfs client
+)
diff --git a/pkg/source/loader/http.go b/pkg/source/loader/http.go
new file mode 100644
index 00000000..a74a3bfe
--- /dev/null
+++ b/pkg/source/loader/http.go
@@ -0,0 +1,21 @@
+/*
+ *     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 loader
+
+import (
+	_ "d7y.io/dragonfly/v2/pkg/source/clients/httpprotocol" // Register http client
+)
diff --git a/pkg/source/loader/oss.go b/pkg/source/loader/oss.go
new file mode 100644
index 00000000..39b15ddb
--- /dev/null
+++ b/pkg/source/loader/oss.go
@@ -0,0 +1,21 @@
+/*
+ *     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 loader
+
+import (
+	_ "d7y.io/dragonfly/v2/pkg/source/clients/ossprotocol" // Register oss client
+)
diff --git a/pkg/source/source_client.go b/pkg/source/source_client.go
index 49d6e8bd..349c1797 100644
--- a/pkg/source/source_client.go
+++ b/pkg/source/source_client.go
@@ -120,14 +120,17 @@ type ResourceLister interface {
 }
 
 type ClientManager interface {
-	// Register a source client with scheme
+	// Register registers a source client with scheme
 	Register(scheme string, resourceClient ResourceClient, adapter requestAdapter, hook ...Hook) error
 
-	// UnRegister a source client from manager
+	// UnRegister revoke a source client from manager
 	UnRegister(scheme string)
 
-	// GetClient a source client by scheme
+	// GetClient gets a source client by scheme
 	GetClient(scheme string, options ...Option) (ResourceClient, bool)
+
+	// ListClients lists all supported client scheme
+	ListClients() []string
 }
 
 // clientManager implements the interface ClientManager
@@ -186,6 +189,16 @@ func (m *clientManager) UnRegister(scheme string) {
 	delete(m.clients, scheme)
 }
 
+func (m *clientManager) ListClients() []string {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var clients []string
+	for c := range m.clients {
+		clients = append(clients, c)
+	}
+	return clients
+}
+
 func (m *clientManager) GetClient(scheme string, options ...Option) (ResourceClient, bool) {
 	logger.Debugf("current clients: %#v", m.clients)
 	m.mu.RLock()
@@ -226,6 +239,10 @@ func UnRegister(scheme string) {
 	_defaultManager.UnRegister(scheme)
 }
 
+func ListClients() []string {
+	return _defaultManager.ListClients()
+}
+
 type requestAdapter func(request *Request) *Request
 
 // Hook TODO hook
diff --git a/scheduler/config/config.go b/scheduler/config/config.go
index 2768b198..4dc83d9b 100644
--- a/scheduler/config/config.go
+++ b/scheduler/config/config.go
@@ -121,6 +121,10 @@ func New() *Config {
 
 // Validate config parameters
 func (c *Config) Validate() error {
+	if c.Server == nil {
+		return errors.New("server requires parameter server")
+	}
+
 	if c.Server.IP == "" {
 		return errors.New("server requires parameter ip")
 	}
@@ -149,6 +153,10 @@ func (c *Config) Validate() error {
 		return errors.New("scheduler requires parameter retryInterval")
 	}
 
+	if c.Scheduler.GC == nil {
+		return errors.New("scheduler requires parameter gc")
+	}
+
 	if c.Scheduler.GC.PeerGCInterval <= 0 {
 		return errors.New("scheduler requires parameter peerGCInterval")
 	}
@@ -181,7 +189,7 @@ func (c *Config) Validate() error {
 		return errors.New("manager requires parameter keepAlive interval")
 	}
 
-	if c.Job.Enable {
+	if c.Job != nil && c.Job.Enable {
 		if c.Job.GlobalWorkerNum == 0 {
 			return errors.New("job requires parameter globalWorkerNum")
 		}
@@ -211,6 +219,10 @@ func (c *Config) Validate() error {
 		}
 	}
 
+	if c.Storage == nil {
+		return errors.New("server requires parameter storage")
+	}
+
 	if c.Storage.MaxSize <= 0 {
 		return errors.New("storage requires parameter maxSize")
 	}
@@ -223,7 +235,7 @@ func (c *Config) Validate() error {
 		return errors.New("storage requires parameter bufferSize")
 	}
 
-	if c.Metrics.Enable {
+	if c.Metrics != nil && c.Metrics.Enable {
 		if c.Metrics.Addr == "" {
 			return errors.New("metrics requires parameter addr")
 		}
diff --git a/scheduler/resource/cdn.go b/scheduler/resource/cdn.go
index 7a169055..2cd9f178 100644
--- a/scheduler/resource/cdn.go
+++ b/scheduler/resource/cdn.go
@@ -27,7 +27,7 @@ import (
 	"google.golang.org/grpc"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base/common"
 	"d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
diff --git a/scheduler/resource/cdn_mock.go b/scheduler/resource/cdn_mock.go
index 67a461af..17138375 100644
--- a/scheduler/resource/cdn_mock.go
+++ b/scheduler/resource/cdn_mock.go
@@ -8,7 +8,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	dfnet "d7y.io/dragonfly/v2/internal/dfnet"
+	dfnet "d7y.io/dragonfly/v2/pkg/dfnet"
 	base "d7y.io/dragonfly/v2/pkg/rpc/base"
 	cdnsystem "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
 	client "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem/client"
diff --git a/scheduler/resource/peer.go b/scheduler/resource/peer.go
index 4648b029..cf9a79be 100644
--- a/scheduler/resource/peer.go
+++ b/scheduler/resource/peer.go
@@ -149,6 +149,20 @@ type Peer struct {
 	// BlockPeers is bad peer ids
 	BlockPeers set.SafeSet
 
+	// NeedBackToSource needs downloaded from source
+	//
+	// When peer is registering, at the same time,
+	// scheduler needs to create the new corresponding task and the cdn is disabled,
+	// NeedBackToSource is set to true
+	NeedBackToSource *atomic.Bool
+
+	// IsBackToSource is downloaded from source
+	//
+	// When peer is scheduling and NeedBackToSource is true,
+	// scheduler needs to return Code_SchedNeedBackSource and
+	// IsBackToSource is set to true
+	IsBackToSource *atomic.Bool
+
 	// CreateAt is peer create time
 	CreateAt *atomic.Time
 
@@ -165,22 +179,24 @@ type Peer struct {
 // New Peer instance
 func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
 	p := &Peer{
-		ID:         id,
-		BizTag:     DefaultBizTag,
-		Pieces:     &bitset.BitSet{},
-		pieceCosts: []int64{},
-		Stream:     &atomic.Value{},
-		Task:       task,
-		Host:       host,
-		Parent:     &atomic.Value{},
-		Children:   &sync.Map{},
-		ChildCount: atomic.NewInt32(0),
-		StealPeers: set.NewSafeSet(),
-		BlockPeers: set.NewSafeSet(),
-		CreateAt:   atomic.NewTime(time.Now()),
-		UpdateAt:   atomic.NewTime(time.Now()),
-		mu:         &sync.RWMutex{},
-		Log:        logger.WithTaskAndPeerID(task.ID, id),
+		ID:               id,
+		BizTag:           DefaultBizTag,
+		Pieces:           &bitset.BitSet{},
+		pieceCosts:       []int64{},
+		Stream:           &atomic.Value{},
+		Task:             task,
+		Host:             host,
+		Parent:           &atomic.Value{},
+		Children:         &sync.Map{},
+		ChildCount:       atomic.NewInt32(0),
+		StealPeers:       set.NewSafeSet(),
+		BlockPeers:       set.NewSafeSet(),
+		NeedBackToSource: atomic.NewBool(false),
+		IsBackToSource:   atomic.NewBool(false),
+		CreateAt:         atomic.NewTime(time.Now()),
+		UpdateAt:         atomic.NewTime(time.Now()),
+		mu:               &sync.RWMutex{},
+		Log:              logger.WithTaskAndPeerID(task.ID, id),
 	}
 
 	// Initialize state machine
@@ -222,6 +238,7 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
 				p.Log.Infof("peer state is %s", e.FSM.Current())
 			},
 			PeerEventDownloadFromBackToSource: func(e *fsm.Event) {
+				p.IsBackToSource.Store(true)
 				p.Task.BackToSourcePeers.Add(p)
 				p.DeleteParent()
 				p.Host.DeletePeer(p.ID)
@@ -353,6 +370,9 @@ 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 {
@@ -369,7 +389,7 @@ func (p *Peer) Depth() int {
 
 		// Prevent traversal tree from infinite loop
 		if p.ID == parent.ID {
-			p.Log.Info("tree structure produces an infinite loop")
+			p.Log.Error("tree structure produces an infinite loop")
 			break
 		}
 
@@ -379,18 +399,47 @@ func (p *Peer) Depth() int {
 	return depth
 }
 
+// Ancestors returns peer's ancestors
+func (p *Peer) Ancestors() []string {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+
+	var ancestors []string
+	node := p
+	for node != nil {
+		parent, ok := node.LoadParent()
+		if !ok {
+			return ancestors
+		}
+
+		// Prevent traversal tree from infinite loop
+		if parent.ID == node.ID {
+			p.Log.Error("tree structure produces an infinite loop")
+			return ancestors
+		}
+
+		node = parent
+		ancestors = append(ancestors, node.ID)
+	}
+
+	return ancestors
+}
+
 // IsDescendant determines whether it is ancestor of peer
 func (p *Peer) IsDescendant(ancestor *Peer) bool {
-	return isDescendant(ancestor, p)
+	return p.isDescendant(ancestor, p)
 }
 
 // IsAncestor determines whether it is descendant of peer
 func (p *Peer) IsAncestor(descendant *Peer) bool {
-	return isDescendant(p, descendant)
+	return p.isDescendant(p, descendant)
 }
 
 // isDescendant determines whether it is ancestor of peer
-func isDescendant(ancestor, descendant *Peer) bool {
+func (p *Peer) isDescendant(ancestor, descendant *Peer) bool {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+
 	node := descendant
 	for node != nil {
 		parent, ok := node.LoadParent()
@@ -402,6 +451,12 @@ func isDescendant(ancestor, descendant *Peer) bool {
 			return true
 		}
 
+		// Prevent traversal tree from infinite loop
+		if parent.ID == descendant.ID {
+			p.Log.Error("tree structure produces an infinite loop")
+			return true
+		}
+
 		node = parent
 	}
 
diff --git a/scheduler/scheduler/evaluator/evaluator_base.go b/scheduler/scheduler/evaluator/evaluator_base.go
index d771bdcd..3145f7f8 100644
--- a/scheduler/scheduler/evaluator/evaluator_base.go
+++ b/scheduler/scheduler/evaluator/evaluator_base.go
@@ -177,7 +177,8 @@ 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) {
+	if peer.FSM.Is(resource.PeerStateFailed) || peer.FSM.Is(resource.PeerStateLeave) || peer.FSM.Is(resource.PeerStatePending) ||
+		peer.FSM.Is(resource.PeerStateReceivedTiny) || peer.FSM.Is(resource.PeerStateReceivedSmall) || peer.FSM.Is(resource.PeerStateReceivedNormal) {
 		peer.Log.Debugf("peer is bad node because peer status is %s", peer.FSM.Current())
 		return true
 	}
diff --git a/scheduler/scheduler/scheduler.go b/scheduler/scheduler/scheduler.go
index c4e4005a..4676c63a 100644
--- a/scheduler/scheduler/scheduler.go
+++ b/scheduler/scheduler/scheduler.go
@@ -82,7 +82,9 @@ func (s *scheduler) ScheduleParent(ctx context.Context, peer *resource.Peer, blo
 
 		// If the scheduling exceeds the RetryBackSourceLimit or the latest cdn peer state is PeerStateFailed,
 		// peer will download the task back-to-source
-		if (n >= s.config.RetryBackSourceLimit || peer.Task.IsCDNFailed()) &&
+		isCDNFailed := peer.Task.IsCDNFailed()
+		needBackToSource := peer.NeedBackToSource.Load()
+		if (n >= s.config.RetryBackSourceLimit || isCDNFailed || needBackToSource) &&
 			peer.Task.CanBackToSource() {
 			stream, ok := peer.LoadStream()
 			if !ok {
@@ -90,13 +92,14 @@ func (s *scheduler) ScheduleParent(ctx context.Context, peer *resource.Peer, blo
 				return
 			}
 
+			peer.Log.Infof("peer downloads back-to-source, scheduling %d times, cdn is failed %t, peer need back-to-source %t",
+				n, isCDNFailed, needBackToSource)
+
 			// Notify peer back-to-source
 			if err := stream.Send(&rpcscheduler.PeerPacket{Code: base.Code_SchedNeedBackSource}); err != nil {
 				peer.Log.Errorf("send packet failed: %s", err.Error())
 				return
 			}
-			peer.Log.Infof("peer scheduling %d times, peer downloads back-to-source %d",
-				n, base.Code_SchedNeedBackSource)
 
 			if err := peer.FSM.Event(resource.PeerEventDownloadFromBackToSource); err != nil {
 				peer.Log.Errorf("peer fsm event failed: %s", err.Error())
@@ -156,19 +159,19 @@ func (s *scheduler) NotifyAndFindParent(ctx context.Context, peer *resource.Peer
 		return []*resource.Peer{}, false
 	}
 
-	// Find the parent that can be scheduled
-	parents := s.filterParents(peer, blocklist)
-	if len(parents) == 0 {
-		peer.Log.Info("can not find parents")
+	// Find the candidate parent that can be scheduled
+	candidateParents := s.filterCandidateParents(peer, blocklist)
+	if len(candidateParents) == 0 {
+		peer.Log.Info("can not find candidate parents")
 		return []*resource.Peer{}, false
 	}
 
-	// Sort parents by evaluation score
+	// Sort candidate parents by evaluation score
 	taskTotalPieceCount := peer.Task.TotalPieceCount.Load()
 	sort.Slice(
-		parents,
+		candidateParents,
 		func(i, j int) bool {
-			return s.evaluator.Evaluate(parents[i], peer, taskTotalPieceCount) > s.evaluator.Evaluate(parents[j], peer, taskTotalPieceCount)
+			return s.evaluator.Evaluate(candidateParents[i], peer, taskTotalPieceCount) > s.evaluator.Evaluate(candidateParents[j], peer, taskTotalPieceCount)
 		},
 	)
 
@@ -179,55 +182,56 @@ func (s *scheduler) NotifyAndFindParent(ctx context.Context, peer *resource.Peer
 		return []*resource.Peer{}, false
 	}
 
-	if err := stream.Send(constructSuccessPeerPacket(s.dynconfig, peer, parents[0], parents[1:])); err != nil {
+	if err := stream.Send(constructSuccessPeerPacket(s.dynconfig, peer, candidateParents[0], candidateParents[1:])); err != nil {
 		peer.Log.Error(err)
 		return []*resource.Peer{}, false
 	}
 
 	// Add steal peers to current peer
 	peer.StealPeers.Clear()
-	for _, parent := range parents[1:] {
-		peer.StealPeers.Add(parent.ID)
+	for _, candidateParent := range candidateParents[1:] {
+		peer.StealPeers.Add(candidateParent.ID)
 	}
 
 	// Replace peer's parent with scheduled parent
-	peer.ReplaceParent(parents[0])
+	peer.ReplaceParent(candidateParents[0])
 	peer.Log.Infof("schedule parent successful, replace parent to %s and steal peers is %v",
-		parents[0].ID, peer.StealPeers.Values())
-	return parents, true
+		candidateParents[0].ID, peer.StealPeers.Values())
+	peer.Log.Debugf("peer ancestors is %v", peer.Ancestors())
+	return candidateParents, true
 }
 
 // FindParent finds parent that best matches the evaluation
 func (s *scheduler) FindParent(ctx context.Context, peer *resource.Peer, blocklist set.SafeSet) (*resource.Peer, bool) {
-	// Filter the parent that can be scheduled
-	parents := s.filterParents(peer, blocklist)
-	if len(parents) == 0 {
-		peer.Log.Info("can not find parents")
+	// Filter the candidate parent that can be scheduled
+	candidateParents := s.filterCandidateParents(peer, blocklist)
+	if len(candidateParents) == 0 {
+		peer.Log.Info("can not find candidate parents")
 		return nil, false
 	}
 
-	// Sort parents by evaluation score
+	// Sort candidate parents by evaluation score
 	taskTotalPieceCount := peer.Task.TotalPieceCount.Load()
 	sort.Slice(
-		parents,
+		candidateParents,
 		func(i, j int) bool {
-			return s.evaluator.Evaluate(parents[i], peer, taskTotalPieceCount) > s.evaluator.Evaluate(parents[j], peer, taskTotalPieceCount)
+			return s.evaluator.Evaluate(candidateParents[i], peer, taskTotalPieceCount) > s.evaluator.Evaluate(candidateParents[j], peer, taskTotalPieceCount)
 		},
 	)
 
-	peer.Log.Infof("find parent %s successful", parents[0].ID)
-	return parents[0], true
+	peer.Log.Infof("find parent %s successful", candidateParents[0].ID)
+	return candidateParents[0], true
 }
 
-// Filter the parent that can be scheduled
-func (s *scheduler) filterParents(peer *resource.Peer, blocklist set.SafeSet) []*resource.Peer {
+// Filter the candidate parent that can be scheduled
+func (s *scheduler) filterCandidateParents(peer *resource.Peer, blocklist set.SafeSet) []*resource.Peer {
 	filterParentLimit := config.DefaultSchedulerFilterParentLimit
 	if config, ok := s.dynconfig.GetSchedulerClusterConfig(); ok && filterParentLimit > 0 {
 		filterParentLimit = int(config.FilterParentLimit)
 	}
 
-	var parents []*resource.Peer
-	var parentIDs []string
+	var candidateParents []*resource.Peer
+	var candidateParentIDs []string
 	var n int
 	peer.Task.Peers.Range(func(_, value interface{}) bool {
 		if n > filterParentLimit {
@@ -235,67 +239,84 @@ func (s *scheduler) filterParents(peer *resource.Peer, blocklist set.SafeSet) []
 		}
 		n++
 
-		parent, ok := value.(*resource.Peer)
+		candidateParent, ok := value.(*resource.Peer)
 		if !ok {
 			return true
 		}
 
-		if blocklist.Contains(parent.ID) {
-			peer.Log.Debugf("parent %s is not selected because it is in blocklist", parent.ID)
+		// Candidate parent is in blocklist
+		if blocklist.Contains(candidateParent.ID) {
+			peer.Log.Debugf("candidate parent %s is not selected because it is in blocklist", candidateParent.ID)
 			return true
 		}
 
-		if parent.StealPeers.Contains(peer.ID) {
-			peer.Log.Debugf("parent %s is not selected because it is in steal peers", parent.ID)
+		// Candidate parent is itself
+		if candidateParent.ID == peer.ID {
+			peer.Log.Debug("candidate parent is not selected because it is same")
 			return true
 		}
 
-		if parent.ID == peer.ID {
-			peer.Log.Debug("parent is not selected because it is same")
+		// Candidate parent is bad node
+		if s.evaluator.IsBadNode(candidateParent) {
+			peer.Log.Debugf("candidate parent %s is not selected because it is bad node", candidateParent.ID)
 			return true
 		}
 
-		if s.evaluator.IsBadNode(parent) {
-			peer.Log.Debugf("parent %s is not selected because it is bad node", parent.ID)
+		// Conditions for candidate parent to be a parent:
+		// 1. candidate parent has parent
+		// 2. candidate parent is CDN
+		// 3. candidate parent has been back-to-source
+		// 4. candidate parent has been succeeded
+		_, ok = candidateParent.LoadParent()
+		isBackToSource := candidateParent.IsBackToSource.Load()
+		if !ok && !candidateParent.Host.IsCDN && !isBackToSource &&
+			!candidateParent.FSM.Is(resource.PeerStateSucceeded) {
+			peer.Log.Debugf("candidate parent %s is not selected, because its download state is %t %t %t %s",
+				candidateParent.ID, ok, candidateParent.Host.IsCDN, isBackToSource, candidateParent.FSM.Current())
 			return true
 		}
 
+		// Candidate parent's depth exceeds available depth
 		peerChildCount := peer.ChildCount.Load()
-		parentDepth := parent.Depth()
+		parentDepth := candidateParent.Depth()
 		if peerChildCount > 0 && parentDepth > defaultAvailableDepth {
-			peer.Log.Debugf("peer has %d children and parent %s depth is %d", peerChildCount, parent.ID, parentDepth)
+			peer.Log.Debugf("candidate peer has %d children and parent %s depth is %d", peerChildCount, candidateParent.ID, parentDepth)
 			return true
 		}
 
+		// Peer's depth exceeds limit depth
 		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)
+			peer.Log.Debugf("exceeds the %d depth limit of the tree, peer depth is %d, candidate parent %s is %d", defaultDepthLimit, peerDepth, candidateParent.ID, parentDepth)
 			return true
 		}
 
-		if parent.IsDescendant(peer) {
-			peer.Log.Debugf("parent %s is not selected because it is descendant", parent.ID)
+		// Candidate parent is an descendant of peer
+		if candidateParent.IsDescendant(peer) {
+			peer.Log.Debugf("candidate parent %s is not selected because it is descendant", candidateParent.ID)
 			return true
 		}
 
-		if parent.IsAncestor(peer) {
-			peer.Log.Debugf("parent %s is not selected because it is ancestor", parent.ID)
+		// Candidate parent is an ancestor of peer
+		if candidateParent.IsAncestor(peer) {
+			peer.Log.Debugf("candidate parent %s is not selected because it is ancestor", candidateParent.ID)
 			return true
 		}
 
-		if parent.Host.FreeUploadLoad() <= 0 {
-			peer.Log.Debugf("parent %s is not selected because its free upload is empty, upload limit is %d, upload peer count is %d",
-				parent.ID, parent.Host.UploadLoadLimit.Load(), parent.Host.UploadPeerCount.Load())
+		// Candidate parent's free upload is empty
+		if candidateParent.Host.FreeUploadLoad() <= 0 {
+			peer.Log.Debugf("candidate parent %s is not selected because its free upload is empty, upload limit is %d, upload peer count is %d",
+				candidateParent.ID, candidateParent.Host.UploadLoadLimit.Load(), candidateParent.Host.UploadPeerCount.Load())
 			return true
 		}
 
-		parents = append(parents, parent)
-		parentIDs = append(parentIDs, parent.ID)
+		candidateParents = append(candidateParents, candidateParent)
+		candidateParentIDs = append(candidateParentIDs, candidateParent.ID)
 		return true
 	})
 
-	peer.Log.Infof("candidate parents include %#v", parentIDs)
-	return parents
+	peer.Log.Infof("candidate parents include %#v", candidateParentIDs)
+	return candidateParents
 }
 
 // Construct peer successful packet
diff --git a/scheduler/service/service.go b/scheduler/service/service.go
index 6d6d0251..5a3a7ffb 100644
--- a/scheduler/service/service.go
+++ b/scheduler/service/service.go
@@ -77,7 +77,7 @@ func New(
 // RegisterPeerTask registers peer and triggers CDN download task
 func (s *Service) RegisterPeerTask(ctx context.Context, req *rpcscheduler.PeerTaskRequest) (*rpcscheduler.RegisterResult, error) {
 	// Register task and trigger cdn download task
-	task, err := s.registerTask(ctx, req)
+	task, needBackToSource, err := s.registerTask(ctx, req)
 	if err != nil {
 		msg := fmt.Sprintf("peer %s register is failed: %s", req.PeerId, err.Error())
 		logger.Error(msg)
@@ -87,6 +87,10 @@ func (s *Service) RegisterPeerTask(ctx context.Context, req *rpcscheduler.PeerTa
 	peer := s.registerPeer(ctx, req.PeerId, task, host, req.UrlMeta.Tag)
 	peer.Log.Infof("register peer task request: %#v %#v %#v", req, req.UrlMeta, req.HostLoad)
 
+	// When the peer registers for the first time and
+	// does not have a CDN, it will back-to-source.
+	peer.NeedBackToSource.Store(needBackToSource)
+
 	// Task has been successful
 	if task.FSM.Is(resource.TaskStateSucceeded) {
 		peer.Log.Info("tasks can be reused")
@@ -163,13 +167,16 @@ func (s *Service) RegisterPeerTask(ctx context.Context, req *rpcscheduler.PeerTa
 				}, nil
 			}
 
-			peer.ReplaceParent(parent)
 			if err := peer.FSM.Event(resource.PeerEventRegisterSmall); err != nil {
 				msg := fmt.Sprintf("peer %s register is failed: %s", req.PeerId, err.Error())
 				peer.Log.Error(msg)
 				return nil, dferrors.New(base.Code_SchedError, msg)
 			}
 
+			peer.ReplaceParent(parent)
+			peer.Log.Infof("schedule parent successful, replace parent to %s ", parent.ID)
+			peer.Log.Debugf("peer ancestors is %v", peer.Ancestors())
+
 			singlePiece := &rpcscheduler.SinglePiece{
 				DstPid:  parent.ID,
 				DstAddr: fmt.Sprintf("%s:%d", parent.Host.IP, parent.Host.DownloadPort),
@@ -501,25 +508,26 @@ func (s *Service) LeaveTask(ctx context.Context, req *rpcscheduler.PeerTarget) e
 }
 
 // registerTask creates a new task or reuses a previous task
-func (s *Service) registerTask(ctx context.Context, req *rpcscheduler.PeerTaskRequest) (*resource.Task, error) {
+func (s *Service) registerTask(ctx context.Context, req *rpcscheduler.PeerTaskRequest) (*resource.Task, bool, error) {
 	task := resource.NewTask(idgen.TaskID(req.Url, req.UrlMeta), req.Url, resource.TaskTypeNormal, req.UrlMeta, resource.WithBackToSourceLimit(int32(s.config.Scheduler.BackSourceCount)))
 	task, loaded := s.resource.TaskManager().LoadOrStore(task)
 	if loaded && !task.FSM.Is(resource.TaskStateFailed) {
 		task.Log.Infof("task state is %s", task.FSM.Current())
-		return task, nil
+		return task, false, nil
 	}
 
 	// Trigger task
 	if err := task.FSM.Event(resource.TaskEventDownload); err != nil {
-		return nil, err
+		return nil, false, err
 	}
 
 	// Start trigger cdn task
 	if s.config.CDN.Enable {
 		go s.triggerCDNTask(ctx, task)
+		return task, false, nil
 	}
 
-	return task, nil
+	return task, true, nil
 }
 
 // registerHost creates a new host or reuses a previous host
@@ -664,7 +672,7 @@ func (s *Service) handlePieceFail(ctx context.Context, peer *resource.Peer, piec
 		peer.Log.Infof("parent %s is cdn", piece.DstPid)
 		fallthrough
 	case base.Code_CDNTaskNotFound:
-		s.handlePeerFail(ctx, parent)
+		s.handleLegacyCDNPeer(ctx, parent)
 
 		// Start trigger cdn task
 		if s.config.CDN.Enable {
@@ -730,6 +738,32 @@ func (s *Service) handlePeerFail(ctx context.Context, peer *resource.Peer) {
 	})
 }
 
+// handleLegacyCDNPeer handles cdn server's task has left,
+// but did not notify the schduler to leave the task
+func (s *Service) handleLegacyCDNPeer(ctx context.Context, peer *resource.Peer) {
+	if err := peer.FSM.Event(resource.PeerEventDownloadFailed); err != nil {
+		peer.Log.Errorf("peer fsm event failed: %s", err.Error())
+		return
+	}
+
+	if err := peer.FSM.Event(resource.PeerEventLeave); err != nil {
+		peer.Log.Errorf("peer fsm event failed: %s", err.Error())
+		return
+	}
+
+	// Reschedule a new parent to children of peer to exclude the current failed peer
+	peer.Children.Range(func(_, value interface{}) bool {
+		child, ok := value.(*resource.Peer)
+		if !ok {
+			return true
+		}
+
+		child.Log.Infof("schedule parent because of parent peer %s is failed", peer.ID)
+		s.scheduler.ScheduleParent(ctx, child, child.BlockPeers)
+		return true
+	})
+}
+
 // Conditions for the task to switch to the TaskStateSucceeded are:
 // 1. CDN downloads the resource successfully
 // 2. Dfdaemon back-to-source to download successfully


Test patch:
diff --git a/.github/workflows/compatibility-e2e.yml b/.github/workflows/compatibility-e2e.yml
index cec1839e..3bb8e09b 100644
--- a/.github/workflows/compatibility-e2e.yml
+++ b/.github/workflows/compatibility-e2e.yml
@@ -7,6 +7,8 @@ on:
   pull_request:
     branches: [main, release-*]
     paths-ignore: ["**.md", "**.png", "**.jpg", "**.svg", "**/docs/**"]
+  schedule:
+    - cron: '0 4 * * *'
 
 env:
   GO_VERSION: 1.17
diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml
index 1f9cbcb4..4a44f34b 100644
--- a/.github/workflows/e2e.yml
+++ b/.github/workflows/e2e.yml
@@ -7,6 +7,8 @@ on:
   pull_request:
     branches: [main, release-*]
     paths-ignore: ["**.md", "**.png", "**.jpg", "**.svg", "**/docs/**"]
+  schedule:
+    - cron: '0 4 * * *'
 
 env:
   GO_VERSION: 1.17
diff --git a/cdn/supervisor/cdn/cache_detector_test.go b/cdn/supervisor/cdn/cache_detector_test.go
index 12bfc628..58b67166 100644
--- a/cdn/supervisor/cdn/cache_detector_test.go
+++ b/cdn/supervisor/cdn/cache_detector_test.go
@@ -35,7 +35,7 @@ import (
 	storageMock "d7y.io/dragonfly/v2/cdn/supervisor/cdn/storage/mock"
 	"d7y.io/dragonfly/v2/cdn/supervisor/task"
 	"d7y.io/dragonfly/v2/pkg/source"
-	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
+	"d7y.io/dragonfly/v2/pkg/source/clients/httpprotocol"
 	sourceMock "d7y.io/dragonfly/v2/pkg/source/mock"
 	"d7y.io/dragonfly/v2/pkg/util/digestutils"
 	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
diff --git a/cdn/supervisor/cdn/manager_test.go b/cdn/supervisor/cdn/manager_test.go
index 6384bb8d..367c65d4 100644
--- a/cdn/supervisor/cdn/manager_test.go
+++ b/cdn/supervisor/cdn/manager_test.go
@@ -36,7 +36,7 @@ import (
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/source"
-	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
+	"d7y.io/dragonfly/v2/pkg/source/clients/httpprotocol"
 	sourceMock "d7y.io/dragonfly/v2/pkg/source/mock"
 	"d7y.io/dragonfly/v2/pkg/util/net/urlutils"
 	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
diff --git a/cdn/supervisor/progress/manager_test.go b/cdn/supervisor/progress/manager_test.go
index 77a0df64..fa6ba417 100644
--- a/cdn/supervisor/progress/manager_test.go
+++ b/cdn/supervisor/progress/manager_test.go
@@ -26,7 +26,7 @@ import (
 
 	"d7y.io/dragonfly/v2/cdn/supervisor/task"
 	"d7y.io/dragonfly/v2/pkg/source"
-	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
+	"d7y.io/dragonfly/v2/pkg/source/clients/httpprotocol"
 	sourcemock "d7y.io/dragonfly/v2/pkg/source/mock"
 	"d7y.io/dragonfly/v2/pkg/util/rangeutils"
 )
diff --git a/cdn/supervisor/task/manager_test.go b/cdn/supervisor/task/manager_test.go
index 572a2d69..ffdf3c34 100644
--- a/cdn/supervisor/task/manager_test.go
+++ b/cdn/supervisor/task/manager_test.go
@@ -29,7 +29,7 @@ import (
 	"d7y.io/dragonfly/v2/internal/util"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/source"
-	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
+	"d7y.io/dragonfly/v2/pkg/source/clients/httpprotocol"
 	sourcemock "d7y.io/dragonfly/v2/pkg/source/mock"
 )
 
diff --git a/client/config/peerhost_test.go b/client/config/peerhost_test.go
index 7a9c2840..6bf2f334 100644
--- a/client/config/peerhost_test.go
+++ b/client/config/peerhost_test.go
@@ -26,7 +26,7 @@ import (
 	"gopkg.in/yaml.v3"
 
 	"d7y.io/dragonfly/v2/client/clientutil"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/unit"
 )
 
diff --git a/client/daemon/daemon_test.go b/client/daemon/daemon_test.go
index f2ed6847..96a40294 100644
--- a/client/daemon/daemon_test.go
+++ b/client/daemon/daemon_test.go
@@ -22,7 +22,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc/manager"
 )
 
diff --git a/client/daemon/peer/peertask_manager_test.go b/client/daemon/peer/peertask_manager_test.go
index 38d2fa65..142f4162 100644
--- a/client/daemon/peer/peertask_manager_test.go
+++ b/client/daemon/peer/peertask_manager_test.go
@@ -49,7 +49,7 @@ import (
 	mock_daemon "d7y.io/dragonfly/v2/client/daemon/test/mock/daemon"
 	"d7y.io/dragonfly/v2/internal/dferrors"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
@@ -59,7 +59,7 @@ import (
 	schedulerclient "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
 	mock_scheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client/mocks"
 	"d7y.io/dragonfly/v2/pkg/source"
-	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
+	"d7y.io/dragonfly/v2/pkg/source/clients/httpprotocol"
 	sourceMock "d7y.io/dragonfly/v2/pkg/source/mock"
 	"d7y.io/dragonfly/v2/pkg/util/digestutils"
 )
diff --git a/client/daemon/peer/peertask_stream_backsource_partial_test.go b/client/daemon/peer/peertask_stream_backsource_partial_test.go
index 607e328c..a81159fd 100644
--- a/client/daemon/peer/peertask_stream_backsource_partial_test.go
+++ b/client/daemon/peer/peertask_stream_backsource_partial_test.go
@@ -44,7 +44,7 @@ import (
 	"d7y.io/dragonfly/v2/client/daemon/test"
 	mock_daemon "d7y.io/dragonfly/v2/client/daemon/test/mock/daemon"
 	"d7y.io/dragonfly/v2/internal/dferrors"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
@@ -53,7 +53,7 @@ import (
 	schedulerclient "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
 	mock_scheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client/mocks"
 	"d7y.io/dragonfly/v2/pkg/source"
-	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
+	"d7y.io/dragonfly/v2/pkg/source/clients/httpprotocol"
 	sourceMock "d7y.io/dragonfly/v2/pkg/source/mock"
 	"d7y.io/dragonfly/v2/pkg/util/digestutils"
 )
diff --git a/client/daemon/peer/piece_downloader_test.go b/client/daemon/peer/piece_downloader_test.go
index 4a71c80f..63917dad 100644
--- a/client/daemon/peer/piece_downloader_test.go
+++ b/client/daemon/peer/piece_downloader_test.go
@@ -40,7 +40,7 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/source"
-	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
+	"d7y.io/dragonfly/v2/pkg/source/clients/httpprotocol"
 )
 
 func TestPieceDownloader_DownloadPiece(t *testing.T) {
diff --git a/client/daemon/peer/piece_manager_test.go b/client/daemon/peer/piece_manager_test.go
index 391136ae..b8d9f0f7 100644
--- a/client/daemon/peer/piece_manager_test.go
+++ b/client/daemon/peer/piece_manager_test.go
@@ -43,7 +43,7 @@ import (
 	_ "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/server"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	"d7y.io/dragonfly/v2/pkg/source"
-	"d7y.io/dragonfly/v2/pkg/source/httpprotocol"
+	"d7y.io/dragonfly/v2/pkg/source/clients/httpprotocol"
 )
 
 func TestPieceManager_DownloadSource(t *testing.T) {
diff --git a/client/daemon/rpcserver/rpcserver_test.go b/client/daemon/rpcserver/rpcserver_test.go
index 8b8e6d82..da6f812c 100644
--- a/client/daemon/rpcserver/rpcserver_test.go
+++ b/client/daemon/rpcserver/rpcserver_test.go
@@ -34,7 +34,7 @@ import (
 	"d7y.io/dragonfly/v2/client/daemon/peer"
 	mock_peer "d7y.io/dragonfly/v2/client/daemon/test/mock/peer"
 	mock_storage "d7y.io/dragonfly/v2/client/daemon/test/mock/storage"
-	"d7y.io/dragonfly/v2/internal/dfnet"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	dfdaemongrpc "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
@@ -48,7 +48,7 @@ func TestMain(m *testing.M) {
 	os.Exit(m.Run())
 }
 
-func TestDownloadManager_ServeDownload(t *testing.T) {
+func Test_ServeDownload(t *testing.T) {
 	assert := testifyassert.New(t)
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -112,7 +112,7 @@ func TestDownloadManager_ServeDownload(t *testing.T) {
 	assert.True(lastResult.Done)
 }
 
-func TestDownloadManager_ServePeer(t *testing.T) {
+func Test_ServePeer(t *testing.T) {
 	assert := testifyassert.New(t)
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -216,7 +216,7 @@ func TestDownloadManager_ServePeer(t *testing.T) {
 	}
 }
 
-func TestDownloadManager_SyncPieceTasks(t *testing.T) {
+func Test_SyncPieceTasks(t *testing.T) {
 	assert := testifyassert.New(t)
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -231,12 +231,12 @@ func TestDownloadManager_SyncPieceTasks(t *testing.T) {
 	}
 	var tests = []struct {
 		name            string
-		existTaskID     string // test for non-exists task
-		existPieces     []pieceRange
+		existTaskID     string       // test for non-exists task
+		existPieces     []pieceRange // already exist pieces in storage
+		followingPieces []pieceRange // following pieces in running task subscribe channel
 		requestPieces   []int32
 		limit           uint32
 		totalPieces     uint32
-		followingPieces []pieceRange
 		success         bool
 		verify          func(t *testing.T, assert *testifyassert.Assertions)
 	}{
@@ -253,6 +253,20 @@ func TestDownloadManager_SyncPieceTasks(t *testing.T) {
 			verify: func(t *testing.T, assert *testifyassert.Assertions) {
 			},
 		},
+		{
+			name: "already exists in storage with extra get piece request",
+			existPieces: []pieceRange{
+				{
+					start: 0,
+					end:   10,
+				},
+			},
+			totalPieces:   11,
+			requestPieces: []int32{1, 3, 5, 7},
+			success:       true,
+			verify: func(t *testing.T, assert *testifyassert.Assertions) {
+			},
+		},
 		{
 			name: "already exists in storage - large",
 			existPieces: []pieceRange{
@@ -266,6 +280,20 @@ func TestDownloadManager_SyncPieceTasks(t *testing.T) {
 			verify: func(t *testing.T, assert *testifyassert.Assertions) {
 			},
 		},
+		{
+			name: "already exists in storage - large with extra get piece request",
+			existPieces: []pieceRange{
+				{
+					start: 0,
+					end:   1000,
+				},
+			},
+			totalPieces:   1001,
+			requestPieces: []int32{1, 3, 5, 7, 100, 500, 1000},
+			success:       true,
+			verify: func(t *testing.T, assert *testifyassert.Assertions) {
+			},
+		},
 		{
 			name: "partial exists in storage",
 			existPieces: []pieceRange{
@@ -452,7 +480,6 @@ func TestDownloadManager_SyncPieceTasks(t *testing.T) {
 				}
 
 				port, client := setupPeerServerAndClient(t, s, assert, s.ServePeer)
-				defer s.peerServer.GracefulStop()
 
 				syncClient, err := client.SyncPieceTasks(
 					context.Background(),
@@ -514,6 +541,7 @@ func TestDownloadManager_SyncPieceTasks(t *testing.T) {
 				if tc.success {
 					assert.Equal(int(maxNum+1), len(total))
 				}
+				s.peerServer.GracefulStop()
 			}
 
 		})
diff --git a/client/daemon/test/mock/scheduler/scheduler_client.go b/client/daemon/test/mock/scheduler/scheduler_client.go
index 2821c1c9..4aa17da3 100644
--- a/client/daemon/test/mock/scheduler/scheduler_client.go
+++ b/client/daemon/test/mock/scheduler/scheduler_client.go
@@ -8,7 +8,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	dfnet "d7y.io/dragonfly/v2/internal/dfnet"
+	dfnet "d7y.io/dragonfly/v2/pkg/dfnet"
 	scheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	client "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
 	gomock "github.com/golang/mock/gomock"
diff --git a/scheduler/resource/cdn_test.go b/scheduler/resource/cdn_test.go
index 2fe1c30d..879f1a96 100644
--- a/scheduler/resource/cdn_test.go
+++ b/scheduler/resource/cdn_test.go
@@ -27,8 +27,8 @@ import (
 	gomock "github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 
-	"d7y.io/dragonfly/v2/internal/dfnet"
 	"d7y.io/dragonfly/v2/manager/types"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	rpcscheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 	"d7y.io/dragonfly/v2/scheduler/config"
 	configmocks "d7y.io/dragonfly/v2/scheduler/config/mocks"
diff --git a/scheduler/resource/peer_test.go b/scheduler/resource/peer_test.go
index be84f453..0f2c763c 100644
--- a/scheduler/resource/peer_test.go
+++ b/scheduler/resource/peer_test.go
@@ -544,6 +544,55 @@ func TestPeer_Depth(t *testing.T) {
 	}
 }
 
+func TestPeer_Ancestors(t *testing.T) {
+	tests := []struct {
+		name    string
+		childID string
+		expect  func(t *testing.T, peer *Peer, mockChildPeer *Peer)
+	}{
+		{
+			name:    "parent is ancestor",
+			childID: idgen.PeerID("127.0.0.1"),
+			expect: func(t *testing.T, peer *Peer, mockChildPeer *Peer) {
+				assert := assert.New(t)
+				peer.StoreChild(mockChildPeer)
+				assert.Equal(len(mockChildPeer.Ancestors()), 1)
+				assert.EqualValues(mockChildPeer.Ancestors(), []string{peer.ID})
+			},
+		},
+		{
+			name:    "child has no parent",
+			childID: idgen.PeerID("127.0.0.1"),
+			expect: func(t *testing.T, peer *Peer, mockChildPeer *Peer) {
+				assert := assert.New(t)
+				assert.Equal(len(mockChildPeer.Ancestors()), 0)
+				assert.EqualValues(mockChildPeer.Ancestors(), []string(nil))
+			},
+		},
+		{
+			name:    "infinite loop",
+			childID: idgen.PeerID("127.0.0.1"),
+			expect: func(t *testing.T, peer *Peer, mockChildPeer *Peer) {
+				assert := assert.New(t)
+				peer.StoreChild(peer)
+				assert.Equal(len(peer.Ancestors()), 0)
+				assert.Equal(peer.Ancestors(), []string(nil))
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			mockHost := NewHost(mockRawHost)
+			mockTask := NewTask(mockTaskID, mockTaskURL, TaskTypeNormal, mockTaskURLMeta, WithBackToSourceLimit(mockTaskBackToSourceLimit))
+			mockChildPeer := NewPeer(tc.childID, mockTask, mockHost)
+			peer := NewPeer(mockPeerID, mockTask, mockHost)
+
+			tc.expect(t, peer, mockChildPeer)
+		})
+	}
+}
+
 func TestPeer_IsDescendant(t *testing.T) {
 	tests := []struct {
 		name    string
diff --git a/scheduler/scheduler/evaluator/evaluator_base_test.go b/scheduler/scheduler/evaluator/evaluator_base_test.go
index 34aaac82..83e6d1f5 100644
--- a/scheduler/scheduler/evaluator/evaluator_base_test.go
+++ b/scheduler/scheduler/evaluator/evaluator_base_test.go
@@ -593,6 +593,54 @@ func TestEvaluatorBase_IsBadNode(t *testing.T) {
 				assert.True(isBadNode)
 			},
 		},
+		{
+			name:            "peer state is PeerStatePending",
+			peer:            resource.NewPeer(mockPeerID, mockTask, mockHost),
+			totalPieceCount: 1,
+			mock: func(peer *resource.Peer) {
+				peer.FSM.SetState(resource.PeerStatePending)
+			},
+			expect: func(t *testing.T, isBadNode bool) {
+				assert := assert.New(t)
+				assert.True(isBadNode)
+			},
+		},
+		{
+			name:            "peer state is PeerStateReceivedTiny",
+			peer:            resource.NewPeer(mockPeerID, mockTask, mockHost),
+			totalPieceCount: 1,
+			mock: func(peer *resource.Peer) {
+				peer.FSM.SetState(resource.PeerStateReceivedTiny)
+			},
+			expect: func(t *testing.T, isBadNode bool) {
+				assert := assert.New(t)
+				assert.True(isBadNode)
+			},
+		},
+		{
+			name:            "peer state is PeerStateReceivedSmall",
+			peer:            resource.NewPeer(mockPeerID, mockTask, mockHost),
+			totalPieceCount: 1,
+			mock: func(peer *resource.Peer) {
+				peer.FSM.SetState(resource.PeerStateReceivedSmall)
+			},
+			expect: func(t *testing.T, isBadNode bool) {
+				assert := assert.New(t)
+				assert.True(isBadNode)
+			},
+		},
+		{
+			name:            "peer state is PeerStateReceivedNormal",
+			peer:            resource.NewPeer(mockPeerID, mockTask, mockHost),
+			totalPieceCount: 1,
+			mock: func(peer *resource.Peer) {
+				peer.FSM.SetState(resource.PeerStateReceivedNormal)
+			},
+			expect: func(t *testing.T, isBadNode bool) {
+				assert := assert.New(t)
+				assert.True(isBadNode)
+			},
+		},
 		{
 			name:            "download costs does not meet the normal distribution and last cost is twenty times more than mean",
 			peer:            resource.NewPeer(mockPeerID, mockTask, mockHost),
diff --git a/scheduler/scheduler/scheduler_test.go b/scheduler/scheduler/scheduler_test.go
index e39c7f8d..17873fff 100644
--- a/scheduler/scheduler/scheduler_test.go
+++ b/scheduler/scheduler/scheduler_test.go
@@ -193,6 +193,25 @@ func TestScheduler_ScheduleParent(t *testing.T) {
 				assert.True(peer.Task.FSM.Is(resource.TaskStatePending))
 			},
 		},
+		{
+			name: "peer need back-to-source and send Code_SchedNeedBackSource code success",
+			mock: func(cancel context.CancelFunc, peer *resource.Peer, cdnPeer *resource.Peer, blocklist set.SafeSet, stream rpcscheduler.Scheduler_ReportPieceResultServer, mr *rpcschedulermocks.MockScheduler_ReportPieceResultServerMockRecorder, md *configmocks.MockDynconfigInterfaceMockRecorder) {
+				task := peer.Task
+				task.StorePeer(peer)
+				peer.NeedBackToSource.Store(true)
+				peer.FSM.SetState(resource.PeerStateRunning)
+				peer.StoreStream(stream)
+
+				mr.Send(gomock.Eq(&rpcscheduler.PeerPacket{Code: base.Code_SchedNeedBackSource})).Return(nil).Times(1)
+			},
+			expect: func(t *testing.T, peer *resource.Peer) {
+				assert := assert.New(t)
+				_, ok := peer.LoadParent()
+				assert.False(ok)
+				assert.True(peer.FSM.Is(resource.PeerStateBackToSource))
+				assert.True(peer.Task.FSM.Is(resource.TaskStatePending))
+			},
+		},
 		{
 			name: "cdn peer state is PeerStateFailed and task state is PeerStateFailed",
 			mock: func(cancel context.CancelFunc, peer *resource.Peer, cdnPeer *resource.Peer, blocklist set.SafeSet, stream rpcscheduler.Scheduler_ReportPieceResultServer, mr *rpcschedulermocks.MockScheduler_ReportPieceResultServerMockRecorder, md *configmocks.MockDynconfigInterfaceMockRecorder) {
@@ -651,6 +670,8 @@ func TestScheduler_NotifyAndFindParent(t *testing.T) {
 			mock: func(peer *resource.Peer, mockPeer *resource.Peer, blocklist set.SafeSet, stream rpcscheduler.Scheduler_ReportPieceResultServer, dynconfig config.DynconfigInterface, ms *rpcschedulermocks.MockScheduler_ReportPieceResultServerMockRecorder, md *configmocks.MockDynconfigInterfaceMockRecorder) {
 				peer.FSM.SetState(resource.PeerStateRunning)
 				mockPeer.FSM.SetState(resource.PeerStateRunning)
+				peer.Task.BackToSourcePeers.Add(mockPeer)
+				mockPeer.IsBackToSource.Store(true)
 				peer.Task.StorePeer(mockPeer)
 				mockPeer.Pieces.Set(0)
 				peer.StoreStream(stream)
@@ -678,6 +699,10 @@ func TestScheduler_NotifyAndFindParent(t *testing.T) {
 				stealPeer.FSM.SetState(resource.PeerStateRunning)
 				peer.Task.StorePeer(mockPeer)
 				peer.Task.StorePeer(stealPeer)
+				peer.Task.BackToSourcePeers.Add(mockPeer)
+				peer.Task.BackToSourcePeers.Add(stealPeer)
+				mockPeer.IsBackToSource.Store(true)
+				stealPeer.IsBackToSource.Store(true)
 				mockPeer.Pieces.Set(0)
 				peer.StoreStream(stream)
 				gomock.InOrder(
@@ -807,13 +832,84 @@ func TestScheduler_FindParent(t *testing.T) {
 			},
 		},
 		{
-			name: "find parent",
+			name: "find back-to-source parent",
+			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)
+				mockPeers[1].FSM.SetState(resource.PeerStateRunning)
+				peer.Task.StorePeer(mockPeers[0])
+				peer.Task.StorePeer(mockPeers[1])
+				peer.Task.BackToSourcePeers.Add(mockPeers[0])
+				peer.Task.BackToSourcePeers.Add(mockPeers[1])
+				mockPeers[0].IsBackToSource.Store(true)
+				mockPeers[1].IsBackToSource.Store(true)
+				mockPeers[0].Pieces.Set(0)
+				mockPeers[1].Pieces.Set(0)
+				mockPeers[1].Pieces.Set(1)
+				mockPeers[1].Pieces.Set(2)
+
+				md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{}, false).Times(1)
+			},
+			expect: func(t *testing.T, mockPeers []*resource.Peer, parent *resource.Peer, ok bool) {
+				assert := assert.New(t)
+				assert.True(ok)
+				assert.Equal(mockPeers[1].ID, parent.ID)
+			},
+		},
+		{
+			name: "find cdn parent",
+			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)
+				mockPeers[1].FSM.SetState(resource.PeerStateRunning)
+				peer.Task.StorePeer(mockPeers[0])
+				peer.Task.StorePeer(mockPeers[1])
+				mockPeers[0].Host.IsCDN = true
+				mockPeers[1].Host.IsCDN = true
+				mockPeers[0].Pieces.Set(0)
+				mockPeers[1].Pieces.Set(0)
+				mockPeers[1].Pieces.Set(1)
+				mockPeers[1].Pieces.Set(2)
+
+				md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{}, false).Times(1)
+			},
+			expect: func(t *testing.T, mockPeers []*resource.Peer, parent *resource.Peer, ok bool) {
+				assert := assert.New(t)
+				assert.True(ok)
+				assert.Equal(mockPeers[1].ID, parent.ID)
+			},
+		},
+		{
+			name: "parent state is PeerStateSucceeded",
+			mock: func(peer *resource.Peer, mockPeers []*resource.Peer, blocklist set.SafeSet, md *configmocks.MockDynconfigInterfaceMockRecorder) {
+				peer.FSM.SetState(resource.PeerStateRunning)
+				mockPeers[0].FSM.SetState(resource.PeerStateSucceeded)
+				mockPeers[1].FSM.SetState(resource.PeerStateSucceeded)
+				peer.Task.StorePeer(mockPeers[0])
+				peer.Task.StorePeer(mockPeers[1])
+				mockPeers[0].Pieces.Set(0)
+				mockPeers[1].Pieces.Set(0)
+				mockPeers[1].Pieces.Set(1)
+				mockPeers[1].Pieces.Set(2)
+
+				md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{}, false).Times(1)
+			},
+			expect: func(t *testing.T, mockPeers []*resource.Peer, parent *resource.Peer, ok bool) {
+				assert := assert.New(t)
+				assert.True(ok)
+				assert.Equal(mockPeers[1].ID, parent.ID)
+			},
+		},
+		{
+			name: "find parent with ancestor",
 			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)
 				mockPeers[1].FSM.SetState(resource.PeerStateRunning)
 				peer.Task.StorePeer(mockPeers[0])
 				peer.Task.StorePeer(mockPeers[1])
+				mockPeers[0].StoreParent(mockPeers[3])
+				mockPeers[1].StoreParent(mockPeers[3])
 				mockPeers[0].Pieces.Set(0)
 				mockPeers[1].Pieces.Set(0)
 				mockPeers[1].Pieces.Set(1)
@@ -835,6 +931,10 @@ func TestScheduler_FindParent(t *testing.T) {
 				mockPeers[1].FSM.SetState(resource.PeerStateRunning)
 				peer.Task.StorePeer(mockPeers[0])
 				peer.Task.StorePeer(mockPeers[1])
+				peer.Task.BackToSourcePeers.Add(mockPeers[0])
+				peer.Task.BackToSourcePeers.Add(mockPeers[1])
+				mockPeers[0].IsBackToSource.Store(true)
+				mockPeers[1].IsBackToSource.Store(true)
 				mockPeers[0].Pieces.Set(0)
 				mockPeers[1].Pieces.Set(0)
 				mockPeers[1].Pieces.Set(1)
diff --git a/scheduler/service/service_test.go b/scheduler/service/service_test.go
index 0910ad9d..db0eeb68 100644
--- a/scheduler/service/service_test.go
+++ b/scheduler/service/service_test.go
@@ -154,6 +154,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				dferr, ok := err.(*dferrors.DfError)
 				assert.True(ok)
 				assert.Equal(dferr.Code, base.Code_SchedTaskStatusError)
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 		{
@@ -185,6 +186,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				assert.NoError(err)
 				assert.Equal(result.TaskId, peer.Task.ID)
 				assert.Equal(result.SizeScope, base.SizeScope_NORMAL)
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 		{
@@ -217,6 +219,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				dferr, ok := err.(*dferrors.DfError)
 				assert.True(ok)
 				assert.Equal(dferr.Code, base.Code_SchedError)
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 		{
@@ -253,6 +256,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				assert.Equal(result.DirectPiece, &rpcscheduler.RegisterResult_PieceContent{
 					PieceContent: peer.Task.DirectPiece,
 				})
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 		{
@@ -287,6 +291,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				dferr, ok := err.(*dferrors.DfError)
 				assert.True(ok)
 				assert.Equal(dferr.Code, base.Code_SchedError)
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 		{
@@ -321,6 +326,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				dferr, ok := err.(*dferrors.DfError)
 				assert.True(ok)
 				assert.Equal(dferr.Code, base.Code_SchedError)
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 		{
@@ -355,6 +361,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				assert.Equal(result.TaskId, peer.Task.ID)
 				assert.Equal(result.SizeScope, base.SizeScope_NORMAL)
 				assert.True(peer.FSM.Is(resource.PeerStateReceivedNormal))
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 		{
@@ -396,6 +403,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				assert.Equal(result.TaskId, peer.Task.ID)
 				assert.Equal(result.SizeScope, base.SizeScope_NORMAL)
 				assert.True(peer.FSM.Is(resource.PeerStateReceivedNormal))
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 		{
@@ -436,6 +444,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				dferr, ok := err.(*dferrors.DfError)
 				assert.True(ok)
 				assert.Equal(dferr.Code, base.Code_SchedError)
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 		{
@@ -476,6 +485,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				dferr, ok := err.(*dferrors.DfError)
 				assert.True(ok)
 				assert.Equal(dferr.Code, base.Code_SchedError)
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 		{
@@ -516,6 +526,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				assert.Equal(result.TaskId, peer.Task.ID)
 				assert.Equal(result.SizeScope, base.SizeScope_SMALL)
 				assert.True(peer.FSM.Is(resource.PeerStateReceivedSmall))
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 		{
@@ -550,6 +561,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				dferr, ok := err.(*dferrors.DfError)
 				assert.True(ok)
 				assert.Equal(dferr.Code, base.Code_SchedError)
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 		{
@@ -584,6 +596,7 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				assert.Equal(result.TaskId, peer.Task.ID)
 				assert.Equal(result.SizeScope, base.SizeScope_NORMAL)
 				assert.True(peer.FSM.Is(resource.PeerStateReceivedNormal))
+				assert.Equal(peer.NeedBackToSource.Load(), false)
 			},
 		},
 	}
@@ -1691,9 +1704,10 @@ func TestService_registerTask(t *testing.T) {
 					mt.LoadOrStore(gomock.Any()).Return(mockTask, true).Times(1),
 				)
 
-				task, err := svc.registerTask(context.Background(), req)
+				task, loaded, err := svc.registerTask(context.Background(), req)
 				assert := assert.New(t)
 				assert.NoError(err)
+				assert.False(loaded)
 				assert.EqualValues(mockTask, task)
 			},
 		},
@@ -1718,9 +1732,10 @@ func TestService_registerTask(t *testing.T) {
 					mt.LoadOrStore(gomock.Any()).Return(mockTask, true).Times(1),
 				)
 
-				task, err := svc.registerTask(context.Background(), req)
+				task, loaded, err := svc.registerTask(context.Background(), req)
 				assert := assert.New(t)
 				assert.NoError(err)
+				assert.False(loaded)
 				assert.EqualValues(mockTask, task)
 			},
 		},
@@ -1745,9 +1760,10 @@ func TestService_registerTask(t *testing.T) {
 					mt.LoadOrStore(gomock.Any()).Return(mockTask, true).Times(1),
 				)
 
-				task, err := svc.registerTask(context.Background(), req)
+				task, loaded, err := svc.registerTask(context.Background(), req)
 				assert := assert.New(t)
 				assert.NoError(err)
+				assert.False(loaded)
 				assert.EqualValues(mockTask, task)
 			},
 		},
@@ -1776,9 +1792,10 @@ func TestService_registerTask(t *testing.T) {
 					mc.TriggerTask(gomock.Any(), gomock.Any()).Do(func(ctx context.Context, task *resource.Task) { wg.Done() }).Return(mockPeer, &rpcscheduler.PeerResult{}, nil).Times(1),
 				)
 
-				task, err := svc.registerTask(context.Background(), req)
+				task, loaded, err := svc.registerTask(context.Background(), req)
 				assert := assert.New(t)
 				assert.NoError(err)
+				assert.False(loaded)
 				assert.EqualValues(mockTask, task)
 			},
 		},
@@ -1807,9 +1824,10 @@ func TestService_registerTask(t *testing.T) {
 					mc.TriggerTask(gomock.Any(), gomock.Any()).Do(func(ctx context.Context, task *resource.Task) { wg.Done() }).Return(mockPeer, &rpcscheduler.PeerResult{}, nil).Times(1),
 				)
 
-				task, err := svc.registerTask(context.Background(), req)
+				task, loaded, err := svc.registerTask(context.Background(), req)
 				assert := assert.New(t)
 				assert.NoError(err)
+				assert.False(loaded)
 				assert.EqualValues(mockTask, task)
 			},
 		},
@@ -1838,9 +1856,10 @@ func TestService_registerTask(t *testing.T) {
 					mc.TriggerTask(gomock.Any(), gomock.Any()).Do(func(ctx context.Context, task *resource.Task) { wg.Done() }).Return(mockPeer, &rpcscheduler.PeerResult{}, errors.New("foo")).Times(1),
 				)
 
-				task, err := svc.registerTask(context.Background(), req)
+				task, loaded, err := svc.registerTask(context.Background(), req)
 				assert := assert.New(t)
 				assert.NoError(err)
+				assert.False(loaded)
 				assert.EqualValues(mockTask, task)
 			},
 		},
@@ -1869,9 +1888,10 @@ func TestService_registerTask(t *testing.T) {
 					mc.TriggerTask(gomock.Any(), gomock.Any()).Do(func(ctx context.Context, task *resource.Task) { wg.Done() }).Return(mockPeer, &rpcscheduler.PeerResult{}, errors.New("foo")).Times(1),
 				)
 
-				task, err := svc.registerTask(context.Background(), req)
+				task, loaded, err := svc.registerTask(context.Background(), req)
 				assert := assert.New(t)
 				assert.NoError(err)
+				assert.False(loaded)
 				assert.EqualValues(mockTask, task)
 			},
 		},
@@ -1888,16 +1908,16 @@ func TestService_registerTask(t *testing.T) {
 				UrlMeta: mockTaskURLMeta,
 			},
 			run: func(t *testing.T, svc *Service, req *rpcscheduler.PeerTaskRequest, mockTask *resource.Task, mockPeer *resource.Peer, taskManager resource.TaskManager, cdn resource.CDN, mr *resource.MockResourceMockRecorder, mt *resource.MockTaskManagerMockRecorder, mc *resource.MockCDNMockRecorder) {
-
 				mockTask.FSM.SetState(resource.TaskStatePending)
 				gomock.InOrder(
 					mr.TaskManager().Return(taskManager).Times(1),
 					mt.LoadOrStore(gomock.Any()).Return(mockTask, false).Times(1),
 				)
 
-				task, err := svc.registerTask(context.Background(), req)
+				task, loaded, err := svc.registerTask(context.Background(), req)
 				assert := assert.New(t)
 				assert.NoError(err)
+				assert.True(loaded)
 				assert.EqualValues(mockTask, task)
 			},
 		},
@@ -1920,9 +1940,10 @@ func TestService_registerTask(t *testing.T) {
 					mt.LoadOrStore(gomock.Any()).Return(mockTask, true).Times(1),
 				)
 
-				task, err := svc.registerTask(context.Background(), req)
+				task, loaded, err := svc.registerTask(context.Background(), req)
 				assert := assert.New(t)
 				assert.NoError(err)
+				assert.True(loaded)
 				assert.EqualValues(mockTask, task)
 			},
 		},
@@ -2562,7 +2583,7 @@ func TestService_handlePieceFail(t *testing.T) {
 
 				svc.handlePieceFail(context.Background(), peer, piece)
 				assert := assert.New(t)
-				assert.True(parent.FSM.Is(resource.PeerStateFailed))
+				assert.True(parent.FSM.Is(resource.PeerStateLeave))
 			},
 		},
 		{
@@ -2592,7 +2613,7 @@ func TestService_handlePieceFail(t *testing.T) {
 				svc.handlePieceFail(context.Background(), peer, piece)
 				assert := assert.New(t)
 				assert.True(peer.FSM.Is(resource.PeerStateRunning))
-				assert.True(parent.FSM.Is(resource.PeerStateFailed))
+				assert.True(parent.FSM.Is(resource.PeerStateLeave))
 			},
 		},
 		{
diff --git a/scheduler/storage/storage_test.go b/scheduler/storage/storage_test.go
index 5ace7c61..97b994f0 100644
--- a/scheduler/storage/storage_test.go
+++ b/scheduler/storage/storage_test.go
@@ -585,34 +585,6 @@ func TestStorage_backups(t *testing.T) {
 				assert.Error(err)
 			},
 		},
-		{
-			name:    "get backups",
-			baseDir: os.TempDir(),
-			options: []Option{WithMaxSize(0), WithBufferSize(1)},
-			mock: func(t *testing.T, s Storage) {
-				if err := s.Create(Record{ID: "1"}); err != nil {
-					t.Fatal(err)
-				}
-
-				if err := s.Create(Record{ID: "2"}); err != nil {
-					t.Fatal(err)
-				}
-
-				if err := s.Create(Record{ID: "3"}); err != nil {
-					t.Fatal(err)
-				}
-			},
-			expect: func(t *testing.T, s Storage, baseDir string) {
-				assert := assert.New(t)
-				backups, err := s.(*storage).backups()
-				assert.NoError(err)
-				assert.Equal(len(backups), 3)
-				assert.Equal(backups[2].Name(), fmt.Sprintf("%s.%s", RecordFilePrefix, RecordFileExt))
-				if err := s.Clear(); err != nil {
-					t.Fatal(err)
-				}
-			},
-		},
 	}
 
 	for _, tc := range tests {
diff --git a/test/e2e/dfget_test.go b/test/e2e/dfget_test.go
index 8b389428..54b016b4 100644
--- a/test/e2e/dfget_test.go
+++ b/test/e2e/dfget_test.go
@@ -99,7 +99,8 @@ func singleDfgetTest(name, ns, label, podNamePrefix, container string, fileDetai
 		pod := e2eutil.NewPodExec(ns, podName, container)
 
 		// install curl
-		_, err = pod.Command("apk", "add", "-U", "curl").CombinedOutput()
+		out, err = pod.Command("apk", "add", "-U", "curl").CombinedOutput()
+		fmt.Println("apk output: " + string(out))
 		Expect(err).NotTo(HaveOccurred())
 
 		for path, size := range fileDetails {