dragonflyoss
/
dragonfly
mirror of https://github.com/dragonflyoss/dragonfly.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: update fsm stable api (#1938) 

Signed-off-by: Gaius <gaius.qi@gmail.com>
This commit is contained in:
Gaius 2022-12-20 14:28:56 +08:00
parent 41a762c2a4
commit d01a742b6f
No known key found for this signature in database
GPG Key ID: 8B4E5D1290FA2FFB
10 changed files with 54 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
go.mod
View File

@ -42,7 +42,7 @@ require (
github.com/jarcoal/httpmock v1.2.0
github.com/johanbrandhorst/certify v1.9.0
github.com/juju/ratelimit v1.0.2
github.com/looplab/fsm v0.3.0
github.com/looplab/fsm v1.0.0
github.com/mcuadros/go-gin-prometheus v0.1.0
github.com/mdlayher/vsock v1.2.0
github.com/mitchellh/mapstructure v1.5.0

4
go.sum
View File

@ -752,8 +752,8 @@ github.com/lib/pq v1.10.2 h1:AqzbZs4ZoCBp+GtejcpCpcxM3zlSMx29dXbUSeVtJb8=
github.com/lib/pq v1.10.2/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
github.com/lightstep/lightstep-tracer-common/golang/gogo v0.0.0-20190605223551-bc2310a04743/go.mod h1:qklhhLq1aX+mtWk9cPHPzaBjWImj5ULL6C7HFJtXQMM=
github.com/lightstep/lightstep-tracer-go v0.18.1/go.mod h1:jlF1pusYV4pidLvZ+XD0UBX0ZE6WURAspgAczcDHrL4=
github.com/looplab/fsm v0.3.0 h1:kIgNS3Yyud1tyxhG8kDqh853B7QqwnlWdgL3TD2s3Sw=
github.com/looplab/fsm v0.3.0/go.mod h1:PmD3fFvQEIsjMEfvZdrCDZ6y8VwKTwWNjlpEr6IKPO4=
github.com/looplab/fsm v1.0.0 h1:dyD2Nf6FforIQt7BHBTSX7Vk4UcVNJ++ouCEgU1i3wA=
github.com/looplab/fsm v1.0.0/go.mod h1:PmD3fFvQEIsjMEfvZdrCDZ6y8VwKTwWNjlpEr6IKPO4=
github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4=
github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
github.com/lyft/protoc-gen-validate v0.0.13/go.mod h1:XbGvPuh87YZc5TdIa2/I4pLk0QoUACkjt2znoq26NVQ=

3
scheduler/resource/host.go
View File

@ -17,6 +17,7 @@
package resource
import (
"context"
"sync"
"time"
@ -187,7 +188,7 @@ func (h *Host) LeavePeers() {
}
peer.Log.Info("host leaves peers, causing the peer to leave")
if err := peer.FSM.Event(PeerEventLeave); err != nil {
if err := peer.FSM.Event(context.Background(), PeerEventLeave); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
return true
}

20
scheduler/resource/peer.go
View File

@ -242,27 +242,27 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
}, Dst: PeerStateLeave},
},
fsm.Callbacks{
PeerEventRegisterEmpty: func(e *fsm.Event) {
PeerEventRegisterEmpty: func(ctx context.Context, e *fsm.Event) {
p.UpdatedAt.Store(time.Now())
p.Log.Infof("peer state is %s", e.FSM.Current())
},
PeerEventRegisterTiny: func(e *fsm.Event) {
PeerEventRegisterTiny: func(ctx context.Context, e *fsm.Event) {
p.UpdatedAt.Store(time.Now())
p.Log.Infof("peer state is %s", e.FSM.Current())
},
PeerEventRegisterSmall: func(e *fsm.Event) {
PeerEventRegisterSmall: func(ctx context.Context, e *fsm.Event) {
p.UpdatedAt.Store(time.Now())
p.Log.Infof("peer state is %s", e.FSM.Current())
},
PeerEventRegisterNormal: func(e *fsm.Event) {
PeerEventRegisterNormal: func(ctx context.Context, e *fsm.Event) {
p.UpdatedAt.Store(time.Now())
p.Log.Infof("peer state is %s", e.FSM.Current())
},
PeerEventDownload: func(e *fsm.Event) {
PeerEventDownload: func(ctx context.Context, e *fsm.Event) {
p.UpdatedAt.Store(time.Now())
p.Log.Infof("peer state is %s", e.FSM.Current())
},
PeerEventDownloadBackToSource: func(e *fsm.Event) {
PeerEventDownloadBackToSource: func(ctx context.Context, e *fsm.Event) {
p.IsBackToSource.Store(true)
p.Task.BackToSourcePeers.Add(p.ID)
@ -273,7 +273,7 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
p.UpdatedAt.Store(time.Now())
p.Log.Infof("peer state is %s", e.FSM.Current())
},
PeerEventDownloadSucceeded: func(e *fsm.Event) {
PeerEventDownloadSucceeded: func(ctx context.Context, e *fsm.Event) {
if e.Src == PeerStateBackToSource {
p.Task.BackToSourcePeers.Delete(p.ID)
}
@ -286,7 +286,7 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
p.UpdatedAt.Store(time.Now())
p.Log.Infof("peer state is %s", e.FSM.Current())
},
PeerEventDownloadFailed: func(e *fsm.Event) {
PeerEventDownloadFailed: func(ctx context.Context, e *fsm.Event) {
if e.Src == PeerStateBackToSource {
p.Task.PeerFailedCount.Inc()
p.Task.BackToSourcePeers.Delete(p.ID)
@ -299,7 +299,7 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
p.UpdatedAt.Store(time.Now())
p.Log.Infof("peer state is %s", e.FSM.Current())
},
PeerEventLeave: func(e *fsm.Event) {
PeerEventLeave: func(ctx context.Context, e *fsm.Event) {
if err := p.Task.DeletePeerInEdges(p.ID); err != nil {
p.Log.Errorf("delete peer inedges failed: %s", err.Error())
}
@ -428,7 +428,7 @@ func (p *Peer) DownloadTinyFile() ([]byte, error) {
func (p *Peer) GetPriority(dynconfig config.DynconfigInterface) commonv1.Priority {
pbApplications, err := dynconfig.GetApplications()
if err != nil {
p.Log.Warn(err)
p.Log.Info(err)
return commonv1.Priority_LEVEL0
}

9
scheduler/resource/peer_manager.go
View File

@ -19,6 +19,7 @@
package resource
import (
"context"
"sync"
"time"
@ -149,7 +150,7 @@ func (p *peerManager) RunGC() error {
elapsed := time.Since(peer.PieceUpdatedAt.Load())
if elapsed > p.pieceDownloadTimeout {
peer.Log.Info("peer elapsed exceeds the timeout of downloading piece, causing the peer to leave")
if err := peer.FSM.Event(PeerEventLeave); err != nil {
if err := peer.FSM.Event(context.Background(), PeerEventLeave); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
return true
}
@ -163,7 +164,7 @@ func (p *peerManager) RunGC() error {
elapsed := time.Since(peer.UpdatedAt.Load())
if elapsed > p.ttl {
peer.Log.Info("peer elapsed exceeds the ttl, causing the peer to leave")
if err := peer.FSM.Event(PeerEventLeave); err != nil {
if err := peer.FSM.Event(context.Background(), PeerEventLeave); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
return true
}
@ -175,7 +176,7 @@ func (p *peerManager) RunGC() error {
// then set the peer state to PeerStateLeave and then delete peer.
if peer.FSM.Is(PeerStateFailed) {
peer.Log.Info("peer state is PeerStateFailed, causing the peer to leave")
if err := peer.FSM.Event(PeerEventLeave); err != nil {
if err := peer.FSM.Event(context.Background(), PeerEventLeave); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
return true
}
@ -196,7 +197,7 @@ func (p *peerManager) RunGC() error {
if peer.Task.PeerCount() > PeerCountLimitForTask &&
peer.FSM.Is(PeerStateSucceeded) && degree == 0 {
peer.Log.Info("task dag size exceeds the limit, causing the peer to leave")
if err := peer.FSM.Event(PeerEventLeave); err != nil {
if err := peer.FSM.Event(context.Background(), PeerEventLeave); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
return true
}

10
scheduler/resource/seed_peer.go
View File

@ -98,7 +98,7 @@ func (s *seedPeer) TriggerTask(ctx context.Context, task *Task) (*Peer, *schedul
// If the peer initialization succeeds and the download fails,
// set peer status is PeerStateFailed.
if peer != nil {
if err := peer.FSM.Event(PeerEventDownloadFailed); err != nil {
if err := peer.FSM.Event(ctx, PeerEventDownloadFailed); err != nil {
return nil, nil, err
}
}
@ -110,7 +110,7 @@ func (s *seedPeer) TriggerTask(ctx context.Context, task *Task) (*Peer, *schedul
initialized = true
// Initialize seed peer.
peer, err = s.initSeedPeer(task, piece)
peer, err = s.initSeedPeer(ctx, task, piece)
if err != nil {
return nil, nil, err
}
@ -120,7 +120,7 @@ func (s *seedPeer) TriggerTask(ctx context.Context, task *Task) (*Peer, *schedul
// Handle begin of piece.
if piece.PieceInfo.PieceNum == common.BeginOfPiece {
peer.Log.Infof("receive begin of piece from seed peer: %#v %#v", piece, piece.PieceInfo)
if err := peer.FSM.Event(PeerEventDownload); err != nil {
if err := peer.FSM.Event(ctx, PeerEventDownload); err != nil {
return nil, nil, err
}
@ -168,7 +168,7 @@ func (s *seedPeer) TriggerTask(ctx context.Context, task *Task) (*Peer, *schedul
}
// Initialize seed peer.
func (s *seedPeer) initSeedPeer(task *Task, ps *cdnsystemv1.PieceSeed) (*Peer, error) {
func (s *seedPeer) initSeedPeer(ctx context.Context, task *Task, ps *cdnsystemv1.PieceSeed) (*Peer, error) {
// Load peer from manager.
peer, ok := s.peerManager.Load(ps.PeerId)
if ok {
@ -188,7 +188,7 @@ func (s *seedPeer) initSeedPeer(task *Task, ps *cdnsystemv1.PieceSeed) (*Peer, e
s.peerManager.Store(peer)
peer.Log.Info("seed peer has been stored")
if err := peer.FSM.Event(PeerEventRegisterNormal); err != nil {
if err := peer.FSM.Event(ctx, PeerEventRegisterNormal); err != nil {
return nil, err
}

11
scheduler/resource/task.go
View File

@ -17,6 +17,7 @@
package resource
import (
"context"
"errors"
"sort"
"sync"
@ -172,19 +173,19 @@ func NewTask(id, url string, taskType commonv1.TaskType, meta *commonv1.UrlMeta,
{Name: TaskEventLeave, Src: []string{TaskStatePending, TaskStateRunning, TaskStateSucceeded, TaskStateFailed}, Dst: TaskStateLeave},
},
fsm.Callbacks{
TaskEventDownload: func(e *fsm.Event) {
TaskEventDownload: func(ctx context.Context, e *fsm.Event) {
t.UpdatedAt.Store(time.Now())
t.Log.Infof("task state is %s", e.FSM.Current())
},
TaskEventDownloadSucceeded: func(e *fsm.Event) {
TaskEventDownloadSucceeded: func(ctx context.Context, e *fsm.Event) {
t.UpdatedAt.Store(time.Now())
t.Log.Infof("task state is %s", e.FSM.Current())
},
TaskEventDownloadFailed: func(e *fsm.Event) {
TaskEventDownloadFailed: func(ctx context.Context, e *fsm.Event) {
t.UpdatedAt.Store(time.Now())
t.Log.Infof("task state is %s", e.FSM.Current())
},
TaskEventLeave: func(e *fsm.Event) {
TaskEventLeave: func(ctx context.Context, e *fsm.Event) {
t.UpdatedAt.Store(time.Now())
t.Log.Infof("task state is %s", e.FSM.Current())
},
@ -458,7 +459,7 @@ func (t *Task) NotifyPeers(peerPacket *schedulerv1.PeerPacket, event string) {
}
t.Log.Infof("task notify peer %s code %s", peer.ID, peerPacket.Code)
if err := peer.FSM.Event(event); err != nil {
if err := peer.FSM.Event(context.Background(), event); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
continue
}

3
scheduler/resource/task_manager.go
View File

@ -19,6 +19,7 @@
package resource
import (
"context"
"sync"
pkggc "d7y.io/dragonfly/v2/pkg/gc"
@ -111,7 +112,7 @@ func (t *taskManager) RunGC() error {
// If there is no peer then switch the task state to TaskStateLeave.
if task.PeerCount() == 0 {
if err := task.FSM.Event(TaskEventLeave); err != nil {
if err := task.FSM.Event(context.Background(), TaskEventLeave); err != nil {
task.Log.Errorf("task fsm event failed: %s", err.Error())
return true
}

4
scheduler/scheduler/scheduler.go
View File

@ -93,7 +93,7 @@ func (s *scheduler) ScheduleParent(ctx context.Context, peer *resource.Peer, blo
return
}
if err := peer.FSM.Event(resource.PeerEventDownloadBackToSource); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventDownloadBackToSource); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
return
}
@ -101,7 +101,7 @@ func (s *scheduler) ScheduleParent(ctx context.Context, peer *resource.Peer, blo
// If the task state is TaskStateFailed,
// peer back-to-source and reset task state to TaskStateRunning.
if peer.Task.FSM.Is(resource.TaskStateFailed) {
if err := peer.Task.FSM.Event(resource.TaskEventDownload); err != nil {
if err := peer.Task.FSM.Event(ctx, resource.TaskEventDownload); err != nil {
peer.Task.Log.Errorf("task fsm event failed: %s", err.Error())
return
}

38
scheduler/service/service.go
View File

@ -327,7 +327,7 @@ func (s *Service) AnnounceTask(ctx context.Context, req *schedulerv1.AnnounceTas
// advance the task state to TaskStateSucceeded.
if !task.FSM.Is(resource.TaskStateSucceeded) {
if task.FSM.Can(resource.TaskEventDownload) {
if err := task.FSM.Event(resource.TaskEventDownload); err != nil {
if err := task.FSM.Event(ctx, resource.TaskEventDownload); err != nil {
msg := fmt.Sprintf("task fsm event failed: %s", err.Error())
peer.Log.Error(msg)
return dferrors.New(commonv1.Code_SchedError, msg)
@ -359,7 +359,7 @@ func (s *Service) AnnounceTask(ctx context.Context, req *schedulerv1.AnnounceTas
// advance the peer state to PeerStateSucceeded.
if !peer.FSM.Is(resource.PeerStateSucceeded) {
if peer.FSM.Is(resource.PeerStatePending) {
if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventRegisterNormal); err != nil {
msg := fmt.Sprintf("peer fsm event failed: %s", err.Error())
peer.Log.Error(msg)
return dferrors.New(commonv1.Code_SchedError, msg)
@ -369,7 +369,7 @@ func (s *Service) AnnounceTask(ctx context.Context, req *schedulerv1.AnnounceTas
if peer.FSM.Is(resource.PeerStateReceivedTiny) ||
peer.FSM.Is(resource.PeerStateReceivedSmall) ||
peer.FSM.Is(resource.PeerStateReceivedNormal) {
if err := peer.FSM.Event(resource.PeerEventDownload); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventDownload); err != nil {
msg := fmt.Sprintf("peer fsm event failed: %s", err.Error())
peer.Log.Error(msg)
return dferrors.New(commonv1.Code_SchedError, msg)
@ -416,7 +416,7 @@ func (s *Service) LeaveTask(ctx context.Context, req *schedulerv1.PeerTarget) er
}
metrics.LeaveTaskCount.WithLabelValues(peer.Tag, peer.Application).Inc()
if err := peer.FSM.Event(resource.PeerEventLeave); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventLeave); err != nil {
metrics.LeaveTaskFailureCount.WithLabelValues(peer.Tag, peer.Application).Inc()
msg := fmt.Sprintf("peer fsm event failed: %s", err.Error())
peer.Log.Error(msg)
@ -500,7 +500,7 @@ func (s *Service) triggerTask(ctx context.Context, req *schedulerv1.PeerTaskRequ
// If the task triggers the TaskEventDownload failed and it has no available peer,
// let the peer do the scheduling.
if !task.FSM.Is(resource.TaskStateRunning) {
if err := task.FSM.Event(resource.TaskEventDownload); err != nil {
if err := task.FSM.Event(ctx, resource.TaskEventDownload); err != nil {
peer.Log.Errorf("task fsm event failed: %s", err.Error())
return err
}
@ -643,7 +643,7 @@ func (s *Service) storePeer(ctx context.Context, peerID string, task *resource.T
// registerEmptyTask registers the empty task.
func (s *Service) registerEmptyTask(ctx context.Context, peer *resource.Peer) (*schedulerv1.RegisterResult, error) {
if err := peer.FSM.Event(resource.PeerEventRegisterEmpty); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventRegisterEmpty); err != nil {
return nil, err
}
@ -659,7 +659,7 @@ func (s *Service) registerEmptyTask(ctx context.Context, peer *resource.Peer) (*
// registerEmptyTask registers the tiny task.
func (s *Service) registerTinyTask(ctx context.Context, peer *resource.Peer) (*schedulerv1.RegisterResult, error) {
if err := peer.FSM.Event(resource.PeerEventRegisterTiny); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventRegisterTiny); err != nil {
return nil, err
}
@ -701,7 +701,7 @@ func (s *Service) registerSmallTask(ctx context.Context, peer *resource.Peer) (*
return nil, err
}
if err := peer.FSM.Event(resource.PeerEventRegisterSmall); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventRegisterSmall); err != nil {
return nil, err
}
@ -728,7 +728,7 @@ func (s *Service) registerSmallTask(ctx context.Context, peer *resource.Peer) (*
// registerNormalTask registers the tiny task.
func (s *Service) registerNormalTask(ctx context.Context, peer *resource.Peer) (*schedulerv1.RegisterResult, error) {
if err := peer.FSM.Event(resource.PeerEventRegisterNormal); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventRegisterNormal); err != nil {
return nil, err
}
@ -741,7 +741,7 @@ func (s *Service) registerNormalTask(ctx context.Context, peer *resource.Peer) (
// handleRegisterFailure handles failure of register.
func (s *Service) handleRegisterFailure(ctx context.Context, peer *resource.Peer) {
if err := peer.FSM.Event(resource.PeerEventLeave); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventLeave); err != nil {
peer.Log.Error(err)
}
@ -761,19 +761,19 @@ func (s *Service) handleBeginOfPiece(ctx context.Context, peer *resource.Peer) {
case resource.PeerStateReceivedTiny:
// When the task is tiny,
// the peer has already returned to piece data when registering.
if err := peer.FSM.Event(resource.PeerEventDownload); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventDownload); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
return
}
case resource.PeerStateReceivedSmall:
// When the task is small,
// the peer has already returned to the parent when registering.
if err := peer.FSM.Event(resource.PeerEventDownload); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventDownload); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
return
}
case resource.PeerStateReceivedNormal:
if err := peer.FSM.Event(resource.PeerEventDownload); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventDownload); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
return
}
@ -841,7 +841,7 @@ func (s *Service) handlePieceFailure(ctx context.Context, peer *resource.Peer, p
switch code {
case commonv1.Code_PeerTaskNotFound:
if err := parent.FSM.Event(resource.PeerEventDownloadFailed); err != nil {
if err := parent.FSM.Event(ctx, resource.PeerEventDownloadFailed); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
break
}
@ -891,7 +891,7 @@ func (s *Service) handlePieceFailure(ctx context.Context, peer *resource.Peer, p
// handlePeerSuccess handles successful peer.
func (s *Service) handlePeerSuccess(ctx context.Context, peer *resource.Peer) {
if err := peer.FSM.Event(resource.PeerEventDownloadSucceeded); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventDownloadSucceeded); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
return
}
@ -926,7 +926,7 @@ func (s *Service) handlePeerSuccess(ctx context.Context, peer *resource.Peer) {
// handlePeerFailure handles failed peer.
func (s *Service) handlePeerFailure(ctx context.Context, peer *resource.Peer) {
if err := peer.FSM.Event(resource.PeerEventDownloadFailed); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventDownloadFailed); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
return
}
@ -941,7 +941,7 @@ func (s *Service) handlePeerFailure(ctx context.Context, peer *resource.Peer) {
// handleLegacySeedPeer handles seed server's task has left,
// but did not notify the scheduler to leave the task.
func (s *Service) handleLegacySeedPeer(ctx context.Context, peer *resource.Peer) {
if err := peer.FSM.Event(resource.PeerEventLeave); err != nil {
if err := peer.FSM.Event(ctx, resource.PeerEventLeave); err != nil {
peer.Log.Errorf("peer fsm event failed: %s", err.Error())
return
}
@ -966,7 +966,7 @@ func (s *Service) handleTaskSuccess(ctx context.Context, task *resource.Task, re
task.TotalPieceCount.Store(req.TotalPieceCount)
task.ContentLength.Store(req.ContentLength)
if err := task.FSM.Event(resource.TaskEventDownloadSucceeded); err != nil {
if err := task.FSM.Event(ctx, resource.TaskEventDownloadSucceeded); err != nil {
task.Log.Errorf("task fsm event failed: %s", err.Error())
return
}
@ -1036,7 +1036,7 @@ func (s *Service) handleTaskFailure(ctx context.Context, task *resource.Task, ba
return
}
if err := task.FSM.Event(resource.TaskEventDownloadFailed); err != nil {
if err := task.FSM.Event(ctx, resource.TaskEventDownloadFailed); err != nil {
task.Log.Errorf("task fsm event failed: %s", err.Error())
return
}