Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187package mill
import ( "sync"
"tangled.org/core/spindle/models"
millv1 "tangled.org/core/spindle/mill/proto/gen")
// the mill's view of a remote attempttype leaseState int32
const ( // won a bid, executor is holding a seat, not yet committed leaseReserved leaseState = iota // the mill sent CommitLease, the executor may already be running, but // the mill may not have seen Committed yet leaseCommitting // the executor acked CommitLease, job is running on the executor leaseRunning // terminal result arrived or we gave up, no further action leaseDone)
type cancelAction int
const ( cancelNoop cancelAction = iota cancelLocal cancelRemote)
// mill-side fencing token for one placed jobtype RemoteLease struct { id string nodeID string epoch string engine string wid models.WorkflowId // job this lease carries, set once placed // restored after a mill restart. no RunStep waits on it, so terminals // and death are authored directly. set before publication, never mutated orphaned bool claimed bool cancelAcked bool cleanedUp bool cleanupRetry bool
mu sync.Mutex state leaseState cancel bool reason string released bool terminal chan *millv1.AttemptResult // buffered(1), RunStep waits here
finishMu sync.Mutex}
func newLease(id, nodeID, epoch, engine string) *RemoteLease { return &RemoteLease{ id: id, nodeID: nodeID, epoch: epoch, engine: engine, state: leaseReserved, terminal: make(chan *millv1.AttemptResult, 1), }}
func (l *RemoteLease) setState(s leaseState) { l.mu.Lock() l.state = s l.mu.Unlock()}
func (l *RemoteLease) markCommitting() bool { l.mu.Lock() defer l.mu.Unlock() if l.state == leaseDone { return false } if l.state == leaseReserved { l.state = leaseCommitting } return true}
func (l *RemoteLease) markRunning() { l.mu.Lock() defer l.mu.Unlock() if l.state != leaseDone { l.state = leaseRunning }}
func (l *RemoteLease) getState() leaseState { l.mu.Lock() defer l.mu.Unlock() return l.state}
// only one caller gets to mark it donefunc (l *RemoteLease) markDone() bool { l.mu.Lock() defer l.mu.Unlock() if l.state == leaseDone { return false } l.state = leaseDone return true}
func (l *RemoteLease) requestCancel(reason string) cancelAction { l.mu.Lock() defer l.mu.Unlock() if l.state == leaseDone { return cancelNoop } l.cancel = true l.reason = reason if l.state == leaseReserved { return cancelLocal } return cancelRemote}
func (l *RemoteLease) cancelRequested() (bool, string) { l.mu.Lock() defer l.mu.Unlock() return l.cancel, l.reason}
func (l *RemoteLease) releaseState() (leaseState, bool) { l.mu.Lock() defer l.mu.Unlock() l.released = true return l.state, l.cancel}
func (l *RemoteLease) cleanupReady() bool { l.mu.Lock() defer l.mu.Unlock() return l.released && l.state == leaseDone}
func (l *RemoteLease) deliverCancelled(reason string) { l.deliverTerminal(&millv1.AttemptResult{ Status: millv1.TerminalStatus_CANCELLED, Error: reason, })}
// hands the terminal to a waiting RunStep without blocking. duplicates// (eg. reconnect replays) just drop, the channel holds one and the lease// is already donefunc (l *RemoteLease) deliverTerminal(res *millv1.AttemptResult) { if !l.markDone() { return } select { case l.terminal <- res: default: }}
// mill's synthetic single step. real steps run on the executor and the// mill never mirrors themtype remoteStep struct{}
func (remoteStep) Name() string { return "remote execution" }func (remoteStep) Command() string { return "" }func (remoteStep) Kind() models.StepKind { return models.StepKindSystem }
// what AcquireWorkflowSlot returns. Release unwinds placementtype millSlot struct { fleet *Mill lease *RemoteLease once sync.Once}
func (s *millSlot) Release() { if s == nil { return } s.once.Do(func() { s.fleet.releaseSlot(s) })}