1// Aggregate listener over multiple physical listeners.
2//
3// AggregateListener is the daemon's global ports.ServerListener over several
4// physical daemonmux connections. The daemon's shared accept loop must not stop
5// when one physical carriage is lost: a daemonmux.Listener Accept fails with
6// ErrListenerLost exactly when its one physical connection reached its terminal
7// outcome, and a raw accept loop treats that as the end of service. The
8// aggregate composes independent child listeners into one accept stream, so the
9// loss or closure of one child is isolated - the aggregate deregisters it and
10// keeps accepting from every other child and from children registered later -
11// and never becomes the aggregate's own terminal error.
12//
13// Each registered child is owned: the aggregate runs exactly one forwarding
14// goroutine per child (bounded by the number of registered children, never by
15// the number of accepted streams) that repeatedly takes one typed
16// ports.ServerConnection from the child and hands it to the aggregate's bounded
17// FIFO. The FIFO is a mutex-guarded slice signalled by a sync.Cond
18// (MaxAggregateQueue, 128, by default): a slow or stalled consumer applies
19// backpressure to the child forwarders instead of growing memory, so the
20// aggregate never holds more than its queue bound plus one in-flight connection
21// per child forwarder, and each child's own bounded accept queue still refuses
22// a stream whose slot it cannot hold.
23//
24// A child whose Accept fails for any terminal reason - ErrListenerLost,
25// ErrListenerClosed, or a transport-specific error - is deregistered and its
26// goroutine joined; the failure is swallowed rather than returned by the
27// aggregate's Accept, because one physical carriage is not the daemon's whole
28// listener. Register and Remove are safe for concurrent use, as are Accept and
29// Close. Close unblocks every blocked Accept with ErrAggregateClosed, closes
30// every owned child (which unblocks and joins that child's own accept loop),
31// joins every forwarding goroutine, and closes every connection that was taken
32// from a child but never handed to the daemon. Removing one child closes that
33// child and joins its goroutine without touching a sibling, a queued connection
34// already taken from it, or a connection already delivered to the daemon.
35//
36// The aggregate owns no framing, session conversion, or carrier: it consumes
37// only the typed ports.ServerListener and ports.ServerConnection contracts, so
38// it composes with the daemonmux Listener, the sessionwire listener, or any
39// later physical listener implementation.
40package daemonmux
41
42import (
43 "errors"
44 "sync"
45
46 "github.com/bnema/vev/internal/ports"
47)
48
49// Aggregate-listener sentinels. They are typed so a caller can classify a
50// configuration, lifecycle, or registration outcome without matching on
51// message text.
52var (
53 // ErrAggregateConfig reports an aggregate listener operation on a nil
54 // receiver, or a registration of a nil child.
55 ErrAggregateConfig = errors.New("daemonmux: invalid aggregate listener configuration")
56 // ErrAggregateClosed reports a registration on, or an Accept of, an
57 // aggregate listener that was closed.
58 ErrAggregateClosed = errors.New("daemonmux: aggregate listener is closed")
59 // ErrAggregateChildTaken reports a registration of a child that is already
60 // registered.
61 ErrAggregateChildTaken = errors.New("daemonmux: aggregate child is already registered")
62 // ErrAggregateUnknownChild reports a removal of a child that is not
63 // registered, whether it was never registered or already deregistered
64 // because its physical carriage reached its terminal outcome.
65 ErrAggregateUnknownChild = errors.New("daemonmux: aggregate child is not registered")
66)
67
68// MaxAggregateQueue bounds the accepted-but-undelivered typed connections the
69// aggregate listener holds across every child. It is the per-physical-listener
70// accept bound, so the aggregate's own memory is never larger than one child's.
71// A child forwarder may additionally hold one connection while the queue is
72// full, so the aggregate's total retained connections are bounded by this value
73// plus the number of registered children.
74const MaxAggregateQueue = MaxAcceptQueue
75
76// AggregateListener is the daemon's global ports.ServerListener over several
77// independent physical listeners. It is safe for concurrent use.
78type AggregateListener struct {
79 limit int
80
81 // mu guards the child registry, the accepted-but-undelivered FIFO, and the
82 // closed flag. cond broadcasts every FIFO and lifecycle change so a blocked
83 // Accept and a child forwarder waiting for space never miss a wakeup.
84 mu sync.Mutex
85 cond *sync.Cond
86 children map[ports.ServerListener]*aggregateChild
87 queue []ports.ServerConnection
88 closed bool
89
90 closeOnce sync.Once
91}
92
93// aggregateChild is one registered physical listener and the lifecycle of its
94// single forwarding goroutine. quit is the per-child stop signal Remove and the
95// forwarder's own shutdown observe; done closes when the forwarder returns.
96type aggregateChild struct {
97 listener ports.ServerListener
98 quit chan struct{}
99 done chan struct{}
100}
101
102var _ ports.ServerListener = (*AggregateListener)(nil)
103
104// NewAggregateListener returns an empty aggregate listener under the package
105// policy: the aggregate queue is bounded by MaxAggregateQueue. Children are
106// registered and removed explicitly; the aggregate owns every registered child
107// and closes it on Close.
108func NewAggregateListener() *AggregateListener {
109 return newAggregateListener(MaxAggregateQueue)
110}
111
112// newAggregateListener builds an aggregate listener from an explicit queue
113// bound, so a test can tighten it and prove the bound and its backpressure
114// without changing live behavior. A non-positive bound falls back to the
115// package policy.
116func newAggregateListener(limit int) *AggregateListener {
117 if limit <= 0 {
118 limit = MaxAggregateQueue
119 }
120 listener := &AggregateListener{
121 limit: limit,
122 children: make(map[ports.ServerListener]*aggregateChild),
123 }
124 listener.cond = sync.NewCond(&listener.mu)
125 return listener
126}
127
128// Addr names the aggregate's carriage for ports.ServerListener. The aggregate
129// multiplexes several physical carriages into one accept stream, so it has no
130// single per-child address.
131func (a *AggregateListener) Addr() string { return "daemonmux" }
132
133// Register takes ownership of one physical child listener and starts its single
134// forwarding goroutine. Registering a nil child is refused with
135// ErrAggregateConfig, a child that is already registered with
136// ErrAggregateChildTaken, and any child after Close with ErrAggregateClosed. A
137// child whose Accept is already terminal is deregistered by its forwarder as
138// soon as it runs, without failing the aggregate. The child is identified by
139// its interface value, so a child is a pointer-like, comparable listener as
140// every ports.ServerListener implementation in this repository is.
141func (a *AggregateListener) Register(child ports.ServerListener) error {
142 if a == nil || child == nil {
143 return ErrAggregateConfig
144 }
145 a.mu.Lock()
146 if a.closed {
147 a.mu.Unlock()
148 return ErrAggregateClosed
149 }
150 if _, exists := a.children[child]; exists {
151 a.mu.Unlock()
152 return ErrAggregateChildTaken
153 }
154 entry := &aggregateChild{
155 listener: child,
156 quit: make(chan struct{}),
157 done: make(chan struct{}),
158 }
159 a.children[child] = entry
160 a.mu.Unlock()
161 go a.forward(entry)
162 return nil
163}
164
165// Remove closes one registered child and joins its forwarding goroutine without
166// touching a sibling, the connections already queued from that child, or a
167// connection already delivered to the daemon. It is safe to call concurrently
168// with Accept, Close, and another Remove; a child that was never registered or
169// already deregistered because its carriage was lost reports
170// ErrAggregateUnknownChild. A removal that races Close may either win (and
171// close the child) or lose (and report the child already gone); either way the
172// child is closed exactly once.
173func (a *AggregateListener) Remove(child ports.ServerListener) error {
174 if a == nil || child == nil {
175 return ErrAggregateConfig
176 }
177 a.mu.Lock()
178 entry, ok := a.children[child]
179 if !ok {
180 a.mu.Unlock()
181 return ErrAggregateUnknownChild
182 }
183 delete(a.children, child)
184 close(entry.quit)
185 // Wake a forwarder that is waiting for queue space, so it observes quit and
186 // releases the connection it was holding instead of waiting forever.
187 a.cond.Broadcast()
188 a.mu.Unlock()
189
190 _ = entry.listener.Close()
191 <-entry.done
192 return nil
193}
194
195// Accept returns the oldest typed connection accepted from any registered child,
196// blocking until one is available, and fails with ErrAggregateClosed after
197// Close. It never surfaces a child's Accept failure: a lost or closed physical
198// carriage is isolated by its forwarder and cannot stop the aggregate's accept
199// stream. A nil receiver is refused with ErrAggregateConfig.
200func (a *AggregateListener) Accept() (ports.ServerConnection, error) {
201 if a == nil {
202 return nil, ErrAggregateConfig
203 }
204 a.mu.Lock()
205 defer a.mu.Unlock()
206 for len(a.queue) == 0 {
207 if a.closed {
208 return nil, ErrAggregateClosed
209 }
210 a.cond.Wait()
211 }
212 connection := a.queue[0]
213 a.queue = a.queue[1:]
214 return connection, nil
215}
216
217// Close stops accepting: it unblocks every blocked Accept with
218// ErrAggregateClosed, clears every owned child (which unblocks and joins that
219// child's own accept loop), joins every forwarding goroutine, and closes every
220// connection that was taken from a child but never handed to the daemon. Close
221// is idempotent and concurrent-safe; closeOnce guarantees a single drain across
222// concurrent callers. It never touches a connection already handed to the
223// daemon, which owns it from then on.
224func (a *AggregateListener) Close() error {
225 if a == nil {
226 return nil
227 }
228 a.closeOnce.Do(func() {
229 a.mu.Lock()
230 a.closed = true
231 queued := a.queue
232 a.queue = nil
233 children := make([]*aggregateChild, 0, len(a.children))
234 for _, entry := range a.children {
235 children = append(children, entry)
236 }
237 a.children = make(map[ports.ServerListener]*aggregateChild)
238 // Wake every blocked Accept (which now returns ErrAggregateClosed) and
239 // every forwarder waiting for queue space (which now releases its
240 // in-flight connection).
241 a.cond.Broadcast()
242 a.mu.Unlock()
243
244 // Close every owned child first: this unblocks a forwarder parked in
245 // the child's Accept so it can observe the closed aggregate and exit.
246 for _, entry := range children {
247 _ = entry.listener.Close()
248 }
249 for _, entry := range children {
250 <-entry.done
251 }
252 for _, connection := range queued {
253 if connection != nil {
254 _ = connection.Close()
255 }
256 }
257 })
258 return nil
259}
260
261// forward is one child's single forwarding goroutine. It takes one typed
262// connection at a time from the child and hands it to the aggregate's bounded
263// FIFO; when the child's Accept fails for any reason it deregisters the child
264// and returns, so the failure is isolated from the aggregate and every sibling.
265// When the aggregate is closed or the child is removed it releases the
266// connection it is holding and returns.
267func (a *AggregateListener) forward(entry *aggregateChild) {
268 defer close(entry.done)
269 for {
270 connection, err := entry.listener.Accept()
271 if err != nil {
272 if connection != nil {
273 _ = connection.Close()
274 }
275 a.deregister(entry)
276 return
277 }
278 if !a.enqueue(entry, connection) {
279 if connection != nil {
280 _ = connection.Close()
281 }
282 a.deregister(entry)
283 return
284 }
285 }
286}
287
288// enqueue appends one accepted connection to the bounded FIFO, waiting for
289// space while the queue is full. It reports false when the aggregate was closed
290// or the child was removed before the connection could be queued, so the caller
291// releases it. It never blocks a child's own accept loop: while the queue is
292// full the child is left to its own bounded queue and refuses a stream it
293// cannot hold.
294func (a *AggregateListener) enqueue(entry *aggregateChild, connection ports.ServerConnection) bool {
295 a.mu.Lock()
296 defer a.mu.Unlock()
297 for len(a.queue) >= a.limit && !a.closed && !childStopped(entry) {
298 a.cond.Wait()
299 }
300 if a.closed || childStopped(entry) {
301 return false
302 }
303 a.queue = append(a.queue, connection)
304 a.cond.Broadcast()
305 return true
306}
307
308// deregister removes one forwarder's child from the registry if it is still
309// registered. It is a no-op for a child already removed by Remove or by Close,
310// so a natural terminal outcome and an explicit removal never double-close.
311func (a *AggregateListener) deregister(entry *aggregateChild) {
312 a.mu.Lock()
313 if a.children[entry.listener] == entry {
314 delete(a.children, entry.listener)
315 }
316 a.mu.Unlock()
317}
318
319// childStopped reports whether one child's forwarder was asked to stop. The
320// caller holds a.mu.
321func childStopped(entry *aggregateChild) bool {
322 select {
323 case <-entry.quit:
324 return true
325 default:
326 return false
327 }
328}