Repository: dragonflyoss/Dragonfly2
Pull request #1426 (2022-07-01 06:46:22)

feat: concurrent multiple pieces back source
Signed-off-by: Jim Ma <majinjing3@gmail.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description

In some network environments, one tcp connection is limited to very low transfer speed like 2M/s.
Concurrent downloading is very useful for this scenario.

This PR make concurrent multiple pieces download configurable and disabled by default.

We need update documents after merged this PR.

## 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)
- [x] 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.
- [x] I have read the **CONTRIBUTING** document.
- [x] I have added tests to cover my changes.



Patch:
diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml
index 63eee23c..711df87f 100644
--- a/.github/workflows/codeql-analysis.yml
+++ b/.github/workflows/codeql-analysis.yml
@@ -25,12 +25,12 @@ jobs:
         uses: actions/checkout@v2
 
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@v1
+        uses: github/codeql-action/init@v2
         with:
           languages: ${{ matrix.language }}
 
       - name: Autobuild
-        uses: github/codeql-action/autobuild@v1
+        uses: github/codeql-action/autobuild@v2
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@v1
+        uses: github/codeql-action/analyze@v2
diff --git a/client/config/peerhost.go b/client/config/peerhost.go
index d80b3398..2f746371 100644
--- a/client/config/peerhost.go
+++ b/client/config/peerhost.go
@@ -224,17 +224,18 @@ type HostOption struct {
 }
 
 type DownloadOption struct {
-	DefaultPattern       string           `mapstructure:"defaultPattern" yaml:"defaultPattern"`
-	TotalRateLimit       util.RateLimit   `mapstructure:"totalRateLimit" yaml:"totalRateLimit"`
-	PerPeerRateLimit     util.RateLimit   `mapstructure:"perPeerRateLimit" yaml:"perPeerRateLimit"`
-	PieceDownloadTimeout time.Duration    `mapstructure:"pieceDownloadTimeout" yaml:"pieceDownloadTimeout"`
-	DownloadGRPC         ListenOption     `mapstructure:"downloadGRPC" yaml:"downloadGRPC"`
-	PeerGRPC             ListenOption     `mapstructure:"peerGRPC" yaml:"peerGRPC"`
-	CalculateDigest      bool             `mapstructure:"calculateDigest" yaml:"calculateDigest"`
-	TransportOption      *TransportOption `mapstructure:"transportOption" yaml:"transportOption"`
-	GetPiecesMaxRetry    int              `mapstructure:"getPiecesMaxRetry" yaml:"getPiecesMaxRetry"`
-	Prefetch             bool             `mapstructure:"prefetch" yaml:"prefetch"`
-	WatchdogTimeout      time.Duration    `mapstructure:"watchdogTimeout" yaml:"watchdogTimeout"`
+	DefaultPattern       string            `mapstructure:"defaultPattern" yaml:"defaultPattern"`
+	TotalRateLimit       util.RateLimit    `mapstructure:"totalRateLimit" yaml:"totalRateLimit"`
+	PerPeerRateLimit     util.RateLimit    `mapstructure:"perPeerRateLimit" yaml:"perPeerRateLimit"`
+	PieceDownloadTimeout time.Duration     `mapstructure:"pieceDownloadTimeout" yaml:"pieceDownloadTimeout"`
+	DownloadGRPC         ListenOption      `mapstructure:"downloadGRPC" yaml:"downloadGRPC"`
+	PeerGRPC             ListenOption      `mapstructure:"peerGRPC" yaml:"peerGRPC"`
+	CalculateDigest      bool              `mapstructure:"calculateDigest" yaml:"calculateDigest"`
+	Transport            *TransportOption  `mapstructure:"transportOption" yaml:"transportOption"`
+	GetPiecesMaxRetry    int               `mapstructure:"getPiecesMaxRetry" yaml:"getPiecesMaxRetry"`
+	Prefetch             bool              `mapstructure:"prefetch" yaml:"prefetch"`
+	WatchdogTimeout      time.Duration     `mapstructure:"watchdogTimeout" yaml:"watchdogTimeout"`
+	Concurrent           *ConcurrentOption `mapstructure:"concurrent" yaml:"concurrent"`
 }
 
 type TransportOption struct {
@@ -247,6 +248,19 @@ type TransportOption struct {
 	ExpectContinueTimeout time.Duration `mapstructure:"expectContinueTimeout" yaml:"expectContinueTimeout"`
 }
 
+type ConcurrentOption struct {
+	// ThresholdSize indicates the threshold to download pieces concurrently
+	ThresholdSize util.Size `mapstructure:"thresholdSize" yaml:"thresholdSize"`
+	// GoroutineCount indicates the concurrent goroutine count for every task
+	GoroutineCount int `mapstructure:"goroutineCount" yaml:"goroutineCount"`
+	// InitBackoff second for every piece failed, default: 0.5
+	InitBackoff float64 `mapstructure:"initBackoff" yaml:"initBackoff"`
+	// MaxBackoff second for every piece failed, default: 3
+	MaxBackoff float64 `mapstructure:"maxBackoff" yaml:"maxBackoff"`
+	// MaxAttempts for every piece failed,default: 3
+	MaxAttempts int `mapstructure:"maxAttempts" yaml:"MaxAttempts"`
+}
+
 type ProxyOption struct {
 	// WARNING: when add more option, please update ProxyOption.unmarshal function
 	ListenOption    `mapstructure:",squash" yaml:",inline"`
diff --git a/client/daemon/daemon.go b/client/daemon/daemon.go
index 19bf4287..36b4449a 100644
--- a/client/daemon/daemon.go
+++ b/client/daemon/daemon.go
@@ -180,7 +180,8 @@ func New(opt *config.DaemonOption, d dfpath.Dfpath) (Daemon, error) {
 	pieceManager, err := peer.NewPieceManager(
 		opt.Download.PieceDownloadTimeout,
 		peer.WithLimiter(rate.NewLimiter(opt.Download.TotalRateLimit.Limit, int(opt.Download.TotalRateLimit.Limit))),
-		peer.WithCalculateDigest(opt.Download.CalculateDigest), peer.WithTransportOption(opt.Download.TransportOption),
+		peer.WithCalculateDigest(opt.Download.CalculateDigest), peer.WithTransportOption(opt.Download.Transport),
+		peer.WithConcurrentOption(opt.Download.Concurrent),
 	)
 	if err != nil {
 		return nil, err
diff --git a/client/daemon/peer/peertask_conductor.go b/client/daemon/peer/peertask_conductor.go
index 91a0b213..4c25d81b 100644
--- a/client/daemon/peer/peertask_conductor.go
+++ b/client/daemon/peer/peertask_conductor.go
@@ -486,7 +486,7 @@ func (pt *peerTaskConductor) backSource() {
 
 	ctx, span := tracer.Start(pt.ctx, config.SpanBackSource)
 	pt.SetContentLength(-1)
-	err := pt.pieceManager.DownloadSource(ctx, pt, pt.request)
+	err := pt.pieceManager.DownloadSource(ctx, pt, pt.request, pt.rg)
 	if err != nil {
 		pt.Errorf("download from source error: %s", err)
 		span.SetAttributes(config.AttributePeerTaskSuccess.Bool(false))
diff --git a/client/daemon/peer/peertask_piecetask_poller.go b/client/daemon/peer/peertask_piecetask_poller.go
index 4eb66821..80213c1b 100644
--- a/client/daemon/peer/peertask_piecetask_poller.go
+++ b/client/daemon/peer/peertask_piecetask_poller.go
@@ -160,7 +160,7 @@ func (poller *pieceTaskPoller) getPieceTasksByPeer(
 		count             int
 		ptc               = poller.peerTaskConductor
 	)
-	p, _, err := retry.Run(ptc.ctx, func() (any, bool, error) {
+	p, _, err := retry.Run(ptc.ctx, 0.05, 0.2, 40, func() (any, bool, error) {
 		// GetPieceTasks must be fast, so short time out is okay
 		ctx, cancel := context.WithTimeout(ptc.ctx, 4*time.Second)
 		defer cancel()
@@ -235,7 +235,7 @@ func (poller *pieceTaskPoller) getPieceTasksByPeer(
 			trace.WithAttributes(config.AttributeGetPieceRetry.Int(count)))
 		ptc.Infof("peer %s returns success but with empty pieces, retry later", peer.PeerId)
 		return nil, false, dferrors.ErrEmptyValue
-	}, 0.05, 0.2, 40, nil)
+	})
 	if peerPacketChanged {
 		return nil, errPeerPacketChanged
 	}
diff --git a/client/daemon/peer/piece_manager.go b/client/daemon/peer/piece_manager.go
index 12d9e7a5..177f6ab8 100644
--- a/client/daemon/peer/piece_manager.go
+++ b/client/daemon/peer/piece_manager.go
@@ -25,8 +25,11 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"sync"
 	"time"
 
+	"github.com/go-http-utils/headers"
+	"go.uber.org/atomic"
 	"golang.org/x/time/rate"
 
 	"d7y.io/dragonfly/v2/client/config"
@@ -35,6 +38,7 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/internal/util"
 	"d7y.io/dragonfly/v2/pkg/digest"
+	"d7y.io/dragonfly/v2/pkg/retry"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/dfdaemon"
 	"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
@@ -42,7 +46,7 @@ import (
 )
 
 type PieceManager interface {
-	DownloadSource(ctx context.Context, pt Task, request *scheduler.PeerTaskRequest) error
+	DownloadSource(ctx context.Context, pt Task, request *scheduler.PeerTaskRequest, parsedRange *clientutil.Range) error
 	DownloadPiece(ctx context.Context, request *DownloadPieceRequest) (*DownloadPieceResult, error)
 	ImportFile(ctx context.Context, ptm storage.PeerTaskMetadata, tsd storage.TaskStorageDriver, req *dfdaemon.ImportTaskRequest) error
 	Import(ctx context.Context, ptm storage.PeerTaskMetadata, tsd storage.TaskStorageDriver, contentLength int64, reader io.Reader) error
@@ -52,8 +56,8 @@ type pieceManager struct {
 	*rate.Limiter
 	pieceDownloader  PieceDownloader
 	computePieceSize func(contentLength int64) uint32
-
-	calculateDigest bool
+	calculateDigest  bool
+	concurrentOption *config.ConcurrentOption
 }
 
 func NewPieceManager(pieceDownloadTimeout time.Duration, opts ...func(*pieceManager)) (PieceManager, error) {
@@ -118,6 +122,27 @@ func WithTransportOption(opt *config.TransportOption) func(*pieceManager) {
 	}
 }
 
+func WithConcurrentOption(opt *config.ConcurrentOption) func(*pieceManager) {
+	return func(manager *pieceManager) {
+		manager.concurrentOption = opt
+		if manager.concurrentOption == nil {
+			return
+		}
+		if manager.concurrentOption.GoroutineCount <= 0 {
+			manager.concurrentOption.GoroutineCount = 4
+		}
+		if manager.concurrentOption.InitBackoff <= 0 {
+			manager.concurrentOption.InitBackoff = 0.5
+		}
+		if manager.concurrentOption.MaxBackoff <= 0 {
+			manager.concurrentOption.MaxBackoff = 3
+		}
+		if manager.concurrentOption.MaxAttempts <= 0 {
+			manager.concurrentOption.MaxAttempts = 3
+		}
+	}
+}
+
 func (pm *pieceManager) DownloadPiece(ctx context.Context, request *DownloadPieceRequest) (*DownloadPieceResult, error) {
 	var result = &DownloadPieceResult{
 		Size:       -1,
@@ -243,27 +268,76 @@ func (pm *pieceManager) processPieceFromSource(pt Task,
 	return
 }
 
-func (pm *pieceManager) DownloadSource(ctx context.Context, pt Task, request *scheduler.PeerTaskRequest) error {
-	if request.UrlMeta == nil {
-		request.UrlMeta = &base.UrlMeta{
+func (pm *pieceManager) DownloadSource(ctx context.Context, pt Task, peerTaskRequest *scheduler.PeerTaskRequest, parsedRange *clientutil.Range) error {
+	if peerTaskRequest.UrlMeta == nil {
+		peerTaskRequest.UrlMeta = &base.UrlMeta{
 			Header: map[string]string{},
 		}
-	} else if request.UrlMeta.Header == nil {
-		request.UrlMeta.Header = map[string]string{}
+	} else if peerTaskRequest.UrlMeta.Header == nil {
+		peerTaskRequest.UrlMeta.Header = map[string]string{}
 	}
-	if request.UrlMeta.Range != "" {
+	if peerTaskRequest.UrlMeta.Range != "" {
 		// in http source package, adapter will update the real range, we inject "X-Dragonfly-Range" here
-		request.UrlMeta.Header[source.Range] = request.UrlMeta.Range
+		peerTaskRequest.UrlMeta.Header[source.Range] = peerTaskRequest.UrlMeta.Range
 	}
+
 	log := pt.Log()
 	log.Infof("start to download from source")
 
-	// 1. download pieces from source
-	downloadRequest, err := source.NewRequestWithContext(ctx, request.Url, request.UrlMeta.Header)
+	backSourceRequest, err := source.NewRequestWithContext(ctx, peerTaskRequest.Url, peerTaskRequest.UrlMeta.Header)
 	if err != nil {
 		return err
 	}
-	response, err := source.Download(downloadRequest)
+
+	if pm.concurrentOption != nil {
+		// check metadata
+		// 1. support range request
+		// 2. target content length is greater than concurrentOption.ThresholdSize
+		metadata, err := source.GetMetadata(backSourceRequest)
+		if err == nil && metadata.SupportRange && metadata.TotalContentLength > -1 {
+			var (
+				targetContentLength int64
+			)
+			if parsedRange != nil {
+				if parsedRange.Length > -1 {
+					targetContentLength = parsedRange.Length
+				} else {
+					// for range like "Range: bytes=1-", complete the length in range
+					targetContentLength = metadata.TotalContentLength - parsedRange.Start
+					// update length from metadata
+					parsedRange.Length = targetContentLength
+				}
+			} else {
+				// for non-ranged request, add a dummy range
+				parsedRange = &clientutil.Range{
+					Start:  0,
+					Length: metadata.TotalContentLength,
+				}
+				targetContentLength = parsedRange.Length
+			}
+
+			if targetContentLength > int64(pm.concurrentOption.ThresholdSize.Limit) {
+				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
+				}
+				// use concurrent piece download mode
+				return pm.concurrentDownloadSource(ctx, pt, peerTaskRequest, parsedRange, metadata)
+			}
+		}
+	}
+
+	// 1. download pieces from source
+	response, err := source.Download(backSourceRequest)
 	// TODO update expire info
 	if err != nil {
 		return err
@@ -274,7 +348,7 @@ func (pm *pieceManager) DownloadSource(ctx context.Context, pt Task, request *sc
 	}
 	contentLength := response.ContentLength
 	if contentLength < 0 {
-		log.Warnf("can not get content length for %s", request.Url)
+		log.Warnf("can not get content length for %s", peerTaskRequest.Url)
 	} else {
 		log.Debugf("back source content length: %d", contentLength)
 		err = pt.GetStorage().UpdateTask(ctx,
@@ -295,7 +369,7 @@ func (pm *pieceManager) DownloadSource(ctx context.Context, pt Task, request *sc
 
 	// calc total
 	if pm.calculateDigest {
-		reader, err = digest.NewReader(response.Body, digest.WithDigest(request.UrlMeta.Digest), digest.WithLogger(pt.Log()))
+		reader, err = digest.NewReader(response.Body, digest.WithDigest(peerTaskRequest.UrlMeta.Digest), digest.WithLogger(pt.Log()))
 		if err != nil {
 			log.Errorf("init digest reader error: %s", err.Error())
 			return err
@@ -315,7 +389,7 @@ func (pm *pieceManager) DownloadSource(ctx context.Context, pt Task, request *sc
 
 func (pm *pieceManager) downloadKnownLengthSource(pt Task, contentLength int64, pieceSize uint32, reader io.Reader) error {
 	log := pt.Log()
-	maxPieceNum := util.ComputePieceNum(contentLength, pieceSize)
+	maxPieceNum := util.ComputePieceCount(contentLength, pieceSize)
 	pt.SetContentLength(contentLength)
 	pt.SetTotalPieces(maxPieceNum)
 
@@ -498,7 +572,7 @@ func (pm *pieceManager) ImportFile(ctx context.Context, ptm storage.PeerTaskMeta
 	}
 	contentLength := stat.Size()
 	pieceSize := pm.computePieceSize(contentLength)
-	maxPieceNum := util.ComputePieceNum(contentLength, pieceSize)
+	maxPieceNum := util.ComputePieceCount(contentLength, pieceSize)
 
 	file, err := os.Open(req.Path)
 	if err != nil {
@@ -563,7 +637,7 @@ func (pm *pieceManager) ImportFile(ctx context.Context, ptm storage.PeerTaskMeta
 func (pm *pieceManager) Import(ctx context.Context, ptm storage.PeerTaskMetadata, tsd storage.TaskStorageDriver, contentLength int64, reader io.Reader) error {
 	log := logger.WithTaskAndPeerID(ptm.TaskID, ptm.PeerID)
 	pieceSize := pm.computePieceSize(contentLength)
-	maxPieceNum := util.ComputePieceNum(contentLength, pieceSize)
+	maxPieceNum := util.ComputePieceCount(contentLength, pieceSize)
 
 	for pieceNum := int32(0); pieceNum < maxPieceNum; pieceNum++ {
 		size := pieceSize
@@ -616,3 +690,161 @@ 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 {
+	// parsedRange is always exist
+	pieceSize := pm.computePieceSize(parsedRange.Length)
+	pieceCount := util.ComputePieceCount(parsedRange.Length, pieceSize)
+	var downloadError atomic.Value
+
+	log := pt.Log()
+	pt.SetContentLength(parsedRange.Length)
+	pt.SetTotalPieces(pieceCount)
+
+	con := pm.concurrentOption.GoroutineCount
+	if int(pieceCount) < con {
+		con = int(pieceCount)
+	}
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	var pieceCh = make(chan int32, con)
+
+	wg := sync.WaitGroup{}
+	wg.Add(int(pieceCount))
+
+	downloadedPieceCount := atomic.NewInt32(0)
+
+	for i := 0; i < con; i++ {
+		go func(i int) {
+			for {
+				select {
+				case <-ctx.Done():
+				case num, ok := <-pieceCh:
+					if !ok {
+						log.Debugf("concurrent worker %d exit", i)
+						return
+					}
+					log.Infof("concurrent worker %d start to download piece %d", i, num)
+					_, _, retryErr := retry.Run(ctx,
+						pm.concurrentOption.InitBackoff,
+						pm.concurrentOption.MaxBackoff,
+						pm.concurrentOption.MaxAttempts,
+						func() (data any, cancel bool, err error) {
+							err = pm.downloadPieceFromSource(ctx, pt, log,
+								peerTaskRequest, pieceSize, num,
+								parsedRange, pieceCount, downloadedPieceCount)
+							return nil, err == context.Canceled, err
+						})
+					if retryErr != nil {
+						// download piece error after many retry, cancel task
+						cancel()
+						downloadError.Store(&backSourceError{err: retryErr})
+						log.Infof("concurrent worker %d failed to download piece %d after %d retries, last error: %s",
+							i, num, pm.concurrentOption.MaxAttempts, retryErr.Error())
+					}
+					wg.Done()
+				}
+			}
+		}(i)
+	}
+
+	for i := int32(0); i < pieceCount; i++ {
+		select {
+		case <-ctx.Done():
+			log.Warnf("context cancelled")
+			if downloadError.Load() != nil {
+				return downloadError.Load().(*backSourceError).err
+			}
+			return ctx.Err()
+		case pieceCh <- i:
+		}
+	}
+
+	// wait all pieces processed
+	// the downloadError will be updated if there is any errors before wg.Wait() returned
+	wg.Wait()
+
+	// let all working goroutines exit
+	close(pieceCh)
+
+	// check error
+	if downloadError.Load() != nil {
+		return downloadError.Load().(*pieceDownloadError).err
+	}
+
+	return nil
+}
+
+func (pm *pieceManager) downloadPieceFromSource(ctx context.Context,
+	pt Task, log *logger.SugaredLoggerOnWith,
+	peerTaskRequest *scheduler.PeerTaskRequest,
+	pieceSize uint32, num int32,
+	parsedRange *clientutil.Range,
+	pieceCount int32,
+	downloadedPieceCount *atomic.Int32) error {
+	backSourceRequest, err := source.NewRequestWithContext(ctx, peerTaskRequest.Url, peerTaskRequest.UrlMeta.Header)
+	if err != nil {
+		log.Errorf("build piece %d back source request error: %s", num, err)
+		return err
+	}
+	size := pieceSize
+	offset := uint64(num) * uint64(pieceSize)
+	// calculate piece size for last piece
+	if int64(offset)+int64(size) > parsedRange.Length {
+		size = uint32(parsedRange.Length - int64(offset))
+	}
+
+	// offset is the position for current peer task, if this peer task already has range
+	// we need add the start to the offset when download from source
+	rg := fmt.Sprintf("bytes=%d-%d", offset+uint64(parsedRange.Start), offset+uint64(parsedRange.Start)+uint64(size)-1)
+	backSourceRequest.Header.Set(headers.Range, rg)
+
+	response, err := source.Download(backSourceRequest)
+	if err != nil {
+		log.Errorf("piece %d back source response error: %s", num, err)
+		return err
+	}
+	defer response.Body.Close()
+
+	err = response.Validate()
+	if err != nil {
+		log.Errorf("piece %d back source response validate error: %s", num, err)
+		return err
+	}
+
+	log.Debugf("piece %d back source response ok", num)
+	result, md5, err := pm.processPieceFromSource(
+		pt, response.Body, parsedRange.Length, num, offset, size,
+		func(int64) (int32, int64, bool) {
+			downloadedPieceCount.Inc()
+			return pieceCount, parsedRange.Length, downloadedPieceCount.Load() == pieceCount
+		})
+	request := &DownloadPieceRequest{
+		TaskID: pt.GetTaskID(),
+		PeerID: pt.GetPeerID(),
+		piece: &base.PieceInfo{
+			PieceNum:    num,
+			RangeStart:  offset,
+			RangeSize:   uint32(result.Size),
+			PieceMd5:    md5,
+			PieceOffset: offset,
+			PieceStyle:  0,
+		},
+	}
+	if err != nil {
+		log.Errorf("download piece %d error: %s", num, err)
+		pt.ReportPieceResult(request, result, detectBackSourceError(err))
+		return err
+	}
+
+	if result.Size != int64(size) {
+		log.Errorf("download piece %d size not match, desired: %d, actual: %d", num, size, result.Size)
+		pt.ReportPieceResult(request, result, detectBackSourceError(err))
+		return storage.ErrShortRead
+	}
+
+	pt.ReportPieceResult(request, result, nil)
+	pt.PublishPieceInfo(num, uint32(result.Size))
+	return nil
+}
diff --git a/client/util/types.go b/client/util/types.go
index 53463d68..fe581550 100644
--- a/client/util/types.go
+++ b/client/util/types.go
@@ -36,6 +36,8 @@ type RateLimit struct {
 	rate.Limit
 }
 
+type Size = RateLimit
+
 func (r *RateLimit) UnmarshalJSON(b []byte) error {
 	return r.unmarshal(json.Unmarshal, b)
 }
diff --git a/go.mod b/go.mod
index 6e4c3de3..d8b8b2ab 100644
--- a/go.mod
+++ b/go.mod
@@ -45,6 +45,7 @@ require (
 	github.com/onsi/gomega v1.19.0
 	github.com/phayes/freeport v0.0.0-20220201140144-74d24b5ae9f5
 	github.com/prometheus/client_golang v1.12.2
+	github.com/prometheus/common v0.35.0
 	github.com/schollz/progressbar/v3 v3.8.6
 	github.com/serialx/hashring v0.0.0-20200727003509-22c0c7ab6b1b
 	github.com/shirou/gopsutil/v3 v3.22.5
@@ -167,7 +168,6 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20220216144756-c35f1ee13d7c // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
-	github.com/prometheus/common v0.35.0 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/robfig/cron/v3 v3.0.1 // indirect
diff --git a/internal/util/util.go b/internal/util/util.go
index e3394cad..ab71547e 100644
--- a/internal/util/util.go
+++ b/internal/util/util.go
@@ -43,7 +43,7 @@ func ComputePieceSize(length int64) uint32 {
 	return uint32(mpSize)
 }
 
-// ComputePieceNum returns piece count with given length and pieceSize
-func ComputePieceNum(length int64, pieceSize uint32) int32 {
+// ComputePieceCount returns piece count with given length and pieceSize
+func ComputePieceCount(length int64, pieceSize uint32) int32 {
 	return int32(math.Ceil(float64(length) / float64(pieceSize)))
 }
diff --git a/manager/service/job.go b/manager/service/job.go
index 98a5b1a6..1c0ee7f1 100644
--- a/manager/service/job.go
+++ b/manager/service/job.go
@@ -100,7 +100,7 @@ func (s *service) CreatePreheatJob(ctx context.Context, json types.CreatePreheat
 func (s *service) pollingJob(ctx context.Context, id uint, taskID string) {
 	var job model.Job
 
-	if _, _, err := retry.Run(ctx, func() (any, bool, error) {
+	if _, _, err := retry.Run(ctx, 5, 10, 120, func() (any, bool, error) {
 		groupJob, err := s.job.GetGroupJobState(taskID)
 		if err != nil {
 			logger.Errorf("polling job %d and task %s failed: %v", id, taskID, err)
@@ -124,7 +124,7 @@ func (s *service) pollingJob(ctx context.Context, id uint, taskID string) {
 		default:
 			return nil, false, fmt.Errorf("polling job %d and task %s status is %s", id, taskID, job.State)
 		}
-	}, 5, 10, 120, nil); err != nil {
+	}); err != nil {
 		logger.Errorf("polling job %d and task %s failed %s", id, taskID, err)
 	}
 
diff --git a/pkg/math/rand.go b/pkg/math/rand.go
index 58c3147e..00f3a8c8 100644
--- a/pkg/math/rand.go
+++ b/pkg/math/rand.go
@@ -50,7 +50,7 @@ func RandString(n int) string {
 	return string(b)
 }
 
-func RandBackoff(initBackoff float64, maxBackoff float64, base float64, exp int) time.Duration {
+func RandBackoffSeconds(initBackoff float64, maxBackoff float64, base float64, exp int) time.Duration {
 	m := math.Max(initBackoff, rand.Float64()*math.Min(math.Pow(base, float64(exp))*initBackoff, maxBackoff))
 	return time.Duration(m * float64(time.Second))
 }
diff --git a/pkg/retry/retry.go b/pkg/retry/retry.go
index 3a267293..560138ac 100644
--- a/pkg/retry/retry.go
+++ b/pkg/retry/retry.go
@@ -24,18 +24,18 @@ import (
 )
 
 func Run(ctx context.Context,
-	f func() (data any, cancel bool, err error),
 	initBackoff float64,
 	maxBackoff float64,
 	maxAttempts int,
-	cause error) (any, bool, error) {
+	f func() (data any, cancel bool, err error)) (any, bool, error) {
 	var (
 		res    any
 		cancel bool
+		cause  error
 	)
 	for i := 0; i < maxAttempts; i++ {
 		if i > 0 {
-			time.Sleep(math.RandBackoff(initBackoff, maxBackoff, 2.0, i))
+			time.Sleep(math.RandBackoffSeconds(initBackoff, maxBackoff, 2.0, i))
 		}
 
 		res, cancel, cause = f()
diff --git a/pkg/rpc/client_util.go b/pkg/rpc/client_util.go
index 27c9554a..484f2861 100644
--- a/pkg/rpc/client_util.go
+++ b/pkg/rpc/client_util.go
@@ -223,7 +223,7 @@ func ExecuteWithRetry(f func() (any, error), initBackoff float64, maxBackoff flo
 			return res, cause
 		}
 		if i > 0 {
-			time.Sleep(math.RandBackoff(initBackoff, maxBackoff, 2.0, i))
+			time.Sleep(math.RandBackoffSeconds(initBackoff, maxBackoff, 2.0, i))
 		}
 
 		res, cause = f()
diff --git a/pkg/source/clients/httpprotocol/http_source_client.go b/pkg/source/clients/httpprotocol/http_source_client.go
index 74d1eee8..311f46a3 100644
--- a/pkg/source/clients/httpprotocol/http_source_client.go
+++ b/pkg/source/clients/httpprotocol/http_source_client.go
@@ -19,13 +19,17 @@ package httpprotocol
 import (
 	"crypto/tls"
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"net/url"
 	"os"
+	"regexp"
+	"strconv"
 	"time"
 
 	"github.com/go-http-utils/headers"
+	"github.com/pkg/errors"
 
 	"d7y.io/dragonfly/v2/pkg/source"
 )
@@ -37,8 +41,17 @@ const (
 	ProxyEnv = "D7Y_SOURCE_PROXY"
 )
 
-var _defaultHTTPClient *http.Client
-var _ source.ResourceClient = (*httpSourceClient)(nil)
+var (
+	_defaultHTTPClient *http.Client
+	_                  source.ResourceClient = (*httpSourceClient)(nil)
+
+	// Syntax:
+	//   Content-Range: <unit> <range-start>-<range-end>/<size> -> Done
+	//   Content-Range: <unit> <range-start>-<range-end>/* -> Done
+	//   Content-Range: <unit> */<size> -> TODO
+	contentRangeRegexp            = regexp.MustCompile(`bytes (?P<Start>\d+)-(?P<End>\d+)/(?P<Length>(\d*|\*))`)
+	contentRangeRegexpLengthIndex = contentRangeRegexp.SubexpIndex("Length")
+)
 
 func init() {
 	// TODO support customize source client
@@ -155,6 +168,52 @@ func (client *httpSourceClient) IsSupportRange(request *source.Request) (bool, e
 	return resp.StatusCode == http.StatusPartialContent, nil
 }
 
+func (client *httpSourceClient) GetMetadata(request *source.Request) (*source.Metadata, error) {
+	// clone a new request to avoid change original request
+	request = request.Clone(request.Context())
+	request.Header.Set(headers.Range, "bytes=0-0")
+
+	// use GET method to get metadata instead of HEAD method
+	// some object service like OSS, only sign url with GET method, so did not use HEAD method
+	resp, err := client.doRequest(http.MethodGet, request)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	var totalContentLength int64 = -1
+	cr := resp.Header.Get(headers.ContentRange)
+	if cr != "" {
+		matches := contentRangeRegexp.FindStringSubmatch(cr)
+		if len(matches) > contentRangeRegexpLengthIndex {
+			length := matches[contentRangeRegexpLengthIndex]
+			if length != "*" {
+				totalContentLength, err = strconv.ParseInt(length, 10, 64)
+				if err != nil {
+					return nil, errors.Wrapf(err, "convert length from string %q error", length)
+				}
+			}
+		}
+
+		// this server supports Range request
+		// we can discard the only one byte for reuse underlay connection
+		_, err = io.Copy(io.Discard, io.LimitReader(resp.Body, 1))
+		if err != io.EOF && err != nil {
+			return nil, err
+		}
+	}
+
+	hdr := source.Header{}
+	for k, v := range exportPassThroughHeader(resp.Header) {
+		hdr.Set(k, v)
+	}
+	return &source.Metadata{
+		Header:             hdr,
+		SupportRange:       resp.StatusCode == http.StatusPartialContent,
+		TotalContentLength: totalContentLength,
+	}, nil
+}
+
 func (client *httpSourceClient) IsExpired(request *source.Request, info *source.ExpireInfo) (bool, error) {
 	if info != nil {
 		if request.Header == nil {
diff --git a/pkg/source/metadata.go b/pkg/source/metadata.go
new file mode 100644
index 00000000..83fb1d5f
--- /dev/null
+++ b/pkg/source/metadata.go
@@ -0,0 +1,30 @@
+/*
+ *     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 source
+
+type Metadata struct {
+	Header Header
+	// SupportRange indicates source supports partial download, like Range in http request
+	SupportRange bool
+	// ContentLength indicates the current content length for the target request
+	// ContentLength int64
+	// TotalContentLength indicates the total content length for the target request
+	// eg, for http response header:
+	//      Content-Range: bytes 0-9/2443
+	// 2443 is the TotalContentLength, 10 is the ContentLength
+	TotalContentLength int64
+}
diff --git a/pkg/source/mocks/mock_source_client.go b/pkg/source/mocks/mock_source_client.go
index cd840791..6cddad6c 100644
--- a/pkg/source/mocks/mock_source_client.go
+++ b/pkg/source/mocks/mock_source_client.go
@@ -110,6 +110,44 @@ func (mr *MockResourceClientMockRecorder) IsSupportRange(request interface{}) *g
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IsSupportRange", reflect.TypeOf((*MockResourceClient)(nil).IsSupportRange), request)
 }
 
+// MockResourceMetadataGetter is a mock of ResourceMetadataGetter interface.
+type MockResourceMetadataGetter struct {
+	ctrl     *gomock.Controller
+	recorder *MockResourceMetadataGetterMockRecorder
+}
+
+// MockResourceMetadataGetterMockRecorder is the mock recorder for MockResourceMetadataGetter.
+type MockResourceMetadataGetterMockRecorder struct {
+	mock *MockResourceMetadataGetter
+}
+
+// NewMockResourceMetadataGetter creates a new mock instance.
+func NewMockResourceMetadataGetter(ctrl *gomock.Controller) *MockResourceMetadataGetter {
+	mock := &MockResourceMetadataGetter{ctrl: ctrl}
+	mock.recorder = &MockResourceMetadataGetterMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockResourceMetadataGetter) EXPECT() *MockResourceMetadataGetterMockRecorder {
+	return m.recorder
+}
+
+// GetMetadata mocks base method.
+func (m *MockResourceMetadataGetter) GetMetadata(request *source.Request) (*source.Metadata, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetMetadata", request)
+	ret0, _ := ret[0].(*source.Metadata)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetMetadata indicates an expected call of GetMetadata.
+func (mr *MockResourceMetadataGetterMockRecorder) GetMetadata(request interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetMetadata", reflect.TypeOf((*MockResourceMetadataGetter)(nil).GetMetadata), request)
+}
+
 // MockResourceLister is a mock of ResourceLister interface.
 type MockResourceLister struct {
 	ctrl     *gomock.Controller
diff --git a/pkg/source/source_client.go b/pkg/source/source_client.go
index f3fd8935..5425ced9 100644
--- a/pkg/source/source_client.go
+++ b/pkg/source/source_client.go
@@ -32,7 +32,7 @@ import (
 )
 
 var (
-	// Clinet back-to-source download timeout
+	// client back-to-source timeout for metadata
 	contextTimeout = 1 * time.Minute
 )
 
@@ -45,6 +45,9 @@ var (
 
 	// ErrClientNotSupportList represents the source client not support list action
 	ErrClientNotSupportList = errors.New("source client not support list")
+
+	// ErrClientNotSupportGetMetadata represents the source client not support get metadata
+	ErrClientNotSupportGetMetadata = errors.New("source client not support get metadata")
 )
 
 // UnexpectedStatusCodeError is returned when a source responds with neither an error
@@ -114,6 +117,12 @@ type ResourceClient interface {
 	GetLastModified(request *Request) (int64, error)
 }
 
+// ResourceMetadataGetter defines the API interface to get metadata for special resource
+// The metadata will be used for concurrent multiple pieces downloading
+type ResourceMetadataGetter interface {
+	GetMetadata(request *Request) (*Metadata, error)
+}
+
 // ResourceLister defines the interface to list all downloadable resources in request url
 type ResourceLister interface {
 	List(request *Request) (urls []*url.URL, err error)
@@ -350,3 +359,15 @@ func List(request *Request) ([]*url.URL, error) {
 	}
 	return lister.List(request)
 }
+
+func GetMetadata(request *Request) (*Metadata, error) {
+	client, ok := _defaultManager.GetClient(request.URL.Scheme)
+	if !ok {
+		return nil, fmt.Errorf("scheme %s: %w", request.URL.Scheme, ErrNoClientFound)
+	}
+	getter, ok := client.(*clientWrapper).rc.(ResourceMetadataGetter)
+	if !ok {
+		return nil, fmt.Errorf("scheme %s: %w", request.URL.Scheme, ErrClientNotSupportGetMetadata)
+	}
+	return getter.GetMetadata(request)
+}


Test patch:
diff --git a/client/daemon/peer/piece_manager_test.go b/client/daemon/peer/piece_manager_test.go
index 75fc187a..8fe88bc6 100644
--- a/client/daemon/peer/piece_manager_test.go
+++ b/client/daemon/peer/piece_manager_test.go
@@ -29,6 +29,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-http-utils/headers"
 	"github.com/golang/mock/gomock"
 	testifyassert "github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -80,6 +81,7 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 		pieceSize         uint32
 		withContentLength bool
 		checkDigest       bool
+		concurrentOption  *config.ConcurrentOption
 	}{
 		{
 			name:              "multiple pieces with content length, check digest",
@@ -93,6 +95,18 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 			checkDigest:       false,
 			withContentLength: true,
 		},
+		{
+			name:              "multiple pieces with content length, concurrent download",
+			pieceSize:         2048,
+			checkDigest:       false,
+			withContentLength: true,
+			concurrentOption: &config.ConcurrentOption{
+				GoroutineCount: 4,
+				ThresholdSize: util.Size{
+					Limit: 1024,
+				},
+			},
+		},
 		{
 			name:              "multiple pieces without content length, check digest",
 			pieceSize:         1024,
@@ -188,16 +202,37 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 			/********** prepare test end **********/
 			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 				if tc.withContentLength {
-					w.Header().Set("Content-Length",
+					w.Header().Set(headers.ContentLength,
 						fmt.Sprintf("%d", len(testBytes)))
 				}
-				n, err := io.Copy(w, bytes.NewBuffer(testBytes))
+				var (
+					buf *bytes.Buffer
+					l   int
+				)
+				rg := r.Header.Get(headers.Range)
+				if rg == "" {
+					buf = bytes.NewBuffer(testBytes)
+					l = len(testBytes)
+				} else {
+					parsedRange, err := util.ParseRange(rg, int64(len(testBytes)))
+					assert.Nil(err)
+					w.Header().Set(headers.ContentRange,
+						fmt.Sprintf("bytes %d-%d/%d",
+							parsedRange[0].Start,
+							parsedRange[0].Start+parsedRange[0].Length-1,
+							len(testBytes)))
+					w.WriteHeader(http.StatusPartialContent)
+
+					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(len(testBytes)), n)
+				assert.Equal(int64(l), n)
 			}))
 			defer ts.Close()
 
-			pm, err := NewPieceManager(pieceDownloadTimeout)
+			pm, err := NewPieceManager(pieceDownloadTimeout, WithConcurrentOption(tc.concurrentOption))
 			assert.Nil(err)
 			pm.(*pieceManager).computePieceSize = func(length int64) uint32 {
 				return tc.pieceSize
@@ -215,7 +250,7 @@ func TestPieceManager_DownloadSource(t *testing.T) {
 				request.UrlMeta.Digest = digest
 			}
 
-			err = pm.DownloadSource(context.Background(), mockPeerTask, request)
+			err = pm.DownloadSource(context.Background(), mockPeerTask, request, nil)
 			assert.Nil(err)
 
 			err = storageManager.Store(context.Background(),
diff --git a/test/testdata/charts/config.yaml b/test/testdata/charts/config.yaml
index 90f6eeb5..cb2920e8 100644
--- a/test/testdata/charts/config.yaml
+++ b/test/testdata/charts/config.yaml
@@ -76,6 +76,9 @@ dfdaemon:
     metrics: 127.0.0.1:8888
     download:
       prefetch: true
+      concurrent:
+        thresholdSize: 10M
+        goroutineCount: 4
     scheduler:
       disableAutoBackSource: true
     proxy:
diff --git a/test/testdata/k8s/proxy.yaml b/test/testdata/k8s/proxy.yaml
index 174f1856..541319b7 100644
--- a/test/testdata/k8s/proxy.yaml
+++ b/test/testdata/k8s/proxy.yaml
@@ -43,6 +43,9 @@ data:
           port: 65000
       perPeerRateLimit: 100Mi
       totalRateLimit: 200Mi
+      concurrent:
+        thresholdSize: 10M
+        goroutineCount: 4
     upload:
       rateLimit: 100Mi
       security: