Repository: dragonflyoss/Dragonfly2
Pull request #1211 (2022-03-30 09:58:29)

feat: if the number of failed peers in the task is greater than FailedPeerCountLimit, then scheduler notifies running peers of failure
if the number of failed peers in the task is greater than FailedPeerCountLimit, then scheduler notifies running peers of failure.

Signed-off-by: Gaius <gaius.qi@gmail.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description
- If the number of failed peers in the task is greater than FailedPeerCountLimit, then scheduler notifies running peers of failure.
<!--- 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.



Patch:
diff --git a/manager/rpcserver/rpcserver.go b/manager/rpcserver/rpcserver.go
index d462d9ce..54c0f914 100644
--- a/manager/rpcserver/rpcserver.go
+++ b/manager/rpcserver/rpcserver.go
@@ -578,6 +578,7 @@ func (s *Server) KeepAlive(stream manager.Manager_KeepAliveServer) error {
 				logger.Infof("%s close keepalive in cluster %d", hostName, clusterID)
 				return nil
 			}
+
 			logger.Errorf("%s keepalive failed in cluster %d: %v", hostName, clusterID, err)
 			return status.Error(codes.Unknown, err.Error())
 		}
diff --git a/scheduler/resource/peer.go b/scheduler/resource/peer.go
index 3d749e9f..b5fab2b8 100644
--- a/scheduler/resource/peer.go
+++ b/scheduler/resource/peer.go
@@ -231,11 +231,13 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
 
 				p.DeleteParent()
 				p.Host.DeletePeer(p.ID)
