feat: auto switch to concurrent back source based on download speed

Signed-off-by: greenhandatsjtu <[email protected]>

Author: greenhandatsjtu

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

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,

client/config/testdata/config/daemon.yaml

@@ -79,6 +79,7 @@ download:
   watchdogTimeout: 1s
   concurrent:
     thresholdSize: 1
+    thresholdSpeed: 1
     goroutineCount: 1
     initBackoff: 1
     maxBackoff: 1

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