| 1 | package schedule |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "sync" |
| 6 | ) |
| 7 | |
| 8 | // HostGate is the one per-host gate shared by the three writers of a host's |
| 9 | // timer rows (moves spec §9.4, round-2 finding 21): the runner holds a host's |
| 10 | // read side for the whole of a delivery, the converger holds it across its |
| 11 | // Frozen check and its writes, and a move's drain takes the write side once |
| 12 | // the row is held. A drain therefore returns only when no delivery or |
| 13 | // convergence that could have missed the hold is still running, and from |
| 14 | // then on the hold check inside the gate refuses both. |
| 15 | // |
| 16 | // Entry is TryRLock, never RLock: Go's RWMutex blocks new readers once a |
| 17 | // writer waits, and a drain can wait for a delivery of up to DeliveryTimeout. |
| 18 | // A blocking entry would park the tick loop, and with it every other host's |
| 19 | // timers, or the converger's pass, behind one held host. A refused entry is |
| 20 | // a host being drained, which is a host about to be frozen: it is skipped |
| 21 | // this tick or pass, with no bookkeeping. |
| 22 | // |
| 23 | // A nil *HostGate is inert (every entry succeeds, a drain returns at once): |
| 24 | // the runner and converger with moves off are main's. |
| 25 | type HostGate struct { |
| 26 | mu sync.Mutex |
| 27 | hosts map[string]*sync.RWMutex |
| 28 | } |
| 29 | |
| 30 | func (g *HostGate) lock(host string) *sync.RWMutex { |
| 31 | g.mu.Lock() |
| 32 | defer g.mu.Unlock() |
| 33 | if g.hosts == nil { |
| 34 | g.hosts = map[string]*sync.RWMutex{} |
| 35 | } |
| 36 | l := g.hosts[host] |
| 37 | if l == nil { |
| 38 | l = &sync.RWMutex{} |
| 39 | g.hosts[host] = l |
| 40 | } |
| 41 | return l |
| 42 | } |
| 43 | |
| 44 | // tryEnter takes host's read side without waiting. false: a drain holds the |
| 45 | // host or waits for it. |
| 46 | func (g *HostGate) tryEnter(host string) bool { |
| 47 | if g == nil { |
| 48 | return true |
| 49 | } |
| 50 | return g.lock(host).TryRLock() |
| 51 | } |
| 52 | |
| 53 | // leave ends an entry tryEnter granted. |
| 54 | func (g *HostGate) leave(host string) { |
| 55 | if g == nil { |
| 56 | return |
| 57 | } |
| 58 | g.lock(host).RUnlock() |
| 59 | } |
| 60 | |
| 61 | // Drain waits until every delivery and convergence in flight for host has |
| 62 | // finished, then lets the host go: the cut's drain, called after SetHold |
| 63 | // (moves spec §9.4 "drained"). Work that starts afterwards re-reads the row |
| 64 | // or the freeze inside the gate and stops. ctx bounds the wait; an ended |
| 65 | // context returns its error, and the pending drain still completes in the |
| 66 | // background (until then the host's new work is refused, which is what a |
| 67 | // held host wants anyway). |
| 68 | func (g *HostGate) Drain(ctx context.Context, host string) error { |
| 69 | if g == nil { |
| 70 | return nil |
| 71 | } |
| 72 | l := g.lock(host) |
| 73 | done := make(chan struct{}) |
| 74 | go func() { |
| 75 | // The barrier is the acquisition itself: nothing runs inside it. |
| 76 | l.Lock() |
| 77 | l.Unlock() |
| 78 | close(done) |
| 79 | }() |
| 80 | select { |
| 81 | case <-done: |
| 82 | return nil |
| 83 | case <-ctx.Done(): |
| 84 | return ctx.Err() |
| 85 | } |
| 86 | } |
| 87 | |
amadan