+				p.Task.PeerFailedCount.Store(0)
 				p.UpdateAt.Store(time.Now())
 				p.Log.Infof("peer state is %s", e.FSM.Current())
 			},
 			PeerEventDownloadFailed: func(e *fsm.Event) {
 				if e.Src == PeerStateBackToSource {
+					p.Task.PeerFailedCount.Inc()
 					p.Task.BackToSourcePeers.Delete(p)
 				}
 
diff --git a/scheduler/resource/task.go b/scheduler/resource/task.go
index 90b75754..60f48783 100644
--- a/scheduler/resource/task.go
+++ b/scheduler/resource/task.go
@@ -27,11 +27,15 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/container/set"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
+	rpcscheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
 )
 
 const (
 	// Tiny file size is 128 bytes
 	TinyFileSize = 128
+
+	// Peer failure limit in task
+	FailedPeerCountLimit = 200
 )
 
 const (
@@ -96,6 +100,10 @@ type Task struct {
 	// PeerCount is peer count
 	PeerCount *atomic.Int32
 
+	// PeerFailedCount is peer failed count,
+	// if one peer succeeds, the value is reset to zero
+	PeerFailedCount *atomic.Int32
+
 	// CreateAt is task create time
 	CreateAt *atomic.Time
 
@@ -119,6 +127,7 @@ func NewTask(id, url string, backToSourceLimit int, meta *base.UrlMeta) *Task {
 		Pieces:            &sync.Map{},
 		Peers:             &sync.Map{},
 		PeerCount:         atomic.NewInt32(0),
+		PeerFailedCount:   atomic.NewInt32(0),
 		CreateAt:          atomic.NewTime(time.Now()),
 		UpdateAt:          atomic.NewTime(time.Now()),
 		Log:               logger.WithTaskIDAndURL(id, url),
@@ -281,3 +290,29 @@ func (t *Task) SizeScope() base.SizeScope {
 func (t *Task) CanBackToSource() bool {
 	return int32(t.BackToSourcePeers.Len()) < t.BackToSourceLimit.Load()
 }
+
+// NotifyPeers notify all peers in the task with the state code
+func (t *Task) NotifyPeers(code base.Code, event string) {
+	t.Peers.Range(func(_, value interface{}) bool {
+		peer := value.(*Peer)
+		if peer.FSM.Is(PeerStateRunning) {
+			stream, ok := peer.LoadStream()
+			if !ok {
+				return true
+			}
+
+			if err := stream.Send(&rpcscheduler.PeerPacket{Code: code}); err != nil {
+				t.Log.Errorf("send packet to peer %s failed: %v", peer.ID, err)
+				return true
+			}
+			t.Log.Infof("task notify peer %s code %s", peer.ID, code)
+
+			if err := peer.FSM.Event(event); err != nil {
+				peer.Log.Errorf("peer fsm event failed: %v", err)
+				return true
+			}
+		}
+
+		return true
+	})
+}
diff --git a/scheduler/service/service.go b/scheduler/service/service.go
index 0496539a..f50e08d6 100644
--- a/scheduler/service/service.go
+++ b/scheduler/service/service.go
@@ -624,6 +624,13 @@ func (s *Service) handleTaskSuccess(ctx context.Context, task *resource.Task, re
 // 1. CDN downloads the resource falied
 // 2. Dfdaemon back-to-source to download failed
 func (s *Service) handleTaskFail(ctx context.Context, task *resource.Task) {
+	// If the number of failed peers in the task is greater than FailedPeerCountLimit,
+	// then scheduler notifies running peers of failure
+	if task.PeerFailedCount.Load() > resource.FailedPeerCountLimit {
+		task.PeerFailedCount.Store(0)
+		task.NotifyPeers(base.Code_SchedTaskStatusError, resource.PeerEventDownloadFailed)
+	}
+
 	if task.FSM.Is(resource.TaskStateFailed) {
 		return
 	}


Test patch:
diff --git a/scheduler/resource/task_test.go b/scheduler/resource/task_test.go
index d0ca9d8b..840fe2f5 100644
--- a/scheduler/resource/task_test.go
+++ b/scheduler/resource/task_test.go
@@ -17,13 +17,17 @@
 package resource
 
 import (
+	"errors"
 	"testing"
 	"time"
 
+	gomock "github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 
 	"d7y.io/dragonfly/v2/pkg/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
+	rpcscheduler "d7y.io/dragonfly/v2/pkg/rpc/scheduler"
+	rpcschedulermocks "d7y.io/dragonfly/v2/pkg/rpc/scheduler/mocks"
 )
 
 var (
@@ -749,3 +753,84 @@ func TestTask_CanBackToSource(t *testing.T) {
 		})
 	}
 }
+
+func TestTask_NotifyPeers(t *testing.T) {
+	tests := []struct {
+		name string
+		run  func(t *testing.T, task *Task, mockPeer *Peer, stream rpcscheduler.Scheduler_ReportPieceResultServer, ms *rpcschedulermocks.MockScheduler_ReportPieceResultServerMockRecorder)
+	}{
+		{
+			name: "peer state is PeerStatePending",
+			run: func(t *testing.T, task *Task, mockPeer *Peer, stream rpcscheduler.Scheduler_ReportPieceResultServer, ms *rpcschedulermocks.MockScheduler_ReportPieceResultServerMockRecorder) {
+				mockPeer.FSM.SetState(PeerStatePending)
+				task.NotifyPeers(base.Code_SchedTaskStatusError, PeerEventDownloadFailed)
+
+				assert := assert.New(t)
+				assert.True(mockPeer.FSM.Is(PeerStatePending))
+			},
+		},
+		{
+			name: "peer state is PeerStateRunning and stream is empty",
+			run: func(t *testing.T, task *Task, mockPeer *Peer, stream rpcscheduler.Scheduler_ReportPieceResultServer, ms *rpcschedulermocks.MockScheduler_ReportPieceResultServerMockRecorder) {
+				mockPeer.FSM.SetState(PeerStateRunning)
+				task.NotifyPeers(base.Code_SchedTaskStatusError, PeerEventDownloadFailed)
+
+				assert := assert.New(t)
+				assert.True(mockPeer.FSM.Is(PeerStateRunning))
+			},
+		},
+		{
+			name: "peer state is PeerStateRunning and stream sending failed",
+			run: func(t *testing.T, task *Task, mockPeer *Peer, stream rpcscheduler.Scheduler_ReportPieceResultServer, ms *rpcschedulermocks.MockScheduler_ReportPieceResultServerMockRecorder) {
+				mockPeer.FSM.SetState(PeerStateRunning)
+				mockPeer.StoreStream(stream)
+				ms.Send(gomock.Eq(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError})).Return(errors.New("foo")).Times(1)
+
+				task.NotifyPeers(base.Code_SchedTaskStatusError, PeerEventDownloadFailed)
+
+				assert := assert.New(t)
+				assert.True(mockPeer.FSM.Is(PeerStateRunning))
+			},
+		},
+		{
+			name: "peer state is PeerStateRunning and state changing failed",
+			run: func(t *testing.T, task *Task, mockPeer *Peer, stream rpcscheduler.Scheduler_ReportPieceResultServer, ms *rpcschedulermocks.MockScheduler_ReportPieceResultServerMockRecorder) {
+				mockPeer.FSM.SetState(PeerStateRunning)
+				mockPeer.StoreStream(stream)
+				ms.Send(gomock.Eq(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError})).Return(errors.New("foo")).Times(1)
+
+				task.NotifyPeers(base.Code_SchedTaskStatusError, PeerEventRegisterNormal)
+
+				assert := assert.New(t)
+				assert.True(mockPeer.FSM.Is(PeerStateRunning))
+			},
+		},
+		{
+			name: "peer state is PeerStateRunning and notify peer successfully",
+			run: func(t *testing.T, task *Task, mockPeer *Peer, stream rpcscheduler.Scheduler_ReportPieceResultServer, ms *rpcschedulermocks.MockScheduler_ReportPieceResultServerMockRecorder) {
+				mockPeer.FSM.SetState(PeerStateRunning)
+				mockPeer.StoreStream(stream)
+				ms.Send(gomock.Eq(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError})).Return(nil).Times(1)
+
+				task.NotifyPeers(base.Code_SchedTaskStatusError, PeerEventDownloadFailed)
+
+				assert := assert.New(t)
+				assert.True(mockPeer.FSM.Is(PeerStateFailed))
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctl := gomock.NewController(t)
+			defer ctl.Finish()
+			stream := rpcschedulermocks.NewMockScheduler_ReportPieceResultServer(ctl)
+
+			mockHost := NewHost(mockRawHost)
+			task := NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
+			mockPeer := NewPeer(mockPeerID, task, mockHost)
+			task.StorePeer(mockPeer)
+			tc.run(t, task, mockPeer, stream, stream.EXPECT())
+		})
+	}
+}
diff --git a/scheduler/service/service_test.go b/scheduler/service/service_test.go
index c6d4fa7f..e9650766 100644
--- a/scheduler/service/service_test.go
+++ b/scheduler/service/service_test.go
@@ -2522,6 +2522,18 @@ func TestService_handleTaskFail(t *testing.T) {
 				assert.True(task.FSM.Is(resource.TaskStateFailed))
 			},
 		},
+		{
+			name: "number of failed peers in the task is greater than FailedPeerCountLimit",
+			mock: func(task *resource.Task) {
+				task.FSM.SetState(resource.TaskStateFailed)
+				task.PeerFailedCount.Store(201)
+			},
+			expect: func(t *testing.T, task *resource.Task) {
+				assert := assert.New(t)
+				assert.True(task.FSM.Is(resource.TaskStateFailed))
+				assert.Equal(task.PeerFailedCount.Load(), int32(0))
+			},
+		},
 	}
 
 	for _, tc := range tests {