internal/agent/reconcile/worker.go
Ref: Size: 6.6 KiB History
package reconcile
import (
"context"
"sync"
"github.com/a73x/eitri/internal/pb"
)
// manager owns one goroutine per VM — the demux half of the reconcile loop.
// Step hands it one assignment per VM, it routes each to that VM's worker, and
// it collects every worker's last-published result for the host report.
//
// Concurrency contract:
// - one goroutine per VM, so every operation on a single VM is serialized for
// free: the worker IS the lock, and there is no per-VM mutex;
// - anything shared BETWEEN VMs is guarded where it lives: the admission
// ledger under Engine.mu (see admit), whatever the host backend shares
// between VMs, under the backend's own locking, and the state store by one
// file per VM written via temp-file rename.
type manager struct {
eng *Engine
mu sync.Mutex
workers map[string]*worker
tombstoned map[string]bool
stopped bool
}
func newManager(eng *Engine) *manager {
return &manager{eng: eng, workers: map[string]*worker{}, tombstoned: map[string]bool{}}
}
// manager returns the engine's worker manager, constructing it on first use.
// Lazy construction rather than a Start method keeps Step the single entry
// point, so syncclient and cmd/eitri-agent need no extra wiring.
func (e *Engine) manager() *manager {
e.mgrOnce.Do(func() { e.mgr = newManager(e) })
return e.mgr
}
// deliver hands a VM its latest assignment, spawning its worker on first sight.
func (m *manager) deliver(id string, a assignment) {
m.mu.Lock()
if m.stopped {
m.mu.Unlock()
return
}
w, ok := m.workers[id]
if !ok {
w = newWorker(m.eng, id)
m.workers[id] = w
go w.run()
}
m.mu.Unlock()
w.mu.Lock()
w.pending = &a
w.cond.Broadcast()
w.mu.Unlock()
}
// reapAbsent stops the workers for VMs present in neither desired state nor
// local records — nothing is left to reconcile. Only IDLE workers are reaped;
// a busy one is deferred to a later tick, which reaping being level-triggered
// makes free.
func (m *manager) reapAbsent(live map[string]assignment) {
m.mu.Lock()
defer m.mu.Unlock()
for id, w := range m.workers {
if _, ok := live[id]; ok {
continue
}
w.mu.Lock()
idle := w.pending == nil && !w.busy
w.mu.Unlock()
if !idle {
continue // still owns this VM; reap it on a later tick
}
delete(m.workers, id)
w.stop()
}
}
func (m *manager) setTombstoned(ids map[string]bool) {
m.mu.Lock()
defer m.mu.Unlock()
m.tombstoned = ids
}
// tombstones returns the current delete set. The map is returned rather than
// copied: setTombstoned replaces it wholesale on every dispatch and nothing
// ever mutates one in place, so a reader can hold an older map safely.
func (m *manager) tombstones() map[string]bool {
m.mu.Lock()
defer m.mu.Unlock()
return m.tombstoned
}
// collect folds every worker's last-published result into rep. Each read takes
// only the worker's lock, which is never held across a pass, so a VM wedged in
// a multi-second operation cannot delay the report — it simply contributes its
// previous result, or nothing if it has not published yet.
func (m *manager) collect(rep *pb.Report) {
for _, w := range m.snapshot() {
w.mu.Lock()
res := w.result
w.mu.Unlock()
// Cloned outside the lock: a published result is never mutated in
// place, so its protos are safe to read once the pointer is in hand.
own := res.clone()
own.merge(rep)
}
}
// snapshot returns the current worker set, so callers can walk it without
// holding the manager lock while touching workers.
func (m *manager) snapshot() []*worker {
m.mu.Lock()
defer m.mu.Unlock()
out := make([]*worker, 0, len(m.workers))
for _, w := range m.workers {
out = append(out, w)
}
return out
}
// worker owns exactly one VM's reconcile. Its goroutine is the serialization
// primitive: at most one operation on that VM is ever in flight.
type worker struct {
eng *Engine
id string
mu sync.Mutex
cond *sync.Cond
pending *assignment
// busy reports that a pass is running (with mu released).
busy bool
// gen names the pass currently in flight. run stamps a fresh one before
// every pass and hands it to that pass's publisher, so a row can be traced
// back to the pass that produced it — see publish.
gen uint64
// stopped ends the goroutine after the current pass.
stopped bool
// result is this VM's last-published contribution to the host report.
result vmResult
}
func newWorker(eng *Engine, id string) *worker {
w := &worker{eng: eng, id: id}
w.cond = sync.NewCond(&w.mu)
return w
}
// run is the worker loop: take the latest assignment, reconcile, publish.
//
// Each pass gets a fresh context.Background rather than the context of the Step
// that dispatched it: a pass outlives its Step by design, so inheriting that
// context would cancel the work the instant the report went out. VMTimeout is
// what bounds a pass (see reconcileOne).
func (w *worker) run() {
for {
w.mu.Lock()
for w.pending == nil && !w.stopped {
w.cond.Wait()
}
if w.stopped {
w.mu.Unlock()
return
}
a := *w.pending
w.pending = nil
w.busy = true
w.gen++
gen := w.gen
w.mu.Unlock()
// LOAD-BEARING: the pass runs with w.mu RELEASED. That is what lets
// deliver and collect run at full speed while this VM does slow work.
// The publisher carries this pass's generation, which is what stops a
// row from a finished pass landing on a later one (see publish).
res, ok := w.eng.reconcileOne(context.Background(), w.id, a,
func(interim vmResult) { w.publish(gen, interim) })
w.mu.Lock()
if ok {
w.result = res
}
w.busy = false
w.cond.Broadcast()
w.mu.Unlock()
}
}
// publish swaps in a result from INSIDE a running pass, so the host report can
// carry a row for work that has not finished. Every other publish happens at
// the end of a pass (see run); this is the one that happens during one, and it
// exists because the pass that most needs reporting on is the one that takes
// minutes — a first create.
//
// It takes the same lock the pass-end publish takes and holds it no longer, so
// a narrating create is no more able to delay a report than a silent one.
//
// gen is the generation run stamped on the pass handed this publisher, and a
// row lands only if that pass is still the one in flight. Narration reaches
// here on goroutines the pass does not own: the image cache shares one download
// between waiters and runs its progress callback on the fetch.
func (w *worker) publish(gen uint64, res vmResult) {
w.mu.Lock()
defer w.mu.Unlock()
if !w.busy || gen != w.gen {
return // a straggler from a pass that has already ended
}
w.result = res
}
func (w *worker) stop() {
w.mu.Lock()
w.stopped = true
w.cond.Broadcast()
w.mu.Unlock()
}