admission_gate.go Raw History Folder

1package process
2
3import (
4	"context"
5	"sync"
6	"time"
7)
8
9// AdmissionGate is a narrow thread-safe admission/work-lease boundary shared
10// by the relay listener and the shutdown/residency policy. A lease must be
11// acquired before sending an acknowledgement; shutdown atomically stops
12// admission while admitted work drains. It never blocks the holder: waiting
13// uses WaitDrained with a caller-provided bounded context.
14type AdmissionGate struct {
15	mu        sync.Mutex
16	closed    bool
17	active    int
18	onDrained []func()
19}
20
21// NewAdmissionGate returns an open gate with no active leases.
22func NewAdmissionGate() *AdmissionGate {
23	return &AdmissionGate{}
24}
25
26// Acquire takes one work lease. It reports false when admission is closed;
27// the caller must then send the existing error response before any
28// acknowledgement and must not dispatch.
29func (g *AdmissionGate) Acquire() bool {
30	g.mu.Lock()
31	defer g.mu.Unlock()
32	if g.closed {
33		return false
34	}
35	g.active++
36	return true
37}
38
39// Release returns one work lease. The final release notifies drain
40// subscribers.
41func (g *AdmissionGate) Release() {
42	g.mu.Lock()
43	var notify []func()
44	if g.active > 0 {
45		g.active--
46	}
47	if g.active == 0 {
48		notify = append([]func(){}, g.onDrained...)
49	}
50	g.mu.Unlock()
51	for _, fn := range notify {
52		fn()
53	}
54}
55
56// Close atomically stops admission. In-flight leases are unaffected; they
57// drain through Release.
58func (g *AdmissionGate) Close() {
59	g.mu.Lock()
60	defer g.mu.Unlock()
61	g.closed = true
62}
63
64// Reopen resumes admission after a seal that no longer applies: an
65// invalidated deferred quit (e.g. a reopened window) must not leave the
66// surviving process permanently deaf to relay requests.
67func (g *AdmissionGate) Reopen() {
68	g.mu.Lock()
69	defer g.mu.Unlock()
70	g.closed = false
71}
72
73// Closed reports whether admission is stopped.
74func (g *AdmissionGate) Closed() bool {
75	g.mu.Lock()
76	defer g.mu.Unlock()
77	return g.closed
78}
79
80// Active reports the number of held leases.
81func (g *AdmissionGate) Active() int {
82	g.mu.Lock()
83	defer g.mu.Unlock()
84	return g.active
85}
86
87// OnDrained registers a callback invoked (without holding the gate lock)
88// every time the active count reaches zero. Registration is not idempotent:
89// each call appends. There is no unsubscribe; gates are process-lifetime
90// owners. Callbacks must not call back into the gate.
91func (g *AdmissionGate) OnDrained(fn func()) {
92	if fn == nil {
93		return
94	}
95	g.mu.Lock()
96	defer g.mu.Unlock()
97	g.onDrained = append(g.onDrained, fn)
98}
99
100// WaitDrained blocks until no leases are held or ctx ends. It reports false
101// on context expiry so shutdown never waits forever. The short poll runs
102// only on shutdown/drain paths, never on any frame or event loop.
103func (g *AdmissionGate) WaitDrained(ctx context.Context) bool {
104	ticker := time.NewTicker(10 * time.Millisecond)
105	defer ticker.Stop()
106	for {
107		g.mu.Lock()
108		drained := g.active == 0
109		g.mu.Unlock()
110		if drained {
111			return true
112		}
113		select {
114		case <-ctx.Done():
115			return false
116		case <-ticker.C:
117		}
118	}
119}

admission_gate.go RawHistoryFolder

admission_gate.go Raw History Folder