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

chore(deps): bump github.com/schollz/progressbar/v3 from 3.8.6 to 3.8.7
Bumps [github.com/schollz/progressbar/v3](https://github.com/schollz/progressbar) from 3.8.6 to 3.8.7.
<details>
<summary>Release notes</summary>
<p><em>Sourced from <a href="https://github.com/schollz/progressbar/releases">github.com/schollz/progressbar/v3's releases</a>.</em></p>
<blockquote>
<h2>v3.8.7</h2>
<ul>
<li>fix bug in printing newline</li>
</ul>
</blockquote>
</details>
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/schollz/progressbar/commit/4ded60010d9c894a047be953b111dbef49926365"><code>4ded600</code></a> Merge pull request <a href="https://github-redirect.dependabot.com/schollz/progressbar/issues/123">#123</a> from bodgit/revert-113</li>
<li><a href="https://github.com/schollz/progressbar/commit/660a82725884488782b6ccf0d872272d3a43b941"><code>660a827</code></a> Revert &quot;fix: Newline printing improperly in some situations&quot;</li>
<li><a href="https://github.com/schollz/progressbar/commit/a982b57ac70f3b4484bbbfdf45b0be436335e3e3"><code>a982b57</code></a> Test nothing gets written to os.Stdout</li>
<li>See full diff in <a href="https://github.com/schollz/progressbar/compare/v3.8.6...v3.8.7">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=github.com/schollz/progressbar/v3&package-manager=go_modules&previous-version=3.8.6&new-version=3.8.7)](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/1502?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1502](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1502?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (35f7b49) 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 **decrease** coverage by `0.24%`.
> The diff coverage is `n/a`.

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

```diff
@@            Coverage Diff             @@
##             main    #1502      +/-   ##
==========================================
- Coverage   43.39%   43.15%   -0.25%     
==========================================
  Files         116      116              
  Lines       10690    10690              
==========================================
- Hits         4639     4613      -26     
- Misses       5567     5594      +27     
+ Partials      484      483       -1     
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `43.15% <ø> (-0.25%)` | :arrow_down: |

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/1502?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [client/daemon/peer/peertask\_piecetask\_poller.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1502/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19wb2xsZXIuZ28=) | `30.43% <0.00%> (-16.67%)` | :arrow_down: |
| [...ent/daemon/peer/peertask\_piecetask\_synchronizer.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1502/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: |


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..b5696b23 100644
--- a/go.mod
+++ b/go.mod
@@ -48,7 +48,7 @@ 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
diff --git a/go.sum b/go.sum
index 0f25c409..8846ccf7 100644
--- a/go.sum
+++ b/go.sum
@@ -919,8 +919,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=


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