1package xturbo
2
3import (
4 "bytes"
5 "encoding/binary"
6 "errors"
7 "syscall"
8 "testing"
9
10 "golang.org/x/sys/unix"
11)
12
13func TestBufRequest(t *testing.T) {
14 b := NewBuf(4)
15 p, seq := b.Request(8, 2, 4)
16 if seq != 1 || len(p) != 4 || !bytes.Equal(b.Bytes(), []byte{8, 2, 2, 0, 0, 0, 0, 0}) {
17 t.Fatalf("%x %d", b.Bytes(), seq)
18 }
19 p[0] = 255
20 b.Reset()
21 p, seq = b.Request(14, 0, 4)
22 if seq != 2 || p[0] != 0 || b.Next() != 3 || b.Len() != 8 {
23 t.Fatal("reset sequence/zeroing")
24 }
25 first := errors.New("first")
26 b.Fail(first)
27 b.Fail(ErrBadLength)
28 p, seq = b.Request(1, 0, 0)
29 if p != nil || seq != 0 || b.Err() != first {
30 t.Fatal("sticky error")
31 }
32}
33
34func TestNewBufDefault(t *testing.T) {
35 b := NewBuf(0)
36 if cap(b.b) != 64<<10 || b.maxReq != 262140 || b.Next() != 1 || b.Len() != 0 {
37 t.Fatalf("cap %d maxReq %d next %d", cap(b.b), b.maxReq, b.Next())
38 }
39}
40
41func TestBufLimits(t *testing.T) {
42 for _, body := range []int{-1, 1, 262140, 1 << 20} {
43 b := NewBuf(0)
44 p, s := b.Request(1, 0, body)
45 if p != nil || s != 0 || b.Err() == nil || b.Len() != 0 || b.Next() != 1 {
46 t.Fatal(body)
47 }
48 }
49 // Largest core request: 0xFFFF units, 4-byte header.
50 b := NewBuf(0)
51 if p, s := b.Request(1, 0, 262136); s != 1 || len(p) != 262136 || binary.LittleEndian.Uint16(b.Bytes()[2:]) != 0xffff {
52 t.Fatal("largest core request")
53 }
54}
55
56func TestBufMaxReq(t *testing.T) {
57 b := NewBuf(0)
58 b.setMaxReqWords(70000) // 280000 bytes
59 if b.maxReq != 280000 {
60 t.Fatal(b.maxReq)
61 }
62 // body+8 == 280000 fits exactly.
63 if p, s := b.Request(1, 0, 280000-8); s != 1 || len(p) != 280000-8 {
64 t.Fatal("max-size big request")
65 }
66 if got := binary.LittleEndian.Uint32(b.Bytes()[4:]); got != 70000 {
67 t.Fatal(got)
68 }
69 if p, s := b.Request(1, 0, 280000-4); p != nil || s != 0 || b.Err() != ErrRequestTooLarge {
70 t.Fatal("over maxReq accepted")
71 }
72 b = NewBuf(0)
73 b.setMaxReqWords(0xffffffff)
74 if b.maxReq != 0xfffffffc {
75 t.Fatal("maxReq clamp", b.maxReq)
76 }
77}
78
79func TestBufSeqOverflow(t *testing.T) {
80 b := NewBuf(0)
81 b.next = 0xffff
82 if p, s := b.Request(1, 0, 0); p != nil || s != 0 || b.Err() != ErrSeqOverflow {
83 t.Fatal("overflow")
84 }
85 b = NewBuf(0)
86 b.next = 0xfffe
87 if _, s := b.Request(1, 0, 0); s != 0xfffe {
88 t.Fatal("last allowed seq", s)
89 }
90 if _, s := b.Request(1, 0, 0); s != 0 || b.Err() != ErrSeqOverflow {
91 t.Fatal("overflow at next-acked == 0xffff")
92 }
93 b = NewBuf(0)
94 b.next = 0xffff
95 b.ack(5)
96 b.ack(3) // stale acks never move acked backwards
97 b.ack(0xffff)
98 if b.acked != 5 {
99 t.Fatal("ack", b.acked)
100 }
101 if _, s := b.Request(1, 0, 0); s != 0xffff || b.Err() != nil {
102 t.Fatal("ack must free sequence space")
103 }
104}
105
106func TestBufGrowth(t *testing.T) {
107 b := NewBuf(16)
108 for i := range 15 {
109 b.Request(1, 0, 4)
110 want := 16
111 for want < (i+1)*8 {
112 want *= 2
113 }
114 if c := cap(b.b); c != want {
115 t.Fatal("cap", i, c, want)
116 }
117 }
118 if cap(b.b) != 128 || b.Len() != 120 {
119 t.Fatal(cap(b.b), b.Len())
120 }
121 // One request bigger than double the capacity grows straight to fit.
122 b.Request(1, 0, 1000)
123 if cap(b.b) != 2048 || b.Len() != 1124 {
124 t.Fatal(cap(b.b), b.Len())
125 }
126 for i, v := range b.Bytes()[:120] {
127 if i%8 == 0 && v != 1 || i%8 == 2 && v != 2 {
128 t.Fatal("growth lost bytes", i)
129 }
130 }
131 b.Reset()
132 if cap(b.b) != 2048 {
133 t.Fatal("shrunk")
134 }
135}
136
137func TestAllocBufRequest(t *testing.T) {
138 for _, body := range []int{4, 65536} {
139 b := NewBuf(128 << 10)
140 if n := testing.AllocsPerRun(1000, func() { b.Reset(); b.Request(1, 0, body) }); n != 0 {
141 t.Fatal(n)
142 }
143 if b.Err() != nil {
144 t.Fatal(b.Err())
145 }
146 }
147}
148
149// socketpair returns a non-blocking stream pair with small kernel buffers so
150// EAGAIN is reached quickly.
151func socketpair(t testing.TB) (w, r int) {
152 t.Helper()
153 fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_NONBLOCK|unix.SOCK_CLOEXEC, 0)
154 if err != nil {
155 t.Fatal(err)
156 }
157 t.Cleanup(func() { unix.Close(fds[0]); unix.Close(fds[1]) })
158 for _, fd := range fds {
159 if err := unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_SNDBUF, 4096); err != nil {
160 t.Fatal(err)
161 }
162 if err := unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_RCVBUF, 4096); err != nil {
163 t.Fatal(err)
164 }
165 }
166 return fds[0], fds[1]
167}
168
169// drain reads everything currently readable from fd into dst.
170func drain(t testing.TB, fd int, dst *bytes.Buffer, scratch []byte) {
171 t.Helper()
172 for {
173 n, err := unix.Read(fd, scratch)
174 if err == unix.EAGAIN {
175 return
176 }
177 if err != nil {
178 t.Fatal(err)
179 }
180 dst.Write(scratch[:n])
181 }
182}
183
184// fill queues requests whose bodies carry their sequence number.
185func fill(b *Buf, n, body int) []byte {
186 for range n {
187 p, seq := b.Request(7, 0, body)
188 binary.LittleEndian.PutUint32(p, uint32(seq))
189 }
190 return bytes.Clone(b.Bytes())
191}
192
193func TestBufFlushPartialResume(t *testing.T) {
194 w, r := socketpair(t)
195 b := NewBuf(0)
196 want := fill(b, 2000, 60) // 128000 bytes, far beyond the socket buffers
197 err := b.flush(w)
198 if err != ErrWouldBlock {
199 t.Fatal("want ErrWouldBlock, got", err)
200 }
201 if b.flushed == 0 || b.Len() == 0 || b.Len() >= len(want) {
202 t.Fatalf("partial write state: flushed %d pending %d", b.flushed, b.Len())
203 }
204 if !bytes.Equal(b.Bytes(), want[len(want)-b.Len():]) {
205 t.Fatal("Bytes must return the unflushed tail")
206 }
207 if b.Err() != nil {
208 t.Fatal("EAGAIN must not be sticky")
209 }
210 // Repeated flush on a full socket keeps the state.
211 pending := b.Len()
212 if err := b.flush(w); err != ErrWouldBlock || b.Len() != pending {
213 t.Fatal("second flush on full socket", err, b.Len())
214 }
215 var got bytes.Buffer
216 scratch := make([]byte, 64<<10)
217 for rounds := 0; ; rounds++ {
218 drain(t, r, &got, scratch)
219 err := b.flush(w)
220 if err == nil {
221 break
222 }
223 if err != ErrWouldBlock || rounds > 10000 {
224 t.Fatal(err, rounds)
225 }
226 }
227 drain(t, r, &got, scratch)
228 if !bytes.Equal(got.Bytes(), want) {
229 t.Fatalf("stream mismatch: got %d bytes want %d", got.Len(), len(want))
230 }
231 if b.Len() != 0 || b.flushed != 0 || len(b.b) != 0 || cap(b.b) != 128<<10 {
232 t.Fatal("full flush must compact", b.Len(), b.flushed, len(b.b), cap(b.b))
233 }
234 if err := b.flush(w); err != nil {
235 t.Fatal("empty flush", err)
236 }
237}
238
239func TestBufCompactsAfterPartialFlush(t *testing.T) {
240 w, _ := socketpair(t)
241 b := NewBuf(0)
242 fill(b, 1000, 60) // 64000 bytes in a 64 KiB buffer
243 if err := b.flush(w); err != ErrWouldBlock {
244 t.Fatal(err)
245 }
246 flushed, pending := b.flushed, b.Len()
247 tail := append([]byte(nil), b.Bytes()...)
248 // Does not fit after len(b.b) but fits once flushed bytes are dropped.
249 body := (cap(b.b) - pending - 4) &^ 3
250 if body+4 <= cap(b.b)-len(b.b) {
251 t.Fatal("test setup: request fits without compaction", flushed)
252 }
253 p, seq := b.Request(9, 0, body)
254 if p == nil || cap(b.b) != 64<<10 || b.flushed != 0 {
255 t.Fatalf("compaction: seq %d cap %d flushed %d body %d", seq, cap(b.b), b.flushed, body)
256 }
257 if !bytes.Equal(b.Bytes()[:pending], tail) || b.Bytes()[pending] != 9 {
258 t.Fatal("compaction lost pending bytes")
259 }
260}
261
262func TestBufGrowthKeepsPendingAfterPartialFlush(t *testing.T) {
263 w, _ := socketpair(t)
264 b := NewBuf(0)
265 want := fill(b, 1000, 60)
266 if err := b.flush(w); err != ErrWouldBlock {
267 t.Fatal(err)
268 }
269 pending := b.Len()
270 b.Request(9, 0, 100000)
271 if cap(b.b) != 256<<10 || b.flushed != 0 || !bytes.Equal(b.Bytes()[:pending], want[len(want)-pending:]) {
272 t.Fatal("growth with flushed prefix", cap(b.b), b.flushed)
273 }
274}
275
276func TestBufFlushClosedPeer(t *testing.T) {
277 w, r := socketpair(t)
278 b := NewBuf(0)
279 fill(b, 1, 4)
280 unix.Close(r)
281 if err := b.flush(w); err != ErrClosed || b.Err() != ErrClosed {
282 t.Fatal("want sticky ErrClosed, got", err)
283 }
284 if p, s := b.Request(1, 0, 0); p != nil || s != 0 {
285 t.Fatal("request after dead connection")
286 }
287 if err := b.flush(w); err != ErrClosed {
288 t.Fatal("flush must return sticky error", err)
289 }
290}
291
292func TestBufFlushOtherError(t *testing.T) {
293 b := NewBuf(0)
294 fill(b, 1, 4)
295 err := b.flush(-1)
296 if !errors.Is(err, syscall.EBADF) || b.Err() != err {
297 t.Fatal("want sticky EBADF, got", err)
298 }
299 if b.flush(-1) != err {
300 t.Fatal("sticky")
301 }
302}
303
304func TestBufFlushStickyBeforeWrite(t *testing.T) {
305 w, r := socketpair(t)
306 b := NewBuf(0)
307 fill(b, 1, 4)
308 b.Fail(ErrBadLength)
309 if err := b.flush(w); err != ErrBadLength {
310 t.Fatal(err)
311 }
312 if n, err := unix.Read(r, make([]byte, 16)); err != unix.EAGAIN {
313 t.Fatal("bytes written after sticky error", n)
314 }
315}
316
317func TestAllocBufFlush(t *testing.T) {
318 w, r := socketpair(t)
319 b := NewBuf(0)
320 scratch := make([]byte, 64<<10)
321 n := testing.AllocsPerRun(1000, func() {
322 for range 8 {
323 b.Request(8, 0, 4)
324 }
325 if err := b.flush(w); err != nil {
326 t.Fatal(err)
327 }
328 if _, err := unix.Read(r, scratch); err != nil {
329 t.Fatal(err)
330 }
331 b.acked = b.next - 1
332 })
333 if n != 0 {
334 t.Fatal(n)
335 }
336 // EAGAIN path.
337 fill(b, 2000, 60)
338 n = testing.AllocsPerRun(100, func() {
339 if err := b.flush(w); err != ErrWouldBlock {
340 t.Fatal(err)
341 }
342 })
343 if n != 0 {
344 t.Fatal("EAGAIN path allocates", n)
345 }
346}
347
348func BenchmarkBufRequest(b *testing.B) {
349 buf := NewBuf(0)
350 b.ReportAllocs()
351 for b.Loop() {
352 buf.Reset()
353 buf.acked = buf.next
354 buf.Request(1, 0, 4)
355 }
356}
357
358// BenchmarkBufRequestFlush queues 64 MapWindow-sized requests and flushes
359// them to a socketpair per iteration.
360func BenchmarkBufRequestFlush(b *testing.B) {
361 fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_NONBLOCK|unix.SOCK_CLOEXEC, 0)
362 if err != nil {
363 b.Fatal(err)
364 }
365 defer unix.Close(fds[0])
366 defer unix.Close(fds[1])
367 buf := NewBuf(0)
368 scratch := make([]byte, 64<<10)
369 b.ReportAllocs()
370 b.SetBytes(64 * 8)
371 for b.Loop() {
372 for range 64 {
373 p, _ := buf.Request(8, 0, 4)
374 binary.LittleEndian.PutUint32(p, 0x0300009f)
375 }
376 if err := buf.flush(fds[0]); err != nil {
377 b.Fatal(err)
378 }
379 if _, err := unix.Read(fds[1], scratch); err != nil {
380 b.Fatal(err)
381 }
382 buf.acked = buf.next - 1
383 }
384}