Repository: dragonflyoss/Dragonfly2
Pull request #1102 (2022-02-28 07:12:39)

feat: optimize depth limit func
Signed-off-by: Gaius <gaius.qi@gmail.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description
- Optimize depth limit func.
<!--- 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)
- [x] 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! -->

- [x] Code compiles correctly.
- [x] My code follows the code style of this project.
- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [x] I have read the **CONTRIBUTING** document.
- [x] I have added tests to cover my changes.
- [x] All new and existing tests passed.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1102?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1102](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1102?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (932546a) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/b93e9b09db07249e9c83d8ae5c6dce95eeca3691?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (b93e9b0) will **increase** coverage by `0.03%`.
> The diff coverage is `n/a`.

> :exclamation: Current head 932546a differs from pull request most recent head 17f7117. Consider uploading reports for the commit 17f7117 to get more accurate results

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1102/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/1102?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main    #1102      +/-   ##
==========================================
+ Coverage   45.39%   45.42%   +0.03%     
==========================================
  Files         130      130              
  Lines        9919     9919              
==========================================
+ Hits         4503     4506       +3     
+ Misses       4891     4888       -3     
  Partials      525      525              
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `45.42% <ø> (+0.03%)` | :arrow_up: |

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/1102?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [client/daemon/peer/peertask\_conductor.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1102/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2NvbmR1Y3Rvci5nbw==) | `57.67% <0.00%> (+0.35%)` | :arrow_up: |


Patch:
diff --git a/manager/database/mysql.go b/manager/database/mysql.go
index 8fb770e6..b849e77f 100644
--- a/manager/database/mysql.go
+++ b/manager/database/mysql.go
@@ -32,8 +32,11 @@ import (
 )
 
 const (
-	defaultCDNLoadLimit    = 300
-	defaultClientLoadLimit = 100
+	// Default number of cdn load limit
+	defaultCDNLoadLimit = 300
+
+	// Default number of client load limit
+	defaultClientLoadLimit = 50
 )
 
 func newMyqsl(cfg *config.MysqlConfig) (*gorm.DB, error) {
diff --git a/scheduler/resource/peer.go b/scheduler/resource/peer.go
index 9544be0a..58c908f2 100644
--- a/scheduler/resource/peer.go
+++ b/scheduler/resource/peer.go
@@ -310,9 +310,6 @@ func (p *Peer) ReplaceParent(parent *Peer) {
 
 // Depth represents depth of tree
 func (p *Peer) Depth() int {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-
 	node := p
 	var depth int
 	for node != nil {
diff --git a/scheduler/scheduler/scheduler.go b/scheduler/scheduler/scheduler.go
index 5914033c..22c2269a 100644
--- a/scheduler/scheduler/scheduler.go
+++ b/scheduler/scheduler/scheduler.go
@@ -253,8 +253,9 @@ func (s *scheduler) filterParents(peer *resource.Peer, blocklist set.SafeSet) []
 			return true
 		}
 
-		if parent.Depth() > defaultDepthLimit {
-			peer.Log.Infof("exceeds the %d depth limit of the tree", defaultDepthLimit)
+		depth := parent.Depth()
+		if depth > defaultDepthLimit {
+			peer.Log.Infof("%d exceeds the %d depth limit of the tree", depth, defaultDepthLimit)
 			return true
 		}
 
diff --git a/scheduler/service/service.go b/scheduler/service/service.go
index 0a159067..cadb4189 100644
--- a/scheduler/service/service.go
+++ b/scheduler/service/service.go
@@ -107,8 +107,7 @@ func (s *Service) RegisterPeerTask(ctx context.Context, req *rpcscheduler.PeerTa
 			fallthrough
 		case base.SizeScope_SMALL:
 			peer.Log.Info("task size scope is small")
-			// If the file is registered as a small type,
-			// there is no need to build a tree, just find the parent and return
+			// There is no need to build a tree, just find the parent and return
 			parent, ok := s.scheduler.FindParent(ctx, peer, set.NewSafeSet())
 			if !ok {
 				peer.Log.Warn("task size scope is small and it can not select parent")
@@ -124,6 +123,22 @@ func (s *Service) RegisterPeerTask(ctx context.Context, req *rpcscheduler.PeerTa
 				}, nil
 			}
 
+			// When task size scope is small, parent must be downloaded successfully
+			// before returning to the parent directly
+			if !parent.FSM.Is(resource.PeerStateSucceeded) {
+				peer.Log.Infof("task size scope is small and download state %s is not PeerStateSucceeded", parent.FSM.Current())
+				if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
+					dferr := dferrors.New(base.Code_SchedError, err.Error())
+					peer.Log.Errorf("peer %s register is failed: %v", req.PeerId, err)
+					return nil, dferr
+				}
+
+				return &rpcscheduler.RegisterResult{
+					TaskId:    task.ID,
+					SizeScope: base.SizeScope_NORMAL,
+				}, nil
+			}
+
 			firstPiece, ok := task.LoadPiece(0)
 			if !ok {
 				peer.Log.Warn("task size scope is small and it can not get first piece")


Test patch:
diff --git a/scheduler/service/service_test.go b/scheduler/service/service_test.go
index 0959b939..0f2deff9 100644
--- a/scheduler/service/service_test.go
+++ b/scheduler/service/service_test.go
@@ -346,6 +346,46 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				assert.True(peer.FSM.Is(resource.PeerStateReceivedNormal))
 			},
 		},
+		{
+			name: "task scope size is SizeScope_SMALL and load piece error, parent state is PeerStateRunning",
+			req: &rpcscheduler.PeerTaskRequest{
+				PeerHost: &rpcscheduler.PeerHost{
+					Uuid: mockRawHost.Uuid,
+				},
+			},
+			mock: func(
+				req *rpcscheduler.PeerTaskRequest, mockPeer *resource.Peer, mockCDNPeer *resource.Peer,
+				scheduler scheduler.Scheduler, res resource.Resource, hostManager resource.HostManager, taskManager resource.TaskManager, peerManager resource.PeerManager,
+				ms *mocks.MockSchedulerMockRecorder, mr *resource.MockResourceMockRecorder, mh *resource.MockHostManagerMockRecorder, mt *resource.MockTaskManagerMockRecorder, mp *resource.MockPeerManagerMockRecorder,
+			) {
+				mockPeer.Task.FSM.SetState(resource.TaskStateSucceeded)
+				mockPeer.Task.StorePeer(mockCDNPeer)
+				mockPeer.Task.ContentLength.Store(129)
+				mockPeer.Task.StorePiece(&base.PieceInfo{
+					PieceNum: 0,
+				})
+				mockPeer.Task.TotalPieceCount.Store(1)
+				mockPeer.FSM.SetState(resource.PeerStatePending)
+				mockCDNPeer.FSM.SetState(resource.PeerStateRunning)
+
+				gomock.InOrder(
+					mr.TaskManager().Return(taskManager).Times(1),
+					mt.LoadOrStore(gomock.Any()).Return(mockPeer.Task, true).Times(1),
+					mr.HostManager().Return(hostManager).Times(1),
+					mh.Load(gomock.Eq(mockPeer.Host.ID)).Return(mockPeer.Host, true).Times(1),
+					mr.PeerManager().Return(peerManager).Times(1),
+					mp.LoadOrStore(gomock.Any()).Return(mockPeer, true).Times(1),
+					ms.FindParent(gomock.Any(), gomock.Any(), gomock.Any()).Return(mockCDNPeer, true).Times(1),
+				)
+			},
+			expect: func(t *testing.T, peer *resource.Peer, result *rpcscheduler.RegisterResult, err error) {
+				assert := assert.New(t)
+				assert.NoError(err)
+				assert.Equal(result.TaskId, peer.Task.ID)
+				assert.Equal(result.SizeScope, base.SizeScope_NORMAL)
+				assert.True(peer.FSM.Is(resource.PeerStateReceivedNormal))
+			},
+		},
 		{
 			name: "task scope size is SizeScope_SMALL and load piece error, peer state is PeerStateFailed",
 			req: &rpcscheduler.PeerTaskRequest{
@@ -366,6 +406,8 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				})
 				mockPeer.Task.TotalPieceCount.Store(1)
 				mockPeer.FSM.SetState(resource.PeerStateFailed)
+				mockCDNPeer.FSM.SetState(resource.PeerStateSucceeded)
+
 				gomock.InOrder(
 					mr.TaskManager().Return(taskManager).Times(1),
 					mt.LoadOrStore(gomock.Any()).Return(mockPeer.Task, true).Times(1),
@@ -403,6 +445,8 @@ func TestService_RegisterPeerTask(t *testing.T) {
 				})
 				mockPeer.Task.TotalPieceCount.Store(1)
 				mockPeer.FSM.SetState(resource.PeerStateFailed)
+				mockCDNPeer.FSM.SetState(resource.PeerStateSucceeded)
+
 				gomock.InOrder(
 					mr.TaskManager().Return(taskManager).Times(1),
 					mt.LoadOrStore(gomock.Any()).Return(mockPeer.Task, true).Times(1),
@@ -439,6 +483,8 @@ func TestService_RegisterPeerTask(t *testing.T) {
 					PieceNum: 0,
 				})
 				mockPeer.Task.TotalPieceCount.Store(1)
+				mockCDNPeer.FSM.SetState(resource.PeerStateSucceeded)
+
 				gomock.InOrder(
 					mr.TaskManager().Return(taskManager).Times(1),
 					mt.LoadOrStore(gomock.Any()).Return(mockPeer.Task, true).Times(1),