Repository: dragonflyoss/Dragonfly2
Pull request #1169 (2022-03-18 03:12:03)

fix: subscriber data race
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! -->

- [ ] Code compiles correctly.
- [ ] My code follows the code style of this project.
- [ ] 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.
- [ ] All new and existing tests passed.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1169](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (cda9dc7) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/7b082c01affc86c04dba563add93554322e5d127?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (7b082c0) will **increase** coverage by `46.11%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169/graphs/tree.svg?width=650&height=150&src=pr&token=PUICJYSAT7&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff            @@
##           main    #1169       +/-   ##
=========================================
+ Coverage      0   46.11%   +46.11%     
=========================================
  Files         0      132      +132     
  Lines         0    10238    +10238     
=========================================
+ Hits          0     4721     +4721     
- Misses        0     4973     +4973     
- Partials      0      544      +544     
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `46.11% <100.00%> (?)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [client/daemon/rpcserver/subscriber.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9ycGNzZXJ2ZXIvc3Vic2NyaWJlci5nbw==) | `53.68% <100.00%> (ø)` | |
| [client/config/deprecated.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2NvbmZpZy9kZXByZWNhdGVkLmdv) | `0.00% <0.00%> (ø)` | |
| [cdn/config/config.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2RuL2NvbmZpZy9jb25maWcuZ28=) | `46.15% <0.00%> (ø)` | |
| [cdn/supervisor/task/manager\_util.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2RuL3N1cGVydmlzb3IvdGFzay9tYW5hZ2VyX3V0aWwuZ28=) | `12.96% <0.00%> (ø)` | |
| [scheduler/resource/cdn.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3Jlc291cmNlL2Nkbi5nbw==) | `62.29% <0.00%> (ø)` | |
| [scheduler/scheduler/evaluator/plugin.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3NjaGVkdWxlci9ldmFsdWF0b3IvcGx1Z2luLmdv) | `50.00% <0.00%> (ø)` | |
| [internal/dynconfig/dynconfig\_manager.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-aW50ZXJuYWwvZHluY29uZmlnL2R5bmNvbmZpZ19tYW5hZ2VyLmdv) | `72.41% <0.00%> (ø)` | |
| [cdn/supervisor/task/task.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2RuL3N1cGVydmlzb3IvdGFzay90YXNrLmdv) | `42.85% <0.00%> (ø)` | |
| [client/daemon/proxy/cert.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wcm94eS9jZXJ0Lmdv) | `0.00% <0.00%> (ø)` | |
| [pkg/util/digestutils/digest\_reader.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-cGtnL3V0aWwvZGlnZXN0dXRpbHMvZGlnZXN0X3JlYWRlci5nbw==) | `76.19% <0.00%> (ø)` | |
| ... and [123 more](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1169/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | |


Patch:
diff --git a/client/daemon/rpcserver/subscriber.go b/client/daemon/rpcserver/subscriber.go
index 56923530..0e392d63 100644
--- a/client/daemon/rpcserver/subscriber.go
+++ b/client/daemon/rpcserver/subscriber.go
@@ -153,11 +153,12 @@ loop:
 				s.Unlock()
 				break loop
 			}
-			s.Unlock()
 			if info.Finished {
+				s.Unlock()
 				break loop
 			}
 			nextPieceNum = s.searchNextPieceNum(nextPieceNum)
+			s.Unlock()
 		case <-s.Success:
 			s.Lock()
 			// all pieces already sent


Test patch:
diff --git a/client/daemon/rpcserver/rpcserver_test.go b/client/daemon/rpcserver/rpcserver_test.go
index 6f7401a3..8b8e6d82 100644
--- a/client/daemon/rpcserver/rpcserver_test.go
+++ b/client/daemon/rpcserver/rpcserver_test.go
@@ -336,180 +336,186 @@ func TestDownloadManager_SyncPieceTasks(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			mockStorageManger := mock_storage.NewMockManager(ctrl)
+			for _, delay := range []bool{false, true} {
+				delay := delay
+				mockStorageManger := mock_storage.NewMockManager(ctrl)
 
-			if tc.limit == 0 {
-				tc.limit = 1024
-			}
+				if tc.limit == 0 {
+					tc.limit = 1024
+				}
 
-			var (
-				totalPieces []*base.PieceInfo
-				lock        sync.Mutex
-			)
+				var (
+					totalPieces []*base.PieceInfo
+					lock        sync.Mutex
+				)
 
-			var addedPieces = make(map[uint32]*base.PieceInfo)
-			for _, p := range tc.existPieces {
-				if p.end == 0 {
-					p.end = p.start
-				}
-				for i := p.start; i <= p.end; i++ {
-					if _, ok := addedPieces[uint32(i)]; ok {
-						continue
+				var addedPieces = make(map[uint32]*base.PieceInfo)
+				for _, p := range tc.existPieces {
+					if p.end == 0 {
+						p.end = p.start
 					}
-					piece := &base.PieceInfo{
-						PieceNum:    int32(i),
-						RangeStart:  uint64(i) * uint64(pieceSize),
-						RangeSize:   pieceSize,
-						PieceOffset: uint64(i) * uint64(pieceSize),
-						PieceStyle:  base.PieceStyle_PLAIN,
+					for i := p.start; i <= p.end; i++ {
+						if _, ok := addedPieces[uint32(i)]; ok {
+							continue
+						}
+						piece := &base.PieceInfo{
+							PieceNum:    int32(i),
+							RangeStart:  uint64(i) * uint64(pieceSize),
+							RangeSize:   pieceSize,
+							PieceOffset: uint64(i) * uint64(pieceSize),
+							PieceStyle:  base.PieceStyle_PLAIN,
+						}
+						totalPieces = append(totalPieces, piece)
+						addedPieces[uint32(i)] = piece
 					}
-					totalPieces = append(totalPieces, piece)
-					addedPieces[uint32(i)] = piece
 				}
-			}
 
-			mockStorageManger.EXPECT().GetPieces(gomock.Any(),
-				gomock.Any()).AnyTimes().DoAndReturn(
-				func(ctx context.Context, req *base.PieceTaskRequest) (*base.PiecePacket, error) {
-					var pieces []*base.PieceInfo
-					lock.Lock()
-					for i := req.StartNum; i < tc.totalPieces; i++ {
-						if piece, ok := addedPieces[i]; ok {
-							if piece.PieceNum >= int32(req.StartNum) && len(pieces) < int(req.Limit) {
-								pieces = append(pieces, piece)
+				mockStorageManger.EXPECT().GetPieces(gomock.Any(),
+					gomock.Any()).AnyTimes().DoAndReturn(
+					func(ctx context.Context, req *base.PieceTaskRequest) (*base.PiecePacket, error) {
+						var pieces []*base.PieceInfo
+						lock.Lock()
+						for i := req.StartNum; i < tc.totalPieces; i++ {
+							if piece, ok := addedPieces[i]; ok {
+								if piece.PieceNum >= int32(req.StartNum) && len(pieces) < int(req.Limit) {
+									pieces = append(pieces, piece)
+								}
 							}
 						}
-					}
-					lock.Unlock()
-					return &base.PiecePacket{
-						TaskId:        req.TaskId,
-						DstPid:        req.DstPid,
-						DstAddr:       "",
-						PieceInfos:    pieces,
-						TotalPiece:    int32(tc.totalPieces),
-						ContentLength: int64(tc.totalPieces) * int64(pieceSize),
-						PieceMd5Sign:  "",
-					}, nil
-				})
-			mockTaskManager := mock_peer.NewMockTaskManager(ctrl)
-			mockTaskManager.EXPECT().Subscribe(gomock.Any()).AnyTimes().DoAndReturn(
-				func(request *base.PieceTaskRequest) (*peer.SubscribeResult, bool) {
-					ch := make(chan *peer.PieceInfo)
-					success := make(chan struct{})
-					fail := make(chan struct{})
+						lock.Unlock()
+						return &base.PiecePacket{
+							TaskId:        req.TaskId,
+							DstPid:        req.DstPid,
+							DstAddr:       "",
+							PieceInfos:    pieces,
+							TotalPiece:    int32(tc.totalPieces),
+							ContentLength: int64(tc.totalPieces) * int64(pieceSize),
+							PieceMd5Sign:  "",
+						}, nil
+					})
+				mockTaskManager := mock_peer.NewMockTaskManager(ctrl)
+				mockTaskManager.EXPECT().Subscribe(gomock.Any()).AnyTimes().DoAndReturn(
+					func(request *base.PieceTaskRequest) (*peer.SubscribeResult, bool) {
+						ch := make(chan *peer.PieceInfo)
+						success := make(chan struct{})
+						fail := make(chan struct{})
 
-					go func(followingPieces []pieceRange) {
-						for i, p := range followingPieces {
-							if p.end == 0 {
-								p.end = p.start
-							}
-							for j := p.start; j <= p.end; j++ {
-								lock.Lock()
-								if _, ok := addedPieces[uint32(j)]; ok {
-									continue
+						go func(followingPieces []pieceRange) {
+							for i, p := range followingPieces {
+								if p.end == 0 {
+									p.end = p.start
 								}
-								piece := &base.PieceInfo{
-									PieceNum:    int32(j),
-									RangeStart:  uint64(j) * uint64(pieceSize),
-									RangeSize:   pieceSize,
-									PieceOffset: uint64(j) * uint64(pieceSize),
-									PieceStyle:  base.PieceStyle_PLAIN,
-								}
-								totalPieces = append(totalPieces, piece)
-								addedPieces[uint32(j)] = piece
-								lock.Unlock()
+								for j := p.start; j <= p.end; j++ {
+									lock.Lock()
+									if _, ok := addedPieces[uint32(j)]; ok {
+										continue
+									}
+									piece := &base.PieceInfo{
+										PieceNum:    int32(j),
+										RangeStart:  uint64(j) * uint64(pieceSize),
+										RangeSize:   pieceSize,
+										PieceOffset: uint64(j) * uint64(pieceSize),
+										PieceStyle:  base.PieceStyle_PLAIN,
+									}
+									totalPieces = append(totalPieces, piece)
+									addedPieces[uint32(j)] = piece
+									lock.Unlock()
 
-								var finished bool
-								if i == len(followingPieces)-1 && j == p.end {
-									finished = true
-								}
-								ch <- &peer.PieceInfo{
-									Num:      int32(j),
-									Finished: finished,
+									var finished bool
+									if i == len(followingPieces)-1 && j == p.end {
+										finished = true
+									}
+									if !delay {
+										ch <- &peer.PieceInfo{
+											Num:      int32(j),
+											Finished: finished,
+										}
+									}
 								}
 							}
-						}
-						close(success)
-					}(tc.followingPieces)
+							close(success)
+						}(tc.followingPieces)
 
-					return &peer.SubscribeResult{
-						Storage:          mockStorageManger,
-						PieceInfoChannel: ch,
-						Success:          success,
-						Fail:             fail,
-					}, true
-				})
+						return &peer.SubscribeResult{
+							Storage:          mockStorageManger,
+							PieceInfoChannel: ch,
+							Success:          success,
+							Fail:             fail,
+						}, true
+					})
 
-			s := &server{
-				KeepAlive:       clientutil.NewKeepAlive("test"),
-				peerHost:        &scheduler.PeerHost{},
-				storageManager:  mockStorageManger,
-				peerTaskManager: mockTaskManager,
-			}
+				s := &server{
+					KeepAlive:       clientutil.NewKeepAlive("test"),
+					peerHost:        &scheduler.PeerHost{},
+					storageManager:  mockStorageManger,
+					peerTaskManager: mockTaskManager,
+				}
 
-			port, client := setupPeerServerAndClient(t, s, assert, s.ServePeer)
-			defer s.peerServer.GracefulStop()
+				port, client := setupPeerServerAndClient(t, s, assert, s.ServePeer)
+				defer s.peerServer.GracefulStop()
 
-			syncClient, err := client.SyncPieceTasks(
-				context.Background(),
-				dfnet.NetAddr{
-					Type: dfnet.TCP,
-					Addr: fmt.Sprintf("127.0.0.1:%d", port),
-				},
-				&base.PieceTaskRequest{
-					TaskId:   tc.name,
-					SrcPid:   idgen.PeerID(iputils.IPv4),
-					DstPid:   idgen.PeerID(iputils.IPv4),
-					StartNum: 0,
-					Limit:    tc.limit,
-				})
-			assert.Nil(err, "client sync piece tasks grpc call should be ok")
+				syncClient, err := client.SyncPieceTasks(
+					context.Background(),
+					dfnet.NetAddr{
+						Type: dfnet.TCP,
+						Addr: fmt.Sprintf("127.0.0.1:%d", port),
+					},
+					&base.PieceTaskRequest{
+						TaskId:   tc.name,
+						SrcPid:   idgen.PeerID(iputils.IPv4),
+						DstPid:   idgen.PeerID(iputils.IPv4),
+						StartNum: 0,
+						Limit:    tc.limit,
+					})
+				assert.Nil(err, "client sync piece tasks grpc call should be ok")
 
-			var (
-				total       = make(map[int32]bool)
-				maxNum      int32
-				requestSent = make(chan bool)
-			)
-			if len(tc.requestPieces) == 0 {
-				close(requestSent)
-			} else {
-				go func() {
-					for _, n := range tc.requestPieces {
-						request := &base.PieceTaskRequest{
-							TaskId:   tc.name,
-							SrcPid:   idgen.PeerID(iputils.IPv4),
-							DstPid:   idgen.PeerID(iputils.IPv4),
-							StartNum: uint32(n),
-							Limit:    tc.limit,
-						}
-						assert.Nil(syncClient.Send(request))
-					}
+				var (
+					total       = make(map[int32]bool)
+					maxNum      int32
+					requestSent = make(chan bool)
+				)
+				if len(tc.requestPieces) == 0 {
 					close(requestSent)
-				}()
-			}
-			for {
-				p, err := syncClient.Recv()
-				if err == io.EOF {
-					break
+				} else {
+					go func() {
+						for _, n := range tc.requestPieces {
+							request := &base.PieceTaskRequest{
+								TaskId:   tc.name,
+								SrcPid:   idgen.PeerID(iputils.IPv4),
+								DstPid:   idgen.PeerID(iputils.IPv4),
+								StartNum: uint32(n),
+								Limit:    tc.limit,
+							}
+							assert.Nil(syncClient.Send(request))
+						}
+						close(requestSent)
+					}()
 				}
-				for _, info := range p.PieceInfos {
-					total[info.PieceNum] = true
-					if info.PieceNum >= maxNum {
-						maxNum = info.PieceNum
+				for {
+					p, err := syncClient.Recv()
+					if err == io.EOF {
+						break
+					}
+					for _, info := range p.PieceInfos {
+						total[info.PieceNum] = true
+						if info.PieceNum >= maxNum {
+							maxNum = info.PieceNum
+						}
+					}
+					if tc.success {
+						assert.Nil(err, "receive piece tasks should be ok")
+					}
+					if int(p.TotalPiece) == len(total) {
+						<-requestSent
+						err = syncClient.CloseSend()
+						assert.Nil(err)
 					}
 				}
 				if tc.success {
-					assert.Nil(err, "receive piece tasks should be ok")
+					assert.Equal(int(maxNum+1), len(total))
 				}
-				if int(p.TotalPiece) == len(total) {
-					<-requestSent
-					err = syncClient.CloseSend()
-					assert.Nil(err)
-				}
-			}
-			if tc.success {
-				assert.Equal(int(maxNum+1), len(total))
 			}
+
 		})
 	}
 }