ringbuf.mx raw
1 //:build linux
2
3 package runtime
4
5 import "unsafe"
6
7 // SPSC ring buffer for lock-free cross-domain channel IPC.
8 // Allocated via mmap(MAP_SHARED) so the buffer survives fork and both
9 // parent and child access the same physical memory. One ring per
10 // channel direction - no multiplexing, no chanID headers.
11 //
12 // Layout: 16-byte header followed by data[size] bytes.
13 // [0:4] writeIdx - monotonically increasing, producer-written
14 // [4:8] readIdx - monotonically increasing, consumer-written
15 // [8:12] size - data area size (power of 2)
16 // [12:16] closed - 0=open, 1=closed
17 // [16:] data
18 //
19 // Messages: [len:4][payload:len][pad to align 8].
20 // Indices wrap via mask (size-1). Both sides use atomic load/store
21 // for cross-process visibility.
22
23 const (
24 ringHeaderSize = 16
25 ringDefaultSize = 1 << 16 // 64KB data area
26 flag_MAP_SHARED = 0x1
27 )
28
29 type ringBuf struct {
30 writeIdx uint32
31 readIdx uint32
32 size uint32
33 closed uint32
34 }
35
36 func ringCreate(dataSize uint32) uintptr {
37 total := uintptr(ringHeaderSize) + uintptr(dataSize)
38 p := mmap(nil, total, flag_PROT_READ|flag_PROT_WRITE,
39 flag_MAP_SHARED|flag_MAP_ANONYMOUS, -1, 0)
40 if uintptr(p) == 0 || uintptr(p) == ^uintptr(0) {
41 return 0
42 }
43 hdr := (*ringBuf)(p)
44 hdr.writeIdx = 0
45 hdr.readIdx = 0
46 hdr.size = dataSize
47 hdr.closed = 0
48 return uintptr(p)
49 }
50
51 func ringDestroy(ring uintptr) {
52 if ring == 0 {
53 return
54 }
55 hdr := (*ringBuf)(unsafe.Pointer(ring))
56 total := uintptr(ringHeaderSize) + uintptr(hdr.size)
57 munmap(unsafe.Pointer(ring), total)
58 }
59
60 func ringDataPtr(ring uintptr) unsafe.Pointer {
61 return unsafe.Pointer(ring + ringHeaderSize)
62 }
63
64 func ringAvail(ring uintptr) uint32 {
65 hdr := (*ringBuf)(unsafe.Pointer(ring))
66 w := atomicLoadU32(&hdr.writeIdx)
67 r := atomicLoadU32(&hdr.readIdx)
68 return w - r
69 }
70
71 func ringSpace(ring uintptr) uint32 {
72 hdr := (*ringBuf)(unsafe.Pointer(ring))
73 w := atomicLoadU32(&hdr.writeIdx)
74 r := atomicLoadU32(&hdr.readIdx)
75 return hdr.size - (w - r)
76 }
77
78 func ringIsClosed(ring uintptr) bool {
79 hdr := (*ringBuf)(unsafe.Pointer(ring))
80 return atomicLoadU32(&hdr.closed) != 0
81 }
82
83 func ringClose(ring uintptr) {
84 hdr := (*ringBuf)(unsafe.Pointer(ring))
85 atomicStoreU32(&hdr.closed, 1)
86 }
87
88 func ringAlign8(n uint32) uint32 {
89 return (n + 7) &^ 7
90 }
91
92 // ringWrite copies src[0:n] into the ring at offset pos, handling wrap.
93 func ringWrite(ring uintptr, pos, n uint32, src unsafe.Pointer) {
94 hdr := (*ringBuf)(unsafe.Pointer(ring))
95 data := ringDataPtr(ring)
96 mask := hdr.size - 1
97 start := pos & mask
98 if start+n <= hdr.size {
99 memcpy(unsafe.Add(data, uintptr(start)), src, uintptr(n))
100 } else {
101 first := hdr.size - start
102 memcpy(unsafe.Add(data, uintptr(start)), src, uintptr(first))
103 memcpy(data, unsafe.Add(src, uintptr(first)), uintptr(n-first))
104 }
105 }
106
107 // ringRead copies n bytes from the ring at offset pos into dst, handling wrap.
108 func ringRead(ring uintptr, pos, n uint32, dst unsafe.Pointer) {
109 hdr := (*ringBuf)(unsafe.Pointer(ring))
110 data := ringDataPtr(ring)
111 mask := hdr.size - 1
112 start := pos & mask
113 if start+n <= hdr.size {
114 memcpy(dst, unsafe.Add(data, uintptr(start)), uintptr(n))
115 } else {
116 first := hdr.size - start
117 memcpy(dst, unsafe.Add(data, uintptr(start)), uintptr(first))
118 memcpy(unsafe.Add(dst, uintptr(first)), data, uintptr(n-first))
119 }
120 }
121
122 // ringFits returns true if payload could ever fit in this ring buffer.
123 func ringFits(ring uintptr, payloadLen int32) bool {
124 if ring == 0 {
125 return false
126 }
127 hdr := (*ringBuf)(unsafe.Pointer(ring))
128 return ringAlign8(4+uint32(payloadLen)) <= hdr.size
129 }
130
131 // ringSend writes a length-prefixed message to the ring. Returns false
132 // if the ring is full (backpressure) or closed.
133 func ringSend(ring uintptr, payload []byte) bool {
134 if ring == 0 || ringIsClosed(ring) {
135 return false
136 }
137 hdr := (*ringBuf)(unsafe.Pointer(ring))
138 msgLen := uint32(len(payload))
139 total := ringAlign8(4 + msgLen)
140 if total > hdr.size {
141 return false
142 }
143 if ringSpace(ring) < total {
144 return false
145 }
146 w := atomicLoadU32(&hdr.writeIdx)
147 // Write length prefix.
148 var lenBuf [4]byte
149 lenBuf[0] = byte(msgLen >> 24)
150 lenBuf[1] = byte(msgLen >> 16)
151 lenBuf[2] = byte(msgLen >> 8)
152 lenBuf[3] = byte(msgLen)
153 ringWrite(ring, w, 4, unsafe.Pointer(&lenBuf[0]))
154 // Write payload.
155 if msgLen > 0 {
156 ringWrite(ring, w+4, msgLen, unsafe.Pointer(&payload[0]))
157 }
158 // Publish: advance writeIdx with release semantics.
159 atomicStoreU32(&hdr.writeIdx, w+total)
160 return true
161 }
162
163 // ringRecv reads one length-prefixed message from the ring. Returns
164 // (data, true) on success, (nil, false) if empty or closed.
165 func ringRecv(ring uintptr) ([]byte, bool) {
166 if ring == 0 {
167 return nil, false
168 }
169 hdr := (*ringBuf)(unsafe.Pointer(ring))
170 if ringAvail(ring) < 4 {
171 if ringIsClosed(ring) {
172 return nil, false
173 }
174 return nil, false
175 }
176 r := atomicLoadU32(&hdr.readIdx)
177 // Read length prefix.
178 var lenBuf [4]byte
179 ringRead(ring, r, 4, unsafe.Pointer(&lenBuf[0]))
180 msgLen := uint32(lenBuf[0])<<24 | uint32(lenBuf[1])<<16 | uint32(lenBuf[2])<<8 | uint32(lenBuf[3])
181 total := ringAlign8(4 + msgLen)
182 if ringAvail(ring) < total {
183 return nil, false
184 }
185 var data []byte
186 if msgLen > 0 {
187 data = []byte{:msgLen}
188 ringRead(ring, r+4, msgLen, unsafe.Pointer(&data[0]))
189 }
190 // Consume: advance readIdx with release semantics.
191 atomicStoreU32(&hdr.readIdx, r+total)
192 return data, true
193 }
194
195 // Atomic operations - use C builtins for cross-process visibility.
196 // On x86_64 regular aligned loads/stores are atomic but we use explicit
197 // atomics for ARM64 correctness and to prevent compiler reordering.
198
199 //:noinline
200 func atomicLoadU32(p *uint32) uint32 {
201 return moxie_atomic_load_u32(p)
202 }
203
204 //:noinline
205 func atomicStoreU32(p *uint32, val uint32) {
206 moxie_atomic_store_u32(p, val)
207 }
208
209 //export moxie_atomic_load_u32
210 func moxie_atomic_load_u32(p *uint32) uint32
211
212 //export moxie_atomic_store_u32
213 func moxie_atomic_store_u32(p *uint32, val uint32)
214
215 // spawnRingSet holds ring buffer pairs for the most recent spawn call.
216 // Set before fork, inherited by child. Each pair: [0]=parent→child, [1]=child→parent.
217 var spawnRingSet [16][2]uintptr
218 var spawnRingCount int32
219