Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225package mill
import ( "sync"
"tangled.org/core/spindle/models" "tangled.org/core/spindle/quota"
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 remoteQuota struct { key string resources quota.Resources committing bool}
type RemoteLease struct { id string nodeID string epoch string engine string cacheNamespace string wid models.WorkflowId // job this lease carries, set once placed // identity comes from the mill's pipeline record ownerDID string repoDID string // keep the winning resources if quota waiting causes a re-bid resources quota.Resources millRecordsTerminalMetrics bool // 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
// quotaLease is nil when only its persisted id survived a restart quotaLease quota.Lease quotaID string // executor-side reservations stay attached until commit or release remoteQuotas map[string]remoteQuota
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), remoteQuotas: make(map[string]remoteQuota), }}
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}
// reconnect can replay the cancel, only the first ack starts the teardown timerfunc (l *RemoteLease) markCancelAcked() bool { l.mu.Lock() defer l.mu.Unlock() if l.cancelAcked { return false } l.cancelAcked = true return true}
func (l *RemoteLease) cancelRequested() (bool, string) { l.mu.Lock() defer l.mu.Unlock() return l.cancel, l.reason}
func (l *RemoteLease) cancelPending() bool { l.mu.Lock() defer l.mu.Unlock() return l.cancel && !l.cancelAcked && l.state != leaseDone}
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_TERMINAL_STATUS_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) })}