feat: add option to cancel idle peer task (#3810)

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

Author: jim3ma

client/config/peerhost.go

@@ -278,6 +278,7 @@ type DownloadOption struct {
 	Concurrent           *ConcurrentOption `mapstructure:"concurrent" yaml:"concurrent"`
 	SyncPieceViaHTTPS    bool              `mapstructure:"syncPieceViaHTTPS" yaml:"syncPieceViaHTTPS"`
 	SplitRunningTasks    bool              `mapstructure:"splitRunningTasks" yaml:"splitRunningTasks"`
+	CancelIdlePeerTask   bool              `mapstructure:"cancelIdlePeerTask" yaml:"cancelIdlePeerTask"`
 	// resource clients option
 	ResourceClients ResourceClientsOption `mapstructure:"resourceClients" yaml:"resourceClients"`
 

client/daemon/daemon.go

@@ -255,14 +255,15 @@ func New(opt *config.DaemonOption, d dfpath.Dfpath) (Daemon, error) {
 
 	peerTaskManagerOption := &peer.TaskManagerOption{
 		TaskOption: peer.TaskOption{
-			PeerHost:        host,
-			SchedulerOption: opt.Scheduler,
-			PieceManager:    pieceManager,
-			StorageManager:  storageManager,
-			WatchdogTimeout: opt.Download.WatchdogTimeout,
-			CalculateDigest: opt.Download.CalculateDigest,
-			GRPCCredentials: rpc.NewInsecureCredentials(),
-			GRPCDialTimeout: opt.Download.GRPCDialTimeout,
+			PeerHost:           host,
+			SchedulerOption:    opt.Scheduler,
+			PieceManager:       pieceManager,
+			StorageManager:     storageManager,
+			WatchdogTimeout:    opt.Download.WatchdogTimeout,
+			CalculateDigest:    opt.Download.CalculateDigest,
+			GRPCCredentials:    rpc.NewInsecureCredentials(),
+			GRPCDialTimeout:    opt.Download.GRPCDialTimeout,
+			CancelIdlePeerTask: opt.Download.CancelIdlePeerTask,
 		},
 		SchedulerClient:       schedulerClient,
 		PerPeerRateLimit:      opt.Download.PerPeerRateLimit.Limit,

client/daemon/peer/peertask_conductor.go

@@ -173,7 +173,8 @@ type TaskOption struct {
 	GRPCCredentials credentials.TransportCredentials
 	GRPCDialTimeout time.Duration
 	// WatchdogTimeout > 0 indicates to start watch dog for every single peer task
-	WatchdogTimeout time.Duration
+	WatchdogTimeout    time.Duration
+	CancelIdlePeerTask bool
 }
 
 func (ptm *peerTaskManager) newPeerTaskConductor(
@@ -183,9 +184,12 @@ func (ptm *peerTaskManager) newPeerTaskConductor(
 	parent *peerTaskConductor,
 	rg *nethttp.Range,
 	seed bool) *peerTaskConductor {
-	// use a new context with span info
-	ctx = trace.ContextWithSpan(context.Background(), trace.SpanFromContext(ctx))
-	ctx, span := tracer.Start(ctx, config.SpanPeerTask, trace.WithSpanKind(trace.SpanKindClient))
+	var span trace.Span
+	if !ptm.TaskOption.CancelIdlePeerTask {
+		// use a new context to avoid cancel idle peer task
+		ctx = trace.ContextWithSpan(context.Background(), trace.SpanFromContext(ctx))
+	}
+	ctx, span = tracer.Start(ctx, config.SpanPeerTask, trace.WithSpanKind(trace.SpanKindClient))
 	span.SetAttributes(config.AttributePeerHost.String(ptm.PeerHost.Id))
 	span.SetAttributes(semconv.NetHostIPKey.String(ptm.PeerHost.Ip))
 	span.SetAttributes(config.AttributePeerID.String(request.PeerId))