Repository: dragonflyoss/Dragonfly2
Pull request #1269 (2022-04-21 10:24:21)

chore: optimize create synchronizer logic
Signed-off-by: Jim Ma <majinjing3@gmail.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description

<!--- Describe your changes in detail -->

## 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)
- [ ] 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.
- [ ] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.



Patch:
diff --git a/client/daemon/peer/peertask_conductor.go b/client/daemon/peer/peertask_conductor.go
index e2f19f92..081c0ed8 100644
--- a/client/daemon/peer/peertask_conductor.go
+++ b/client/daemon/peer/peertask_conductor.go
@@ -681,21 +681,21 @@ loop:
 
 // updateSynchronizer will convert peers to synchronizer, if failed, will update failed peers to scheduler.PeerPacket
 func (pt *peerTaskConductor) updateSynchronizer(lastNum int32, p *scheduler.PeerPacket) int32 {
-	num, ok := pt.getNextNotReadyPieceNum(lastNum)
+	desiredPiece, ok := pt.getNextNotReadyPieceNum(lastNum)
 	if !ok {
 		pt.Infof("all pieces is ready, peer task completed, skip to synchronize")
 		p.MainPeer = nil
 		p.StealPeers = nil
-		return num
+		return desiredPiece
 	}
 	var peers = []*scheduler.PeerPacket_DestPeer{p.MainPeer}
 	peers = append(peers, p.StealPeers...)
 
-	legacyPeers := pt.pieceTaskSyncManager.newMultiPieceTaskSynchronizer(peers, num)
+	legacyPeers := pt.pieceTaskSyncManager.newMultiPieceTaskSynchronizer(peers, desiredPiece)
 
 	p.MainPeer = nil
 	p.StealPeers = legacyPeers
-	return num
+	return desiredPiece
 }
 
 func (pt *peerTaskConductor) confirmReceivePeerPacketError(err error) {
@@ -1368,6 +1368,7 @@ func (pt *peerTaskConductor) done() {
 	defer func() {
 		pt.broker.Stop()
 		pt.span.End()
+		pt.pieceDownloadCancel()
 		if pt.pieceTaskSyncManager != nil {
 			pt.pieceTaskSyncManager.cancel()
 		}
@@ -1459,6 +1460,7 @@ func (pt *peerTaskConductor) fail() {
 		close(pt.failCh)
 		pt.broker.Stop()
 		pt.span.End()
+		pt.pieceDownloadCancel()
 		if pt.pieceTaskSyncManager != nil {
 			pt.pieceTaskSyncManager.cancel()
 		}
diff --git a/client/daemon/peer/peertask_piecetask_synchronizer.go b/client/daemon/peer/peertask_piecetask_synchronizer.go
index 8b1e90dd..49c1f6ec 100644
--- a/client/daemon/peer/peertask_piecetask_synchronizer.go
+++ b/client/daemon/peer/peertask_piecetask_synchronizer.go
@@ -60,44 +60,49 @@ type pieceTaskSynchronizerError struct {
 }
 
 // FIXME for compatibility, sync will be called after the dfclient.GetPieceTasks deprecated and the pieceTaskPoller removed
-func (s *pieceTaskSyncManager) sync(pp *scheduler.PeerPacket, request *base.PieceTaskRequest) error {
+func (s *pieceTaskSyncManager) sync(pp *scheduler.PeerPacket, desiredPiece int32) error {
 	var (
-		peers  = map[string]bool{}
-		errors []error
+		peers = map[string]bool{}
+		errs  []error
 	)
 	peers[pp.MainPeer.PeerId] = true
 	// TODO if the worker failed, reconnect and retry
 	s.Lock()
 	defer s.Unlock()
 	if _, ok := s.workers[pp.MainPeer.PeerId]; !ok {
-		err := s.newPieceTaskSynchronizer(s.ctx, pp.MainPeer, request)
+		err := s.newPieceTaskSynchronizer(s.ctx, pp.MainPeer, desiredPiece)
 		if err != nil {
 			s.peerTaskConductor.Errorf("main peer SyncPieceTasks error: %s", err)
-			errors = append(errors, err)
+			errs = append(errs, err)
 		}
 	}
 	for _, p := range pp.StealPeers {
 		peers[p.PeerId] = true
 		if _, ok := s.workers[p.PeerId]; !ok {
-			err := s.newPieceTaskSynchronizer(s.ctx, p, request)
+			err := s.newPieceTaskSynchronizer(s.ctx, p, desiredPiece)
 			if err != nil {
 				s.peerTaskConductor.Errorf("steal peer SyncPieceTasks error: %s", err)
-				errors = append(errors, err)
+				errs = append(errs, err)
 			}
 		}
 	}
 
 	// cancel old workers
 	if len(s.workers) != len(peers) {
+		var peersToRemove []string
 		for p, worker := range s.workers {
 			if !peers[p] {
 				worker.close()
+				peersToRemove = append(peersToRemove, p)
 			}
 		}
+		for _, p := range peersToRemove {
+			delete(s.workers, p)
+		}
 	}
 
-	if len(errors) > 0 {
-		return errors[0]
+	if len(errs) > 0 {
+		return errs[0]
 	}
 	return nil
 }
@@ -128,9 +133,14 @@ func (s *pieceTaskSyncManager) cleanStaleWorker(destPeers []*scheduler.PeerPacke
 func (s *pieceTaskSyncManager) newPieceTaskSynchronizer(
 	ctx context.Context,
 	dstPeer *scheduler.PeerPacket_DestPeer,
-	request *base.PieceTaskRequest) error {
-
-	request.DstPid = dstPeer.PeerId
+	lastNum int32) error {
+	request := &base.PieceTaskRequest{
+		TaskId:   s.peerTaskConductor.taskID,
+		SrcPid:   s.peerTaskConductor.peerID,
+		DstPid:   dstPeer.PeerId,
+		StartNum: uint32(lastNum),
+		Limit:    16,
+	}
 	if worker, ok := s.workers[dstPeer.PeerId]; ok {
 		// worker is okay, keep it go on
 		if worker.error.Load() == nil {
@@ -141,7 +151,6 @@ func (s *pieceTaskSyncManager) newPieceTaskSynchronizer(
 		delete(s.workers, dstPeer.PeerId)
 	}
 
-	request.DstPid = dstPeer.PeerId
 	client, err := dfclient.SyncPieceTasks(ctx, dstPeer, request)
 	// Refer: https://github.com/grpc/grpc-go/blob/v1.44.0/stream.go#L104
 	// When receive io.EOF, the real error should be discovered using RecvMsg, here is client.Recv() here
@@ -179,18 +188,11 @@ func (s *pieceTaskSyncManager) newPieceTaskSynchronizer(
 
 func (s *pieceTaskSyncManager) newMultiPieceTaskSynchronizer(
 	destPeers []*scheduler.PeerPacket_DestPeer,
-	lastNum int32) (legacyPeers []*scheduler.PeerPacket_DestPeer) {
+	desiredPiece int32) (legacyPeers []*scheduler.PeerPacket_DestPeer) {
 	s.Lock()
 	defer s.Unlock()
 	for _, peer := range destPeers {
-		request := &base.PieceTaskRequest{
-			TaskId:   s.peerTaskConductor.taskID,
-			SrcPid:   s.peerTaskConductor.peerID,
-			DstPid:   "",
-			StartNum: uint32(lastNum),
-			Limit:    16,
-		}
-		err := s.newPieceTaskSynchronizer(s.ctx, peer, request)
+		err := s.newPieceTaskSynchronizer(s.ctx, peer, desiredPiece)
 		if err == nil {
 			s.peerTaskConductor.Infof("connected to peer: %s", peer.PeerId)
 			continue
@@ -345,16 +347,14 @@ func (s *pieceTaskSynchronizer) acquire(request *base.PieceTaskRequest) error {
 		s.Debugf("synchronizer already error %s, skip acquire more pieces", err)
 		return err
 	}
+	request.DstPid = s.dstPeer.PeerId
 	err := s.client.Send(request)
 	s.span.AddEvent(fmt.Sprintf("send piece #%d request", request.StartNum))
 	if err != nil {
+		// send should always ok
 		s.error.Store(&pieceTaskSynchronizerError{err})
-		if s.canceled(err) {
-			s.Debugf("synchronizer sends canceled")
-		} else {
-			s.Errorf("synchronizer sends with error: %s", err)
-			s.reportError(err)
-		}
+		s.Errorf("synchronizer sends with error: %s", err)
+		s.reportError(err)
 	}
 	return err
 }


Test patch:
diff --git a/scheduler/storage/storage_test.go b/scheduler/storage/storage_test.go
index 5ace7c61..97b994f0 100644
--- a/scheduler/storage/storage_test.go
+++ b/scheduler/storage/storage_test.go
@@ -585,34 +585,6 @@ func TestStorage_backups(t *testing.T) {
 				assert.Error(err)
 			},
 		},
-		{
-			name:    "get backups",
-			baseDir: os.TempDir(),
-			options: []Option{WithMaxSize(0), WithBufferSize(1)},
-			mock: func(t *testing.T, s Storage) {
-				if err := s.Create(Record{ID: "1"}); err != nil {
-					t.Fatal(err)
-				}
-
-				if err := s.Create(Record{ID: "2"}); err != nil {
-					t.Fatal(err)
-				}
-
-				if err := s.Create(Record{ID: "3"}); err != nil {
-					t.Fatal(err)
-				}
-			},
-			expect: func(t *testing.T, s Storage, baseDir string) {
-				assert := assert.New(t)
-				backups, err := s.(*storage).backups()
-				assert.NoError(err)
-				assert.Equal(len(backups), 3)
-				assert.Equal(backups[2].Name(), fmt.Sprintf("%s.%s", RecordFilePrefix, RecordFileExt))
-				if err := s.Clear(); err != nil {
-					t.Fatal(err)
-				}
-			},
-		},
 	}
 
 	for _, tc := range tests {