alice.pds.demo.boltless.dev
This repository has no description
1package mill
2
3import (
4 "sync"
5
6 "tangled.org/core/spindle/models"
7
8 millv1 "tangled.org/core/spindle/mill/proto/gen"
9)
10
11// the mill's view of a remote attempt
12type leaseState int32
13
14const (
15 // won a bid, executor is holding a seat, not yet committed
16 leaseReserved leaseState = iota
17 // CommitLease sent. the executor may already be running, but the mill
18 // may not have seen Committed yet
19 leaseCommitting
20 // CommitLease sent and acked. job is running on the executor
21 leaseRunning
22 // terminal result arrived or we gave up, no further action
23 leaseDone
24)
25
26type cancelAction int
27
28const (
29 cancelNoop cancelAction = iota
30 cancelLocal
31 cancelRemote
32)
33
34// mill-side fencing token for one placed job
35type RemoteLease struct {
36 id string
37 nodeID string
38 epoch string
39 engine string
40 wid models.WorkflowId // job this lease carries, set once placed
41 // restored after a mill restart. no RunStep waits on it, so terminals
42 // and death are authored directly. set before publication, never mutated
43 orphaned bool
44 claimed bool
45 cancelAcked bool
46 cleanedUp bool
47 cleanupRetry bool
48
49 mu sync.Mutex
50 state leaseState
51 cancel bool
52 reason string
53 released bool
54 terminal chan *millv1.AttemptResult // buffered(1), RunStep waits here
55
56 finishMu sync.Mutex
57}
58
59func newLease(id, nodeID, epoch, engine string) *RemoteLease {
60 return &RemoteLease{
61 id: id,
62 nodeID: nodeID,
63 epoch: epoch,
64 engine: engine,
65 state: leaseReserved,
66 terminal: make(chan *millv1.AttemptResult, 1),
67 }
68}
69
70func (l *RemoteLease) setState(s leaseState) {
71 l.mu.Lock()
72 l.state = s
73 l.mu.Unlock()
74}
75
76func (l *RemoteLease) markCommitting() bool {
77 l.mu.Lock()
78 defer l.mu.Unlock()
79 if l.state == leaseDone {
80 return false
81 }
82 if l.state == leaseReserved {
83 l.state = leaseCommitting
84 }
85 return true
86}
87
88func (l *RemoteLease) markRunning() {
89 l.mu.Lock()
90 defer l.mu.Unlock()
91 if l.state != leaseDone {
92 l.state = leaseRunning
93 }
94}
95
96func (l *RemoteLease) getState() leaseState {
97 l.mu.Lock()
98 defer l.mu.Unlock()
99 return l.state
100}
101
102// only one caller gets to mark it done
103func (l *RemoteLease) markDone() bool {
104 l.mu.Lock()
105 defer l.mu.Unlock()
106 if l.state == leaseDone {
107 return false
108 }
109 l.state = leaseDone
110 return true
111}
112
113func (l *RemoteLease) requestCancel(reason string) cancelAction {
114 l.mu.Lock()
115 defer l.mu.Unlock()
116 if l.state == leaseDone {
117 return cancelNoop
118 }
119 l.cancel = true
120 l.reason = reason
121 if l.state == leaseReserved {
122 return cancelLocal
123 }
124 return cancelRemote
125}
126
127func (l *RemoteLease) cancelRequested() (bool, string) {
128 l.mu.Lock()
129 defer l.mu.Unlock()
130 return l.cancel, l.reason
131}
132
133func (l *RemoteLease) releaseState() (leaseState, bool) {
134 l.mu.Lock()
135 defer l.mu.Unlock()
136 l.released = true
137 return l.state, l.cancel
138}
139
140func (l *RemoteLease) cleanupReady() bool {
141 l.mu.Lock()
142 defer l.mu.Unlock()
143 return l.released && l.state == leaseDone
144}
145
146func (l *RemoteLease) deliverCancelled(reason string) {
147 l.deliverTerminal(&millv1.AttemptResult{
148 Status: millv1.TerminalStatus_CANCELLED,
149 Error: reason,
150 })
151}
152
153// hands the terminal to a waiting RunStep without blocking. duplicates
154// (eg. reconnect replays) just drop, the channel holds one and the lease
155// is already done
156func (l *RemoteLease) deliverTerminal(res *millv1.AttemptResult) {
157 if !l.markDone() {
158 return
159 }
160 select {
161 case l.terminal <- res:
162 default:
163 }
164}
165
166// mill's synthetic single step. real steps run on the executor and the
167// mill never mirrors them
168type remoteStep struct{}
169
170func (remoteStep) Name() string { return "remote execution" }
171func (remoteStep) Command() string { return "" }
172func (remoteStep) Kind() models.StepKind { return models.StepKindSystem }
173
174// what AcquireWorkflowSlot returns. Release unwinds placement
175type millSlot struct {
176 fleet *Mill
177 lease *RemoteLease
178 once sync.Once
179}
180
181func (s *millSlot) Release() {
182 if s == nil {
183 return
184 }
185 s.once.Do(func() { s.fleet.releaseSlot(s) })
186}