Merge branch 'main' of github.com:dragonflyoss/Dragonfly2 into featur…

…e/lock-task

Signed-off-by: Gaius <gaius.qi@gmail.com>

client/daemon/peer/peertask_conductor.go

@@ -160,7 +160,8 @@ func (ptm *peerTaskManager) newPeerTaskConductor(
 	defer cancel()
 	regCtx, regSpan := tracer.Start(regCtx, config.SpanRegisterTask)
 	logger.Infof("step 1: peer %s start to register", request.PeerId)
-	result, err := ptm.schedulerClient.RegisterPeerTask(regCtx, request)
+	schedulerClient := ptm.schedulerClient
+	result, err := schedulerClient.RegisterPeerTask(regCtx, request)
 	regSpan.RecordError(err)
 	regSpan.End()
 
@@ -178,7 +179,7 @@ func (ptm *peerTaskManager) newPeerTaskConductor(
 		}
 		needBackSource = true
 		// can not detect source or scheduler error, create a new dummy scheduler client
-		ptm.schedulerClient = &dummySchedulerClient{}
+		schedulerClient = &dummySchedulerClient{}
 		result = &scheduler.RegisterResult{TaskId: idgen.TaskID(request.Url, request.UrlMeta)}
 		logger.Warnf("register peer task failed: %s, peer id: %s, try to back source", err, request.PeerId)
 	}
@@ -243,7 +244,7 @@ func (ptm *peerTaskManager) newPeerTaskConductor(
 		}
 	}
 
-	peerPacketStream, err := ptm.schedulerClient.ReportPieceResult(ctx, result.TaskId, request)
+	peerPacketStream, err := schedulerClient.ReportPieceResult(ctx, result.TaskId, request)
 	log.Infof("step 2: start report piece result")
 	if err != nil {
 		defer span.End()
@@ -280,7 +281,7 @@ func (ptm *peerTaskManager) newPeerTaskConductor(
 		pieceParallelCount:  atomic.NewInt32(0),
 		totalPiece:          -1,
 		schedulerOption:     ptm.schedulerOption,
-		schedulerClient:     ptm.schedulerClient,
+		schedulerClient:     schedulerClient,
 		limiter:             rate.NewLimiter(limit, int(limit)),
 		completedLength:     atomic.NewInt64(0),
 		usedTraffic:         atomic.NewUint64(0),

client/daemon/peer/peertask_dummy.go

@@ -28,6 +28,7 @@ import (
 	schedulerclient "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
 )
 
+// when scheduler is not available, use dummySchedulerClient to back source
 type dummySchedulerClient struct {
 }
 

client/daemon/peer/peertask_manager_test.go

@@ -453,6 +453,8 @@ func TestPeerTaskManager_TaskSuite(t *testing.T) {
 
 	for _, _tc := range testCases {
 		t.Run(_tc.name, func(t *testing.T) {
+			assert := testifyassert.New(t)
+			require := testifyrequire.New(t)
 			for _, typ := range taskTypes {
 				// dup a new test case with the task type
 				tc := _tc
@@ -492,20 +494,23 @@ func TestPeerTaskManager_TaskSuite(t *testing.T) {
 						sourceClient = tc.mockHTTPSourceClient(ctrl, tc.taskData, tc.url)
 					}
 
-					mm := setupMockManager(
-						ctrl, &tc,
-						componentsOption{
-							taskID:             taskID,
-							contentLength:      int64(mockContentLength),
-							pieceSize:          uint32(tc.pieceSize),
-							pieceParallelCount: tc.pieceParallelCount,
-							pieceDownloader:    downloader,
-							sourceClient:       sourceClient,
-							content:            tc.taskData,
-							scope:              tc.sizeScope,
-							peerPacketDelay:    tc.peerPacketDelay,
-							backSource:         tc.backSource,
-						})
+					option := componentsOption{
+						taskID:             taskID,
+						contentLength:      int64(mockContentLength),
+						pieceSize:          uint32(tc.pieceSize),
+						pieceParallelCount: tc.pieceParallelCount,
+						pieceDownloader:    downloader,
+						sourceClient:       sourceClient,
+						content:            tc.taskData,
+						scope:              tc.sizeScope,
+						peerPacketDelay:    tc.peerPacketDelay,
+						backSource:         tc.backSource,
+					}
+					// keep peer task running in enough time to check "getOrCreatePeerTaskConductor" always return same
+					if tc.taskType == taskTypeConductor {
+						option.peerPacketDelay = []time.Duration{time.Second}
+					}
+					mm := setupMockManager(ctrl, &tc, option)
 					defer mm.CleanUp()
 
 					tc.run(assert, require, mm, urlMeta)
@@ -652,8 +657,8 @@ func (ts *testSpec) runConductorTest(assert *testifyassert.Assertions, require *
 	}
 
 	var (
-		runningTaskCount int
-		success          bool
+		noRunningTask = true
+		success       bool
 	)
 	select {
 	case <-ptc.successCh:
@@ -663,11 +668,17 @@ func (ts *testSpec) runConductorTest(assert *testifyassert.Assertions, require *
 	}
 	assert.True(success, "task should success")
 
-	ptm.runningPeerTasks.Range(func(key, value interface{}) bool {
-		runningTaskCount++
-		return true
-	})
-	assert.Equal(0, runningTaskCount, "no running tasks")
+	for i := 0; i < 3; i++ {
+		ptm.runningPeerTasks.Range(func(key, value interface{}) bool {
+			noRunningTask = false
+			return false
+		})
+		if noRunningTask {
+			break
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	assert.True(noRunningTask, "no running tasks")
 
 	// test reuse stream task
 	rc, _, ok := ptm.tryReuseStreamPeerTask(context.Background(), request)

scheduler/resource/task.go

@@ -92,9 +92,6 @@ type Task struct {
 	// Peer sync map
 	Peers *sync.Map
 
-	// Task mutex
-	MU *sync.Mutex
-
 	// CreateAt is task create time
 	CreateAt *atomic.Time
 
@@ -119,7 +116,6 @@ func NewTask(id, url string, backToSourceLimit int, meta *base.UrlMeta) *Task {
 		Peers:             &sync.Map{},
 		CreateAt:          atomic.NewTime(time.Now()),
 		UpdateAt:          atomic.NewTime(time.Now()),
-		MU:                &sync.Mutex{},
 		Log:               logger.WithTaskIDAndURL(id, url),
 	}
 

scheduler/service/service.go

@@ -87,9 +87,6 @@ func (s *service) CDN() resource.CDN {
 
 func (s *service) RegisterTask(ctx context.Context, req *rpcscheduler.PeerTaskRequest) (*resource.Task, error) {
 	task := resource.NewTask(idgen.TaskID(req.Url, req.UrlMeta), req.Url, s.config.Scheduler.BackSourceCount, req.UrlMeta)
-
-	task.MU.Lock()
-	defer task.MU.Unlock()
 	task, ok := s.resource.TaskManager().LoadOrStore(task)
 	if ok && (task.FSM.Is(resource.TaskStateRunning) || task.FSM.Is(resource.TaskStateSucceeded)) {
 		// Task is healthy and can be reused