Repository: dragonflyoss/Dragonfly2
Pull request #435 (2021-07-08 04:28:31)

Fix: add process level for scheduler peer task status
## Description

scheduler only processes one peer task status at the same time, and will discard others. it is wrong.

we should set status diffrent priority level, while status with higher  level should not be discarded.

## 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! -->

- [ ] 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/435?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#435](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/435?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (65950eb) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/09076481e41f7b68702d527a9c337d017e02a74b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (0907648) will **decrease** coverage by `1.29%`.
> The diff coverage is `70.00%`.

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

```diff
@@            Coverage Diff             @@
##             main     #435      +/-   ##
==========================================
- Coverage   38.08%   36.78%   -1.30%     
==========================================
  Files          80       84       +4     
  Lines        6278     6581     +303     
==========================================
+ Hits         2391     2421      +30     
- Misses       3550     3817     +267     
- Partials      337      343       +6     
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `36.78% <70.00%> (-1.30%)` | :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/435?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [scheduler/types/peer\_task.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/435/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3R5cGVzL3BlZXJfdGFzay5nbw==) | `12.55% <70.00%> (ø)` | |
| [client/daemon/peerhost.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/435/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyaG9zdC5nbw==) | `0.00% <0.00%> (ø)` | |
| [scheduler/types/piece.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/435/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3R5cGVzL3BpZWNlLmdv) | `0.00% <0.00%> (ø)` | |
| [scheduler/types/task.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/435/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3R5cGVzL3Rhc2suZ28=) | `0.00% <0.00%> (ø)` | |
| [scheduler/types/host.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/435/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3R5cGVzL2hvc3QuZ28=) | `0.00% <0.00%> (ø)` | |
| [client/daemon/peer/peertask\_stream.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/435/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3N0cmVhbS5nbw==) | `56.82% <0.00%> (+0.73%)` | :arrow_up: |


Patch:
diff --git a/go.mod b/go.mod
index feb031bb..5a8386c4 100644
--- a/go.mod
+++ b/go.mod
@@ -47,7 +47,6 @@ require (
 	go.opentelemetry.io/otel/trace v0.20.0
 	go.uber.org/atomic v1.6.0
 	go.uber.org/zap v1.16.0
-	golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	golang.org/x/sys v0.0.0-20210510120138-977fb7262007
 	golang.org/x/term v0.0.0-20201210144234-2321bbc49cbf
diff --git a/scheduler/service/worker/woker.go b/scheduler/service/worker/woker.go
index 77b92a38..1b048b79 100644
--- a/scheduler/service/worker/woker.go
+++ b/scheduler/service/worker/woker.go
@@ -202,11 +202,11 @@ func (w *Worker) doSchedule(peerTask *types.PeerTask) {
 	case types.PeerTaskStatusAddParent:
 		parent, _ := peerTask.GetJobData().(*types.PeerTask)
 		if parent == nil {
-			peerTask.SetNodeStatus(types.PeerTaskStatusHealth)
+			peerTask.SetNodeStatusHealth(types.PeerTaskStatusAddParent)
 			return
 		}
 		peerTask.AddParent(parent, 1)
-		peerTask.SetNodeStatus(types.PeerTaskStatusHealth)
+		peerTask.SetNodeStatusHealth(types.PeerTaskStatusAddParent)
 		return
 	case types.PeerTaskStatusNeedParent:
 		parent, _, err := w.schedulerService.Scheduler.ScheduleParent(peerTask)
@@ -218,7 +218,7 @@ func (w *Worker) doSchedule(peerTask *types.PeerTask) {
 			w.sendJobLater(peerTask)
 		} else {
 			w.sendScheduleResult(peerTask)
-			peerTask.SetNodeStatus(types.PeerTaskStatusHealth)
+			peerTask.SetNodeStatusHealth(types.PeerTaskStatusNeedParent)
 		}
 		w.schedulerService.TaskManager.PeerTask.RefreshDownloadMonitor(peerTask)
 	case types.PeerTaskStatusNeedChildren:
@@ -235,7 +235,7 @@ func (w *Worker) doSchedule(peerTask *types.PeerTask) {
 				w.sendJob(children[i])
 			}
 		}
-		peerTask.SetNodeStatus(types.PeerTaskStatusHealth)
+		peerTask.SetNodeStatusHealth(types.PeerTaskStatusNeedChildren)
 
 	case types.PeerTaskStatusBadNode:
 		adjustNodes, err := w.schedulerService.Scheduler.ScheduleBadNode(peerTask)
@@ -265,7 +265,7 @@ func (w *Worker) doSchedule(peerTask *types.PeerTask) {
 			return
 		}
 		w.sendScheduleResult(peerTask)
-		peerTask.SetNodeStatus(types.PeerTaskStatusHealth)
+		peerTask.SetNodeStatusHealth(types.PeerTaskStatusNeedAdjustNode)
 
 	case types.PeerTaskStatusNeedCheckNode:
 		if w.schedulerService.Scheduler.IsNodeBad(peerTask) && peerTask.GetSubTreeNodesNum() > 1 {
@@ -284,7 +284,7 @@ func (w *Worker) doSchedule(peerTask *types.PeerTask) {
 					w.sendJob(adjustNodes[i])
 				}
 			}
-			peerTask.SetNodeStatus(types.PeerTaskStatusHealth)
+			peerTask.SetNodeStatusHealth(types.PeerTaskStatusNeedCheckNode)
 		} else if w.schedulerService.Scheduler.NeedAdjustParent(peerTask) {
 			_, _, err := w.schedulerService.Scheduler.ScheduleAdjustParentNode(peerTask)
 			if err != nil {
@@ -292,7 +292,7 @@ func (w *Worker) doSchedule(peerTask *types.PeerTask) {
 				return
 			}
 			w.sendScheduleResult(peerTask)
-			peerTask.SetNodeStatus(types.PeerTaskStatusHealth)
+			peerTask.SetNodeStatusHealth(types.PeerTaskStatusNeedCheckNode)
 		}
 
 	case types.PeerTaskStatusDone:
@@ -306,6 +306,7 @@ func (w *Worker) doSchedule(peerTask *types.PeerTask) {
 			parent.SetNodeStatus(types.PeerTaskStatusNeedChildren)
 			w.sendJob(parent)
 		}
+		peerTask.SetNodeStatusHealth(types.PeerTaskStatusDone)
 
 	case types.PeerTaskStatusLeaveNode, types.PeerTaskStatusNodeGone:
 		adjustNodes, err := w.schedulerService.Scheduler.ScheduleLeaveNode(peerTask)
diff --git a/scheduler/types/peer_task.go b/scheduler/types/peer_task.go
index c158cf98..dc7b52c4 100644
--- a/scheduler/types/peer_task.go
+++ b/scheduler/types/peer_task.go
@@ -32,15 +32,15 @@ type PeerTaskStatus int8
 
 const (
 	PeerTaskStatusHealth         PeerTaskStatus = 0
-	PeerTaskStatusNeedParent     PeerTaskStatus = 1
-	PeerTaskStatusNeedChildren   PeerTaskStatus = 2
-	PeerTaskStatusBadNode        PeerTaskStatus = 3
-	PeerTaskStatusNeedAdjustNode PeerTaskStatus = 4
-	PeerTaskStatusNeedCheckNode  PeerTaskStatus = 5
-	PeerTaskStatusDone           PeerTaskStatus = 6
-	PeerTaskStatusLeaveNode      PeerTaskStatus = 7
-	PeerTaskStatusAddParent      PeerTaskStatus = 8
-	PeerTaskStatusNodeGone       PeerTaskStatus = 9
+	PeerTaskStatusBadNode        PeerTaskStatus = 11
+	PeerTaskStatusNodeGone       PeerTaskStatus = 12
+	PeerTaskStatusNeedCheckNode  PeerTaskStatus = 13
+	PeerTaskStatusNeedAdjustNode PeerTaskStatus = 21
+	PeerTaskStatusNeedChildren   PeerTaskStatus = 22
+	PeerTaskStatusNeedParent     PeerTaskStatus = 23
+	PeerTaskStatusAddParent      PeerTaskStatus = 24
+	PeerTaskStatusDone           PeerTaskStatus = 31
+	PeerTaskStatusLeaveNode      PeerTaskStatus = 32
 )
 
 type PeerTask struct {
@@ -433,6 +433,10 @@ func (pt *PeerTask) GetJobData() interface{} {
 func (pt *PeerTask) SetNodeStatus(status PeerTaskStatus, data ...interface{}) {
 	pt.lock.Lock()
 	defer pt.lock.Unlock()
+	s := pt.status
+	if status/10 < s/10 {
+		return
+	}
 	pt.status = status
 	if len(data) > 0 {
 		pt.jobData = data[0]
@@ -441,6 +445,18 @@ func (pt *PeerTask) SetNodeStatus(status PeerTaskStatus, data ...interface{}) {
 	}
 }
 
+func (pt *PeerTask) SetNodeStatusHealth(swapStatus ...PeerTaskStatus) {
+	pt.lock.Lock()
+	defer pt.lock.Unlock()
+	if len(swapStatus) == 0 {
+		pt.status = PeerTaskStatusHealth
+		return
+	}
+	if pt.status == swapStatus[0] {
+		pt.status = PeerTaskStatusHealth
+	}
+}
+
 func (pt *PeerTask) IsWaiting() bool {
 	if pt == nil || pt.parent == nil || pt.parent.DstPeerTask == nil {
 		return false


Test patch:
diff --git a/scheduler/types/peer_task_test.go b/scheduler/types/peer_task_test.go
new file mode 100644
index 00000000..731a04bd
--- /dev/null
+++ b/scheduler/types/peer_task_test.go
@@ -0,0 +1,74 @@
+package types
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPeerTask_SetStatus(t *testing.T) {
+	tests := []struct {
+		name         string
+		oldStatus    PeerTaskStatus
+		newStatus    PeerTaskStatus
+		swapStatus   PeerTaskStatus
+		expectStatus PeerTaskStatus
+	}{
+		{
+			name:         "accept change status 1",
+			oldStatus:    PeerTaskStatusHealth,
+			newStatus:    PeerTaskStatusBadNode,
+			expectStatus: PeerTaskStatusBadNode,
+		}, {
+			name:         "accept change status 2",
+			oldStatus:    PeerTaskStatusBadNode,
+			newStatus:    PeerTaskStatusNodeGone,
+			expectStatus: PeerTaskStatusNodeGone,
+		}, {
+			name:         "accept change status 3",
+			oldStatus:    PeerTaskStatusNeedCheckNode,
+			newStatus:    PeerTaskStatusNeedAdjustNode,
+			expectStatus: PeerTaskStatusNeedAdjustNode,
+		}, {
+			name:         "accept change status 4",
+			oldStatus:    PeerTaskStatusNodeGone,
+			newStatus:    PeerTaskStatusDone,
+			expectStatus: PeerTaskStatusDone,
+		}, {
+			name:         "reject change status 1",
+			oldStatus:    PeerTaskStatusDone,
+			newStatus:    PeerTaskStatusNodeGone,
+			expectStatus: PeerTaskStatusDone,
+		}, {
+			name:         "reject change status 2",
+			oldStatus:    PeerTaskStatusNeedChildren,
+			newStatus:    PeerTaskStatusNodeGone,
+			expectStatus: PeerTaskStatusNeedChildren,
+		}, {
+			name:         "accept change status health",
+			oldStatus:    PeerTaskStatusNeedChildren,
+			newStatus:    PeerTaskStatusHealth,
+			swapStatus:   PeerTaskStatusNeedChildren,
+			expectStatus: PeerTaskStatusHealth,
+		}, {
+			name:         "reject change status health",
+			oldStatus:    PeerTaskStatusNeedChildren,
+			newStatus:    PeerTaskStatusHealth,
+			swapStatus:   PeerTaskStatusNodeGone,
+			expectStatus: PeerTaskStatusNeedChildren,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			mockPT := NewPeerTask("test", nil, nil, func(task *PeerTask) {})
+			mockPT.status = tc.oldStatus
+			if tc.newStatus == PeerTaskStatusHealth {
+				mockPT.SetNodeStatusHealth(tc.swapStatus)
+			} else {
+				mockPT.SetNodeStatus(tc.newStatus)
+			}
+			assert.New(t).Equal(tc.expectStatus, mockPT.GetNodeStatus())
+		})
+	}
+}