Repository: dragonflyoss/Dragonfly2
Pull request #1505 (2022-07-29 08:09:37)

chore(deps): bump github.com/swaggo/gin-swagger from 1.5.0 to 1.5.1
Bumps [github.com/swaggo/gin-swagger](https://github.com/swaggo/gin-swagger) from 1.5.0 to 1.5.1.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/swaggo/gin-swagger/releases">github.com/swaggo/gin-swagger's releases</a>.</em></p>
<blockquote>
<h2>v1.5.1</h2>
<h2>Changelog</h2>
<p>b9e926c fix: update gin-contrib/gzip to latest version (<a href="https://github-redirect.dependabot.com/swaggo/gin-swagger/issues/220">#220</a>)</p>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/swaggo/gin-swagger/commit/b9e926c528e8b8269d0241b3053144235eef4e90"><code>b9e926c</code></a> fix: update gin-contrib/gzip to latest version (<a href="https://github-redirect.dependabot.com/swaggo/gin-swagger/issues/220">#220</a>)</li>
<li>See full diff in <a href="https://github.com/swaggo/gin-swagger/compare/v1.5.0...v1.5.1">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=github.com/swaggo/gin-swagger&package-manager=go_modules&previous-version=1.5.0&new-version=1.5.1)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>

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

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

```diff
@@            Coverage Diff             @@
##             main    #1505      +/-   ##
==========================================
+ Coverage   43.39%   43.41%   +0.01%     
==========================================
  Files         116      116              
  Lines       10690    10690              
==========================================
+ Hits         4639     4641       +2     
+ Misses       5567     5563       -4     
- Partials      484      486       +2     
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `43.41% <ø> (+0.01%)` | :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/1505?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/1505/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19zeW5jaHJvbml6ZXIuZ28=) | `47.61% <0.00%> (-1.30%)` | :arrow_down: |
| [client/daemon/peer/peertask\_conductor.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1505/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2NvbmR1Y3Rvci5nbw==) | `56.12% <0.00%> (+0.48%)` | :arrow_up: |


Patch:
diff --git a/client/config/peerhost.go b/client/config/peerhost.go
index 760b0c5e..d4cddb5c 100644
--- a/client/config/peerhost.go
+++ b/client/config/peerhost.go
@@ -254,6 +254,8 @@ type TransportOption struct {
 type ConcurrentOption struct {
 	// ThresholdSize indicates the threshold to download pieces concurrently
 	ThresholdSize util.Size `mapstructure:"thresholdSize" yaml:"thresholdSize"`
+	// ThresholdSpeed indicates the threshold download speed to download pieces concurrently
+	ThresholdSpeed unit.Bytes `mapstructure:"thresholdSpeed" yaml:"thresholdSpeed"`
 	// GoroutineCount indicates the concurrent goroutine count for every task
 	GoroutineCount int `mapstructure:"goroutineCount" yaml:"goroutineCount"`
 	// InitBackoff second for every piece failed, default: 0.5
diff --git a/client/daemon/peer/piece_manager.go b/client/daemon/peer/piece_manager.go
index af7f53c0..17a60012 100644
--- a/client/daemon/peer/piece_manager.go
+++ b/client/daemon/peer/piece_manager.go
@@ -291,19 +291,21 @@ func (pm *pieceManager) DownloadSource(ctx context.Context, pt Task, peerTaskReq
 	if err != nil {
 		return err
 	}
-
+	var (
+		metadata            *source.Metadata
+		supportConcurrent   bool
+		targetContentLength int64
+	)
 	if pm.concurrentOption != nil {
 		// check metadata
 		// 1. support range request
 		// 2. target content length is greater than concurrentOption.ThresholdSize
-		metadata, err := source.GetMetadata(backSourceRequest)
+		metadata, err = source.GetMetadata(backSourceRequest)
 		if err == nil && metadata.Validate != nil && metadata.Validate() == nil {
 			if !metadata.SupportRange || metadata.TotalContentLength == -1 {
 				goto singleDownload
 			}
-			var (
-				targetContentLength int64
-			)
+			supportConcurrent = true
 			if parsedRange != nil {
 				if parsedRange.Length > -1 {
 					targetContentLength = parsedRange.Length
@@ -337,7 +339,7 @@ func (pm *pieceManager) DownloadSource(ctx context.Context, pt Task, peerTaskReq
 					return err
 				}
 				// use concurrent piece download mode
-				return pm.concurrentDownloadSource(ctx, pt, peerTaskRequest, parsedRange, metadata)
+				return pm.concurrentDownloadSource(ctx, pt, peerTaskRequest, parsedRange, metadata, 0)
 			}
 		}
 	}
@@ -417,10 +419,10 @@ singleDownload:
 		return pm.downloadUnknownLengthSource(pt, pieceSize, reader)
 	}
 
-	return pm.downloadKnownLengthSource(pt, contentLength, pieceSize, reader)
+	return pm.downloadKnownLengthSource(ctx, pt, contentLength, pieceSize, reader, response, peerTaskRequest, parsedRange, metadata, supportConcurrent, targetContentLength)
 }
 
-func (pm *pieceManager) downloadKnownLengthSource(pt Task, contentLength int64, pieceSize uint32, reader io.Reader) error {
+func (pm *pieceManager) downloadKnownLengthSource(ctx context.Context, pt Task, contentLength int64, pieceSize uint32, reader io.Reader, response *source.Response, peerTaskRequest *scheduler.PeerTaskRequest, parsedRange *clientutil.Range, metadata *source.Metadata, supportConcurrent bool, targetContentLength int64) error {
 	log := pt.Log()
 	maxPieceNum := util.ComputePieceCount(contentLength, pieceSize)
 	pt.SetContentLength(contentLength)
@@ -466,6 +468,26 @@ func (pm *pieceManager) downloadKnownLengthSource(pt Task, contentLength int64,
 
 		pt.ReportPieceResult(request, result, nil)
 		pt.PublishPieceInfo(pieceNum, uint32(result.Size))
+		if supportConcurrent {
+			speed := int64(pieceSize) / (result.FinishTime - result.BeginTime)
+			if speed < int64(pm.concurrentOption.ThresholdSpeed) {
+				err = pt.GetStorage().UpdateTask(ctx,
+					&storage.UpdateTaskRequest{
+						PeerTaskMetadata: storage.PeerTaskMetadata{
+							PeerID: pt.GetPeerID(),
+							TaskID: pt.GetTaskID(),
+						},
+						ContentLength: targetContentLength,
+						Header:        &metadata.Header,
+					})
+				if err != nil {
+					log.Errorf("update task error: %s", err)
+					return err
+				}
+				response.Body.Close()
+				return pm.concurrentDownloadSource(ctx, pt, peerTaskRequest, parsedRange, metadata, pieceNum+1)
+			}
+		}
 	}
 
 	log.Infof("download from source ok")
@@ -724,7 +746,7 @@ func (pm *pieceManager) Import(ctx context.Context, ptm storage.PeerTaskMetadata
 	return nil
 }
 
-func (pm *pieceManager) concurrentDownloadSource(ctx context.Context, pt Task, peerTaskRequest *scheduler.PeerTaskRequest, parsedRange *clientutil.Range, metadata *source.Metadata) error {
+func (pm *pieceManager) concurrentDownloadSource(ctx context.Context, pt Task, peerTaskRequest *scheduler.PeerTaskRequest, parsedRange *clientutil.Range, metadata *source.Metadata, startPieceNum int32) error {
 	// parsedRange is always exist
 	pieceSize := pm.computePieceSize(parsedRange.Length)
 	pieceCount := util.ComputePieceCount(parsedRange.Length, pieceSize)
@@ -744,7 +766,7 @@ func (pm *pieceManager) concurrentDownloadSource(ctx context.Context, pt Task, p
 	var pieceCh = make(chan int32, con)
 
 	wg := sync.WaitGroup{}
-	wg.Add(int(pieceCount))
+	wg.Add(int(pieceCount - startPieceNum))
 
 	downloadedPieceCount := atomic.NewInt32(0)
 
@@ -784,7 +806,7 @@ func (pm *pieceManager) concurrentDownloadSource(ctx context.Context, pt Task, p
 		}(i)
 	}
 
-	for i := int32(0); i < pieceCount; i++ {
+	for i := startPieceNum; i < pieceCount; i++ {
 		select {
 		case <-ctx.Done():
 			log.Warnf("context cancelled")
diff --git a/deploy/docker-compose/template/dfget.template.yaml b/deploy/docker-compose/template/dfget.template.yaml
index ae978865..7e23e5f4 100644
--- a/deploy/docker-compose/template/dfget.template.yaml
+++ b/deploy/docker-compose/template/dfget.template.yaml
@@ -89,6 +89,7 @@ host:
 download:
   concurrent:
     thresholdSize: 10M
+    thresholdSpeed: 2M
     goroutineCount: 4
   # calculate digest when transfer files, set false to save memory
   calculateDigest: true
diff --git a/go.mod b/go.mod
index 322f2fb9..751742ea 100644
--- a/go.mod
+++ b/go.mod
@@ -48,15 +48,15 @@ require (
 	github.com/orcaman/concurrent-map/v2 v2.0.0
 	github.com/phayes/freeport v0.0.0-20220201140144-74d24b5ae9f5
 	github.com/prometheus/client_golang v1.12.2
-	github.com/schollz/progressbar/v3 v3.8.6
+	github.com/schollz/progressbar/v3 v3.8.7
 	github.com/serialx/hashring v0.0.0-20200727003509-22c0c7ab6b1b
 	github.com/shirou/gopsutil/v3 v3.22.5
 	github.com/sirupsen/logrus v1.8.1
 	github.com/spf13/cobra v1.5.0
 	github.com/spf13/viper v1.12.0
-	github.com/stretchr/testify v1.7.2
+	github.com/stretchr/testify v1.8.0
 	github.com/swaggo/files v0.0.0-20220610200504-28940afbdbfe
-	github.com/swaggo/gin-swagger v1.5.0
+	github.com/swaggo/gin-swagger v1.5.1
 	github.com/swaggo/swag v1.8.3
 	go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin v0.32.0
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.32.0
diff --git a/go.sum b/go.sum
index 0f25c409..dc393238 100644
--- a/go.sum
+++ b/go.sum
@@ -255,15 +255,14 @@ github.com/getkin/kin-openapi v0.76.0/go.mod h1:660oXbgy5JFMKreazJaQTw7o+X00qeSy
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/gin-contrib/cors v1.3.1 h1:doAsuITavI4IOcd0Y19U4B+O0dNWihRyX//nn4sEmgA=
 github.com/gin-contrib/cors v1.3.1/go.mod h1:jjEJ4268OPZUcU7k9Pm653S7lXUGcqMADzFA61xsmDk=
-github.com/gin-contrib/gzip v0.0.5 h1:mhnVU32YnnBh2LPH2iqRqsA/eR7SAqRaD388jL2s/j0=
-github.com/gin-contrib/gzip v0.0.5/go.mod h1:OPIK6HR0Um2vNmBUTlayD7qle4yVVRZT0PyhdUigrKk=
+github.com/gin-contrib/gzip v0.0.6 h1:NjcunTcGAj5CO1gn4N8jHOSIeRFHIbn51z6K+xaN4d4=
+github.com/gin-contrib/gzip v0.0.6/go.mod h1:QOJlmV2xmayAjkNS2Y8NQsMneuRShOU/kjovCXNuzzk=
 github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE=
 github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI=
 github.com/gin-contrib/static v0.0.1 h1:JVxuvHPuUfkoul12N7dtQw7KRn/pSMq7Ue1Va9Swm1U=
 github.com/gin-contrib/static v0.0.1/go.mod h1:CSxeF+wep05e0kCOsqWdAWbSszmc31zTIbD8TvWl7Hs=
 github.com/gin-gonic/gin v1.5.0/go.mod h1:Nd6IXA8m5kNZdNEHMBd93KT+mdY3+bewLgRvmCsR2Do=
 github.com/gin-gonic/gin v1.6.3/go.mod h1:75u5sXoLsGZoRN5Sgbi1eraJ4GU3++wFwWzhwvtwp4M=
-github.com/gin-gonic/gin v1.7.4/go.mod h1:jD2toBW3GZUr5UMcdrwQA10I7RuaFOl/SGeDjXkfUtY=
 github.com/gin-gonic/gin v1.7.7/go.mod h1:axIBovoeJpVj8S3BwE0uPMTeReE4+AfFtqpqaZ1qq1U=
 github.com/gin-gonic/gin v1.8.1 h1:4+fr/el88TOO3ewCmQr8cx/CtZ/umlIRIs5M4NTNjf8=
 github.com/gin-gonic/gin v1.8.1/go.mod h1:ji8BvRH1azfM+SYow9zQ6SZMvR8qOMZHmsCuWR9tTTk=
@@ -324,6 +323,7 @@ github.com/go-playground/universal-translator v0.18.0 h1:82dyy6p4OuJq4/CByFNOn/j
 github.com/go-playground/universal-translator v0.18.0/go.mod h1:UvRDBj+xPUEGrFYl+lu/H90nyDXpg0fqeB/AQUGNTVA=
 github.com/go-playground/validator/v10 v10.2.0/go.mod h1:uOYAAleCW8F/7oMFd6aG0GOhaH6EGOAJShg8Id5JGkI=
 github.com/go-playground/validator/v10 v10.4.1/go.mod h1:nlOn6nFhuKACm19sB/8EGNn9GlaMV7XkbRSipzJ0Ii4=
+github.com/go-playground/validator/v10 v10.10.0/go.mod h1:74x4gJWsvQexRdW8Pn3dXSGrTK4nAUsbPlLADvpJkos=
 github.com/go-playground/validator/v10 v10.11.0 h1:0W+xRM511GY47Yy3bZUbJVitCNg2BOGlCyvTqsp/xIw=
 github.com/go-playground/validator/v10 v10.11.0/go.mod h1:i+3WkQ1FvaUjjxh1kSvIA4dMGDBiPU55YFDl0WbKdWU=
 github.com/go-redis/cache/v8 v8.4.3 h1:+RZ0pQM+zOd6h/oWCsOl3+nsCgii9rn26oCYmU87kN8=
@@ -836,6 +836,7 @@ github.com/pelletier/go-toml v1.7.0/go.mod h1:vwGMzjaWMwyfHwgIBhI2YUM4fB6nL6lVAv
 github.com/pelletier/go-toml v1.9.3/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c=
 github.com/pelletier/go-toml v1.9.5 h1:4yBQzkHv+7BHq2PQUZF3Mx0IYxG7LsP222s7Agd3ve8=
 github.com/pelletier/go-toml v1.9.5/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c=
+github.com/pelletier/go-toml/v2 v2.0.1/go.mod h1:r9LEWfGN8R5k0VXJ+0BkIe7MYkRdwZOjgMj2KwnJFUo=
 github.com/pelletier/go-toml/v2 v2.0.2 h1:+jQXlF3scKIcSEKkdHzXhCTDLPFi5r1wnK6yPS+49Gw=
 github.com/pelletier/go-toml/v2 v2.0.2/go.mod h1:MovirKjgVRESsAvNZlAjtFwV867yGuwRkXbG66OzopI=
 github.com/performancecopilot/speed v3.0.0+incompatible/go.mod h1:/CLtqpZ5gBg1M9iaPbIdPPGyKcA8hKdoy6hAWba7Yac=
@@ -919,8 +920,8 @@ github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb
 github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
 github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
-github.com/schollz/progressbar/v3 v3.8.6 h1:QruMUdzZ1TbEP++S1m73OqRJk20ON11m6Wqv4EoGg8c=
-github.com/schollz/progressbar/v3 v3.8.6/go.mod h1:W5IEwbJecncFGBvuEh4A7HT1nZZ6WNIL2i3qbnI0WKY=
+github.com/schollz/progressbar/v3 v3.8.7 h1:rtje4lnXVD1Dy/RtPpGd2ijLCmQ7Su3G2ia8dJcRKIo=
+github.com/schollz/progressbar/v3 v3.8.7/go.mod h1:W5IEwbJecncFGBvuEh4A7HT1nZZ6WNIL2i3qbnI0WKY=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
 github.com/serialx/hashring v0.0.0-20200727003509-22c0c7ab6b1b h1:h+3JX2VoWTFuyQEo87pStk/a99dzIO1mM9KxIyLPGTU=
 github.com/serialx/hashring v0.0.0-20200727003509-22c0c7ab6b1b/go.mod h1:/yeG0My1xr/u+HZrFQ1tOQQQQrOawfyMUH13ai5brBc=
@@ -971,8 +972,9 @@ github.com/streadway/amqp v1.0.0/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1Sd
 github.com/streadway/handy v0.0.0-20190108123426-d5acb3125c2a/go.mod h1:qNTQ5P5JnDBl6z3cMAg/SywNDC5ABu5ApDIw6lUbRmI=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/objx v0.2.0 h1:Hbg2NidpLE8veEBkEZTL3CvlkUIVzuU9jDplZO54c48=
 github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE=
+github.com/stretchr/objx v0.4.0 h1:M2gUjqZET1qApGOWNSnZ49BAIMX4F/1plDv3+l31EJ4=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
@@ -981,8 +983,9 @@ github.com/stretchr/testify v1.6.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.7.2 h1:4jaiDzPyXQvSd7D0EjG45355tLlV3VOECpq10pLC+8s=
 github.com/stretchr/testify v1.7.2/go.mod h1:R6va5+xMeoiuVRoj+gSkQ7d3FALtqAAGI1FQKckRals=
+github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stvp/tempredis v0.0.0-20181119212430-b82af8480203 h1:QVqDTf3h2WHt08YuiTGPZLls0Wq99X9bWd0Q5ZSBesM=
 github.com/stvp/tempredis v0.0.0-20181119212430-b82af8480203/go.mod h1:oqN97ltKNihBbwlX8dLpwxCl3+HnXKV/R0e+sRLd9C8=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
@@ -990,8 +993,8 @@ github.com/subosito/gotenv v1.4.0 h1:yAzM1+SmVcz5R4tXGsNMu1jUl2aOJXoiWUCEwwnGrvs
 github.com/subosito/gotenv v1.4.0/go.mod h1:mZd6rFysKEcUhUHXJk0C/08wAgyDBFuwEYL7vWWGaGo=
 github.com/swaggo/files v0.0.0-20220610200504-28940afbdbfe h1:K8pHPVoTgxFJt1lXuIzzOX7zZhZFldJQK/CgKx9BFIc=
 github.com/swaggo/files v0.0.0-20220610200504-28940afbdbfe/go.mod h1:lKJPbtWzJ9JhsTN1k1gZgleJWY/cqq0psdoMmaThG3w=
-github.com/swaggo/gin-swagger v1.5.0 h1:hlLbxPj6qvbtX2wpbsZuOIlcnPRCUDGccA0zMKVNpME=
-github.com/swaggo/gin-swagger v1.5.0/go.mod h1:3mKpZClKx7mnUGsiwJeEkNhnr1VHMkMaTAXIoFYUXrA=
+github.com/swaggo/gin-swagger v1.5.1 h1:PFmlJU1LPn8DjrR0meVLX5gyFdgcPOkLcoFRRFx7WcY=
+github.com/swaggo/gin-swagger v1.5.1/go.mod h1:Cbj/MlHApPOjZdf4joWFXLLgmZVPyh54GPvPPyVjVZM=
 github.com/swaggo/swag v1.8.1/go.mod h1:ugemnJsPZm/kRwFUnzBlbHRd0JY9zE1M4F+uy2pAaPQ=
 github.com/swaggo/swag v1.8.3 h1:3pZSSCQ//gAH88lfmxM3Cd1+JCsxV8Md6f36b9hrZ5s=
 github.com/swaggo/swag v1.8.3/go.mod h1:jMLeXOOmYyjk8PvHTsXBdrubsNd9gUJTTCzL5iBnseg=


Test patch:
diff --git a/client/config/peerhost_test.go b/client/config/peerhost_test.go
index ef5774f4..72a24c1e 100644
--- a/client/config/peerhost_test.go
+++ b/client/config/peerhost_test.go
@@ -337,6 +337,7 @@ func TestPeerHostOption_Load(t *testing.T) {
 				ThresholdSize: util.Size{
 					Limit: 1,
 				},
+				ThresholdSpeed: unit.Bytes(1),
 				GoroutineCount: 1,
 				InitBackoff:    1,
 				MaxBackoff:     1,
diff --git a/client/config/testdata/config/daemon.yaml b/client/config/testdata/config/daemon.yaml
index 4066af53..48c33faa 100644
--- a/client/config/testdata/config/daemon.yaml
+++ b/client/config/testdata/config/daemon.yaml
@@ -79,6 +79,7 @@ download:
   watchdogTimeout: 1s
   concurrent:
     thresholdSize: 1
+    thresholdSpeed: 1
     goroutineCount: 1
     initBackoff: 1
     maxBackoff: 1
diff --git a/client/daemon/peer/piece_manager_test.go b/client/daemon/peer/piece_manager_test.go
index 8fe88bc6..d33dc40d 100644
--- a/client/daemon/peer/piece_manager_test.go
+++ b/client/daemon/peer/piece_manager_test.go
@@ -34,12 +34,14 @@ import (
 	testifyassert "github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
+	"golang.org/x/time/rate"
 
 	"d7y.io/dragonfly/v2/client/config"
 	"d7y.io/dragonfly/v2/client/daemon/storage"
 	"d7y.io/dragonfly/v2/client/daemon/test"
-	"d7y.io/dragonfly/v2/client/util"
+	clientutil "d7y.io/dragonfly/v2/client/util"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
+	"d7y.io/dragonfly/v2/internal/util"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	_ "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/server"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
@@ -67,7 +69,7 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 		config.SimpleLocalTaskStoreStrategy,
 		&config.StorageOption{
 			DataPath: t.TempDir(),
-			TaskExpireTime: util.Duration{
+			TaskExpireTime: clientutil.Duration{
 				Duration: -1 * time.Second,
 			},
 		}, func(request storage.CommonTaskRequest) {})
@@ -77,11 +79,13 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 	digest := hex.EncodeToString(hash.Sum(nil)[:16])
 
 	testCases := []struct {
-		name              string
-		pieceSize         uint32
-		withContentLength bool
-		checkDigest       bool
-		concurrentOption  *config.ConcurrentOption
+		name               string
+		pieceSize          uint32
+		withContentLength  bool
+		checkDigest        bool
+		recordDownloadTime bool
+		bandwidth          clientutil.Size
+		concurrentOption   *config.ConcurrentOption
 	}{
 		{
 			name:              "multiple pieces with content length, check digest",
@@ -90,23 +94,201 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 			withContentLength: true,
 		},
 		{
-			name:              "multiple pieces with content length",
-			pieceSize:         1024,
-			checkDigest:       false,
-			withContentLength: true,
+			name:               "multiple pieces with content length",
+			pieceSize:          1024,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
 		},
 		{
-			name:              "multiple pieces with content length, concurrent download",
-			pieceSize:         2048,
-			checkDigest:       false,
-			withContentLength: true,
+			name:               "multiple pieces with content length, concurrent download with 2 goroutines",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			concurrentOption: &config.ConcurrentOption{
+				GoroutineCount: 2,
+				ThresholdSize: clientutil.Size{
+					Limit: 1024,
+				},
+			},
+		},
+		{
+			name:               "multiple pieces with content length, concurrent download with 4 goroutines",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
 			concurrentOption: &config.ConcurrentOption{
 				GoroutineCount: 4,
-				ThresholdSize: util.Size{
+				ThresholdSize: clientutil.Size{
+					Limit: 1024,
+				},
+			},
+		},
+		{
+			name:               "multiple pieces with content length, concurrent download with 8 goroutines",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			concurrentOption: &config.ConcurrentOption{
+				GoroutineCount: 8,
+				ThresholdSize: clientutil.Size{
 					Limit: 1024,
 				},
 			},
 		},
+		{
+			name:               "multiple pieces with content length, concurrent download with 16 goroutines",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			concurrentOption: &config.ConcurrentOption{
+				GoroutineCount: 16,
+				ThresholdSize: clientutil.Size{
+					Limit: 1024,
+				},
+			},
+		},
+		{
+			name:               "multiple pieces with content length, single-thread download with download bandwidth 2KB/s",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			bandwidth:          clientutil.Size{Limit: 2048},
+		},
+		{
+			name:               "multiple pieces with content length, concurrent download with 2 goroutines, download bandwidth 2KB/s",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			bandwidth:          clientutil.Size{Limit: 2048},
+			concurrentOption: &config.ConcurrentOption{
+				GoroutineCount: 2,
+				ThresholdSize: clientutil.Size{
+					Limit: 1024 * 1024,
+				},
+				ThresholdSpeed: 8192,
+			},
+		},
+		{
+			name:               "multiple pieces with content length, concurrent download with 4 goroutines, download bandwidth 2KB/s",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			bandwidth:          clientutil.Size{Limit: 2048},
+			concurrentOption: &config.ConcurrentOption{
+				GoroutineCount: 4,
+				ThresholdSize: clientutil.Size{
+					Limit: 1024 * 1024,
+				},
+				ThresholdSpeed: 8192,
+			},
+		},
+		{
+			name:               "multiple pieces with content length, concurrent download with 8 goroutines, download bandwidth 2KB/s",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			bandwidth:          clientutil.Size{Limit: 2048},
+			concurrentOption: &config.ConcurrentOption{
+				GoroutineCount: 8,
+				ThresholdSize: clientutil.Size{
+					Limit: 1024 * 1024,
+				},
+				ThresholdSpeed: 8192,
+			},
+		},
+		{
+			name:               "multiple pieces with content length, concurrent download with 16 goroutines, download bandwidth 2KB/s",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			bandwidth:          clientutil.Size{Limit: 2048},
+			concurrentOption: &config.ConcurrentOption{
+				GoroutineCount: 16,
+				ThresholdSize: clientutil.Size{
+					Limit: 1024 * 1024,
+				},
+				ThresholdSpeed: 8192,
+			},
+		},
+		{
+			name:               "multiple pieces with content length, single-thread download with download bandwidth 4KB/s",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			bandwidth:          clientutil.Size{Limit: 4096},
+		},
+		{
+			name:               "multiple pieces with content length, concurrent download with download bandwidth 4KB/s",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			bandwidth:          clientutil.Size{Limit: 4096},
+			concurrentOption: &config.ConcurrentOption{
+				GoroutineCount: 4,
+				ThresholdSize: clientutil.Size{
+					Limit: 1024 * 1024,
+				},
+				ThresholdSpeed: 8192,
+			},
+		},
+		{
+			name:               "multiple pieces with content length, single-thread download with download bandwidth 8KB/s",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			bandwidth:          clientutil.Size{Limit: 8192},
+		},
+		{
+			name:               "multiple pieces with content length, concurrent download with download bandwidth 8KB/s",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			bandwidth:          clientutil.Size{Limit: 8192},
+			concurrentOption: &config.ConcurrentOption{
+				GoroutineCount: 4,
+				ThresholdSize: clientutil.Size{
+					Limit: 1024 * 1024,
+				},
+				ThresholdSpeed: 8192,
+			},
+		},
+		{
+			name:               "multiple pieces with content length, single-thread download with download bandwidth 16KB/s",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			bandwidth:          clientutil.Size{Limit: 16384},
+		},
+		{
+			name:               "multiple pieces with content length, concurrent download with download bandwidth 16KB/s",
+			pieceSize:          2048,
+			checkDigest:        false,
+			withContentLength:  true,
+			recordDownloadTime: true,
+			bandwidth:          clientutil.Size{Limit: 16384},
+			concurrentOption: &config.ConcurrentOption{
+				GoroutineCount: 4,
+				ThresholdSize: clientutil.Size{
+					Limit: 1024 * 1024,
+				},
+				ThresholdSpeed: 8192,
+			},
+		},
 		{
 			name:              "multiple pieces without content length, check digest",
 			pieceSize:         1024,
@@ -214,7 +396,7 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 					buf = bytes.NewBuffer(testBytes)
 					l = len(testBytes)
 				} else {
-					parsedRange, err := util.ParseRange(rg, int64(len(testBytes)))
+					parsedRange, err := clientutil.ParseRange(rg, int64(len(testBytes)))
 					assert.Nil(err)
 					w.Header().Set(headers.ContentRange,
 						fmt.Sprintf("bytes %d-%d/%d",
@@ -226,9 +408,32 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 					buf = bytes.NewBuffer(testBytes[parsedRange[0].Start : parsedRange[0].Start+parsedRange[0].Length])
 					l = int(parsedRange[0].Length)
 				}
-				n, err := io.Copy(w, buf)
-				assert.Nil(err)
-				assert.Equal(int64(l), n)
+				if tc.bandwidth.Limit > 0 {
+					ctx := context.Background()
+					limiter := rate.NewLimiter(tc.bandwidth.Limit, int(tc.bandwidth.Limit))
+					limiter.AllowN(time.Now(), int(tc.bandwidth.Limit))
+					maxPieceNum := util.ComputePieceCount(int64(l), uint32(tc.bandwidth.Limit))
+					for pieceNum := int32(0); pieceNum < maxPieceNum; pieceNum++ {
+						size := int(tc.bandwidth.Limit)
+						offset := uint64(pieceNum) * uint64(size)
+						// calculate piece size for last piece
+						if int64(offset)+int64(size) > int64(l) {
+							size = int(int64(l) - int64(offset))
+						}
+						err := limiter.WaitN(ctx, size)
+						assert.Nil(err)
+						n, err := io.CopyN(w, buf, int64(size))
+						if err != nil {
+							// broken pipe due to auto switch to concurrent download, which will close conn
+							return
+						}
+						assert.Equal(int64(size), n)
+					}
+				} else {
+					n, err := io.Copy(w, buf)
+					assert.Nil(err)
+					assert.Equal(int64(l), n)
+				}
 			}))
 			defer ts.Close()
 
@@ -249,9 +454,17 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 			if tc.checkDigest {
 				request.UrlMeta.Digest = digest
 			}
-
+			var start time.Time
+			if tc.recordDownloadTime {
+				start = time.Now()
+			}
 			err = pm.DownloadSource(context.Background(), mockPeerTask, request, nil)
 			assert.Nil(err)
+			if tc.recordDownloadTime {
+				elapsed := time.Since(start)
+				log := mockPeerTask.Log()
+				log.Infof("download took %s", elapsed)
+			}
 
 			err = storageManager.Store(context.Background(),
 				&storage.StoreRequest{
diff --git a/test/testdata/charts/config.yaml b/test/testdata/charts/config.yaml
index cb2920e8..89ab857a 100644
--- a/test/testdata/charts/config.yaml
+++ b/test/testdata/charts/config.yaml
@@ -78,6 +78,7 @@ dfdaemon:
       prefetch: true
       concurrent:
         thresholdSize: 10M
+        thresholdSpeed: 2M
         goroutineCount: 4
     scheduler:
       disableAutoBackSource: true
diff --git a/test/testdata/k8s/proxy.yaml b/test/testdata/k8s/proxy.yaml
index 8cc7aa7f..15481307 100644
--- a/test/testdata/k8s/proxy.yaml
+++ b/test/testdata/k8s/proxy.yaml
@@ -43,6 +43,7 @@ data:
       totalRateLimit: 200Mi
       concurrent:
         thresholdSize: 10M
+        thresholdSpeed: 2M
         goroutineCount: 4
     upload:
       rateLimit: 100Mi