Repository: dragonflyoss/Dragonfly2
Pull request #1277 (2022-04-24 11:12:11)

fix: filter parent condition
Signed-off-by: Gaius <gaius.qi@gmail.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description
- Fix filter parent condition.
<!--- 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: -->

- [x] 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.
- [x] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1277?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1277](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1277?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (6b54855) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/d3180416f9439673aa331b601f626c36ad713734?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (d318041) will **decrease** coverage by `0.03%`.
> The diff coverage is `66.66%`.

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

```diff
@@            Coverage Diff             @@
##             main    #1277      +/-   ##
==========================================
- Coverage   46.07%   46.04%   -0.04%     
==========================================
  Files         135      135              
  Lines       11353    11355       +2     
==========================================
- Hits         5231     5228       -3     
- Misses       5551     5556       +5     
  Partials      571      571              
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `46.04% <66.66%> (-0.04%)` | :arrow_down: |

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/1277?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [scheduler/resource/peer.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1277/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3Jlc291cmNlL3BlZXIuZ28=) | `77.55% <50.00%> (-0.29%)` | :arrow_down: |
| [scheduler/scheduler/scheduler.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1277/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3NjaGVkdWxlci9zY2hlZHVsZXIuZ28=) | `85.35% <100.00%> (ø)` | |
| [cdn/supervisor/progress/progress.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1277/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2RuL3N1cGVydmlzb3IvcHJvZ3Jlc3MvcHJvZ3Jlc3MuZ28=) | `90.62% <0.00%> (-6.25%)` | :arrow_down: |
| [client/daemon/peer/peertask\_conductor.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1277/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2NvbmR1Y3Rvci5nbw==) | `58.26% <0.00%> (ø)` | |


Patch:
diff --git a/scheduler/resource/peer.go b/scheduler/resource/peer.go
index 9b5cfbdb..cf9a79be 100644
--- a/scheduler/resource/peer.go
+++ b/scheduler/resource/peer.go
@@ -150,8 +150,19 @@ type Peer struct {
 	BlockPeers set.SafeSet
 
 	// NeedBackToSource needs downloaded from source
+	//
+	// When peer is registering, at the same time,
+	// scheduler needs to create the new corresponding task and the cdn is disabled,
+	// NeedBackToSource is set to true
 	NeedBackToSource *atomic.Bool
 
+	// IsBackToSource is downloaded from source
+	//
+	// When peer is scheduling and NeedBackToSource is true,
+	// scheduler needs to return Code_SchedNeedBackSource and
+	// IsBackToSource is set to true
+	IsBackToSource *atomic.Bool
+
 	// CreateAt is peer create time
 	CreateAt *atomic.Time
 
@@ -181,6 +192,7 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
 		StealPeers:       set.NewSafeSet(),
 		BlockPeers:       set.NewSafeSet(),
 		NeedBackToSource: atomic.NewBool(false),
+		IsBackToSource:   atomic.NewBool(false),
 		CreateAt:         atomic.NewTime(time.Now()),
 		UpdateAt:         atomic.NewTime(time.Now()),
 		mu:               &sync.RWMutex{},
@@ -226,6 +238,7 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
 				p.Log.Infof("peer state is %s", e.FSM.Current())
 			},
 			PeerEventDownloadFromBackToSource: func(e *fsm.Event) {
+				p.IsBackToSource.Store(true)
 				p.Task.BackToSourcePeers.Add(p)
 				p.DeleteParent()
 				p.Host.DeletePeer(p.ID)
diff --git a/scheduler/scheduler/scheduler.go b/scheduler/scheduler/scheduler.go
index 7be79dbc..8f0daa2e 100644
--- a/scheduler/scheduler/scheduler.go
+++ b/scheduler/scheduler/scheduler.go
@@ -159,19 +159,19 @@ func (s *scheduler) NotifyAndFindParent(ctx context.Context, peer *resource.Peer
 		return []*resource.Peer{}, false
 	}
 
-	// Find the parent that can be scheduled
-	parents := s.filterParents(peer, blocklist)
-	if len(parents) == 0 {
-		peer.Log.Info("can not find parents")
+	// Find the candidate parent that can be scheduled
+	candidateParents := s.filterCandidateParents(peer, blocklist)
+	if len(candidateParents) == 0 {
+		peer.Log.Info("can not find candidate parents")
 		return []*resource.Peer{}, false
 	}
 
-	// Sort parents by evaluation score
+	// Sort candidate parents by evaluation score
 	taskTotalPieceCount := peer.Task.TotalPieceCount.Load()
 	sort.Slice(
-		parents,
+		candidateParents,
 		func(i, j int) bool {
-			return s.evaluator.Evaluate(parents[i], peer, taskTotalPieceCount) > s.evaluator.Evaluate(parents[j], peer, taskTotalPieceCount)
+			return s.evaluator.Evaluate(candidateParents[i], peer, taskTotalPieceCount) > s.evaluator.Evaluate(candidateParents[j], peer, taskTotalPieceCount)
 		},
 	)
 
@@ -182,56 +182,56 @@ func (s *scheduler) NotifyAndFindParent(ctx context.Context, peer *resource.Peer
 		return []*resource.Peer{}, false
 	}
 
-	if err := stream.Send(constructSuccessPeerPacket(s.dynconfig, peer, parents[0], parents[1:])); err != nil {
+	if err := stream.Send(constructSuccessPeerPacket(s.dynconfig, peer, candidateParents[0], candidateParents[1:])); err != nil {
 		peer.Log.Error(err)
 		return []*resource.Peer{}, false
 	}
 
 	// Add steal peers to current peer
 	peer.StealPeers.Clear()
-	for _, parent := range parents[1:] {
-		peer.StealPeers.Add(parent.ID)
+	for _, candidateParent := range candidateParents[1:] {
+		peer.StealPeers.Add(candidateParent.ID)
 	}
 
 	// Replace peer's parent with scheduled parent
-	peer.ReplaceParent(parents[0])
+	peer.ReplaceParent(candidateParents[0])
 	peer.Log.Infof("schedule parent successful, replace parent to %s and steal peers is %v",
-		parents[0].ID, peer.StealPeers.Values())
+		candidateParents[0].ID, peer.StealPeers.Values())
 	peer.Log.Debugf("peer ancestors is %v", peer.Ancestors())
-	return parents, true
+	return candidateParents, true
 }
 
 // FindParent finds parent that best matches the evaluation
 func (s *scheduler) FindParent(ctx context.Context, peer *resource.Peer, blocklist set.SafeSet) (*resource.Peer, bool) {
-	// Filter the parent that can be scheduled
-	parents := s.filterParents(peer, blocklist)
-	if len(parents) == 0 {
-		peer.Log.Info("can not find parents")
+	// Filter the candidate parent that can be scheduled
+	candidateParents := s.filterCandidateParents(peer, blocklist)
+	if len(candidateParents) == 0 {
+		peer.Log.Info("can not find candidate parents")
 		return nil, false
 	}
 
-	// Sort parents by evaluation score
+	// Sort candidate parents by evaluation score
 	taskTotalPieceCount := peer.Task.TotalPieceCount.Load()
 	sort.Slice(
-		parents,
+		candidateParents,
 		func(i, j int) bool {
-			return s.evaluator.Evaluate(parents[i], peer, taskTotalPieceCount) > s.evaluator.Evaluate(parents[j], peer, taskTotalPieceCount)
+			return s.evaluator.Evaluate(candidateParents[i], peer, taskTotalPieceCount) > s.evaluator.Evaluate(candidateParents[j], peer, taskTotalPieceCount)
 		},
 	)
 
-	peer.Log.Infof("find parent %s successful", parents[0].ID)
-	return parents[0], true
+	peer.Log.Infof("find parent %s successful", candidateParents[0].ID)
+	return candidateParents[0], true
 }
 
-// Filter the parent that can be scheduled
-func (s *scheduler) filterParents(peer *resource.Peer, blocklist set.SafeSet) []*resource.Peer {
+// Filter the candidate parent that can be scheduled
+func (s *scheduler) filterCandidateParents(peer *resource.Peer, blocklist set.SafeSet) []*resource.Peer {
 	filterParentLimit := config.DefaultSchedulerFilterParentLimit
 	if config, ok := s.dynconfig.GetSchedulerClusterConfig(); ok && filterParentLimit > 0 {
 		filterParentLimit = int(config.FilterParentLimit)
 	}
 
-	var parents []*resource.Peer
-	var parentIDs []string
+	var candidateParents []*resource.Peer
+	var candidateParentIDs []string
 	var n int
 	peer.Task.Peers.Range(func(_, value interface{}) bool {
 		if n > filterParentLimit {
@@ -239,70 +239,82 @@ func (s *scheduler) filterParents(peer *resource.Peer, blocklist set.SafeSet) []
 		}
 		n++
 
-		parent, ok := value.(*resource.Peer)
+		candidateParent, ok := value.(*resource.Peer)
 		if !ok {
 			return true
 		}
 
-		if blocklist.Contains(parent.ID) {
-			peer.Log.Debugf("parent %s is not selected because it is in blocklist", parent.ID)
+		// Candidate parent is in blocklist
+		if blocklist.Contains(candidateParent.ID) {
+			peer.Log.Debugf("candidate parent %s is not selected because it is in blocklist", candidateParent.ID)
 			return true
 		}
 
-		if parent.ID == peer.ID {
-			peer.Log.Debug("parent is not selected because it is same")
+		// Candidate parent is itself
+		if candidateParent.ID == peer.ID {
+			peer.Log.Debug("candidate parent is not selected because it is same")
 			return true
 		}
 
-		if s.evaluator.IsBadNode(parent) {
-			peer.Log.Debugf("parent %s is not selected because it is bad node", parent.ID)
+		// Candidate parent is bad node
+		if s.evaluator.IsBadNode(candidateParent) {
+			peer.Log.Debugf("candidate parent %s is not selected because it is bad node", candidateParent.ID)
 			return true
 		}
 
-		_, ok = parent.LoadParent()
-		isBackToSource := peer.Task.BackToSourcePeers.Contains(parent)
-		if !ok && !parent.Host.IsCDN && !isBackToSource {
-			peer.Log.Debugf("parent %s is not selected, because its download state is %t %t %t",
-				parent.ID, ok, parent.Host.IsCDN, isBackToSource)
+		// Conditions for candidate parent to be a parent:
+		// 1. candidate parent has parent
+		// 2. candidate parent is CDN
+		// 3. candidate parent has been back-to-source
+		_, ok = candidateParent.LoadParent()
+		isBackToSource := candidateParent.IsBackToSource.Load()
+		if !ok && !candidateParent.Host.IsCDN && !isBackToSource {
+			peer.Log.Debugf("candidate parent %s is not selected, because its download state is %t %t %t",
+				candidateParent.ID, ok, candidateParent.Host.IsCDN, isBackToSource)
 			return true
 		}
 
+		// Candidate parent's depth exceeds available depth
 		peerChildCount := peer.ChildCount.Load()
-		parentDepth := parent.Depth()
+		parentDepth := candidateParent.Depth()
 		if peerChildCount > 0 && parentDepth > defaultAvailableDepth {
-			peer.Log.Debugf("peer has %d children and parent %s depth is %d", peerChildCount, parent.ID, parentDepth)
+			peer.Log.Debugf("candidate peer has %d children and parent %s depth is %d", peerChildCount, candidateParent.ID, parentDepth)
 			return true
 		}
 
+		// Peer's depth exceeds limit depth
 		peerDepth := peer.Depth()
 		if parentDepth+peerDepth > defaultDepthLimit {
-			peer.Log.Debugf("exceeds the %d depth limit of the tree, peer depth is %d, parent %s is %d", defaultDepthLimit, peerDepth, parent.ID, parentDepth)
+			peer.Log.Debugf("exceeds the %d depth limit of the tree, peer depth is %d, candidate parent %s is %d", defaultDepthLimit, peerDepth, candidateParent.ID, parentDepth)
 			return true
 		}
 
-		if parent.IsDescendant(peer) {
-			peer.Log.Debugf("parent %s is not selected because it is descendant", parent.ID)
+		// Candidate parent is an descendant of peer
+		if candidateParent.IsDescendant(peer) {
+			peer.Log.Debugf("candidate parent %s is not selected because it is descendant", candidateParent.ID)
 			return true
 		}
 
-		if parent.IsAncestor(peer) {
-			peer.Log.Debugf("parent %s is not selected because it is ancestor", parent.ID)
+		// Candidate parent is an ancestor of peer
+		if candidateParent.IsAncestor(peer) {
+			peer.Log.Debugf("candidate parent %s is not selected because it is ancestor", candidateParent.ID)
 			return true
 		}
 
-		if parent.Host.FreeUploadLoad() <= 0 {
-			peer.Log.Debugf("parent %s is not selected because its free upload is empty, upload limit is %d, upload peer count is %d",
-				parent.ID, parent.Host.UploadLoadLimit.Load(), parent.Host.UploadPeerCount.Load())
+		// Candidate parent's free upload is empty
+		if candidateParent.Host.FreeUploadLoad() <= 0 {
+			peer.Log.Debugf("candidate parent %s is not selected because its free upload is empty, upload limit is %d, upload peer count is %d",
+				candidateParent.ID, candidateParent.Host.UploadLoadLimit.Load(), candidateParent.Host.UploadPeerCount.Load())
 			return true
 		}
 
-		parents = append(parents, parent)
-		parentIDs = append(parentIDs, parent.ID)
+		candidateParents = append(candidateParents, candidateParent)
+		candidateParentIDs = append(candidateParentIDs, candidateParent.ID)
 		return true
 	})
 
-	peer.Log.Infof("candidate parents include %#v", parentIDs)
-	return parents
+	peer.Log.Infof("candidate parents include %#v", candidateParentIDs)
+	return candidateParents
 }
 
 // Construct peer successful packet


Test patch:
diff --git a/scheduler/scheduler/scheduler_test.go b/scheduler/scheduler/scheduler_test.go
index 494645cf..cc1684ab 100644
--- a/scheduler/scheduler/scheduler_test.go
+++ b/scheduler/scheduler/scheduler_test.go
@@ -671,6 +671,7 @@ func TestScheduler_NotifyAndFindParent(t *testing.T) {
 				peer.FSM.SetState(resource.PeerStateRunning)
 				mockPeer.FSM.SetState(resource.PeerStateRunning)
 				peer.Task.BackToSourcePeers.Add(mockPeer)
+				mockPeer.IsBackToSource.Store(true)
 				peer.Task.StorePeer(mockPeer)
 				mockPeer.Pieces.Set(0)
 				peer.StoreStream(stream)
@@ -700,6 +701,8 @@ func TestScheduler_NotifyAndFindParent(t *testing.T) {
 				peer.Task.StorePeer(stealPeer)
 				peer.Task.BackToSourcePeers.Add(mockPeer)
 				peer.Task.BackToSourcePeers.Add(stealPeer)
+				mockPeer.IsBackToSource.Store(true)
+				stealPeer.IsBackToSource.Store(true)
 				mockPeer.Pieces.Set(0)
 				peer.StoreStream(stream)
 				gomock.InOrder(
@@ -838,6 +841,8 @@ func TestScheduler_FindParent(t *testing.T) {
 				peer.Task.StorePeer(mockPeers[1])
 				peer.Task.BackToSourcePeers.Add(mockPeers[0])
 				peer.Task.BackToSourcePeers.Add(mockPeers[1])
+				mockPeers[0].IsBackToSource.Store(true)
+				mockPeers[1].IsBackToSource.Store(true)
 				mockPeers[0].Pieces.Set(0)
 				mockPeers[1].Pieces.Set(0)
 				mockPeers[1].Pieces.Set(1)
@@ -907,6 +912,8 @@ func TestScheduler_FindParent(t *testing.T) {
 				peer.Task.StorePeer(mockPeers[1])
 				peer.Task.BackToSourcePeers.Add(mockPeers[0])
 				peer.Task.BackToSourcePeers.Add(mockPeers[1])
+				mockPeers[0].IsBackToSource.Store(true)
+				mockPeers[1].IsBackToSource.Store(true)
 				mockPeers[0].Pieces.Set(0)
 				mockPeers[1].Pieces.Set(0)
 				mockPeers[1].Pieces.Set(1)