1package ipc
2
3import (
4 "encoding/binary"
5 "errors"
6 "io"
7 "net"
8 "sync"
9
10 "github.com/bnema/vev/internal/ports"
11)
12
13// frameHeaderLen is the size, in bytes, of the length prefix that precedes
14// every frame on the wire.
15const frameHeaderLen = 4
16
17// ErrZeroLengthFrame is returned by Recv when a frame's length field is
18// zero. A valid frame always carries at least its type byte.
19var ErrZeroLengthFrame = errors.New("ipc: zero-length frame")
20
21// ErrFrameTooLarge is returned by Recv when a frame's length field exceeds
22// ports.MaxFrameLen, and by Send when a payload is too large to encode.
23var ErrFrameTooLarge = errors.New("ipc: frame exceeds maximum length")
24
25// ErrBackpressure means the bounded IPC egress queue is full. Callers using
26// SendAsync must treat it as a failed render emission rather than spawning an
27// unbounded queue of their own.
28var ErrBackpressure = errors.New("ipc: egress queue full")
29
30var errClosed = errors.New("ipc: closed")
31
32// sendQueueCapacity bounds frames accepted while a peer is not reading. It is
33// deliberately small: render output can be large and must never turn a wedged
34// peer into an unbounded memory commitment.
35const sendQueueCapacity = 8
36
37// unixTransport implements ports.Transport over a net.Conn (in practice an
38// AF_UNIX SOCK_STREAM connection, but any net.Conn works — this also lets
39// tests exercise it over net.Pipe).
40//
41// unixTransport has one writer goroutine which owns every conn.Write. Send
42// waits for that owner's wire attempt; SendAsync only performs a bounded,
43// ordered enqueue. Recv still has one caller at a time.
44type Option func(*unixTransport)
45
46// WithRuntimeObserver enables process-local transport marks. It accepts no
47// clock: timestamp ownership belongs to the configured observer.
48func WithRuntimeObserver(observer ports.SerializedRuntimeObserver) Option {
49 return func(t *unixTransport) { t.observer = observer }
50}
51
52type unixTransport struct {
53 conn net.Conn
54 observer ports.SerializedRuntimeObserver
55
56 operationMu sync.Mutex
57 operationCount int
58 observedClosing bool
59 operationsDone chan struct{}
60
61 // egressMu synchronizes Close with enqueue. Synchronous enqueuers are
62 // tracked separately so the writer cannot drain the queue until each has
63 // observed closure or finished enqueueing.
64 egressMu sync.Mutex
65 egressClosing bool
66 egressSenders int
67 egressSendersDone chan struct{}
68 egress chan sendRequest
69 done chan struct{}
70 writerDone chan struct{}
71 closeOnce sync.Once
72 closeErr error
73
74 // readBuf is reused across Recv calls (grow-once strategy): it grows
75 // to fit the largest frame seen so far and is never shrunk. The Frame
76 // returned to the caller is always a freshly allocated, right-sized
77 // copy, so callers may retain it indefinitely.
78 readBuf []byte
79}
80
81// NewTransport wraps conn as a ports.Transport speaking vev's framed
82// binary protocol.
83func NewTransport(conn net.Conn, opts ...Option) ports.Transport {
84 t := &unixTransport{
85 conn: conn,
86 egress: make(chan sendRequest, sendQueueCapacity),
87 done: make(chan struct{}),
88 writerDone: make(chan struct{}),
89 }
90 for _, opt := range opts {
91 if opt != nil {
92 opt(t)
93 }
94 }
95 go t.writeLoop()
96 return t
97}
98
99type sendRequest struct {
100 data []byte
101 done chan error // nil for SendAsync
102 end func(bool)
103}
104
105// Send queues f for the sole writer and waits for its wire attempt.
106func (t *unixTransport) Send(f ports.Frame) error {
107 data, err := marshalFrame(f)
108 if err != nil {
109 return err
110 }
111 end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
112 result := make(chan error, 1)
113 if err := t.enqueueWait(sendRequest{data: data, done: result, end: end}); err != nil {
114 end(false)
115 return err
116 }
117 select {
118 case err := <-result:
119 return err
120 case <-t.done:
121 return errClosed
122 }
123}
124
125// SendAsync accepts f for ordered background transmission without waiting for
126// a socket write. The queue is bounded and ownership includes a payload copy.
127func (t *unixTransport) SendAsync(f ports.Frame) error {
128 data, err := marshalFrame(f)
129 if err != nil {
130 return err
131 }
132 end := t.beginOperation(ports.RuntimeAdapterSendStart, uint64(len(f.Payload)))
133 if err := t.enqueueAsync(sendRequest{data: data, end: end}); err != nil {
134 end(false)
135 return err
136 }
137 return nil
138}
139
140func marshalFrame(f ports.Frame) ([]byte, error) {
141 n := 1 + len(f.Payload) // type + payload
142 if n > ports.MaxFrameLen {
143 return nil, ErrFrameTooLarge
144 }
145 buf := make([]byte, frameHeaderLen+n)
146 binary.BigEndian.PutUint32(buf[:frameHeaderLen], uint32(n))
147 buf[frameHeaderLen] = byte(f.Type)
148 copy(buf[frameHeaderLen+1:], f.Payload)
149 return buf, nil
150}
151
152// enqueueWait waits for space in the egress queue. Close wakes waiters and
153// the writer waits for all in-flight enqueues before draining queued requests.
154func (t *unixTransport) enqueueWait(req sendRequest) error {
155 t.egressMu.Lock()
156 if t.egressClosing {
157 t.egressMu.Unlock()
158 return errClosed
159 }
160 if t.egressSenders == 0 {
161 t.egressSendersDone = make(chan struct{})
162 }
163 t.egressSenders++
164 t.egressMu.Unlock()
165
166 select {
167 case <-t.done:
168 t.finishEgressSender()
169 return errClosed
170 case t.egress <- req:
171 t.finishEgressSender()
172 return nil
173 }
174}
175
176// enqueueAsync performs a bounded, nonblocking enqueue for SendAsync.
177func (t *unixTransport) enqueueAsync(req sendRequest) error {
178 t.egressMu.Lock()
179 defer t.egressMu.Unlock()
180 if t.egressClosing {
181 return errClosed
182 }
183 select {
184 case t.egress <- req:
185 return nil
186 default:
187 return ErrBackpressure
188 }
189}
190
191func (t *unixTransport) finishEgressSender() {
192 t.egressMu.Lock()
193 defer t.egressMu.Unlock()
194 t.egressSenders--
195 if t.egressSenders == 0 {
196 close(t.egressSendersDone)
197 }
198}
199
200func (t *unixTransport) waitForEgressSenders() {
201 t.egressMu.Lock()
202 done := t.egressSendersDone
203 t.egressMu.Unlock()
204 if done != nil {
205 <-done
206 }
207}
208
209func (t *unixTransport) writeLoop() {
210 defer close(t.writerDone)
211 for {
212 select {
213 case req := <-t.egress:
214 t.completeWrite(req, writeAll(t.conn, req.data))
215 case <-t.done:
216 t.waitForEgressSenders()
217 t.egressMu.Lock()
218 for {
219 select {
220 case req := <-t.egress:
221 t.completeWrite(req, errClosed)
222 default:
223 t.egressMu.Unlock()
224 return
225 }
226 }
227 }
228 }
229}
230
231func (t *unixTransport) completeWrite(req sendRequest, err error) {
232 if req.end != nil {
233 req.end(err == nil)
234 }
235 if req.done != nil {
236 req.done <- err
237 }
238}
239
240func writeAll(w io.Writer, data []byte) error {
241 for len(data) > 0 {
242 n, err := w.Write(data)
243 if n > 0 {
244 data = data[n:]
245 }
246 if err != nil {
247 return err
248 }
249 if n == 0 {
250 return io.ErrShortWrite
251 }
252 }
253 return nil
254}
255
256// Recv reads one frame: a 4-byte length, then a body of that many bytes
257// (type byte + payload). It blocks until a full frame arrives, the
258// connection is closed (io.EOF), or an error occurs.
259func (t *unixTransport) Recv() (ports.Frame, error) {
260 end := t.beginOperation(ports.RuntimeAdapterReceiveStart, 0)
261 var hdr [frameHeaderLen]byte
262 if _, err := io.ReadFull(t.conn, hdr[:]); err != nil {
263 end(false)
264 return ports.Frame{}, err
265 }
266
267 n := binary.BigEndian.Uint32(hdr[:])
268 if n == 0 {
269 end(false)
270 return ports.Frame{}, ErrZeroLengthFrame
271 }
272 if n > ports.MaxFrameLen {
273 end(false)
274 return ports.Frame{}, ErrFrameTooLarge
275 }
276
277 if cap(t.readBuf) < int(n) {
278 t.readBuf = make([]byte, n)
279 } else {
280 t.readBuf = t.readBuf[:n]
281 }
282 if _, err := io.ReadFull(t.conn, t.readBuf); err != nil {
283 end(false)
284 return ports.Frame{}, err
285 }
286
287 var payload []byte
288 if n > 1 {
289 payload = make([]byte, n-1)
290 copy(payload, t.readBuf[1:])
291 }
292 end(true)
293 return ports.Frame{Type: ports.MsgType(t.readBuf[0]), Payload: payload}, nil
294}
295
296func (t *unixTransport) beginOperation(start ports.RuntimeMarkKind, bytes uint64) func(bool) {
297 if t.observer == nil {
298 return func(bool) {}
299 }
300 if !t.beginObservedOperation() {
301 return func(bool) {}
302 }
303 correlation := ports.NewRuntimeCorrelation()
304 t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, start, bytes, true))
305 end := ports.RuntimeAdapterSendEnd
306 if start == ports.RuntimeAdapterReceiveStart {
307 end = ports.RuntimeAdapterReceiveEnd
308 }
309 return func(valid bool) {
310 defer t.finishObservedOperation()
311 t.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("ipc", correlation, end, bytes, valid))
312 }
313}
314
315func (t *unixTransport) beginObservedOperation() bool {
316 t.operationMu.Lock()
317 defer t.operationMu.Unlock()
318 if t.observedClosing {
319 return false
320 }
321 if t.operationCount == 0 {
322 t.operationsDone = make(chan struct{})
323 }
324 t.operationCount++
325 return true
326}
327
328func (t *unixTransport) finishObservedOperation() {
329 t.operationMu.Lock()
330 defer t.operationMu.Unlock()
331 t.operationCount--
332 if t.operationCount == 0 {
333 close(t.operationsDone)
334 }
335}
336
337func (t *unixTransport) beginShutdown() <-chan struct{} {
338 t.operationMu.Lock()
339 defer t.operationMu.Unlock()
340 t.observedClosing = true
341 if t.operationCount == 0 {
342 return nil
343 }
344 return t.operationsDone
345}
346
347// Close interrupts Recv and the writer's current Write, fails queued frames,
348// and waits until the sole writer and all observed operations have stopped.
349func (t *unixTransport) Close() error {
350 t.closeOnce.Do(func() {
351 operationsDone := t.beginShutdown()
352 t.egressMu.Lock()
353 t.egressClosing = true
354 close(t.done)
355 t.egressMu.Unlock()
356 t.closeErr = t.conn.Close()
357 <-t.writerDone
358 if operationsDone != nil {
359 <-operationsDone
360 }
361 })
362 return t.closeErr
363}