detached fc44844

feat(runtime): complete Buf with growth, BIG-REQUESTS limits and non-blocking flush

commit
fc448448c3b3c18fc2b24db8b24eb1cc36a4f65a
author
brice · · 2026-10-22 08:04 UTC
parent
d08e097
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
8 files · +525 -30
Split
UnifiedSplit
A bigreq_test.go +36 -0
@@ -0,0 +1,36 @@
1+package xturbo_test
2+
3+import (
4+ "bytes"
5+ "testing"
6+
7+ "bnema.dev/xturbo"
8+ "bnema.dev/xturbo/internal/xtest"
9+ "bnema.dev/xturbo/xproto"
10+)
11+
12+func TestBigRequestGolden(t *testing.T) {
13+ prefix, err := xtest.LoadHex("testdata/golden/xproto/ChangePropertyBig.req.hex")
14+ if err != nil {
15+ t.Fatal(err)
16+ }
17+ data := make([]byte, 262120)
18+ for i := range data {
19+ data[i] = byte(i)
20+ }
21+ want := append(prefix, data...)
22+
23+ b := xturbo.NewBuf(0)
24+ if xproto.ChangeProperty(b, xproto.PropModeReplace, 0x0300009f, 0xe7, 0x1f, 8, uint32(len(data)), data) != 0 || b.Err() != xturbo.ErrRequestTooLarge || b.Len() != 0 {
25+ t.Fatal("core limit must reject a big request before BIG-REQUESTS is enabled")
26+ }
27+
28+ b = xturbo.NewBuf(0)
29+ xturbo.SetMaxReqWords(b, 0x400000)
30+ if xproto.ChangeProperty(b, xproto.PropModeReplace, 0x0300009f, 0xe7, 0x1f, 8, uint32(len(data)), data) != 1 || b.Err() != nil {
31+ t.Fatal(b.Err())
32+ }
33+ if !bytes.Equal(b.Bytes(), want) {
34+ t.Fatalf("got %x want %x", b.Bytes()[:32], want[:32])
35+ }
36+}
@@ -0,0 +1,36 @@
1+package xturbo_test
2+
3+import (
4+ "bytes"
5+ "testing"
6+
7+ "bnema.dev/xturbo"
8+ "bnema.dev/xturbo/internal/xtest"
9+ "bnema.dev/xturbo/xproto"
10+)
11+
12+func TestBigRequestGolden(t *testing.T) {
13+ prefix, err := xtest.LoadHex("testdata/golden/xproto/ChangePropertyBig.req.hex")
14+ if err != nil {
15+ t.Fatal(err)
16+ }
17+ data := make([]byte, 262120)
18+ for i := range data {
19+ data[i] = byte(i)
20+ }
21+ want := append(prefix, data...)
22+
23+ b := xturbo.NewBuf(0)
24+ if xproto.ChangeProperty(b, xproto.PropModeReplace, 0x0300009f, 0xe7, 0x1f, 8, uint32(len(data)), data) != 0 || b.Err() != xturbo.ErrRequestTooLarge || b.Len() != 0 {
25+ t.Fatal("core limit must reject a big request before BIG-REQUESTS is enabled")
26+ }
27+
28+ b = xturbo.NewBuf(0)
29+ xturbo.SetMaxReqWords(b, 0x400000)
30+ if xproto.ChangeProperty(b, xproto.PropModeReplace, 0x0300009f, 0xe7, 0x1f, 8, uint32(len(data)), data) != 1 || b.Err() != nil {
31+ t.Fatal(b.Err())
32+ }
33+ if !bytes.Equal(b.Bytes(), want) {
34+ t.Fatalf("got %x want %x", b.Bytes()[:32], want[:32])
35+ }
36+}
M buf.go +113 -24
@@ -3,20 +3,35 @@ package xturbo
33 import (
44 "encoding/binary"
55 "errors"
6+ "io"
7+ "os"
8+
9+ "golang.org/x/sys/unix"
610 )
711
812 // Seq identifies a request; zero is invalid.
913 type Seq uint64
1014
1115 var (
16+ ErrWouldBlock = errors.New("xturbo: write would block")
17+ ErrClosed = errors.New("xturbo: connection closed")
1218 ErrRequestTooLarge = errors.New("xturbo: request too large")
1319 ErrExtMissing = errors.New("xturbo: extension missing")
1420 ErrBadLength = errors.New("xturbo: bad length")
1521 ErrSeqOverflow = errors.New("xturbo: sequence overflow")
1622 )
1723
18-// Buf is the caller-owned request encoder. The transport will own flushing and
19-// acknowledgment in Phase 3. It is not safe for concurrent use.
24+const (
25+ defaultBufSize = 64 << 10
26+ // coreMaxReq is the largest core request in bytes (0xFFFF four-byte units).
27+ coreMaxReq = 0xffff * 4
28+ // maxSeqAhead bounds unacknowledged requests so 16-bit wire sequence
29+ // numbers widen unambiguously.
30+ maxSeqAhead = 0xffff
31+)
32+
33+// Buf is the caller-owned request encoder. Conn owns flushing and
34+// acknowledgment. It is not safe for concurrent use.
2035 type Buf struct {
2136 Ext *ExtTable
2237 b []byte
@@ -26,25 +41,36 @@ type Buf struct {
2641 err error
2742 }
2843
44+// NewBuf returns an encoder with size bytes preallocated; size <= 0 means 64 KiB.
2945 func NewBuf(size int) *Buf {
3046 if size <= 0 {
31- size = 64 << 10
47+ size = defaultBufSize
3248 }
33- return &Buf{b: make([]byte, 0, size), next: 1, maxReq: 262140}
49+ return &Buf{b: make([]byte, 0, size), next: 1, maxReq: coreMaxReq}
3450 }
51+
52+// Fail records err as the sticky error; the first error wins.
3553 func (b *Buf) Fail(err error) {
3654 if b.err == nil {
3755 b.err = err
3856 }
3957 }
40-func (b *Buf) Err() error { return b.err }
41-func (b *Buf) Reset() { b.b = b.b[:0]; b.flushed = 0 }
58+func (b *Buf) Err() error { return b.err }
59+
60+// Reset drops unflushed bytes and keeps sequence counters. Tests only.
61+func (b *Buf) Reset() { b.b = b.b[:0]; b.flushed = 0 }
62+
63+// Bytes returns the unflushed bytes.
4264 func (b *Buf) Bytes() []byte { return b.b[b.flushed:] }
4365 func (b *Buf) Len() int { return len(b.b) - b.flushed }
44-func (b *Buf) Next() Seq { return b.next }
4566
46-// Request reserves a zeroed, four-byte-aligned body after the wire header.
47-// BIG-REQUESTS enablement is transport-owned; maxReq starts at the core limit.
67+// Next returns the sequence number the next Request will get.
68+func (b *Buf) Next() Seq { return b.next }
69+
70+// Request reserves a zeroed body of body bytes after the wire header and
71+// returns it with the request's sequence number. body must be a multiple of 4.
72+// Requests longer than 0xFFFF units get the BIG-REQUESTS 8-byte header.
73+// On failure it records a sticky error and returns (nil, 0).
4874 func (b *Buf) Request(major, data uint8, body int) ([]byte, Seq) {
4975 if b.err != nil {
5076 return nil, 0
@@ -53,29 +79,21 @@ func (b *Buf) Request(major, data uint8, body int) ([]byte, Seq) {
5379 b.Fail(ErrBadLength)
5480 return nil, 0
5581 }
56- if uint64(body)+4 > uint64(b.maxReq) {
82+ header := 4
83+ if uint64(body)+4 > coreMaxReq {
84+ header = 8
85+ }
86+ if uint64(body)+uint64(header) > uint64(b.maxReq) {
5787 b.Fail(ErrRequestTooLarge)
5888 return nil, 0
5989 }
60- if b.next-b.acked >= 0xffff {
90+ if b.next-b.acked >= maxSeqAhead {
6191 b.Fail(ErrSeqOverflow)
6292 return nil, 0
6393 }
64- header := 4
65- if (body+4)/4 > 0xffff {
66- header = 8
67- }
6894 total := body + header
95+ b.reserve(total)
6996 start := len(b.b)
70- if total > cap(b.b)-start {
71- capacity := cap(b.b) * 2
72- if capacity < start+total {
73- capacity = start + total
74- }
75- grown := make([]byte, start, capacity)
76- copy(grown, b.b)
77- b.b = grown
78- }
7997 b.b = b.b[:start+total]
8098 p := b.b[start:]
8199 clear(p)
@@ -90,3 +108,74 @@ func (b *Buf) Request(major, data uint8, body int) ([]byte, Seq) {
90108 b.next++
91109 return p[header:], seq
92110 }
111+
112+// reserve makes room for n more bytes. Flushed bytes are dropped first; the
113+// buffer then doubles until n fits. It never shrinks.
114+func (b *Buf) reserve(n int) {
115+ if n <= cap(b.b)-len(b.b) {
116+ return
117+ }
118+ pending := len(b.b) - b.flushed
119+ if b.flushed > 0 && n <= cap(b.b)-pending {
120+ copy(b.b, b.b[b.flushed:])
121+ b.b = b.b[:pending]
122+ b.flushed = 0
123+ return
124+ }
125+ capacity := max(cap(b.b), 1)
126+ for capacity-pending < n {
127+ capacity *= 2
128+ }
129+ grown := make([]byte, pending, capacity)
130+ copy(grown, b.b[b.flushed:])
131+ b.b = grown
132+ b.flushed = 0
133+}
134+
135+// setMaxReqWords sets the request size limit from the BIG-REQUESTS
136+// maximum_request_length (four-byte units). Conn calls it after enabling it.
137+func (b *Buf) setMaxReqWords(words uint32) {
138+ b.maxReq = uint32(min(uint64(words)*4, 0xfffffffc))
139+}
140+
141+// ack records seq as the highest sequence number seen on the read side.
142+func (b *Buf) ack(seq Seq) {
143+ if seq > b.acked && seq < b.next {
144+ b.acked = seq
145+ }
146+}
147+
148+// flush writes unflushed bytes to the non-blocking fd. It returns
149+// ErrWouldBlock when the socket is full (the remainder stays queued) and nil
150+// once everything is written. Any other failure becomes the sticky error.
151+// Conn.Flush wraps it.
152+func (b *Buf) flush(fd int) error {
153+ if b.err != nil {
154+ return b.err
155+ }
156+ for b.flushed < len(b.b) {
157+ n, err := unix.Write(fd, b.b[b.flushed:])
158+ switch {
159+ case err == unix.EINTR:
160+ continue
161+ case err == unix.EAGAIN:
162+ return ErrWouldBlock
163+ case err == unix.EPIPE || err == unix.ECONNRESET:
164+ b.Fail(ErrClosed)
165+ return b.err
166+ case err != nil:
167+ b.Fail(os.NewSyscallError("write", err))
168+ return b.err
169+ case n <= 0:
170+ b.Fail(io.ErrShortWrite)
171+ return b.err
172+ }
173+ if poison {
174+ fillPoison(b.b[b.flushed : b.flushed+n])
175+ }
176+ b.flushed += n
177+ }
178+ b.b = b.b[:0]
179+ b.flushed = 0
180+ return nil
181+}
@@ -3,20 +3,35 @@ package xturbo
3 import (3 import (
4 "encoding/binary"4 "encoding/binary"
5 "errors"5 "errors"
6+ "io"
7+ "os"
8+
9+ "golang.org/x/sys/unix"
6 )10 )
7 11
8 // Seq identifies a request; zero is invalid.12 // Seq identifies a request; zero is invalid.
9 type Seq uint6413 type Seq uint64
10 14
11 var (15 var (
16+ ErrWouldBlock = errors.New("xturbo: write would block")
17+ ErrClosed = errors.New("xturbo: connection closed")
12 ErrRequestTooLarge = errors.New("xturbo: request too large")18 ErrRequestTooLarge = errors.New("xturbo: request too large")
13 ErrExtMissing = errors.New("xturbo: extension missing")19 ErrExtMissing = errors.New("xturbo: extension missing")
14 ErrBadLength = errors.New("xturbo: bad length")20 ErrBadLength = errors.New("xturbo: bad length")
15 ErrSeqOverflow = errors.New("xturbo: sequence overflow")21 ErrSeqOverflow = errors.New("xturbo: sequence overflow")
16 )22 )
17 23
18-// Buf is the caller-owned request encoder. The transport will own flushing and24+const (
19-// acknowledgment in Phase 3. It is not safe for concurrent use.25+ defaultBufSize = 64 << 10
26+ // coreMaxReq is the largest core request in bytes (0xFFFF four-byte units).
27+ coreMaxReq = 0xffff * 4
28+ // maxSeqAhead bounds unacknowledged requests so 16-bit wire sequence
29+ // numbers widen unambiguously.
30+ maxSeqAhead = 0xffff
31+)
32+
33+// Buf is the caller-owned request encoder. Conn owns flushing and
34+// acknowledgment. It is not safe for concurrent use.
20 type Buf struct {35 type Buf struct {
21 Ext *ExtTable36 Ext *ExtTable
22 b []byte37 b []byte
@@ -26,25 +41,36 @@ type Buf struct {
26 err error41 err error
27 }42 }
28 43
44+// NewBuf returns an encoder with size bytes preallocated; size <= 0 means 64 KiB.
29 func NewBuf(size int) *Buf {45 func NewBuf(size int) *Buf {
30 if size <= 0 {46 if size <= 0 {
31- size = 64 << 1047+ size = defaultBufSize
32 }48 }
33- return &Buf{b: make([]byte, 0, size), next: 1, maxReq: 262140}49+ return &Buf{b: make([]byte, 0, size), next: 1, maxReq: coreMaxReq}
34 }50 }
51+
52+// Fail records err as the sticky error; the first error wins.
35 func (b *Buf) Fail(err error) {53 func (b *Buf) Fail(err error) {
36 if b.err == nil {54 if b.err == nil {
37 b.err = err55 b.err = err
38 }56 }
39 }57 }
40-func (b *Buf) Err() error { return b.err }58+func (b *Buf) Err() error { return b.err }
41-func (b *Buf) Reset() { b.b = b.b[:0]; b.flushed = 0 }59+
60+// Reset drops unflushed bytes and keeps sequence counters. Tests only.
61+func (b *Buf) Reset() { b.b = b.b[:0]; b.flushed = 0 }
62+
63+// Bytes returns the unflushed bytes.
42 func (b *Buf) Bytes() []byte { return b.b[b.flushed:] }64 func (b *Buf) Bytes() []byte { return b.b[b.flushed:] }
43 func (b *Buf) Len() int { return len(b.b) - b.flushed }65 func (b *Buf) Len() int { return len(b.b) - b.flushed }
44-func (b *Buf) Next() Seq { return b.next }
45 66
46-// Request reserves a zeroed, four-byte-aligned body after the wire header.67+// Next returns the sequence number the next Request will get.
47-// BIG-REQUESTS enablement is transport-owned; maxReq starts at the core limit.68+func (b *Buf) Next() Seq { return b.next }
69+
70+// Request reserves a zeroed body of body bytes after the wire header and
71+// returns it with the request's sequence number. body must be a multiple of 4.
72+// Requests longer than 0xFFFF units get the BIG-REQUESTS 8-byte header.
73+// On failure it records a sticky error and returns (nil, 0).
48 func (b *Buf) Request(major, data uint8, body int) ([]byte, Seq) {74 func (b *Buf) Request(major, data uint8, body int) ([]byte, Seq) {
49 if b.err != nil {75 if b.err != nil {
50 return nil, 076 return nil, 0
@@ -53,29 +79,21 @@ func (b *Buf) Request(major, data uint8, body int) ([]byte, Seq) {
53 b.Fail(ErrBadLength)79 b.Fail(ErrBadLength)
54 return nil, 080 return nil, 0
55 }81 }
56- if uint64(body)+4 > uint64(b.maxReq) {82+ header := 4
83+ if uint64(body)+4 > coreMaxReq {
84+ header = 8
85+ }
86+ if uint64(body)+uint64(header) > uint64(b.maxReq) {
57 b.Fail(ErrRequestTooLarge)87 b.Fail(ErrRequestTooLarge)
58 return nil, 088 return nil, 0
59 }89 }
60- if b.next-b.acked >= 0xffff {90+ if b.next-b.acked >= maxSeqAhead {
61 b.Fail(ErrSeqOverflow)91 b.Fail(ErrSeqOverflow)
62 return nil, 092 return nil, 0
63 }93 }
64- header := 4
65- if (body+4)/4 > 0xffff {
66- header = 8
67- }
68 total := body + header94 total := body + header
95+ b.reserve(total)
69 start := len(b.b)96 start := len(b.b)
70- if total > cap(b.b)-start {
71- capacity := cap(b.b) * 2
72- if capacity < start+total {
73- capacity = start + total
74- }
75- grown := make([]byte, start, capacity)
76- copy(grown, b.b)
77- b.b = grown
78- }
79 b.b = b.b[:start+total]97 b.b = b.b[:start+total]
80 p := b.b[start:]98 p := b.b[start:]
81 clear(p)99 clear(p)
@@ -90,3 +108,74 @@ func (b *Buf) Request(major, data uint8, body int) ([]byte, Seq) {
90 b.next++108 b.next++
91 return p[header:], seq109 return p[header:], seq
92 }110 }
111+
112+// reserve makes room for n more bytes. Flushed bytes are dropped first; the
113+// buffer then doubles until n fits. It never shrinks.
114+func (b *Buf) reserve(n int) {
115+ if n <= cap(b.b)-len(b.b) {
116+ return
117+ }
118+ pending := len(b.b) - b.flushed
119+ if b.flushed > 0 && n <= cap(b.b)-pending {
120+ copy(b.b, b.b[b.flushed:])
121+ b.b = b.b[:pending]
122+ b.flushed = 0
123+ return
124+ }
125+ capacity := max(cap(b.b), 1)
126+ for capacity-pending < n {
127+ capacity *= 2
128+ }
129+ grown := make([]byte, pending, capacity)
130+ copy(grown, b.b[b.flushed:])
131+ b.b = grown
132+ b.flushed = 0
133+}
134+
135+// setMaxReqWords sets the request size limit from the BIG-REQUESTS
136+// maximum_request_length (four-byte units). Conn calls it after enabling it.
137+func (b *Buf) setMaxReqWords(words uint32) {
138+ b.maxReq = uint32(min(uint64(words)*4, 0xfffffffc))
139+}
140+
141+// ack records seq as the highest sequence number seen on the read side.
142+func (b *Buf) ack(seq Seq) {
143+ if seq > b.acked && seq < b.next {
144+ b.acked = seq
145+ }
146+}
147+
148+// flush writes unflushed bytes to the non-blocking fd. It returns
149+// ErrWouldBlock when the socket is full (the remainder stays queued) and nil
150+// once everything is written. Any other failure becomes the sticky error.
151+// Conn.Flush wraps it.
152+func (b *Buf) flush(fd int) error {
153+ if b.err != nil {
154+ return b.err
155+ }
156+ for b.flushed < len(b.b) {
157+ n, err := unix.Write(fd, b.b[b.flushed:])
158+ switch {
159+ case err == unix.EINTR:
160+ continue
161+ case err == unix.EAGAIN:
162+ return ErrWouldBlock
163+ case err == unix.EPIPE || err == unix.ECONNRESET:
164+ b.Fail(ErrClosed)
165+ return b.err
166+ case err != nil:
167+ b.Fail(os.NewSyscallError("write", err))
168+ return b.err
169+ case n <= 0:
170+ b.Fail(io.ErrShortWrite)
171+ return b.err
172+ }
173+ if poison {
174+ fillPoison(b.b[b.flushed : b.flushed+n])
175+ }
176+ b.flushed += n
177+ }
178+ b.b = b.b[:0]
179+ b.flushed = 0
180+ return nil
181+}
M buf_test.go +321 -6
@@ -4,7 +4,10 @@ import (
44 "bytes"
55 "encoding/binary"
66 "errors"
7+ "syscall"
78 "testing"
9+
10+ "golang.org/x/sys/unix"
811 )
912
1013 func TestBufRequest(t *testing.T) {
@@ -27,26 +30,110 @@ func TestBufRequest(t *testing.T) {
2730 t.Fatal("sticky error")
2831 }
2932 }
33+
34+func 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+
3041 func TestBufLimits(t *testing.T) {
31- for _, body := range []int{-1, 1, 262140} {
42+ for _, body := range []int{-1, 1, 262140, 1 << 20} {
3243 b := NewBuf(0)
3344 p, s := b.Request(1, 0, body)
34- if p != nil || s != 0 || b.Err() == nil || b.Len() != 0 {
45+ if p != nil || s != 0 || b.Err() == nil || b.Len() != 0 || b.Next() != 1 {
3546 t.Fatal(body)
3647 }
3748 }
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+
56+func 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+
79+func TestBufSeqOverflow(t *testing.T) {
3880 b := NewBuf(0)
3981 b.next = 0xffff
4082 if p, s := b.Request(1, 0, 0); p != nil || s != 0 || b.Err() != ErrSeqOverflow {
4183 t.Fatal("overflow")
4284 }
4385 b = NewBuf(0)
44- b.maxReq = 1 << 20
45- p, s := b.Request(1, 0, 262140)
46- if s != 1 || len(p) != 262140 || binary.LittleEndian.Uint16(b.Bytes()[2:]) != 0 || binary.LittleEndian.Uint32(b.Bytes()[4:]) != 65537 {
47- t.Fatal("BIG-REQUESTS header")
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")
48103 }
49104 }
105+
106+func 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+
50137 func TestAllocBufRequest(t *testing.T) {
51138 for _, body := range []int{4, 65536} {
52139 b := NewBuf(128 << 10)
@@ -58,6 +145,206 @@ func TestAllocBufRequest(t *testing.T) {
58145 }
59146 }
60147 }
148+
149+// socketpair returns a non-blocking stream pair with small kernel buffers so
150+// EAGAIN is reached quickly.
151+func 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.
170+func 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.
185+func 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+
193+func 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+
239+func 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+
262+func 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+
276+func 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+
292+func 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+
304+func 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+
317+func 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+
61348 func BenchmarkBufRequest(b *testing.B) {
62349 buf := NewBuf(0)
63350 b.ReportAllocs()
@@ -67,3 +354,31 @@ func BenchmarkBufRequest(b *testing.B) {
67354 buf.Request(1, 0, 4)
68355 }
69356 }
357+
358+// BenchmarkBufRequestFlush queues 64 MapWindow-sized requests and flushes
359+// them to a socketpair per iteration.
360+func 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+}
@@ -4,7 +4,10 @@ import (
4 "bytes"4 "bytes"
5 "encoding/binary"5 "encoding/binary"
6 "errors"6 "errors"
7+ "syscall"
7 "testing"8 "testing"
9+
10+ "golang.org/x/sys/unix"
8 )11 )
9 12
10 func TestBufRequest(t *testing.T) {13 func TestBufRequest(t *testing.T) {
@@ -27,26 +30,110 @@ func TestBufRequest(t *testing.T) {
27 t.Fatal("sticky error")30 t.Fatal("sticky error")
28 }31 }
29 }32 }
33+
34+func 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+
30 func TestBufLimits(t *testing.T) {41 func TestBufLimits(t *testing.T) {
31- for _, body := range []int{-1, 1, 262140} {42+ for _, body := range []int{-1, 1, 262140, 1 << 20} {
32 b := NewBuf(0)43 b := NewBuf(0)
33 p, s := b.Request(1, 0, body)44 p, s := b.Request(1, 0, body)
34- if p != nil || s != 0 || b.Err() == nil || b.Len() != 0 {45+ if p != nil || s != 0 || b.Err() == nil || b.Len() != 0 || b.Next() != 1 {
35 t.Fatal(body)46 t.Fatal(body)
36 }47 }
37 }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+
56+func 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+
79+func TestBufSeqOverflow(t *testing.T) {
38 b := NewBuf(0)80 b := NewBuf(0)
39 b.next = 0xffff81 b.next = 0xffff
40 if p, s := b.Request(1, 0, 0); p != nil || s != 0 || b.Err() != ErrSeqOverflow {82 if p, s := b.Request(1, 0, 0); p != nil || s != 0 || b.Err() != ErrSeqOverflow {
41 t.Fatal("overflow")83 t.Fatal("overflow")
42 }84 }
43 b = NewBuf(0)85 b = NewBuf(0)
44- b.maxReq = 1 << 2086+ b.next = 0xfffe
45- p, s := b.Request(1, 0, 262140)87+ if _, s := b.Request(1, 0, 0); s != 0xfffe {
46- if s != 1 || len(p) != 262140 || binary.LittleEndian.Uint16(b.Bytes()[2:]) != 0 || binary.LittleEndian.Uint32(b.Bytes()[4:]) != 65537 {88+ t.Fatal("last allowed seq", s)
47- t.Fatal("BIG-REQUESTS header")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")
48 }103 }
49 }104 }
105+
106+func 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+
50 func TestAllocBufRequest(t *testing.T) {137 func TestAllocBufRequest(t *testing.T) {
51 for _, body := range []int{4, 65536} {138 for _, body := range []int{4, 65536} {
52 b := NewBuf(128 << 10)139 b := NewBuf(128 << 10)
@@ -58,6 +145,206 @@ func TestAllocBufRequest(t *testing.T) {
58 }145 }
59 }146 }
60 }147 }
148+
149+// socketpair returns a non-blocking stream pair with small kernel buffers so
150+// EAGAIN is reached quickly.
151+func 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.
170+func 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.
185+func 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+
193+func 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+
239+func 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+
262+func 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+
276+func 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+
292+func 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+
304+func 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+
317+func 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+
61 func BenchmarkBufRequest(b *testing.B) {348 func BenchmarkBufRequest(b *testing.B) {
62 buf := NewBuf(0)349 buf := NewBuf(0)
63 b.ReportAllocs()350 b.ReportAllocs()
@@ -67,3 +354,31 @@ func BenchmarkBufRequest(b *testing.B) {
67 buf.Request(1, 0, 4)354 buf.Request(1, 0, 4)
68 }355 }
69 }356 }
357+
358+// BenchmarkBufRequestFlush queues 64 MapWindow-sized requests and flushes
359+// them to a socketpair per iteration.
360+func 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+}
A export_test.go +4 -0
@@ -0,0 +1,4 @@
1+package xturbo
2+
3+// SetMaxReqWords exposes the Conn-owned BIG-REQUESTS limit setter to external tests.
4+var SetMaxReqWords = (*Buf).setMaxReqWords
@@ -0,0 +1,4 @@
1+package xturbo
2+
3+// SetMaxReqWords exposes the Conn-owned BIG-REQUESTS limit setter to external tests.
4+var SetMaxReqWords = (*Buf).setMaxReqWords
A poison.go +13 -0
@@ -0,0 +1,13 @@
1+//go:build xturbo_poison
2+
3+package xturbo
4+
5+// poison makes the runtime overwrite consumed buffer regions so retained
6+// views fail loudly in tests.
7+const poison = true
8+
9+func fillPoison(p []byte) {
10+ for i := range p {
11+ p[i] = 0xa5
12+ }
13+}
@@ -0,0 +1,13 @@
1+//go:build xturbo_poison
2+
3+package xturbo
4+
5+// poison makes the runtime overwrite consumed buffer regions so retained
6+// views fail loudly in tests.
7+const poison = true
8+
9+func fillPoison(p []byte) {
10+ for i := range p {
11+ p[i] = 0xa5
12+ }
13+}
A poison_off.go +7 -0
@@ -0,0 +1,7 @@
1+//go:build !xturbo_poison
2+
3+package xturbo
4+
5+const poison = false
6+
7+func fillPoison([]byte) {}
@@ -0,0 +1,7 @@
1+//go:build !xturbo_poison
2+
3+package xturbo
4+
5+const poison = false
6+
7+func fillPoison([]byte) {}
A poison_test.go +25 -0
@@ -0,0 +1,25 @@
1+//go:build xturbo_poison
2+
3+package xturbo
4+
5+import "testing"
6+
7+func TestPoisonFlushedBytes(t *testing.T) {
8+ w, _ := socketpair(t)
9+ b := NewBuf(0)
10+ fill(b, 2000, 60)
11+ retained := b.b[:4]
12+ if err := b.flush(w); err != ErrWouldBlock {
13+ t.Fatal(err)
14+ }
15+ for _, v := range retained {
16+ if v != 0xa5 {
17+ t.Fatal("flushed bytes must be poisoned")
18+ }
19+ }
20+ for _, v := range b.Bytes()[:4] {
21+ if v == 0xa5 {
22+ t.Fatal("pending bytes poisoned")
23+ }
24+ }
25+}
@@ -0,0 +1,25 @@
1+//go:build xturbo_poison
2+
3+package xturbo
4+
5+import "testing"
6+
7+func TestPoisonFlushedBytes(t *testing.T) {
8+ w, _ := socketpair(t)
9+ b := NewBuf(0)
10+ fill(b, 2000, 60)
11+ retained := b.b[:4]
12+ if err := b.flush(w); err != ErrWouldBlock {
13+ t.Fatal(err)
14+ }
15+ for _, v := range retained {
16+ if v != 0xa5 {
17+ t.Fatal("flushed bytes must be poisoned")
18+ }
19+ }
20+ for _, v := range b.Bytes()[:4] {
21+ if v == 0xa5 {
22+ t.Fatal("pending bytes poisoned")
23+ }
24+ }
25+}
A testdata/golden/xproto/ChangePropertyBig.req.hex +6 -0
@@ -0,0 +1,6 @@
1+# hand: BIG-REQUESTS 8-byte header; Xvfb captures need the bigreq handshake (T3.5)
2+# ChangeProperty Replace window=0x0300009f property=0xe7 type=STRING format=8 len=262120
3+# total = 8 header + 20 fixed + 262120 data = 262148 bytes = 65537 units
4+# prefix only: the 262120 data bytes follow (test fills them with i&0xff)
5+12 00 0000 01000100
6+9f000003 e7000000 1f000000 08 000000 e8ff0300
@@ -0,0 +1,6 @@
1+# hand: BIG-REQUESTS 8-byte header; Xvfb captures need the bigreq handshake (T3.5)
2+# ChangeProperty Replace window=0x0300009f property=0xe7 type=STRING format=8 len=262120
3+# total = 8 header + 20 fixed + 262120 data = 262148 bytes = 65537 units
4+# prefix only: the 262120 data bytes follow (test fills them with i&0xff)
5+12 00 0000 01000100
6+9f000003 e7000000 1f000000 08 000000 e8ff0300

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 1

Tags 1

↑↓ moveenter openesc close