feat: concurrent multiple pieces back source (#1426)

* feat: concurrent multiple pieces back source
* chore: update http source client

Signed-off-by: Jim Ma <[email protected]>

Author: jim3ma

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"`

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

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))

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
 	}