detached 1aebbe1

#5 feat(runtime): dispatch packets and attribute replies through an Expect ring

state
merged · 1aebbe1 · fast-forward
branch
feat/runtime-dispatch → main
opened
bnema ·
updated
CommitSubjectAuthorDateSignatureStatus
1aebbe1fix(runtime): bound and harden Dispatch; reject unsent ExpectbriceGPG Verified
350da19feat(runtime): attribute replies and errors through an Expect ring with connection statsbriceGPG Verified
8a907e5feat(runtime): dispatch server packets from a growing read ring with widened sequencesbriceGPG Verified

Checks 4

7 files · +1402 -28
Unified
UnifiedSplit
4/4
⋯

Merge settings

modes
fast-forward, squash, merge commit
checks
failing or unfinished checks block the merge
M conn.go +32 -7
@@ -16,6 +16,24 @@ type Conn struct {
1616 root uint32
1717
1818 xidBase, xidMask uint32 // resource-id allocation, used by NewXID
19+
20+ // Read side, owned by Dispatch: rb[rpos:rend] is received, unconsumed.
21+ rb []byte
22+ rpos, rend int
23+ maxPacket int // MaxPacket; lowered by tests
24+ rerr error // sticky read failure
25+ dispatching bool // inside Dispatch: rejects nested calls
26+ seq Seq // last widened sequence number seen
27+
28+ // Reply ring: Expect registrations indexed by seq&ringMask.
29+ ring [ringSize]slot
30+ live int // registered slots
31+ stats Stats
32+
33+ // Callback arguments, reused per packet so nothing escapes.
34+ ev Event
35+ reply Reply
36+ xerr XError
1937 }
2038
2139 // NewConn takes ownership of fd, a connected stream socket to an X server,
@@ -46,16 +64,23 @@ func setupConn(fd int, authName string, authData []byte) (*Conn, error) {
4664 if err := unix.SetNonblock(fd, true); err != nil {
4765 return nil, os.NewSyscallError("fcntl", err)
4866 }
67+ return newConnState(fd, setup, info), nil
68+}
69+
70+// newConnState builds a Conn over an fd whose setup already completed.
71+func newConnState(fd int, setup []byte, info setupInfo) *Conn {
4972 c := &Conn{
50- fd: fd,
51- buf: NewBuf(0),
52- setup: setup,
53- root: info.root,
54- xidBase: info.xidBase,
55- xidMask: info.xidMask,
73+ fd: fd,
74+ buf: NewBuf(0),
75+ setup: setup,
76+ root: info.root,
77+ xidBase: info.xidBase,
78+ xidMask: info.xidMask,
79+ rb: make([]byte, readBufSize),
80+ maxPacket: MaxPacket,
5681 }
5782 c.buf.Ext = &c.ext
58- return c, nil
83+ return c
5984 }
6085
6186 // Close closes the socket. Later calls, and Flush after Close, return
@@ -16,6 +16,24 @@ type Conn struct {
16 root uint3216 root uint32
17 17
18 xidBase, xidMask uint32 // resource-id allocation, used by NewXID18 xidBase, xidMask uint32 // resource-id allocation, used by NewXID
19+
20+ // Read side, owned by Dispatch: rb[rpos:rend] is received, unconsumed.
21+ rb []byte
22+ rpos, rend int
23+ maxPacket int // MaxPacket; lowered by tests
24+ rerr error // sticky read failure
25+ dispatching bool // inside Dispatch: rejects nested calls
26+ seq Seq // last widened sequence number seen
27+
28+ // Reply ring: Expect registrations indexed by seq&ringMask.
29+ ring [ringSize]slot
30+ live int // registered slots
31+ stats Stats
32+
33+ // Callback arguments, reused per packet so nothing escapes.
34+ ev Event
35+ reply Reply
36+ xerr XError
19 }37 }
20 38
21 // NewConn takes ownership of fd, a connected stream socket to an X server,39 // NewConn takes ownership of fd, a connected stream socket to an X server,
@@ -46,16 +64,23 @@ func setupConn(fd int, authName string, authData []byte) (*Conn, error) {
46 if err := unix.SetNonblock(fd, true); err != nil {64 if err := unix.SetNonblock(fd, true); err != nil {
47 return nil, os.NewSyscallError("fcntl", err)65 return nil, os.NewSyscallError("fcntl", err)
48 }66 }
67+ return newConnState(fd, setup, info), nil
68+}
69+
70+// newConnState builds a Conn over an fd whose setup already completed.
71+func newConnState(fd int, setup []byte, info setupInfo) *Conn {
49 c := &Conn{72 c := &Conn{
50- fd: fd,73+ fd: fd,
51- buf: NewBuf(0),74+ buf: NewBuf(0),
52- setup: setup,75+ setup: setup,
53- root: info.root,76+ root: info.root,
54- xidBase: info.xidBase,77+ xidBase: info.xidBase,
55- xidMask: info.xidMask,78+ xidMask: info.xidMask,
79+ rb: make([]byte, readBufSize),
80+ maxPacket: MaxPacket,
56 }81 }
57 c.buf.Ext = &c.ext82 c.buf.Ext = &c.ext
58- return c, nil83+ return c
59 }84 }
60 85
61 // Close closes the socket. Later calls, and Flush after Close, return86 // Close closes the socket. Later calls, and Flush after Close, return
M conn_xserver_test.go +35 -21
@@ -120,42 +120,56 @@ func TestXserverDial(t *testing.T) {
120120 t.Fatalf("Root %#x, setup view says otherwise", c.Root())
121121 }
122122
123- // A trivial round trip: GetInputFocus, then read its reply.
123+ // Round trips through Dispatch: GetInputFocus's reply, then the error
124+ // of GetGeometry on window None, both attributed through Expect.
125+ const kindFocus, kindGeom = 16, 17
124126 seq := xproto.GetInputFocus(c.Buf())
125- if seq != 1 {
126- t.Fatalf("seq %d", seq)
127+ bad := xproto.GetGeometry(c.Buf(), xproto.Drawable(xproto.WindowNone))
128+ if seq != 1 || bad != 2 {
129+ t.Fatalf("seqs %d %d", seq, bad)
130+ }
131+ if err := c.Expect(seq, kindFocus, 1, 2, 3, 4); err != nil {
132+ t.Fatal(err)
133+ }
134+ if err := c.Expect(bad, kindGeom, 5, 0, 0, 0); err != nil {
135+ t.Fatal(err)
127136 }
128137 if err := c.Flush(); err != nil {
129138 t.Fatal(err)
130139 }
131- reply := readPacket(t, c.Fd())
132- if reply[0] != 1 || binary.LittleEndian.Uint16(reply[2:]) != uint16(seq) {
133- t.Fatalf("reply %x", reply)
140+ var reply xturbo.Reply
141+ var xerr xturbo.XError
142+ h := xturbo.Handler{
143+ OnReply: func(r *xturbo.Reply) { reply = *r; reply.Raw = nil },
144+ OnError: func(e *xturbo.XError) { xerr = *e },
145+ }
146+ dispatchUntil(t, c, &h, func() bool { return xerr.Seq != 0 })
147+ if reply.Seq != seq || reply.Kind != kindFocus || reply.Args != [4]uint32{1, 2, 3, 4} {
148+ t.Fatalf("reply %+v", reply)
149+ }
150+ if xerr.Seq != bad || xerr.Kind != kindGeom || xerr.Args[0] != 5 || xerr.Code != xproto.BadDrawable ||
151+ xerr.Major != 14 || xerr.Ext != xturbo.ExtCore || xerr.Bad != 0 {
152+ t.Fatalf("error %+v", xerr)
153+ }
154+ if st := c.Stats(); st.DroppedReplies != 0 || st.UnhandledErrors != 0 {
155+ t.Fatalf("%+v", st)
134156 }
135157 }
136158
137-// readPacket polls the non-blocking fd and reads one 32-byte packet.
138-func readPacket(t *testing.T, fd int) [32]byte {
159+// dispatchUntil polls the connection and dispatches until done reports true.
160+func dispatchUntil(t *testing.T, c *xturbo.Conn, h *xturbo.Handler, done func() bool) {
139161 t.Helper()
140- var p [32]byte
141- got := 0
142162 deadline := time.Now().Add(5 * time.Second)
143- for got < len(p) {
163+ for !done() {
144164 if time.Now().After(deadline) {
145- t.Fatal("timeout reading reply")
165+ t.Fatal("timeout waiting for the server")
146166 }
147- fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}
167+ fds := []unix.PollFd{{Fd: int32(c.Fd()), Events: unix.POLLIN}}
148168 if _, err := unix.Poll(fds, 1000); err != nil && err != unix.EINTR {
149169 t.Fatal(err)
150170 }
151- n, err := unix.Read(fd, p[got:])
152- if err == unix.EAGAIN || err == unix.EINTR {
153- continue
154- }
155- if err != nil || n == 0 {
156- t.Fatalf("read: %d %v", n, err)
171+ if err := c.Dispatch(h); err != nil {
172+ t.Fatal(err)
157173 }
158- got += n
159174 }
160- return p
161175 }
@@ -120,42 +120,56 @@ func TestXserverDial(t *testing.T) {
120 t.Fatalf("Root %#x, setup view says otherwise", c.Root())120 t.Fatalf("Root %#x, setup view says otherwise", c.Root())
121 }121 }
122 122
123- // A trivial round trip: GetInputFocus, then read its reply.123+ // Round trips through Dispatch: GetInputFocus's reply, then the error
124+ // of GetGeometry on window None, both attributed through Expect.
125+ const kindFocus, kindGeom = 16, 17
124 seq := xproto.GetInputFocus(c.Buf())126 seq := xproto.GetInputFocus(c.Buf())
125- if seq != 1 {127+ bad := xproto.GetGeometry(c.Buf(), xproto.Drawable(xproto.WindowNone))
126- t.Fatalf("seq %d", seq)128+ if seq != 1 || bad != 2 {
129+ t.Fatalf("seqs %d %d", seq, bad)
130+ }
131+ if err := c.Expect(seq, kindFocus, 1, 2, 3, 4); err != nil {
132+ t.Fatal(err)
133+ }
134+ if err := c.Expect(bad, kindGeom, 5, 0, 0, 0); err != nil {
135+ t.Fatal(err)
127 }136 }
128 if err := c.Flush(); err != nil {137 if err := c.Flush(); err != nil {
129 t.Fatal(err)138 t.Fatal(err)
130 }139 }
131- reply := readPacket(t, c.Fd())140+ var reply xturbo.Reply
132- if reply[0] != 1 || binary.LittleEndian.Uint16(reply[2:]) != uint16(seq) {141+ var xerr xturbo.XError
133- t.Fatalf("reply %x", reply)142+ h := xturbo.Handler{
143+ OnReply: func(r *xturbo.Reply) { reply = *r; reply.Raw = nil },
144+ OnError: func(e *xturbo.XError) { xerr = *e },
145+ }
146+ dispatchUntil(t, c, &h, func() bool { return xerr.Seq != 0 })
147+ if reply.Seq != seq || reply.Kind != kindFocus || reply.Args != [4]uint32{1, 2, 3, 4} {
148+ t.Fatalf("reply %+v", reply)
149+ }
150+ if xerr.Seq != bad || xerr.Kind != kindGeom || xerr.Args[0] != 5 || xerr.Code != xproto.BadDrawable ||
151+ xerr.Major != 14 || xerr.Ext != xturbo.ExtCore || xerr.Bad != 0 {
152+ t.Fatalf("error %+v", xerr)
153+ }
154+ if st := c.Stats(); st.DroppedReplies != 0 || st.UnhandledErrors != 0 {
155+ t.Fatalf("%+v", st)
134 }156 }
135 }157 }
136 158
137-// readPacket polls the non-blocking fd and reads one 32-byte packet.159+// dispatchUntil polls the connection and dispatches until done reports true.
138-func readPacket(t *testing.T, fd int) [32]byte {160+func dispatchUntil(t *testing.T, c *xturbo.Conn, h *xturbo.Handler, done func() bool) {
139 t.Helper()161 t.Helper()
140- var p [32]byte
141- got := 0
142 deadline := time.Now().Add(5 * time.Second)162 deadline := time.Now().Add(5 * time.Second)
143- for got < len(p) {163+ for !done() {
144 if time.Now().After(deadline) {164 if time.Now().After(deadline) {
145- t.Fatal("timeout reading reply")165+ t.Fatal("timeout waiting for the server")
146 }166 }
147- fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}}167+ fds := []unix.PollFd{{Fd: int32(c.Fd()), Events: unix.POLLIN}}
148 if _, err := unix.Poll(fds, 1000); err != nil && err != unix.EINTR {168 if _, err := unix.Poll(fds, 1000); err != nil && err != unix.EINTR {
149 t.Fatal(err)169 t.Fatal(err)
150 }170 }
151- n, err := unix.Read(fd, p[got:])171+ if err := c.Dispatch(h); err != nil {
152- if err == unix.EAGAIN || err == unix.EINTR {172+ t.Fatal(err)
153- continue
154- }
155- if err != nil || n == 0 {
156- t.Fatalf("read: %d %v", n, err)
157 }173 }
158- got += n
159 }174 }
160- return p
161 }175 }
A dispatch.go +251 -0
@@ -0,0 +1,251 @@
1+package xturbo
2+
3+import (
4+ "encoding/binary"
5+ "errors"
6+ "io"
7+ "os"
8+
9+ "golang.org/x/sys/unix"
10+)
11+
12+// MaxPacket bounds one server packet (a reply or an XGE event) in bytes. The
13+// read ring grows up to this size; a larger packet fails Dispatch with
14+// ErrPacketTooLarge and the connection is dead.
15+const MaxPacket = 16 << 20
16+
17+// ErrPacketTooLarge reports a server packet longer than MaxPacket.
18+var ErrPacketTooLarge = errors.New("xturbo: packet too large")
19+
20+// errReentrantDispatch is returned by a Dispatch called from a Handler
21+// callback. It is unexported: it reports a programming error, not a condition
22+// callers handle. The outer Dispatch is unaffected.
23+var errReentrantDispatch = errors.New("xturbo: Dispatch called from a Handler callback")
24+
25+const (
26+ readBufSize = 64 << 10
27+ packetHeadSize = 32
28+)
29+
30+// Handler receives decoded packets from Dispatch, in wire order, on the
31+// goroutine calling Dispatch. OnReply sees only replies registered with
32+// Expect; OnError sees every error (Kind 0 when unregistered). A nil field
33+// drops that packet class (nil OnError counts Stats.UnhandledErrors). Fill it
34+// once at init; the pointers passed to callbacks are owned by the Conn and are
35+// reused for the next packet, so callbacks must not retain them, nor the
36+// views they hold (Reply.Raw, Event.Tail). Callbacks may queue requests and
37+// call Expect but must not call Dispatch: a nested Dispatch returns an error
38+// and changes nothing.
39+type Handler struct {
40+ OnEvent func(ev *Event)
41+ OnReply func(r *Reply)
42+ OnError func(e *XError)
43+}
44+
45+// Reply is one reply packet to a request registered with Expect. Seq is
46+// widened; Kind and Args are the registration's. Raw is the whole reply,
47+// 32+4*length bytes, valid only until the callback returns.
48+type Reply struct {
49+ Seq Seq
50+ Kind uint16
51+ Args [4]uint32
52+ Raw []byte
53+}
54+
55+var noHandler Handler
56+
57+// Dispatch reads what the server has sent without blocking and calls h for
58+// every complete packet, in wire order. It returns nil once the socket would
59+// block or after a bounded amount of input (about one read ring, so a flooding
60+// server cannot starve the caller's event loop): call it again while the fd is
61+// still readable, and Flush requests queued by callbacks in between. It
62+// returns io.EOF when the server closed the connection and ErrPacketTooLarge
63+// on a packet beyond MaxPacket. Read errors are sticky and also fail the Buf.
64+// Dispatch never flushes. A nil h drops every packet.
65+func (c *Conn) Dispatch(h *Handler) error {
66+ if c.dispatching {
67+ return errReentrantDispatch
68+ }
69+ c.dispatching = true
70+ defer func() { c.dispatching = false }()
71+ if c.fd < 0 {
72+ return ErrClosed
73+ }
74+ if c.rerr != nil {
75+ return c.rerr
76+ }
77+ if h == nil {
78+ h = &noHandler
79+ }
80+ budget := len(c.rb)
81+ for {
82+ if err := c.drain(h); err != nil {
83+ return c.failRead(err)
84+ }
85+ if c.fd < 0 { // a callback closed the Conn
86+ return ErrClosed
87+ }
88+ if budget <= 0 {
89+ return nil
90+ }
91+ n, err := unix.Read(c.fd, c.rb[c.rend:])
92+ switch {
93+ case err == unix.EINTR:
94+ continue
95+ case err == unix.EAGAIN:
96+ return nil
97+ case err == unix.ECONNRESET:
98+ return c.failRead(io.EOF)
99+ case err != nil:
100+ return c.failRead(os.NewSyscallError("read", err))
101+ case n == 0:
102+ return c.failRead(io.EOF)
103+ }
104+ c.rend += n
105+ budget -= n
106+ }
107+}
108+
109+func (c *Conn) failRead(err error) error {
110+ c.rerr = err
111+ if err == io.EOF {
112+ c.buf.Fail(ErrClosed)
113+ } else {
114+ c.buf.Fail(err)
115+ }
116+ return err
117+}
118+
119+// drain handles every complete packet in the ring, then makes room for the
120+// next one (compacting, growing up to maxPacket).
121+func (c *Conn) drain(h *Handler) error {
122+ for c.fd >= 0 {
123+ avail := c.rend - c.rpos
124+ need := packetHeadSize
125+ if avail >= packetHeadSize {
126+ size, ok := packetSize(c.rb[c.rpos:], c.maxPacket)
127+ if !ok {
128+ return ErrPacketTooLarge
129+ }
130+ need = size
131+ if avail >= need {
132+ end := c.rpos + need
133+ p := c.rb[c.rpos:end:end]
134+ c.packet(h, p)
135+ if poison {
136+ fillPoison(p)
137+ }
138+ c.rpos = end
139+ continue
140+ }
141+ }
142+ c.makeRoom(need)
143+ break
144+ }
145+ return nil
146+}
147+
148+// packetSize returns the length of the packet starting at p (len(p) >= 32):
149+// 32 for errors and events, 32+4*length for replies and XGE events.
150+func packetSize(p []byte, limit int) (int, bool) {
151+ n := uint64(packetHeadSize)
152+ if p[0] == 1 || p[0]&0x7f == codeGenericEvent {
153+ n += 4 * uint64(binary.LittleEndian.Uint32(p[4:]))
154+ }
155+ if n > uint64(limit) {
156+ return 0, false
157+ }
158+ return int(n), true
159+}
160+
161+// makeRoom moves the unconsumed bytes to the start of the ring and grows it
162+// by doubling until need bytes fit. need <= maxPacket.
163+func (c *Conn) makeRoom(need int) {
164+ if c.rpos > 0 {
165+ k := copy(c.rb, c.rb[c.rpos:c.rend])
166+ if poison {
167+ fillPoison(c.rb[k:c.rend])
168+ }
169+ c.rpos, c.rend = 0, k
170+ }
171+ if need <= len(c.rb) {
172+ return
173+ }
174+ size := max(len(c.rb), packetHeadSize)
175+ for size < need {
176+ size *= 2
177+ }
178+ grown := make([]byte, min(size, c.maxPacket))
179+ copy(grown, c.rb[:c.rend])
180+ if poison {
181+ fillPoison(c.rb)
182+ }
183+ c.rb = grown
184+}
185+
186+// packet decodes one complete packet and calls h.
187+func (c *Conn) packet(h *Handler, p []byte) {
188+ switch p[0] {
189+ case 0:
190+ decodeError(p, &c.ext, &c.xerr)
191+ c.xerr.Seq = c.widen(c.xerr.Seq)
192+ c.xerr.Kind, c.xerr.Args, _ = c.take(c.xerr.Seq)
193+ switch {
194+ case c.xerr.Kind != kindNone && c.xerr.Kind < kindUser:
195+ if !c.consumeError(&c.xerr) && h.OnError != nil {
196+ h.OnError(&c.xerr)
197+ }
198+ case h.OnError != nil:
199+ h.OnError(&c.xerr)
200+ default:
201+ c.stats.UnhandledErrors++
202+ }
203+ case 1:
204+ seq := c.widen(Seq(binary.LittleEndian.Uint16(p[2:])))
205+ kind, args, ok := c.take(seq)
206+ c.reply = Reply{Seq: seq, Kind: kind, Args: args, Raw: p}
207+ switch {
208+ case !ok:
209+ c.stats.DroppedReplies++
210+ case kind < kindUser:
211+ c.consumeReply(&c.reply)
212+ case h.OnReply != nil:
213+ h.OnReply(&c.reply)
214+ }
215+ c.reply.Raw = nil // do not retain the ring
216+ default:
217+ ev, _, ok := DecodeEvent(p, &c.ext)
218+ if !ok { // a sent event claiming code 0 or 1
219+ ev = Event{Kind: KindUnknown, SendEvent: true, Seq: Seq(binary.LittleEndian.Uint16(p[2:]))}
220+ copy(ev.Raw[:], p)
221+ }
222+ if p[0]&0x7f == codeKeymapNotify {
223+ ev.Seq = c.seq
224+ } else {
225+ ev.Seq = c.widen(ev.Seq)
226+ }
227+ if ev.Kind == KindUnknown {
228+ c.stats.UnknownEvents++
229+ }
230+ c.ev = ev
231+ if h.OnEvent != nil {
232+ h.OnEvent(&c.ev)
233+ }
234+ c.ev.Tail = nil // do not retain the ring
235+ }
236+}
237+
238+// widen turns a 16-bit wire sequence into the full Seq, relative to the last
239+// one seen, acknowledges it to the Buf and retires the Expect slots of the
240+// requests it proves complete.
241+func (c *Conn) widen(wire Seq) Seq {
242+ prev := c.seq
243+ full := prev&^0xffff | wire&0xffff
244+ if full < prev {
245+ full += 0x10000
246+ }
247+ c.seq = full
248+ c.buf.ack(full)
249+ c.retire(prev, full)
250+ return full
251+}
@@ -0,0 +1,251 @@
1+package xturbo
2+
3+import (
4+ "encoding/binary"
5+ "errors"
6+ "io"
7+ "os"
8+
9+ "golang.org/x/sys/unix"
10+)
11+
12+// MaxPacket bounds one server packet (a reply or an XGE event) in bytes. The
13+// read ring grows up to this size; a larger packet fails Dispatch with
14+// ErrPacketTooLarge and the connection is dead.
15+const MaxPacket = 16 << 20
16+
17+// ErrPacketTooLarge reports a server packet longer than MaxPacket.
18+var ErrPacketTooLarge = errors.New("xturbo: packet too large")
19+
20+// errReentrantDispatch is returned by a Dispatch called from a Handler
21+// callback. It is unexported: it reports a programming error, not a condition
22+// callers handle. The outer Dispatch is unaffected.
23+var errReentrantDispatch = errors.New("xturbo: Dispatch called from a Handler callback")
24+
25+const (
26+ readBufSize = 64 << 10
27+ packetHeadSize = 32
28+)
29+
30+// Handler receives decoded packets from Dispatch, in wire order, on the
31+// goroutine calling Dispatch. OnReply sees only replies registered with
32+// Expect; OnError sees every error (Kind 0 when unregistered). A nil field
33+// drops that packet class (nil OnError counts Stats.UnhandledErrors). Fill it
34+// once at init; the pointers passed to callbacks are owned by the Conn and are
35+// reused for the next packet, so callbacks must not retain them, nor the
36+// views they hold (Reply.Raw, Event.Tail). Callbacks may queue requests and
37+// call Expect but must not call Dispatch: a nested Dispatch returns an error
38+// and changes nothing.
39+type Handler struct {
40+ OnEvent func(ev *Event)
41+ OnReply func(r *Reply)
42+ OnError func(e *XError)
43+}
44+
45+// Reply is one reply packet to a request registered with Expect. Seq is
46+// widened; Kind and Args are the registration's. Raw is the whole reply,
47+// 32+4*length bytes, valid only until the callback returns.
48+type Reply struct {
49+ Seq Seq
50+ Kind uint16
51+ Args [4]uint32
52+ Raw []byte
53+}
54+
55+var noHandler Handler
56+
57+// Dispatch reads what the server has sent without blocking and calls h for
58+// every complete packet, in wire order. It returns nil once the socket would
59+// block or after a bounded amount of input (about one read ring, so a flooding
60+// server cannot starve the caller's event loop): call it again while the fd is
61+// still readable, and Flush requests queued by callbacks in between. It
62+// returns io.EOF when the server closed the connection and ErrPacketTooLarge
63+// on a packet beyond MaxPacket. Read errors are sticky and also fail the Buf.
64+// Dispatch never flushes. A nil h drops every packet.
65+func (c *Conn) Dispatch(h *Handler) error {
66+ if c.dispatching {
67+ return errReentrantDispatch
68+ }
69+ c.dispatching = true
70+ defer func() { c.dispatching = false }()
71+ if c.fd < 0 {
72+ return ErrClosed
73+ }
74+ if c.rerr != nil {
75+ return c.rerr
76+ }
77+ if h == nil {
78+ h = &noHandler
79+ }
80+ budget := len(c.rb)
81+ for {
82+ if err := c.drain(h); err != nil {
83+ return c.failRead(err)
84+ }
85+ if c.fd < 0 { // a callback closed the Conn
86+ return ErrClosed
87+ }
88+ if budget <= 0 {
89+ return nil
90+ }
91+ n, err := unix.Read(c.fd, c.rb[c.rend:])
92+ switch {
93+ case err == unix.EINTR:
94+ continue
95+ case err == unix.EAGAIN:
96+ return nil
97+ case err == unix.ECONNRESET:
98+ return c.failRead(io.EOF)
99+ case err != nil:
100+ return c.failRead(os.NewSyscallError("read", err))
101+ case n == 0:
102+ return c.failRead(io.EOF)
103+ }
104+ c.rend += n
105+ budget -= n
106+ }
107+}
108+
109+func (c *Conn) failRead(err error) error {
110+ c.rerr = err
111+ if err == io.EOF {
112+ c.buf.Fail(ErrClosed)
113+ } else {
114+ c.buf.Fail(err)
115+ }
116+ return err
117+}
118+
119+// drain handles every complete packet in the ring, then makes room for the
120+// next one (compacting, growing up to maxPacket).
121+func (c *Conn) drain(h *Handler) error {
122+ for c.fd >= 0 {
123+ avail := c.rend - c.rpos
124+ need := packetHeadSize
125+ if avail >= packetHeadSize {
126+ size, ok := packetSize(c.rb[c.rpos:], c.maxPacket)
127+ if !ok {
128+ return ErrPacketTooLarge
129+ }
130+ need = size
131+ if avail >= need {
132+ end := c.rpos + need
133+ p := c.rb[c.rpos:end:end]
134+ c.packet(h, p)
135+ if poison {
136+ fillPoison(p)
137+ }
138+ c.rpos = end
139+ continue
140+ }
141+ }
142+ c.makeRoom(need)
143+ break
144+ }
145+ return nil
146+}
147+
148+// packetSize returns the length of the packet starting at p (len(p) >= 32):
149+// 32 for errors and events, 32+4*length for replies and XGE events.
150+func packetSize(p []byte, limit int) (int, bool) {
151+ n := uint64(packetHeadSize)
152+ if p[0] == 1 || p[0]&0x7f == codeGenericEvent {
153+ n += 4 * uint64(binary.LittleEndian.Uint32(p[4:]))
154+ }
155+ if n > uint64(limit) {
156+ return 0, false
157+ }
158+ return int(n), true
159+}
160+
161+// makeRoom moves the unconsumed bytes to the start of the ring and grows it
162+// by doubling until need bytes fit. need <= maxPacket.
163+func (c *Conn) makeRoom(need int) {
164+ if c.rpos > 0 {
165+ k := copy(c.rb, c.rb[c.rpos:c.rend])
166+ if poison {
167+ fillPoison(c.rb[k:c.rend])
168+ }
169+ c.rpos, c.rend = 0, k
170+ }
171+ if need <= len(c.rb) {
172+ return
173+ }
174+ size := max(len(c.rb), packetHeadSize)
175+ for size < need {
176+ size *= 2
177+ }
178+ grown := make([]byte, min(size, c.maxPacket))
179+ copy(grown, c.rb[:c.rend])
180+ if poison {
181+ fillPoison(c.rb)
182+ }
183+ c.rb = grown
184+}
185+
186+// packet decodes one complete packet and calls h.
187+func (c *Conn) packet(h *Handler, p []byte) {
188+ switch p[0] {
189+ case 0:
190+ decodeError(p, &c.ext, &c.xerr)
191+ c.xerr.Seq = c.widen(c.xerr.Seq)
192+ c.xerr.Kind, c.xerr.Args, _ = c.take(c.xerr.Seq)
193+ switch {
194+ case c.xerr.Kind != kindNone && c.xerr.Kind < kindUser:
195+ if !c.consumeError(&c.xerr) && h.OnError != nil {
196+ h.OnError(&c.xerr)
197+ }
198+ case h.OnError != nil:
199+ h.OnError(&c.xerr)
200+ default:
201+ c.stats.UnhandledErrors++
202+ }
203+ case 1:
204+ seq := c.widen(Seq(binary.LittleEndian.Uint16(p[2:])))
205+ kind, args, ok := c.take(seq)
206+ c.reply = Reply{Seq: seq, Kind: kind, Args: args, Raw: p}
207+ switch {
208+ case !ok:
209+ c.stats.DroppedReplies++
210+ case kind < kindUser:
211+ c.consumeReply(&c.reply)
212+ case h.OnReply != nil:
213+ h.OnReply(&c.reply)
214+ }
215+ c.reply.Raw = nil // do not retain the ring
216+ default:
217+ ev, _, ok := DecodeEvent(p, &c.ext)
218+ if !ok { // a sent event claiming code 0 or 1
219+ ev = Event{Kind: KindUnknown, SendEvent: true, Seq: Seq(binary.LittleEndian.Uint16(p[2:]))}
220+ copy(ev.Raw[:], p)
221+ }
222+ if p[0]&0x7f == codeKeymapNotify {
223+ ev.Seq = c.seq
224+ } else {
225+ ev.Seq = c.widen(ev.Seq)
226+ }
227+ if ev.Kind == KindUnknown {
228+ c.stats.UnknownEvents++
229+ }
230+ c.ev = ev
231+ if h.OnEvent != nil {
232+ h.OnEvent(&c.ev)
233+ }
234+ c.ev.Tail = nil // do not retain the ring
235+ }
236+}
237+
238+// widen turns a 16-bit wire sequence into the full Seq, relative to the last
239+// one seen, acknowledges it to the Buf and retires the Expect slots of the
240+// requests it proves complete.
241+func (c *Conn) widen(wire Seq) Seq {
242+ prev := c.seq
243+ full := prev&^0xffff | wire&0xffff
244+ if full < prev {
245+ full += 0x10000
246+ }
247+ c.seq = full
248+ c.buf.ack(full)
249+ c.retire(prev, full)
250+ return full
251+}
A dispatch_poison_test.go +37 -0
@@ -0,0 +1,37 @@
1+//go:build xturbo_poison
2+
3+package xturbo
4+
5+import "testing"
6+
7+// TestPoisonRetainedViews keeps Reply.Raw and Event.Tail past their callback
8+// (a lifetime bug) and checks the runtime overwrote them.
9+func TestPoisonRetainedViews(t *testing.T) {
10+ c, srv := dispatchConn(t)
11+ var raw, tail []byte
12+ h := Handler{
13+ OnReply: func(r *Reply) { raw = r.Raw },
14+ OnEvent: func(ev *Event) { tail = ev.Tail },
15+ }
16+ stream := append(pktReply(1, 64), pktXGE(2, 1, 64)...)
17+ expect(t, c, 1, 3, 4)
18+ feed(t, c, srv, stream, &h)
19+ if len(raw) != 96 || len(tail) != 64 {
20+ t.Fatal(len(raw), len(tail))
21+ }
22+ for _, p := range [][]byte{raw, tail} {
23+ for i, v := range p {
24+ if v != 0xa5 {
25+ t.Fatalf("retained byte %d = %#x, want poison", i, v)
26+ }
27+ }
28+ }
29+
30+ // Growth poisons the old ring too.
31+ feed(t, c, srv, pktReply(3, 8), &h)
32+ old := c.rb
33+ feed(t, c, srv, pktReply(4, 200<<10), &h)
34+ if &old[0] == &c.rb[0] || old[0] != 0xa5 || old[len(old)-1] != 0xa5 {
35+ t.Fatal("old ring not poisoned after growth")
36+ }
37+}
@@ -0,0 +1,37 @@
1+//go:build xturbo_poison
2+
3+package xturbo
4+
5+import "testing"
6+
7+// TestPoisonRetainedViews keeps Reply.Raw and Event.Tail past their callback
8+// (a lifetime bug) and checks the runtime overwrote them.
9+func TestPoisonRetainedViews(t *testing.T) {
10+ c, srv := dispatchConn(t)
11+ var raw, tail []byte
12+ h := Handler{
13+ OnReply: func(r *Reply) { raw = r.Raw },
14+ OnEvent: func(ev *Event) { tail = ev.Tail },
15+ }
16+ stream := append(pktReply(1, 64), pktXGE(2, 1, 64)...)
17+ expect(t, c, 1, 3, 4)
18+ feed(t, c, srv, stream, &h)
19+ if len(raw) != 96 || len(tail) != 64 {
20+ t.Fatal(len(raw), len(tail))
21+ }
22+ for _, p := range [][]byte{raw, tail} {
23+ for i, v := range p {
24+ if v != 0xa5 {
25+ t.Fatalf("retained byte %d = %#x, want poison", i, v)
26+ }
27+ }
28+ }
29+
30+ // Growth poisons the old ring too.
31+ feed(t, c, srv, pktReply(3, 8), &h)
32+ old := c.rb
33+ feed(t, c, srv, pktReply(4, 200<<10), &h)
34+ if &old[0] == &c.rb[0] || old[0] != 0xa5 || old[len(old)-1] != 0xa5 {
35+ t.Fatal("old ring not poisoned after growth")
36+ }
37+}
A dispatch_test.go +643 -0
@@ -0,0 +1,643 @@
1+package xturbo
2+
3+import (
4+ "bytes"
5+ "encoding/binary"
6+ "errors"
7+ "fmt"
8+ "io"
9+ "testing"
10+ "time"
11+
12+ "bnema.dev/xturbo/internal/xtest"
13+ "golang.org/x/sys/unix"
14+)
15+
16+// dispatchConn returns a Conn over one end of a non-blocking socketpair with
17+// default kernel buffers, and the other end acting as the server. The XFIXES
18+// and Generic Event extensions are negotiated at the golden capture's bases.
19+func dispatchConn(t testing.TB) (*Conn, int) {
20+ t.Helper()
21+ fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_NONBLOCK|unix.SOCK_CLOEXEC, 0)
22+ if err != nil {
23+ t.Fatal(err)
24+ }
25+ c := newConnState(fds[0], nil, setupInfo{})
26+ t.Cleanup(func() { _ = c.Close(); _ = unix.Close(fds[1]) })
27+ c.ext.Set(ExtXFixes, 138, 87, 140)
28+ c.ext.Set(ExtGenericEvent, 0x83, 0, 0)
29+ c.buf.next = 1 << 40 // every widened seq in the tests is acknowledgeable
30+ return c, fds[1]
31+}
32+
33+// Packet builders. seq is the 16-bit wire value.
34+func pktEvent(code uint8, seq uint16, fill byte) []byte {
35+ p := bytes.Repeat([]byte{fill}, 32)
36+ p[0] = code
37+ binary.LittleEndian.PutUint16(p[2:], seq)
38+ return p
39+}
40+
41+func pktReply(seq uint16, data int) []byte {
42+ words := (data + 3) / 4
43+ p := make([]byte, 32+4*words)
44+ p[0], p[1] = 1, 7
45+ binary.LittleEndian.PutUint16(p[2:], seq)
46+ binary.LittleEndian.PutUint32(p[4:], uint32(words))
47+ for i := 8; i < len(p); i++ {
48+ p[i] = byte(i * 7)
49+ }
50+ return p
51+}
52+
53+func pktError(code uint8, seq uint16, major uint8) []byte {
54+ p := make([]byte, 32)
55+ p[1] = code
56+ binary.LittleEndian.PutUint16(p[2:], seq)
57+ binary.LittleEndian.PutUint32(p[4:], 0xdead)
58+ p[10] = major
59+ return p
60+}
61+
62+func pktXGE(seq uint16, evtype uint16, tail int) []byte {
63+ words := (tail + 3) / 4
64+ p := make([]byte, 32+4*words)
65+ p[0], p[1] = codeGenericEvent, 0x83
66+ binary.LittleEndian.PutUint16(p[2:], seq)
67+ binary.LittleEndian.PutUint32(p[4:], uint32(words))
68+ binary.LittleEndian.PutUint16(p[8:], evtype)
69+ for i := 32; i < len(p); i++ {
70+ p[i] = byte(i)
71+ }
72+ return p
73+}
74+
75+// record is one callback as seen by a test handler; data is a copy of the
76+// packet bytes (Raw, plus Tail for events).
77+type record struct {
78+ class byte // 'e', 'r', 'x'
79+ seq Seq
80+ kind uint32
81+ data []byte
82+}
83+
84+func (r record) String() string {
85+ return fmt.Sprintf("%c seq %d kind %#x %d bytes", r.class, r.seq, r.kind, len(r.data))
86+}
87+
88+type recorder struct {
89+ got []record
90+ h Handler
91+}
92+
93+// newRecorder records every callback of c. Replies reach OnReply only when
94+// registered: on each MapRequest event (the first packet of scriptedStream)
95+// it registers the stream's replies, at ev.Seq+1, +4 and +6.
96+func newRecorder(c *Conn) *recorder {
97+ r := &recorder{}
98+ r.h = Handler{
99+ OnEvent: func(ev *Event) {
100+ expectScripted(c, ev)
101+ r.got = append(r.got, record{'e', ev.Seq, uint32(ev.Kind), append(bytes.Clone(ev.Raw[:]), ev.Tail...)})
102+ },
103+ OnReply: func(rp *Reply) {
104+ r.got = append(r.got, record{'r', rp.Seq, uint32(rp.Kind), bytes.Clone(rp.Raw)})
105+ },
106+ OnError: func(e *XError) {
107+ r.got = append(r.got, record{'x', e.Seq, uint32(e.Code), bytes.Clone(e.Raw[:])})
108+ },
109+ }
110+ return r
111+}
112+
113+// feed writes data to the server end, dispatching whenever the socket fills,
114+// then dispatches what remains.
115+func feed(t testing.TB, c *Conn, srv int, data []byte, h *Handler) {
116+ t.Helper()
117+ for len(data) > 0 {
118+ n, err := unix.Write(srv, data)
119+ if err == unix.EAGAIN {
120+ n = 0
121+ } else if err != nil {
122+ t.Fatal(err)
123+ }
124+ data = data[n:]
125+ pump(t, c, h)
126+ }
127+ pump(t, c, h)
128+}
129+
130+// pump dispatches until the socket holds nothing more: one Dispatch reads a
131+// bounded amount.
132+func pump(t testing.TB, c *Conn, h *Handler) {
133+ t.Helper()
134+ for {
135+ if err := c.Dispatch(h); err != nil {
136+ t.Fatal(err)
137+ }
138+ fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
139+ if n, _ := unix.Poll(fds, 0); n == 0 {
140+ return
141+ }
142+ }
143+}
144+
145+func golden(t testing.TB, name string) []byte {
146+ t.Helper()
147+ raw, err := xtest.LoadHex("testdata/golden/" + name + ".hex")
148+ if err != nil {
149+ t.Fatal(err)
150+ }
151+ return raw
152+}
153+
154+// scriptedStream is a mixed stream of goldens and synthetic packets with
155+// increasing wire sequence numbers, and the callbacks it must produce.
156+func scriptedStream(t testing.TB) ([]byte, []record) {
157+ var stream []byte
158+ var want []record
159+ add := func(class byte, kind uint32, p []byte, seq uint16) {
160+ binary.LittleEndian.PutUint16(p[2:], seq)
161+ stream = append(stream, p...)
162+ want = append(want, record{class, Seq(seq), kind, p})
163+ }
164+ add('e', 20, golden(t, "xproto/MapRequest.event"), 3)
165+ add('r', uint32(kindUser), golden(t, "xproto/GetGeometry.reply"), 4)
166+ add('x', 3, golden(t, "xproto/BadWindow.error"), 5)
167+ add('e', uint32(ExtGenericEvent)<<8|0x81, golden(t, "xproto/GeGeneric.event"), 6)
168+ add('e', uint32(ExtXFixes)<<8, golden(t, "xfixes/SelectionNotify.event"), 6)
169+ add('r', uint32(kindUser), golden(t, "xproto/GetProperty.reply"), 7)
170+ km := golden(t, "xproto/KeymapNotify.event") // no sequence: inherits 7
171+ stream = append(stream, km...)
172+ want = append(want, record{'e', 7, codeKeymapNotify, km})
173+ add('e', uint32(KindUnknown), pktEvent(80, 0, 0x11), 8) // unbound code
174+ add('e', uint32(ExtGenericEvent)<<8|0x80, pktXGE(0, 0, 0), 8)
175+ add('r', uint32(kindUser), pktReply(0, 0), 9)
176+ sent := pktEvent(0x80|33, 0, 0x22) // SendEvent ClientMessage
177+ add('e', 33, sent, 9)
178+ return stream, want
179+}
180+
181+// expectScripted registers the replies of scriptedStream when its first
182+// event arrives. Allocation-free, so the alloc tests use it too.
183+func expectScripted(c *Conn, ev *Event) {
184+ if ev.Kind != 20 {
185+ return
186+ }
187+ for _, d := range [...]Seq{1, 4, 6} {
188+ if err := c.Expect(ev.Seq+d, kindUser, uint32(d), 0, 0, 0); err != nil {
189+ panic(err)
190+ }
191+ }
192+}
193+
194+// expect registers seqs with kind kindUser.
195+func expect(t testing.TB, c *Conn, seqs ...Seq) {
196+ t.Helper()
197+ for _, s := range seqs {
198+ if err := c.Expect(s, kindUser, 0, 0, 0, 0); err != nil {
199+ t.Fatal(err)
200+ }
201+ }
202+}
203+
204+func checkRecords(t testing.TB, got, want []record) {
205+ t.Helper()
206+ if len(got) != len(want) {
207+ t.Fatalf("%d callbacks, want %d: %v", len(got), len(want), got)
208+ }
209+ for i := range want {
210+ g, w := got[i], want[i]
211+ if g.class != w.class || g.seq != w.seq || g.kind != w.kind || !bytes.Equal(g.data, w.data) {
212+ t.Fatalf("callback %d: %v, want %v", i, g, w)
213+ }
214+ }
215+}
216+
217+func TestDispatchStream(t *testing.T) {
218+ stream, want := scriptedStream(t)
219+ t.Run("one read", func(t *testing.T) {
220+ c, srv := dispatchConn(t)
221+ r := newRecorder(c)
222+ feed(t, c, srv, stream, &r.h)
223+ checkRecords(t, r.got, want)
224+ if c.buf.acked != 9 || c.seq != 9 {
225+ t.Fatalf("acked %d seq %d", c.buf.acked, c.seq)
226+ }
227+ if st := c.Stats(); st.UnknownEvents != 1 || st.DroppedReplies != 0 || st.UnhandledErrors != 0 || st.MaxRingUse != 3 {
228+ t.Fatalf("%+v", st)
229+ }
230+ })
231+ t.Run("split at every byte", func(t *testing.T) {
232+ for k := range len(stream) + 1 {
233+ c, srv := dispatchConn(t)
234+ r := newRecorder(c)
235+ feed(t, c, srv, stream[:k], &r.h)
236+ feed(t, c, srv, stream[k:], &r.h)
237+ checkRecords(t, r.got, want)
238+ }
239+ })
240+ t.Run("byte by byte", func(t *testing.T) {
241+ c, srv := dispatchConn(t)
242+ r := newRecorder(c)
243+ for i := range stream {
244+ feed(t, c, srv, stream[i:i+1], &r.h)
245+ }
246+ checkRecords(t, r.got, want)
247+ })
248+ t.Run("nil handler and nil fields", func(t *testing.T) {
249+ c, srv := dispatchConn(t)
250+ feed(t, c, srv, stream, nil)
251+ feed(t, c, srv, stream, &Handler{})
252+ if c.rpos != c.rend {
253+ t.Fatal("stream not consumed")
254+ }
255+ })
256+}
257+
258+func TestDispatchSeqWidening(t *testing.T) {
259+ c, srv := dispatchConn(t)
260+ r := newRecorder(c)
261+ // Wire seqs climbing through two 16-bit wraps.
262+ wires := []uint16{1, 0x7fff, 0xfffe, 0xffff, 0, 3, 0x8000, 0xffff, 2}
263+ wantSeq := []Seq{1, 0x7fff, 0xfffe, 0xffff, 0x10000, 0x10003, 0x18000, 0x1ffff, 0x20002}
264+ var stream []byte
265+ for i, w := range wires {
266+ switch i % 3 {
267+ case 0:
268+ stream = append(stream, pktReply(w, 4)...)
269+ case 1:
270+ stream = append(stream, pktEvent(22, w, 0)...)
271+ default:
272+ stream = append(stream, pktError(3, w, 12)...)
273+ }
274+ }
275+ stream = append(stream, golden(t, "xproto/KeymapNotify.event")...)
276+ expect(t, c, 1, 0xffff, 0x18000)
277+ feed(t, c, srv, stream, &r.h)
278+ if len(r.got) != len(wires)+1 {
279+ t.Fatal(r.got)
280+ }
281+ for i, s := range wantSeq {
282+ if r.got[i].seq != s {
283+ t.Fatalf("packet %d: seq %#x, want %#x", i, r.got[i].seq, s)
284+ }
285+ }
286+ if last := r.got[len(wires)]; last.seq != 0x20002 || c.buf.acked != 0x20002 {
287+ t.Fatalf("KeymapNotify seq %#x, acked %#x", last.seq, c.buf.acked)
288+ }
289+
290+ // ack never passes requests that were not sent.
291+ c2, srv2 := dispatchConn(t)
292+ c2.buf.next = 5
293+ if err := c2.expect(9, kindUser, [4]uint32{}); err != nil { // internal: Expect rejects unsent seqs
294+ t.Fatal(err)
295+ }
296+ feed(t, c2, srv2, pktReply(9, 0), nil)
297+ if c2.buf.acked != 0 || c2.seq != 9 {
298+ t.Fatalf("acked %d seq %d", c2.buf.acked, c2.seq)
299+ }
300+}
301+
302+func TestDispatchGrowth(t *testing.T) {
303+ c, srv := dispatchConn(t)
304+ r := newRecorder(c)
305+ big := pktReply(1, 300<<10) // larger than the 64 KiB ring
306+ xge := pktXGE(2, 5, 100<<10)
307+ stream := append(append(append([]byte{}, pktEvent(22, 1, 0)...), big...), xge...)
308+ expect(t, c, 1)
309+ feed(t, c, srv, stream, &r.h)
310+ if len(r.got) != 3 || !bytes.Equal(r.got[1].data, big) || !bytes.Equal(r.got[2].data, xge) {
311+ t.Fatalf("%v", r.got)
312+ }
313+ if len(c.rb) != 512<<10 {
314+ t.Fatalf("ring %d bytes, want 512 KiB", len(c.rb))
315+ }
316+}
317+
318+func TestDispatchMaxPacket(t *testing.T) {
319+ t.Run("exactly MaxPacket", func(t *testing.T) {
320+ c, srv := dispatchConn(t)
321+ c.maxPacket = 256 << 10
322+ r := newRecorder(c)
323+ p := pktReply(1, c.maxPacket-32)
324+ expect(t, c, 1)
325+ feed(t, c, srv, p, &r.h)
326+ if len(r.got) != 1 || len(r.got[0].data) != c.maxPacket || len(c.rb) != c.maxPacket {
327+ t.Fatalf("%v ring %d", r.got, len(c.rb))
328+ }
329+ })
330+ for _, tc := range []struct {
331+ name string
332+ p []byte
333+ }{
334+ {"reply", pktReply(1, 0)},
335+ {"XGE", pktXGE(1, 0, 0)},
336+ } {
337+ t.Run(tc.name+" beyond MaxPacket", func(t *testing.T) {
338+ c, srv := dispatchConn(t)
339+ r := newRecorder(c)
340+ p := tc.p
341+ binary.LittleEndian.PutUint32(p[4:], (MaxPacket-32)/4+1)
342+ stream := append(pktEvent(22, 1, 0), p...)
343+ if _, err := unix.Write(srv, stream); err != nil {
344+ t.Fatal(err)
345+ }
346+ if err := c.Dispatch(&r.h); !errors.Is(err, ErrPacketTooLarge) {
347+ t.Fatal(err)
348+ }
349+ if len(r.got) != 1 || !errors.Is(c.Dispatch(&r.h), ErrPacketTooLarge) || !errors.Is(c.buf.Err(), ErrPacketTooLarge) {
350+ t.Fatal("not sticky")
351+ }
352+ })
353+ }
354+ t.Run("length overflowing 32 bits", func(t *testing.T) {
355+ if n, ok := packetSize(bytes.Repeat([]byte{1, 0, 0, 0, 0xff, 0xff, 0xff, 0xff}, 4), MaxPacket); ok {
356+ t.Fatal(n)
357+ }
358+ })
359+}
360+
361+func TestDispatchEOF(t *testing.T) {
362+ for _, tail := range [][]byte{nil, pktReply(2, 64)[:40]} {
363+ c, srv := dispatchConn(t)
364+ r := newRecorder(c)
365+ data := append(pktEvent(22, 1, 0), tail...)
366+ if _, err := unix.Write(srv, data); err != nil {
367+ t.Fatal(err)
368+ }
369+ unix.Shutdown(srv, unix.SHUT_WR)
370+ if err := c.Dispatch(&r.h); err != io.EOF {
371+ t.Fatal(err)
372+ }
373+ if len(r.got) != 1 || c.Dispatch(&r.h) != io.EOF || !errors.Is(c.buf.Err(), ErrClosed) {
374+ t.Fatal("EOF not sticky")
375+ }
376+ }
377+ c, _ := dispatchConn(t)
378+ c.Close()
379+ if !errors.Is(c.Dispatch(nil), ErrClosed) {
380+ t.Fatal("Dispatch after Close")
381+ }
382+}
383+
384+func TestDispatchCloseInCallback(t *testing.T) {
385+ c, srv := dispatchConn(t)
386+ n := 0
387+ h := Handler{OnEvent: func(*Event) { n++; c.Close() }}
388+ if _, err := unix.Write(srv, append(pktEvent(22, 1, 0), pktEvent(22, 2, 0)...)); err != nil {
389+ t.Fatal(err)
390+ }
391+ if err := c.Dispatch(&h); !errors.Is(err, ErrClosed) || n != 1 {
392+ t.Fatal(err, n)
393+ }
394+}
395+
396+func TestDispatchReentrant(t *testing.T) {
397+ c, srv := dispatchConn(t)
398+ var nested error
399+ var got []Seq
400+ h := Handler{OnEvent: func(ev *Event) {
401+ got = append(got, ev.Seq)
402+ if len(got) == 1 {
403+ nested = c.Dispatch(nil)
404+ }
405+ }}
406+ var stream []byte
407+ for i := range 5 {
408+ stream = append(stream, pktEvent(22, uint16(i+1), 0)...)
409+ }
410+ if _, err := unix.Write(srv, stream); err != nil {
411+ t.Fatal(err)
412+ }
413+ if err := c.Dispatch(&h); err != nil {
414+ t.Fatal(err)
415+ }
416+ if !errors.Is(nested, errReentrantDispatch) || len(got) != 5 || got[4] != 5 {
417+ t.Fatal(nested, got)
418+ }
419+ if err := c.Dispatch(&h); err != nil || c.dispatching { // usable afterwards
420+ t.Fatal(err)
421+ }
422+}
423+
424+func TestDispatchBounded(t *testing.T) {
425+ c, srv := dispatchConn(t)
426+ stop := make(chan struct{})
427+ done := make(chan struct{})
428+ go func() { // keep the socket full until told to stop
429+ defer close(done)
430+ buf := bytes.Repeat(pktEvent(22, 1, 0), 2048)
431+ for {
432+ select {
433+ case <-stop:
434+ return
435+ default:
436+ }
437+ if _, err := unix.Write(srv, buf); err != nil && err != unix.EAGAIN {
438+ return
439+ }
440+ }
441+ }()
442+ defer func() { close(stop); <-done }()
443+ events := 0
444+ h := Handler{OnEvent: func(*Event) { events++ }}
445+ fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
446+ for n, _ := unix.Poll(fds, 1000); n == 0; n, _ = unix.Poll(fds, 1000) {
447+ t.Fatal("writer never filled the socket")
448+ }
449+ res := make(chan error, 1)
450+ go func() { res <- c.Dispatch(&h) }()
451+ select {
452+ case err := <-res:
453+ if err != nil {
454+ t.Fatal(err)
455+ }
456+ case <-time.After(5 * time.Second):
457+ t.Fatal("Dispatch did not return under a flooding server")
458+ }
459+ // one budget of reads, plus at most the bytes already in the ring
460+ if max := 2 * len(c.rb) / 32; events == 0 || events > max {
461+ t.Fatalf("%d events, want 1..%d", events, max)
462+ }
463+}
464+
465+func TestDispatchReservedKindError(t *testing.T) {
466+ c, srv := dispatchConn(t)
467+ r := newRecorder(c)
468+ if err := c.expect(1, kindDiscard, [4]uint32{}); err != nil {
469+ t.Fatal(err)
470+ }
471+ feed(t, c, srv, pktError(3, 1, 0), &r.h)
472+ if len(r.got) != 0 || c.Stats().UnhandledErrors != 0 {
473+ t.Fatalf("reserved-kind error reached OnError: %v", r.got)
474+ }
475+ if c.live != 0 {
476+ t.Fatal("slot not freed")
477+ }
478+}
479+
480+func TestDispatchClearsBorrowedViews(t *testing.T) {
481+ c, srv := dispatchConn(t)
482+ var h Handler
483+ h.OnReply = func(*Reply) {}
484+ expect(t, c, 2)
485+ stream := append(append(pktReply(1, 64), pktReply(2, 64)...), pktXGE(3, 5, 64)...)
486+ feed(t, c, srv, stream, &h) // reply 1 is dropped, 2 delivered, 3 an event with Tail
487+ if c.reply.Raw != nil || c.ev.Tail != nil {
488+ t.Fatalf("retained views: reply %d ev %d", len(c.reply.Raw), len(c.ev.Tail))
489+ }
490+ feed(t, c, srv, pktReply(4, 8), nil) // dropped reply
491+ if c.reply.Raw != nil {
492+ t.Fatal("dropped reply retained")
493+ }
494+}
495+
496+// recordedStream repeats the scripted stream (events, replies, errors) up to
497+// about size bytes.
498+func recordedStream(t testing.TB, size int) ([]byte, int) {
499+ unit, want := scriptedStream(t)
500+ var s []byte
501+ n := 0
502+ for len(s) < size {
503+ s = append(s, unit...)
504+ n += len(want)
505+ }
506+ return s, n
507+}
508+
509+func TestAllocDispatch(t *testing.T) {
510+ stream, packets := recordedStream(t, 1<<20)
511+ c, srv := dispatchConn(t)
512+ var events, replies, errs int
513+ var sink byte
514+ h := Handler{
515+ OnEvent: func(ev *Event) { expectScripted(c, ev); events++; sink ^= ev.Raw[1]; sink ^= byte(len(ev.Tail)) },
516+ OnReply: func(r *Reply) { replies++; sink ^= r.Raw[len(r.Raw)-1] },
517+ OnError: func(e *XError) { errs++; sink ^= e.Code },
518+ }
519+ feed(t, c, srv, stream, &h) // warm the ring
520+ if n := testing.AllocsPerRun(5, func() { feed(t, c, srv, stream, &h) }); n != 0 {
521+ t.Fatalf("Dispatch allocates %v times per 1 MiB", n)
522+ }
523+ if events+replies+errs != 7*packets || errs == 0 || replies == 0 || c.Stats().DroppedReplies != 0 {
524+ t.Fatal(events, replies, errs, c.Stats())
525+ }
526+ _ = sink
527+}
528+
529+// mixedPackets is a 10k-packet stream: core events, replies, errors, XFIXES
530+// and XGE events.
531+func mixedPackets(t testing.TB) []byte {
532+ var s []byte
533+ for i := range 10000 {
534+ seq := uint16(i)
535+ switch i % 5 {
536+ case 0:
537+ s = append(s, pktEvent(22, seq, 0)...)
538+ case 1:
539+ s = append(s, pktReply(seq, 24)...)
540+ case 2:
541+ s = append(s, pktError(3, seq, 12)...)
542+ case 3:
543+ s = append(s, pktEvent(87, seq, 1)...)
544+ default:
545+ s = append(s, pktXGE(seq, 1, 8)...)
546+ }
547+ }
548+ return s
549+}
550+
551+func BenchmarkDispatch10k(b *testing.B) {
552+ stream := mixedPackets(b)
553+ c, srv := dispatchConn(b)
554+ var sink int
555+ h := Handler{
556+ OnEvent: func(ev *Event) {
557+ if ev.Kind == 22 {
558+ _ = c.Expect(ev.Seq+1, kindUser, 0, 0, 0, 0)
559+ }
560+ sink += int(ev.Kind)
561+ },
562+ OnReply: func(r *Reply) { sink += len(r.Raw) },
563+ OnError: func(e *XError) { sink += int(e.Code) },
564+ }
565+ feed(b, c, srv, stream, &h)
566+ b.SetBytes(int64(len(stream)))
567+ b.ReportAllocs()
568+ for b.Loop() {
569+ feed(b, c, srv, stream, &h)
570+ }
571+ if c.Stats().DroppedReplies != 0 {
572+ b.Fatal(c.Stats())
573+ }
574+ _ = sink
575+}
576+
577+// FuzzDispatchStream feeds an arbitrary byte stream in arbitrary chunks.
578+// Invariants: no panic; the callbacks' bytes concatenate to the stream's
579+// complete packets minus the (unregistered, so dropped) replies, which
580+// Stats.DroppedReplies counts; Seq never decreases; Dispatch ends with io.EOF
581+// (or ErrPacketTooLarge) after the server hangs up.
582+func FuzzDispatchStream(f *testing.F) {
583+ stream, _ := scriptedStream(f)
584+ f.Add(stream, uint16(7))
585+ f.Add(stream, uint16(0x2000|7))
586+ f.Add(stream, uint16(0))
587+ f.Add(append(pktReply(1, 1024), pktXGE(2, 1, 64)...), uint16(33))
588+ f.Fuzz(func(t *testing.T, stream []byte, chunk uint16) {
589+ if len(stream) > 1<<16 {
590+ return
591+ }
592+ c, srv := dispatchConn(t)
593+ c.maxPacket = 8 << 10
594+ c.rb = make([]byte, packetHeadSize+int(chunk>>8)%96) // tiny ring: fuzz growth
595+ var got []byte
596+ var last Seq
597+ mono := func(s Seq) {
598+ if s < last {
599+ t.Fatalf("seq %d after %d", s, last)
600+ }
601+ last = s
602+ }
603+ h := Handler{
604+ OnEvent: func(ev *Event) { mono(ev.Seq); got = append(append(got, ev.Raw[:]...), ev.Tail...) },
605+ OnReply: func(r *Reply) { mono(r.Seq); got = append(got, r.Raw...) },
606+ OnError: func(e *XError) { mono(e.Seq); got = append(got, e.Raw[:]...) },
607+ }
608+ step := int(chunk)%97 + 1
609+ var err error
610+ for p := stream; len(p) > 0 && err == nil; {
611+ k := min(step, len(p))
612+ if _, werr := unix.Write(srv, p[:k]); werr != nil {
613+ t.Fatal(werr)
614+ }
615+ p = p[k:]
616+ err = c.Dispatch(&h)
617+ }
618+ unix.Shutdown(srv, unix.SHUT_WR)
619+ for err == nil {
620+ err = c.Dispatch(&h)
621+ }
622+ if err != io.EOF && !errors.Is(err, ErrPacketTooLarge) {
623+ t.Fatal(err)
624+ }
625+ var want []byte
626+ var replies uint64
627+ for p := stream; len(p) >= 32; {
628+ n, ok := packetSize(p, c.maxPacket)
629+ if !ok || n > len(p) {
630+ break
631+ }
632+ if p[0] == 1 {
633+ replies++
634+ } else {
635+ want = append(want, p[:n]...)
636+ }
637+ p = p[n:]
638+ }
639+ if !bytes.Equal(got, want) || c.Stats().DroppedReplies != replies {
640+ t.Fatalf("callbacks diverge from the stream: %d/%d bytes, %d dropped of %d replies", len(got), len(want), c.Stats().DroppedReplies, replies)
641+ }
642+ })
643+}
@@ -0,0 +1,643 @@
1+package xturbo
2+
3+import (
4+ "bytes"
5+ "encoding/binary"
6+ "errors"
7+ "fmt"
8+ "io"
9+ "testing"
10+ "time"
11+
12+ "bnema.dev/xturbo/internal/xtest"
13+ "golang.org/x/sys/unix"
14+)
15+
16+// dispatchConn returns a Conn over one end of a non-blocking socketpair with
17+// default kernel buffers, and the other end acting as the server. The XFIXES
18+// and Generic Event extensions are negotiated at the golden capture's bases.
19+func dispatchConn(t testing.TB) (*Conn, int) {
20+ t.Helper()
21+ fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_NONBLOCK|unix.SOCK_CLOEXEC, 0)
22+ if err != nil {
23+ t.Fatal(err)
24+ }
25+ c := newConnState(fds[0], nil, setupInfo{})
26+ t.Cleanup(func() { _ = c.Close(); _ = unix.Close(fds[1]) })
27+ c.ext.Set(ExtXFixes, 138, 87, 140)
28+ c.ext.Set(ExtGenericEvent, 0x83, 0, 0)
29+ c.buf.next = 1 << 40 // every widened seq in the tests is acknowledgeable
30+ return c, fds[1]
31+}
32+
33+// Packet builders. seq is the 16-bit wire value.
34+func pktEvent(code uint8, seq uint16, fill byte) []byte {
35+ p := bytes.Repeat([]byte{fill}, 32)
36+ p[0] = code
37+ binary.LittleEndian.PutUint16(p[2:], seq)
38+ return p
39+}
40+
41+func pktReply(seq uint16, data int) []byte {
42+ words := (data + 3) / 4
43+ p := make([]byte, 32+4*words)
44+ p[0], p[1] = 1, 7
45+ binary.LittleEndian.PutUint16(p[2:], seq)
46+ binary.LittleEndian.PutUint32(p[4:], uint32(words))
47+ for i := 8; i < len(p); i++ {
48+ p[i] = byte(i * 7)
49+ }
50+ return p
51+}
52+
53+func pktError(code uint8, seq uint16, major uint8) []byte {
54+ p := make([]byte, 32)
55+ p[1] = code
56+ binary.LittleEndian.PutUint16(p[2:], seq)
57+ binary.LittleEndian.PutUint32(p[4:], 0xdead)
58+ p[10] = major
59+ return p
60+}
61+
62+func pktXGE(seq uint16, evtype uint16, tail int) []byte {
63+ words := (tail + 3) / 4
64+ p := make([]byte, 32+4*words)
65+ p[0], p[1] = codeGenericEvent, 0x83
66+ binary.LittleEndian.PutUint16(p[2:], seq)
67+ binary.LittleEndian.PutUint32(p[4:], uint32(words))
68+ binary.LittleEndian.PutUint16(p[8:], evtype)
69+ for i := 32; i < len(p); i++ {
70+ p[i] = byte(i)
71+ }
72+ return p
73+}
74+
75+// record is one callback as seen by a test handler; data is a copy of the
76+// packet bytes (Raw, plus Tail for events).
77+type record struct {
78+ class byte // 'e', 'r', 'x'
79+ seq Seq
80+ kind uint32
81+ data []byte
82+}
83+
84+func (r record) String() string {
85+ return fmt.Sprintf("%c seq %d kind %#x %d bytes", r.class, r.seq, r.kind, len(r.data))
86+}
87+
88+type recorder struct {
89+ got []record
90+ h Handler
91+}
92+
93+// newRecorder records every callback of c. Replies reach OnReply only when
94+// registered: on each MapRequest event (the first packet of scriptedStream)
95+// it registers the stream's replies, at ev.Seq+1, +4 and +6.
96+func newRecorder(c *Conn) *recorder {
97+ r := &recorder{}
98+ r.h = Handler{
99+ OnEvent: func(ev *Event) {
100+ expectScripted(c, ev)
101+ r.got = append(r.got, record{'e', ev.Seq, uint32(ev.Kind), append(bytes.Clone(ev.Raw[:]), ev.Tail...)})
102+ },
103+ OnReply: func(rp *Reply) {
104+ r.got = append(r.got, record{'r', rp.Seq, uint32(rp.Kind), bytes.Clone(rp.Raw)})
105+ },
106+ OnError: func(e *XError) {
107+ r.got = append(r.got, record{'x', e.Seq, uint32(e.Code), bytes.Clone(e.Raw[:])})
108+ },
109+ }
110+ return r
111+}
112+
113+// feed writes data to the server end, dispatching whenever the socket fills,
114+// then dispatches what remains.
115+func feed(t testing.TB, c *Conn, srv int, data []byte, h *Handler) {
116+ t.Helper()
117+ for len(data) > 0 {
118+ n, err := unix.Write(srv, data)
119+ if err == unix.EAGAIN {
120+ n = 0
121+ } else if err != nil {
122+ t.Fatal(err)
123+ }
124+ data = data[n:]
125+ pump(t, c, h)
126+ }
127+ pump(t, c, h)
128+}
129+
130+// pump dispatches until the socket holds nothing more: one Dispatch reads a
131+// bounded amount.
132+func pump(t testing.TB, c *Conn, h *Handler) {
133+ t.Helper()
134+ for {
135+ if err := c.Dispatch(h); err != nil {
136+ t.Fatal(err)
137+ }
138+ fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
139+ if n, _ := unix.Poll(fds, 0); n == 0 {
140+ return
141+ }
142+ }
143+}
144+
145+func golden(t testing.TB, name string) []byte {
146+ t.Helper()
147+ raw, err := xtest.LoadHex("testdata/golden/" + name + ".hex")
148+ if err != nil {
149+ t.Fatal(err)
150+ }
151+ return raw
152+}
153+
154+// scriptedStream is a mixed stream of goldens and synthetic packets with
155+// increasing wire sequence numbers, and the callbacks it must produce.
156+func scriptedStream(t testing.TB) ([]byte, []record) {
157+ var stream []byte
158+ var want []record
159+ add := func(class byte, kind uint32, p []byte, seq uint16) {
160+ binary.LittleEndian.PutUint16(p[2:], seq)
161+ stream = append(stream, p...)
162+ want = append(want, record{class, Seq(seq), kind, p})
163+ }
164+ add('e', 20, golden(t, "xproto/MapRequest.event"), 3)
165+ add('r', uint32(kindUser), golden(t, "xproto/GetGeometry.reply"), 4)
166+ add('x', 3, golden(t, "xproto/BadWindow.error"), 5)
167+ add('e', uint32(ExtGenericEvent)<<8|0x81, golden(t, "xproto/GeGeneric.event"), 6)
168+ add('e', uint32(ExtXFixes)<<8, golden(t, "xfixes/SelectionNotify.event"), 6)
169+ add('r', uint32(kindUser), golden(t, "xproto/GetProperty.reply"), 7)
170+ km := golden(t, "xproto/KeymapNotify.event") // no sequence: inherits 7
171+ stream = append(stream, km...)
172+ want = append(want, record{'e', 7, codeKeymapNotify, km})
173+ add('e', uint32(KindUnknown), pktEvent(80, 0, 0x11), 8) // unbound code
174+ add('e', uint32(ExtGenericEvent)<<8|0x80, pktXGE(0, 0, 0), 8)
175+ add('r', uint32(kindUser), pktReply(0, 0), 9)
176+ sent := pktEvent(0x80|33, 0, 0x22) // SendEvent ClientMessage
177+ add('e', 33, sent, 9)
178+ return stream, want
179+}
180+
181+// expectScripted registers the replies of scriptedStream when its first
182+// event arrives. Allocation-free, so the alloc tests use it too.
183+func expectScripted(c *Conn, ev *Event) {
184+ if ev.Kind != 20 {
185+ return
186+ }
187+ for _, d := range [...]Seq{1, 4, 6} {
188+ if err := c.Expect(ev.Seq+d, kindUser, uint32(d), 0, 0, 0); err != nil {
189+ panic(err)
190+ }
191+ }
192+}
193+
194+// expect registers seqs with kind kindUser.
195+func expect(t testing.TB, c *Conn, seqs ...Seq) {
196+ t.Helper()
197+ for _, s := range seqs {
198+ if err := c.Expect(s, kindUser, 0, 0, 0, 0); err != nil {
199+ t.Fatal(err)
200+ }
201+ }
202+}
203+
204+func checkRecords(t testing.TB, got, want []record) {
205+ t.Helper()
206+ if len(got) != len(want) {
207+ t.Fatalf("%d callbacks, want %d: %v", len(got), len(want), got)
208+ }
209+ for i := range want {
210+ g, w := got[i], want[i]
211+ if g.class != w.class || g.seq != w.seq || g.kind != w.kind || !bytes.Equal(g.data, w.data) {
212+ t.Fatalf("callback %d: %v, want %v", i, g, w)
213+ }
214+ }
215+}
216+
217+func TestDispatchStream(t *testing.T) {
218+ stream, want := scriptedStream(t)
219+ t.Run("one read", func(t *testing.T) {
220+ c, srv := dispatchConn(t)
221+ r := newRecorder(c)
222+ feed(t, c, srv, stream, &r.h)
223+ checkRecords(t, r.got, want)
224+ if c.buf.acked != 9 || c.seq != 9 {
225+ t.Fatalf("acked %d seq %d", c.buf.acked, c.seq)
226+ }
227+ if st := c.Stats(); st.UnknownEvents != 1 || st.DroppedReplies != 0 || st.UnhandledErrors != 0 || st.MaxRingUse != 3 {
228+ t.Fatalf("%+v", st)
229+ }
230+ })
231+ t.Run("split at every byte", func(t *testing.T) {
232+ for k := range len(stream) + 1 {
233+ c, srv := dispatchConn(t)
234+ r := newRecorder(c)
235+ feed(t, c, srv, stream[:k], &r.h)
236+ feed(t, c, srv, stream[k:], &r.h)
237+ checkRecords(t, r.got, want)
238+ }
239+ })
240+ t.Run("byte by byte", func(t *testing.T) {
241+ c, srv := dispatchConn(t)
242+ r := newRecorder(c)
243+ for i := range stream {
244+ feed(t, c, srv, stream[i:i+1], &r.h)
245+ }
246+ checkRecords(t, r.got, want)
247+ })
248+ t.Run("nil handler and nil fields", func(t *testing.T) {
249+ c, srv := dispatchConn(t)
250+ feed(t, c, srv, stream, nil)
251+ feed(t, c, srv, stream, &Handler{})
252+ if c.rpos != c.rend {
253+ t.Fatal("stream not consumed")
254+ }
255+ })
256+}
257+
258+func TestDispatchSeqWidening(t *testing.T) {
259+ c, srv := dispatchConn(t)
260+ r := newRecorder(c)
261+ // Wire seqs climbing through two 16-bit wraps.
262+ wires := []uint16{1, 0x7fff, 0xfffe, 0xffff, 0, 3, 0x8000, 0xffff, 2}
263+ wantSeq := []Seq{1, 0x7fff, 0xfffe, 0xffff, 0x10000, 0x10003, 0x18000, 0x1ffff, 0x20002}
264+ var stream []byte
265+ for i, w := range wires {
266+ switch i % 3 {
267+ case 0:
268+ stream = append(stream, pktReply(w, 4)...)
269+ case 1:
270+ stream = append(stream, pktEvent(22, w, 0)...)
271+ default:
272+ stream = append(stream, pktError(3, w, 12)...)
273+ }
274+ }
275+ stream = append(stream, golden(t, "xproto/KeymapNotify.event")...)
276+ expect(t, c, 1, 0xffff, 0x18000)
277+ feed(t, c, srv, stream, &r.h)
278+ if len(r.got) != len(wires)+1 {
279+ t.Fatal(r.got)
280+ }
281+ for i, s := range wantSeq {
282+ if r.got[i].seq != s {
283+ t.Fatalf("packet %d: seq %#x, want %#x", i, r.got[i].seq, s)
284+ }
285+ }
286+ if last := r.got[len(wires)]; last.seq != 0x20002 || c.buf.acked != 0x20002 {
287+ t.Fatalf("KeymapNotify seq %#x, acked %#x", last.seq, c.buf.acked)
288+ }
289+
290+ // ack never passes requests that were not sent.
291+ c2, srv2 := dispatchConn(t)
292+ c2.buf.next = 5
293+ if err := c2.expect(9, kindUser, [4]uint32{}); err != nil { // internal: Expect rejects unsent seqs
294+ t.Fatal(err)
295+ }
296+ feed(t, c2, srv2, pktReply(9, 0), nil)
297+ if c2.buf.acked != 0 || c2.seq != 9 {
298+ t.Fatalf("acked %d seq %d", c2.buf.acked, c2.seq)
299+ }
300+}
301+
302+func TestDispatchGrowth(t *testing.T) {
303+ c, srv := dispatchConn(t)
304+ r := newRecorder(c)
305+ big := pktReply(1, 300<<10) // larger than the 64 KiB ring
306+ xge := pktXGE(2, 5, 100<<10)
307+ stream := append(append(append([]byte{}, pktEvent(22, 1, 0)...), big...), xge...)
308+ expect(t, c, 1)
309+ feed(t, c, srv, stream, &r.h)
310+ if len(r.got) != 3 || !bytes.Equal(r.got[1].data, big) || !bytes.Equal(r.got[2].data, xge) {
311+ t.Fatalf("%v", r.got)
312+ }
313+ if len(c.rb) != 512<<10 {
314+ t.Fatalf("ring %d bytes, want 512 KiB", len(c.rb))
315+ }
316+}
317+
318+func TestDispatchMaxPacket(t *testing.T) {
319+ t.Run("exactly MaxPacket", func(t *testing.T) {
320+ c, srv := dispatchConn(t)
321+ c.maxPacket = 256 << 10
322+ r := newRecorder(c)
323+ p := pktReply(1, c.maxPacket-32)
324+ expect(t, c, 1)
325+ feed(t, c, srv, p, &r.h)
326+ if len(r.got) != 1 || len(r.got[0].data) != c.maxPacket || len(c.rb) != c.maxPacket {
327+ t.Fatalf("%v ring %d", r.got, len(c.rb))
328+ }
329+ })
330+ for _, tc := range []struct {
331+ name string
332+ p []byte
333+ }{
334+ {"reply", pktReply(1, 0)},
335+ {"XGE", pktXGE(1, 0, 0)},
336+ } {
337+ t.Run(tc.name+" beyond MaxPacket", func(t *testing.T) {
338+ c, srv := dispatchConn(t)
339+ r := newRecorder(c)
340+ p := tc.p
341+ binary.LittleEndian.PutUint32(p[4:], (MaxPacket-32)/4+1)
342+ stream := append(pktEvent(22, 1, 0), p...)
343+ if _, err := unix.Write(srv, stream); err != nil {
344+ t.Fatal(err)
345+ }
346+ if err := c.Dispatch(&r.h); !errors.Is(err, ErrPacketTooLarge) {
347+ t.Fatal(err)
348+ }
349+ if len(r.got) != 1 || !errors.Is(c.Dispatch(&r.h), ErrPacketTooLarge) || !errors.Is(c.buf.Err(), ErrPacketTooLarge) {
350+ t.Fatal("not sticky")
351+ }
352+ })
353+ }
354+ t.Run("length overflowing 32 bits", func(t *testing.T) {
355+ if n, ok := packetSize(bytes.Repeat([]byte{1, 0, 0, 0, 0xff, 0xff, 0xff, 0xff}, 4), MaxPacket); ok {
356+ t.Fatal(n)
357+ }
358+ })
359+}
360+
361+func TestDispatchEOF(t *testing.T) {
362+ for _, tail := range [][]byte{nil, pktReply(2, 64)[:40]} {
363+ c, srv := dispatchConn(t)
364+ r := newRecorder(c)
365+ data := append(pktEvent(22, 1, 0), tail...)
366+ if _, err := unix.Write(srv, data); err != nil {
367+ t.Fatal(err)
368+ }
369+ unix.Shutdown(srv, unix.SHUT_WR)
370+ if err := c.Dispatch(&r.h); err != io.EOF {
371+ t.Fatal(err)
372+ }
373+ if len(r.got) != 1 || c.Dispatch(&r.h) != io.EOF || !errors.Is(c.buf.Err(), ErrClosed) {
374+ t.Fatal("EOF not sticky")
375+ }
376+ }
377+ c, _ := dispatchConn(t)
378+ c.Close()
379+ if !errors.Is(c.Dispatch(nil), ErrClosed) {
380+ t.Fatal("Dispatch after Close")
381+ }
382+}
383+
384+func TestDispatchCloseInCallback(t *testing.T) {
385+ c, srv := dispatchConn(t)
386+ n := 0
387+ h := Handler{OnEvent: func(*Event) { n++; c.Close() }}
388+ if _, err := unix.Write(srv, append(pktEvent(22, 1, 0), pktEvent(22, 2, 0)...)); err != nil {
389+ t.Fatal(err)
390+ }
391+ if err := c.Dispatch(&h); !errors.Is(err, ErrClosed) || n != 1 {
392+ t.Fatal(err, n)
393+ }
394+}
395+
396+func TestDispatchReentrant(t *testing.T) {
397+ c, srv := dispatchConn(t)
398+ var nested error
399+ var got []Seq
400+ h := Handler{OnEvent: func(ev *Event) {
401+ got = append(got, ev.Seq)
402+ if len(got) == 1 {
403+ nested = c.Dispatch(nil)
404+ }
405+ }}
406+ var stream []byte
407+ for i := range 5 {
408+ stream = append(stream, pktEvent(22, uint16(i+1), 0)...)
409+ }
410+ if _, err := unix.Write(srv, stream); err != nil {
411+ t.Fatal(err)
412+ }
413+ if err := c.Dispatch(&h); err != nil {
414+ t.Fatal(err)
415+ }
416+ if !errors.Is(nested, errReentrantDispatch) || len(got) != 5 || got[4] != 5 {
417+ t.Fatal(nested, got)
418+ }
419+ if err := c.Dispatch(&h); err != nil || c.dispatching { // usable afterwards
420+ t.Fatal(err)
421+ }
422+}
423+
424+func TestDispatchBounded(t *testing.T) {
425+ c, srv := dispatchConn(t)
426+ stop := make(chan struct{})
427+ done := make(chan struct{})
428+ go func() { // keep the socket full until told to stop
429+ defer close(done)
430+ buf := bytes.Repeat(pktEvent(22, 1, 0), 2048)
431+ for {
432+ select {
433+ case <-stop:
434+ return
435+ default:
436+ }
437+ if _, err := unix.Write(srv, buf); err != nil && err != unix.EAGAIN {
438+ return
439+ }
440+ }
441+ }()
442+ defer func() { close(stop); <-done }()
443+ events := 0
444+ h := Handler{OnEvent: func(*Event) { events++ }}
445+ fds := []unix.PollFd{{Fd: int32(c.fd), Events: unix.POLLIN}}
446+ for n, _ := unix.Poll(fds, 1000); n == 0; n, _ = unix.Poll(fds, 1000) {
447+ t.Fatal("writer never filled the socket")
448+ }
449+ res := make(chan error, 1)
450+ go func() { res <- c.Dispatch(&h) }()
451+ select {
452+ case err := <-res:
453+ if err != nil {
454+ t.Fatal(err)
455+ }
456+ case <-time.After(5 * time.Second):
457+ t.Fatal("Dispatch did not return under a flooding server")
458+ }
459+ // one budget of reads, plus at most the bytes already in the ring
460+ if max := 2 * len(c.rb) / 32; events == 0 || events > max {
461+ t.Fatalf("%d events, want 1..%d", events, max)
462+ }
463+}
464+
465+func TestDispatchReservedKindError(t *testing.T) {
466+ c, srv := dispatchConn(t)
467+ r := newRecorder(c)
468+ if err := c.expect(1, kindDiscard, [4]uint32{}); err != nil {
469+ t.Fatal(err)
470+ }
471+ feed(t, c, srv, pktError(3, 1, 0), &r.h)
472+ if len(r.got) != 0 || c.Stats().UnhandledErrors != 0 {
473+ t.Fatalf("reserved-kind error reached OnError: %v", r.got)
474+ }
475+ if c.live != 0 {
476+ t.Fatal("slot not freed")
477+ }
478+}
479+
480+func TestDispatchClearsBorrowedViews(t *testing.T) {
481+ c, srv := dispatchConn(t)
482+ var h Handler
483+ h.OnReply = func(*Reply) {}
484+ expect(t, c, 2)
485+ stream := append(append(pktReply(1, 64), pktReply(2, 64)...), pktXGE(3, 5, 64)...)
486+ feed(t, c, srv, stream, &h) // reply 1 is dropped, 2 delivered, 3 an event with Tail
487+ if c.reply.Raw != nil || c.ev.Tail != nil {
488+ t.Fatalf("retained views: reply %d ev %d", len(c.reply.Raw), len(c.ev.Tail))
489+ }
490+ feed(t, c, srv, pktReply(4, 8), nil) // dropped reply
491+ if c.reply.Raw != nil {
492+ t.Fatal("dropped reply retained")
493+ }
494+}
495+
496+// recordedStream repeats the scripted stream (events, replies, errors) up to
497+// about size bytes.
498+func recordedStream(t testing.TB, size int) ([]byte, int) {
499+ unit, want := scriptedStream(t)
500+ var s []byte
501+ n := 0
502+ for len(s) < size {
503+ s = append(s, unit...)
504+ n += len(want)
505+ }
506+ return s, n
507+}
508+
509+func TestAllocDispatch(t *testing.T) {
510+ stream, packets := recordedStream(t, 1<<20)
511+ c, srv := dispatchConn(t)
512+ var events, replies, errs int
513+ var sink byte
514+ h := Handler{
515+ OnEvent: func(ev *Event) { expectScripted(c, ev); events++; sink ^= ev.Raw[1]; sink ^= byte(len(ev.Tail)) },
516+ OnReply: func(r *Reply) { replies++; sink ^= r.Raw[len(r.Raw)-1] },
517+ OnError: func(e *XError) { errs++; sink ^= e.Code },
518+ }
519+ feed(t, c, srv, stream, &h) // warm the ring
520+ if n := testing.AllocsPerRun(5, func() { feed(t, c, srv, stream, &h) }); n != 0 {
521+ t.Fatalf("Dispatch allocates %v times per 1 MiB", n)
522+ }
523+ if events+replies+errs != 7*packets || errs == 0 || replies == 0 || c.Stats().DroppedReplies != 0 {
524+ t.Fatal(events, replies, errs, c.Stats())
525+ }
526+ _ = sink
527+}
528+
529+// mixedPackets is a 10k-packet stream: core events, replies, errors, XFIXES
530+// and XGE events.
531+func mixedPackets(t testing.TB) []byte {
532+ var s []byte
533+ for i := range 10000 {
534+ seq := uint16(i)
535+ switch i % 5 {
536+ case 0:
537+ s = append(s, pktEvent(22, seq, 0)...)
538+ case 1:
539+ s = append(s, pktReply(seq, 24)...)
540+ case 2:
541+ s = append(s, pktError(3, seq, 12)...)
542+ case 3:
543+ s = append(s, pktEvent(87, seq, 1)...)
544+ default:
545+ s = append(s, pktXGE(seq, 1, 8)...)
546+ }
547+ }
548+ return s
549+}
550+
551+func BenchmarkDispatch10k(b *testing.B) {
552+ stream := mixedPackets(b)
553+ c, srv := dispatchConn(b)
554+ var sink int
555+ h := Handler{
556+ OnEvent: func(ev *Event) {
557+ if ev.Kind == 22 {
558+ _ = c.Expect(ev.Seq+1, kindUser, 0, 0, 0, 0)
559+ }
560+ sink += int(ev.Kind)
561+ },
562+ OnReply: func(r *Reply) { sink += len(r.Raw) },
563+ OnError: func(e *XError) { sink += int(e.Code) },
564+ }
565+ feed(b, c, srv, stream, &h)
566+ b.SetBytes(int64(len(stream)))
567+ b.ReportAllocs()
568+ for b.Loop() {
569+ feed(b, c, srv, stream, &h)
570+ }
571+ if c.Stats().DroppedReplies != 0 {
572+ b.Fatal(c.Stats())
573+ }
574+ _ = sink
575+}
576+
577+// FuzzDispatchStream feeds an arbitrary byte stream in arbitrary chunks.
578+// Invariants: no panic; the callbacks' bytes concatenate to the stream's
579+// complete packets minus the (unregistered, so dropped) replies, which
580+// Stats.DroppedReplies counts; Seq never decreases; Dispatch ends with io.EOF
581+// (or ErrPacketTooLarge) after the server hangs up.
582+func FuzzDispatchStream(f *testing.F) {
583+ stream, _ := scriptedStream(f)
584+ f.Add(stream, uint16(7))
585+ f.Add(stream, uint16(0x2000|7))
586+ f.Add(stream, uint16(0))
587+ f.Add(append(pktReply(1, 1024), pktXGE(2, 1, 64)...), uint16(33))
588+ f.Fuzz(func(t *testing.T, stream []byte, chunk uint16) {
589+ if len(stream) > 1<<16 {
590+ return
591+ }
592+ c, srv := dispatchConn(t)
593+ c.maxPacket = 8 << 10
594+ c.rb = make([]byte, packetHeadSize+int(chunk>>8)%96) // tiny ring: fuzz growth
595+ var got []byte
596+ var last Seq
597+ mono := func(s Seq) {
598+ if s < last {
599+ t.Fatalf("seq %d after %d", s, last)
600+ }
601+ last = s
602+ }
603+ h := Handler{
604+ OnEvent: func(ev *Event) { mono(ev.Seq); got = append(append(got, ev.Raw[:]...), ev.Tail...) },
605+ OnReply: func(r *Reply) { mono(r.Seq); got = append(got, r.Raw...) },
606+ OnError: func(e *XError) { mono(e.Seq); got = append(got, e.Raw[:]...) },
607+ }
608+ step := int(chunk)%97 + 1
609+ var err error
610+ for p := stream; len(p) > 0 && err == nil; {
611+ k := min(step, len(p))
612+ if _, werr := unix.Write(srv, p[:k]); werr != nil {
613+ t.Fatal(werr)
614+ }
615+ p = p[k:]
616+ err = c.Dispatch(&h)
617+ }
618+ unix.Shutdown(srv, unix.SHUT_WR)
619+ for err == nil {
620+ err = c.Dispatch(&h)
621+ }
622+ if err != io.EOF && !errors.Is(err, ErrPacketTooLarge) {
623+ t.Fatal(err)
624+ }
625+ var want []byte
626+ var replies uint64
627+ for p := stream; len(p) >= 32; {
628+ n, ok := packetSize(p, c.maxPacket)
629+ if !ok || n > len(p) {
630+ break
631+ }
632+ if p[0] == 1 {
633+ replies++
634+ } else {
635+ want = append(want, p[:n]...)
636+ }
637+ p = p[n:]
638+ }
639+ if !bytes.Equal(got, want) || c.Stats().DroppedReplies != replies {
640+ t.Fatalf("callbacks diverge from the stream: %d/%d bytes, %d dropped of %d replies", len(got), len(want), c.Stats().DroppedReplies, replies)
641+ }
642+ })
643+}
A ring.go +133 -0
@@ -0,0 +1,133 @@
1+package xturbo
2+
3+import "errors"
4+
5+// ErrRingFull reports more than ringSize registered requests outstanding:
6+// the Expect slot of seq still belongs to an earlier request that has not
7+// completed. Treat it as a bug: Sync, then retry.
8+var ErrRingFull = errors.New("xturbo: reply ring full")
9+
10+var errExpectUnsent = errors.New("xturbo: Expect of a sequence number not yet sent")
11+
12+var errReservedKind = errors.New("xturbo: reply kinds below 16 are reserved")
13+
14+const (
15+ ringSize = 1024
16+ ringMask = ringSize - 1
17+)
18+
19+// Reply kinds below kindUser are reserved by root and consumed internally;
20+// they never reach Handler.OnReply.
21+const (
22+ kindNone uint16 = iota // unregistered
23+ kindDiscard // consume and drop (resync GetInputFocus)
24+ kindInternAtom // InternAtoms: atom into dst
25+ kindQueryExtension // connect: fill the ExtTable
26+ kindXIDRange // XC-MISC GetXIDRange refill
27+ kindBigreqEnable // BIG-REQUESTS Enable: Buf limit
28+ kindUser uint16 = 16
29+)
30+
31+// slot is one Expect registration; seq 0 means free.
32+type slot struct {
33+ seq Seq
34+ kind uint16
35+ args [4]uint32
36+}
37+
38+// Stats are connection counters, for diagnostics.
39+type Stats struct {
40+ DroppedReplies uint64 // replies with no Expect slot
41+ UnhandledErrors uint64 // errors dispatched with a nil Handler.OnError
42+ UnknownEvents uint64 // events decoded as KindUnknown
43+ Resyncs uint64 // GetInputFocus appended by Flush to re-sync sequences
44+ MaxRingUse int // most Expect slots outstanding at once
45+}
46+
47+// Stats returns the connection counters.
48+func (c *Conn) Stats() Stats { return c.stats }
49+
50+// Expect registers the request seq so its reply, or its error, is delivered
51+// with kind and args a0..a3 (Reply.Kind/Args, XError.Kind/Args). Call it
52+// before the next Dispatch. kind must be 16 or more; lower kinds are reserved.
53+// A reply without a registration is dropped (Stats.DroppedReplies); a
54+// registered request that completes without reply or error frees its slot
55+// silently once a later sequence number arrives. seq 0 (a failed request) and
56+// requests already completed are ignored. A seq that was never sent (at or
57+// beyond Buf.Next) is a caller bug and fails. ErrRingFull: more than 1024
58+// registered requests outstanding.
59+func (c *Conn) Expect(seq Seq, kind uint16, a0, a1, a2, a3 uint32) error {
60+ if kind < kindUser {
61+ return errReservedKind
62+ }
63+ if seq >= c.buf.Next() {
64+ return errExpectUnsent
65+ }
66+ return c.expect(seq, kind, [4]uint32{a0, a1, a2, a3})
67+}
68+
69+func (c *Conn) expect(seq Seq, kind uint16, args [4]uint32) error {
70+ if seq == 0 || seq < c.seq {
71+ return nil
72+ }
73+ s := &c.ring[seq&ringMask]
74+ switch s.seq {
75+ case seq:
76+ case 0:
77+ c.live++
78+ c.stats.MaxRingUse = max(c.stats.MaxRingUse, c.live)
79+ default:
80+ return ErrRingFull
81+ }
82+ *s = slot{seq: seq, kind: kind, args: args}
83+ return nil
84+}
85+
86+// retire frees the slots of requests in [prev, full): a packet with sequence
87+// full proves they completed. The slot of full itself stays, its reply or
88+// error may still follow its events.
89+func (c *Conn) retire(prev, full Seq) {
90+ if c.live == 0 || full <= prev {
91+ return
92+ }
93+ if full-prev >= ringSize {
94+ for i := range c.ring {
95+ if s := c.ring[i].seq; s != 0 && s < full {
96+ c.ring[i] = slot{}
97+ c.live--
98+ }
99+ }
100+ return
101+ }
102+ for seq := max(prev, 1); seq < full; seq++ {
103+ if s := &c.ring[seq&ringMask]; s.seq == seq {
104+ *s = slot{}
105+ c.live--
106+ }
107+ }
108+}
109+
110+// take frees and returns the slot of seq, or ok=false when none is registered.
111+func (c *Conn) take(seq Seq) (kind uint16, args [4]uint32, ok bool) {
112+ s := &c.ring[seq&ringMask]
113+ if s.seq != seq || seq == 0 {
114+ return 0, [4]uint32{}, false
115+ }
116+ kind, args = s.kind, s.args
117+ *s = slot{}
118+ c.live--
119+ return kind, args, true
120+}
121+
122+// consumeError handles an error whose request was registered with a reserved
123+// kind (1..15). It reports whether it consumed the error; an unconsumed error
124+// is delivered to OnError. The bootstrap consumers arrive with the connection
125+// bootstrap; until then every reserved-kind error is consumed and dropped,
126+// like consumeReply drops its replies.
127+func (c *Conn) consumeError(*XError) bool { return true }
128+
129+// consumeReply handles a reply registered with a reserved kind. kindDiscard
130+// replies are dropped; the bootstrap kinds (kindInternAtom,
131+// kindQueryExtension, kindXIDRange, kindBigreqEnable) get their consumers
132+// with the connection bootstrap and are dropped until then.
133+func (c *Conn) consumeReply(*Reply) {}
@@ -0,0 +1,133 @@
1+package xturbo
2+
3+import "errors"
4+
5+// ErrRingFull reports more than ringSize registered requests outstanding:
6+// the Expect slot of seq still belongs to an earlier request that has not
7+// completed. Treat it as a bug: Sync, then retry.
8+var ErrRingFull = errors.New("xturbo: reply ring full")
9+
10+var errExpectUnsent = errors.New("xturbo: Expect of a sequence number not yet sent")
11+
12+var errReservedKind = errors.New("xturbo: reply kinds below 16 are reserved")
13+
14+const (
15+ ringSize = 1024
16+ ringMask = ringSize - 1
17+)
18+
19+// Reply kinds below kindUser are reserved by root and consumed internally;
20+// they never reach Handler.OnReply.
21+const (
22+ kindNone uint16 = iota // unregistered
23+ kindDiscard // consume and drop (resync GetInputFocus)
24+ kindInternAtom // InternAtoms: atom into dst
25+ kindQueryExtension // connect: fill the ExtTable
26+ kindXIDRange // XC-MISC GetXIDRange refill
27+ kindBigreqEnable // BIG-REQUESTS Enable: Buf limit
28+ kindUser uint16 = 16
29+)
30+
31+// slot is one Expect registration; seq 0 means free.
32+type slot struct {
33+ seq Seq
34+ kind uint16
35+ args [4]uint32
36+}
37+
38+// Stats are connection counters, for diagnostics.
39+type Stats struct {
40+ DroppedReplies uint64 // replies with no Expect slot
41+ UnhandledErrors uint64 // errors dispatched with a nil Handler.OnError
42+ UnknownEvents uint64 // events decoded as KindUnknown
43+ Resyncs uint64 // GetInputFocus appended by Flush to re-sync sequences
44+ MaxRingUse int // most Expect slots outstanding at once
45+}
46+
47+// Stats returns the connection counters.
48+func (c *Conn) Stats() Stats { return c.stats }
49+
50+// Expect registers the request seq so its reply, or its error, is delivered
51+// with kind and args a0..a3 (Reply.Kind/Args, XError.Kind/Args). Call it
52+// before the next Dispatch. kind must be 16 or more; lower kinds are reserved.
53+// A reply without a registration is dropped (Stats.DroppedReplies); a
54+// registered request that completes without reply or error frees its slot
55+// silently once a later sequence number arrives. seq 0 (a failed request) and
56+// requests already completed are ignored. A seq that was never sent (at or
57+// beyond Buf.Next) is a caller bug and fails. ErrRingFull: more than 1024
58+// registered requests outstanding.
59+func (c *Conn) Expect(seq Seq, kind uint16, a0, a1, a2, a3 uint32) error {
60+ if kind < kindUser {
61+ return errReservedKind
62+ }
63+ if seq >= c.buf.Next() {
64+ return errExpectUnsent
65+ }
66+ return c.expect(seq, kind, [4]uint32{a0, a1, a2, a3})
67+}
68+
69+func (c *Conn) expect(seq Seq, kind uint16, args [4]uint32) error {
70+ if seq == 0 || seq < c.seq {
71+ return nil
72+ }
73+ s := &c.ring[seq&ringMask]
74+ switch s.seq {
75+ case seq:
76+ case 0:
77+ c.live++
78+ c.stats.MaxRingUse = max(c.stats.MaxRingUse, c.live)
79+ default:
80+ return ErrRingFull
81+ }
82+ *s = slot{seq: seq, kind: kind, args: args}
83+ return nil
84+}
85+
86+// retire frees the slots of requests in [prev, full): a packet with sequence
87+// full proves they completed. The slot of full itself stays, its reply or
88+// error may still follow its events.
89+func (c *Conn) retire(prev, full Seq) {
90+ if c.live == 0 || full <= prev {
91+ return
92+ }
93+ if full-prev >= ringSize {
94+ for i := range c.ring {
95+ if s := c.ring[i].seq; s != 0 && s < full {
96+ c.ring[i] = slot{}
97+ c.live--
98+ }
99+ }
100+ return
101+ }
102+ for seq := max(prev, 1); seq < full; seq++ {
103+ if s := &c.ring[seq&ringMask]; s.seq == seq {
104+ *s = slot{}
105+ c.live--
106+ }
107+ }
108+}
109+
110+// take frees and returns the slot of seq, or ok=false when none is registered.
111+func (c *Conn) take(seq Seq) (kind uint16, args [4]uint32, ok bool) {
112+ s := &c.ring[seq&ringMask]
113+ if s.seq != seq || seq == 0 {
114+ return 0, [4]uint32{}, false
115+ }
116+ kind, args = s.kind, s.args
117+ *s = slot{}
118+ c.live--
119+ return kind, args, true
120+}
121+
122+// consumeError handles an error whose request was registered with a reserved
123+// kind (1..15). It reports whether it consumed the error; an unconsumed error
124+// is delivered to OnError. The bootstrap consumers arrive with the connection
125+// bootstrap; until then every reserved-kind error is consumed and dropped,
126+// like consumeReply drops its replies.
127+func (c *Conn) consumeError(*XError) bool { return true }
128+
129+// consumeReply handles a reply registered with a reserved kind. kindDiscard
130+// replies are dropped; the bootstrap kinds (kindInternAtom,
131+// kindQueryExtension, kindXIDRange, kindBigreqEnable) get their consumers
132+// with the connection bootstrap and are dropped until then.
133+func (c *Conn) consumeReply(*Reply) {}
A ring_test.go +271 -0
@@ -0,0 +1,271 @@
1+package xturbo
2+
3+import (
4+ "encoding/binary"
5+ "errors"
6+ "testing"
7+)
8+
9+const kindGeom = kindUser + 1
10+
11+func TestExpectUnsentSeq(t *testing.T) {
12+ c, _ := dispatchConn(t)
13+ c.buf.next = 5 // requests 1..4 were sent
14+ if err := c.Expect(4, kindGeom, 0, 0, 0, 0); err != nil {
15+ t.Fatal(err)
16+ }
17+ for _, seq := range []Seq{5, 6, 1 << 20} {
18+ if err := c.Expect(seq, kindGeom, 0, 0, 0, 0); !errors.Is(err, errExpectUnsent) {
19+ t.Fatalf("Expect(%d) = %v", seq, err)
20+ }
21+ }
22+ if c.live != 1 {
23+ t.Fatalf("live %d", c.live)
24+ }
25+}
26+
27+func TestExpectReplyRouting(t *testing.T) {
28+ c, srv := dispatchConn(t)
29+ r := newRecorder(c)
30+ var args [4]uint32
31+ r.h.OnReply = func(rp *Reply) {
32+ r.got = append(r.got, record{'r', rp.Seq, uint32(rp.Kind), nil})
33+ args = rp.Args
34+ }
35+ if err := c.Expect(2, kindGeom, 10, 11, 12, 13); err != nil {
36+ t.Fatal(err)
37+ }
38+ if err := c.Expect(3, kindGeom+1, 0, 0, 0, 0); err != nil {
39+ t.Fatal(err)
40+ }
41+ if st := c.Stats(); st.MaxRingUse != 2 {
42+ t.Fatal(st)
43+ }
44+ // 1: unregistered reply, dropped. 2: registered. 3: registered no-reply
45+ // request, retired by the later reply 4, which is unregistered.
46+ stream := append(append(append(pktReply(1, 0), pktReply(2, 8)...), pktEvent(22, 3, 0)...), pktReply(4, 0)...)
47+ feed(t, c, srv, stream, &r.h)
48+ if len(r.got) != 2 || r.got[0].class != 'r' || r.got[0].seq != 2 || r.got[0].kind != uint32(kindGeom) || args != [4]uint32{10, 11, 12, 13} {
49+ t.Fatalf("%v %v", r.got, args)
50+ }
51+ if st := c.Stats(); st.DroppedReplies != 2 || c.live != 0 {
52+ t.Fatalf("%+v live %d", st, c.live)
53+ }
54+ // A second reply to the same seq (or one after completion) is dropped.
55+ feed(t, c, srv, pktReply(4, 0), &r.h)
56+ if c.Stats().DroppedReplies != 3 {
57+ t.Fatal(c.Stats())
58+ }
59+ // Expect on a completed request, or seq 0, is ignored.
60+ if c.Expect(2, kindGeom, 0, 0, 0, 0) != nil || c.Expect(0, kindGeom, 0, 0, 0, 0) != nil || c.live != 0 {
61+ t.Fatal("stale Expect registered")
62+ }
63+ if err := c.Expect(9, kindDiscard, 0, 0, 0, 0); err == nil {
64+ t.Fatal("reserved kind accepted")
65+ }
66+}
67+
68+func TestExpectErrorRouting(t *testing.T) {
69+ c, srv := dispatchConn(t)
70+ var got []XError
71+ h := Handler{OnError: func(e *XError) { got = append(got, *e) }}
72+ if err := c.Expect(5, kindGeom, 7, 0, 0, 1); err != nil {
73+ t.Fatal(err)
74+ }
75+ stream := append(append(pktError(3, 4, 14), pktError(3, 5, 14)...), pktError(140, 6, 138)...)
76+ feed(t, c, srv, stream, &h)
77+ if len(got) != 3 {
78+ t.Fatal(got)
79+ }
80+ if got[0].Kind != 0 || got[0].Args != [4]uint32{} || got[0].Seq != 4 {
81+ t.Fatalf("unregistered: %+v", got[0])
82+ }
83+ if got[1].Kind != kindGeom || got[1].Args != [4]uint32{7, 0, 0, 1} || got[1].Seq != 5 || got[1].Major != 14 || got[1].Bad != 0xdead {
84+ t.Fatalf("registered: %+v", got[1])
85+ }
86+ if got[2].Ext != ExtXFixes || got[2].Rel != 0 || got[2].Kind != 0 {
87+ t.Fatalf("extension: %+v", got[2])
88+ }
89+ if c.live != 0 || c.Stats().UnhandledErrors != 0 {
90+ t.Fatal(c.live, c.Stats())
91+ }
92+ // A nil OnError counts.
93+ feed(t, c, srv, pktError(3, 7, 14), nil)
94+ if st := c.Stats(); st.UnhandledErrors != 1 {
95+ t.Fatal(st)
96+ }
97+}
98+
99+func TestReservedKindsConsumed(t *testing.T) {
100+ c, srv := dispatchConn(t)
101+ r := newRecorder(c)
102+ for i, k := range []uint16{kindDiscard, kindInternAtom, kindQueryExtension, kindXIDRange, kindBigreqEnable, 15} {
103+ if err := c.expect(Seq(i+1), k, [4]uint32{}); err != nil {
104+ t.Fatal(err)
105+ }
106+ }
107+ var stream []byte
108+ for i := range 6 {
109+ stream = append(stream, pktReply(uint16(i+1), 4)...)
110+ }
111+ feed(t, c, srv, stream, &r.h)
112+ if len(r.got) != 0 || c.live != 0 || c.Stats().DroppedReplies != 0 {
113+ t.Fatalf("%v live %d %+v", r.got, c.live, c.Stats())
114+ }
115+}
116+
117+func TestExpectRingFull(t *testing.T) {
118+ c, srv := dispatchConn(t)
119+ for seq := Seq(1); seq <= ringSize; seq++ {
120+ if err := c.Expect(seq, kindGeom, 0, 0, 0, 0); err != nil {
121+ t.Fatal(seq, err)
122+ }
123+ }
124+ if err := c.Expect(ringSize+1, kindGeom, 0, 0, 0, 0); !errors.Is(err, ErrRingFull) {
125+ t.Fatal(err)
126+ }
127+ // Re-registering the same seq overwrites.
128+ if err := c.Expect(1, kindGeom+2, 0, 0, 0, 0); err != nil || c.live != ringSize || c.Stats().MaxRingUse != ringSize {
129+ t.Fatal(err, c.live)
130+ }
131+ // An event with seq 1 does not free slot 1 (its reply may follow); one
132+ // with seq 2 does.
133+ feed(t, c, srv, pktEvent(22, 1, 0), nil)
134+ if !errors.Is(c.Expect(ringSize+1, kindGeom, 0, 0, 0, 0), ErrRingFull) {
135+ t.Fatal("slot freed early")
136+ }
137+ feed(t, c, srv, pktEvent(22, 2, 0), nil)
138+ if err := c.Expect(ringSize+1, kindGeom, 0, 0, 0, 0); err != nil || c.live != ringSize {
139+ t.Fatal(err, c.live)
140+ }
141+ // A jump far ahead retires every older slot.
142+ feed(t, c, srv, pktEvent(22, 0x9000, 0), nil)
143+ if c.live != 0 {
144+ t.Fatal(c.live)
145+ }
146+}
147+
148+func TestAllocExpect(t *testing.T) {
149+ c, srv := dispatchConn(t)
150+ var sink uint32
151+ h := Handler{
152+ OnReply: func(r *Reply) { sink += r.Args[0] + uint32(r.Kind) },
153+ OnError: func(e *XError) { sink += e.Args[0] + uint32(e.Kind) },
154+ }
155+ // Each round: 3 registered requests, a reply, an error, a void request
156+ // retired by the next round's first packet.
157+ round := append(append(pktReply(0, 0), pktError(3, 0, 12)...), pktEvent(22, 0, 0)...)
158+ next := Seq(1)
159+ run := func() {
160+ for i := range Seq(3) {
161+ if err := c.Expect(next+i, kindGeom, uint32(i), 0, 0, 0); err != nil {
162+ t.Fatal(err)
163+ }
164+ binary.LittleEndian.PutUint16(round[32*i+2:], uint16(next+i))
165+ }
166+ next += 3
167+ feed(t, c, srv, round, &h)
168+ }
169+ run()
170+ if n := testing.AllocsPerRun(1000, run); n != 0 {
171+ t.Fatalf("Expect + reply/error match allocates %v times", n)
172+ }
173+ if c.Stats().MaxRingUse > 4 {
174+ t.Fatal(c.Stats())
175+ }
176+ _ = sink
177+}
178+
179+// FuzzReplyRing drives Expect and packet arrival against a map model. Each op
180+// byte: low 2 bits choose Expect / reply / error / event, the rest a sequence
181+// step. Invariants: Expect fails only with more than ringSize live
182+// registrations of distinct incomplete seqs; each registered reply/error is
183+// delivered once with its kind and args; live count matches the model.
184+func FuzzReplyRing(f *testing.F) {
185+ f.Add([]byte{0, 4, 1, 8, 2, 0xff, 3, 0x40})
186+ f.Add([]byte{0, 4, 8, 0xfd, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1})
187+ f.Add([]byte{0x80, 0x84, 0x88, 1, 5, 9, 3})
188+ f.Fuzz(func(t *testing.T, ops []byte) {
189+ c, srv := dispatchConn(t)
190+ model := map[Seq]uint32{} // registered seq -> args[0]
191+ var gotSeq Seq
192+ var gotArg uint32
193+ var gotKind uint16
194+ delivered := 0
195+ h := Handler{
196+ OnReply: func(r *Reply) { gotSeq, gotArg, gotKind = r.Seq, r.Args[0], r.Kind; delivered++ },
197+ OnError: func(e *XError) { gotSeq, gotArg, gotKind = e.Seq, e.Args[0], e.Kind; delivered++ },
198+ }
199+ wire := Seq(0) // last sequence sent by the "server"
200+ for i, op := range ops {
201+ step := Seq(op >> 2)
202+ jump := step // replies may leap past the ring and the 16-bit wrap
203+ if step >= 56 {
204+ jump = (step - 55) * 700
205+ }
206+ switch op & 3 {
207+ case 0: // Expect wire+1+step*13
208+ seq := wire + 1 + step*13
209+ err := c.Expect(seq, kindGeom, uint32(i), 0, 0, 0)
210+ clash := false
211+ for s := range model {
212+ if s != seq && s&ringMask == seq&ringMask {
213+ clash = true
214+ }
215+ }
216+ if clash != errors.Is(err, ErrRingFull) || (err != nil && !clash) {
217+ t.Fatalf("Expect(%d) = %v, clash %v", seq, err, clash)
218+ }
219+ if err == nil {
220+ model[seq] = uint32(i)
221+ }
222+ default: // a reply, error or event for wire+step
223+ seq := wire + jump
224+ if seq == 0 {
225+ seq = 1
226+ }
227+ var p []byte
228+ switch op & 3 {
229+ case 1:
230+ p = pktReply(uint16(seq), 0)
231+ case 2:
232+ p = pktError(3, uint16(seq), 1)
233+ default:
234+ p = pktEvent(22, uint16(seq), 0)
235+ }
236+ delivered = 0
237+ gotKind = 0
238+ feed(t, c, srv, p, &h)
239+ if c.seq != seq {
240+ t.Fatalf("widened %d, want %d", c.seq, seq)
241+ }
242+ arg, reg := model[seq]
243+ switch op & 3 {
244+ case 1:
245+ if reg != (delivered == 1) || reg && (gotSeq != seq || gotArg != arg || gotKind != kindGeom) {
246+ t.Fatalf("reply %d: reg %v delivered %d", seq, reg, delivered)
247+ }
248+ case 2:
249+ if delivered != 1 || gotSeq != seq || reg != (gotKind == kindGeom) || reg && gotArg != arg {
250+ t.Fatalf("error %d: reg %v kind %d", seq, reg, gotKind)
251+ }
252+ }
253+ if op&3 != 3 {
254+ delete(model, seq)
255+ }
256+ for s := range model {
257+ if s < seq {
258+ delete(model, s)
259+ }
260+ }
261+ wire = seq
262+ }
263+ if c.live != len(model) || c.live > ringSize {
264+ t.Fatalf("live %d, model %d", c.live, len(model))
265+ }
266+ if c.Stats().MaxRingUse > ringSize {
267+ t.Fatal(c.Stats())
268+ }
269+ }
270+ })
271+}
@@ -0,0 +1,271 @@
1+package xturbo
2+
3+import (
4+ "encoding/binary"
5+ "errors"
6+ "testing"
7+)
8+
9+const kindGeom = kindUser + 1
10+
11+func TestExpectUnsentSeq(t *testing.T) {
12+ c, _ := dispatchConn(t)
13+ c.buf.next = 5 // requests 1..4 were sent
14+ if err := c.Expect(4, kindGeom, 0, 0, 0, 0); err != nil {
15+ t.Fatal(err)
16+ }
17+ for _, seq := range []Seq{5, 6, 1 << 20} {
18+ if err := c.Expect(seq, kindGeom, 0, 0, 0, 0); !errors.Is(err, errExpectUnsent) {
19+ t.Fatalf("Expect(%d) = %v", seq, err)
20+ }
21+ }
22+ if c.live != 1 {
23+ t.Fatalf("live %d", c.live)
24+ }
25+}
26+
27+func TestExpectReplyRouting(t *testing.T) {
28+ c, srv := dispatchConn(t)
29+ r := newRecorder(c)
30+ var args [4]uint32
31+ r.h.OnReply = func(rp *Reply) {
32+ r.got = append(r.got, record{'r', rp.Seq, uint32(rp.Kind), nil})
33+ args = rp.Args
34+ }
35+ if err := c.Expect(2, kindGeom, 10, 11, 12, 13); err != nil {
36+ t.Fatal(err)
37+ }
38+ if err := c.Expect(3, kindGeom+1, 0, 0, 0, 0); err != nil {
39+ t.Fatal(err)
40+ }
41+ if st := c.Stats(); st.MaxRingUse != 2 {
42+ t.Fatal(st)
43+ }
44+ // 1: unregistered reply, dropped. 2: registered. 3: registered no-reply
45+ // request, retired by the later reply 4, which is unregistered.
46+ stream := append(append(append(pktReply(1, 0), pktReply(2, 8)...), pktEvent(22, 3, 0)...), pktReply(4, 0)...)
47+ feed(t, c, srv, stream, &r.h)
48+ if len(r.got) != 2 || r.got[0].class != 'r' || r.got[0].seq != 2 || r.got[0].kind != uint32(kindGeom) || args != [4]uint32{10, 11, 12, 13} {
49+ t.Fatalf("%v %v", r.got, args)
50+ }
51+ if st := c.Stats(); st.DroppedReplies != 2 || c.live != 0 {
52+ t.Fatalf("%+v live %d", st, c.live)
53+ }
54+ // A second reply to the same seq (or one after completion) is dropped.
55+ feed(t, c, srv, pktReply(4, 0), &r.h)
56+ if c.Stats().DroppedReplies != 3 {
57+ t.Fatal(c.Stats())
58+ }
59+ // Expect on a completed request, or seq 0, is ignored.
60+ if c.Expect(2, kindGeom, 0, 0, 0, 0) != nil || c.Expect(0, kindGeom, 0, 0, 0, 0) != nil || c.live != 0 {
61+ t.Fatal("stale Expect registered")
62+ }
63+ if err := c.Expect(9, kindDiscard, 0, 0, 0, 0); err == nil {
64+ t.Fatal("reserved kind accepted")
65+ }
66+}
67+
68+func TestExpectErrorRouting(t *testing.T) {
69+ c, srv := dispatchConn(t)
70+ var got []XError
71+ h := Handler{OnError: func(e *XError) { got = append(got, *e) }}
72+ if err := c.Expect(5, kindGeom, 7, 0, 0, 1); err != nil {
73+ t.Fatal(err)
74+ }
75+ stream := append(append(pktError(3, 4, 14), pktError(3, 5, 14)...), pktError(140, 6, 138)...)
76+ feed(t, c, srv, stream, &h)
77+ if len(got) != 3 {
78+ t.Fatal(got)
79+ }
80+ if got[0].Kind != 0 || got[0].Args != [4]uint32{} || got[0].Seq != 4 {
81+ t.Fatalf("unregistered: %+v", got[0])
82+ }
83+ if got[1].Kind != kindGeom || got[1].Args != [4]uint32{7, 0, 0, 1} || got[1].Seq != 5 || got[1].Major != 14 || got[1].Bad != 0xdead {
84+ t.Fatalf("registered: %+v", got[1])
85+ }
86+ if got[2].Ext != ExtXFixes || got[2].Rel != 0 || got[2].Kind != 0 {
87+ t.Fatalf("extension: %+v", got[2])
88+ }
89+ if c.live != 0 || c.Stats().UnhandledErrors != 0 {
90+ t.Fatal(c.live, c.Stats())
91+ }
92+ // A nil OnError counts.
93+ feed(t, c, srv, pktError(3, 7, 14), nil)
94+ if st := c.Stats(); st.UnhandledErrors != 1 {
95+ t.Fatal(st)
96+ }
97+}
98+
99+func TestReservedKindsConsumed(t *testing.T) {
100+ c, srv := dispatchConn(t)
101+ r := newRecorder(c)
102+ for i, k := range []uint16{kindDiscard, kindInternAtom, kindQueryExtension, kindXIDRange, kindBigreqEnable, 15} {
103+ if err := c.expect(Seq(i+1), k, [4]uint32{}); err != nil {
104+ t.Fatal(err)
105+ }
106+ }
107+ var stream []byte
108+ for i := range 6 {
109+ stream = append(stream, pktReply(uint16(i+1), 4)...)
110+ }
111+ feed(t, c, srv, stream, &r.h)
112+ if len(r.got) != 0 || c.live != 0 || c.Stats().DroppedReplies != 0 {
113+ t.Fatalf("%v live %d %+v", r.got, c.live, c.Stats())
114+ }
115+}
116+
117+func TestExpectRingFull(t *testing.T) {
118+ c, srv := dispatchConn(t)
119+ for seq := Seq(1); seq <= ringSize; seq++ {
120+ if err := c.Expect(seq, kindGeom, 0, 0, 0, 0); err != nil {
121+ t.Fatal(seq, err)
122+ }
123+ }
124+ if err := c.Expect(ringSize+1, kindGeom, 0, 0, 0, 0); !errors.Is(err, ErrRingFull) {
125+ t.Fatal(err)
126+ }
127+ // Re-registering the same seq overwrites.
128+ if err := c.Expect(1, kindGeom+2, 0, 0, 0, 0); err != nil || c.live != ringSize || c.Stats().MaxRingUse != ringSize {
129+ t.Fatal(err, c.live)
130+ }
131+ // An event with seq 1 does not free slot 1 (its reply may follow); one
132+ // with seq 2 does.
133+ feed(t, c, srv, pktEvent(22, 1, 0), nil)
134+ if !errors.Is(c.Expect(ringSize+1, kindGeom, 0, 0, 0, 0), ErrRingFull) {
135+ t.Fatal("slot freed early")
136+ }
137+ feed(t, c, srv, pktEvent(22, 2, 0), nil)
138+ if err := c.Expect(ringSize+1, kindGeom, 0, 0, 0, 0); err != nil || c.live != ringSize {
139+ t.Fatal(err, c.live)
140+ }
141+ // A jump far ahead retires every older slot.
142+ feed(t, c, srv, pktEvent(22, 0x9000, 0), nil)
143+ if c.live != 0 {
144+ t.Fatal(c.live)
145+ }
146+}
147+
148+func TestAllocExpect(t *testing.T) {
149+ c, srv := dispatchConn(t)
150+ var sink uint32
151+ h := Handler{
152+ OnReply: func(r *Reply) { sink += r.Args[0] + uint32(r.Kind) },
153+ OnError: func(e *XError) { sink += e.Args[0] + uint32(e.Kind) },
154+ }
155+ // Each round: 3 registered requests, a reply, an error, a void request
156+ // retired by the next round's first packet.
157+ round := append(append(pktReply(0, 0), pktError(3, 0, 12)...), pktEvent(22, 0, 0)...)
158+ next := Seq(1)
159+ run := func() {
160+ for i := range Seq(3) {
161+ if err := c.Expect(next+i, kindGeom, uint32(i), 0, 0, 0); err != nil {
162+ t.Fatal(err)
163+ }
164+ binary.LittleEndian.PutUint16(round[32*i+2:], uint16(next+i))
165+ }
166+ next += 3
167+ feed(t, c, srv, round, &h)
168+ }
169+ run()
170+ if n := testing.AllocsPerRun(1000, run); n != 0 {
171+ t.Fatalf("Expect + reply/error match allocates %v times", n)
172+ }
173+ if c.Stats().MaxRingUse > 4 {
174+ t.Fatal(c.Stats())
175+ }
176+ _ = sink
177+}
178+
179+// FuzzReplyRing drives Expect and packet arrival against a map model. Each op
180+// byte: low 2 bits choose Expect / reply / error / event, the rest a sequence
181+// step. Invariants: Expect fails only with more than ringSize live
182+// registrations of distinct incomplete seqs; each registered reply/error is
183+// delivered once with its kind and args; live count matches the model.
184+func FuzzReplyRing(f *testing.F) {
185+ f.Add([]byte{0, 4, 1, 8, 2, 0xff, 3, 0x40})
186+ f.Add([]byte{0, 4, 8, 0xfd, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1, 0xfc, 1})
187+ f.Add([]byte{0x80, 0x84, 0x88, 1, 5, 9, 3})
188+ f.Fuzz(func(t *testing.T, ops []byte) {
189+ c, srv := dispatchConn(t)
190+ model := map[Seq]uint32{} // registered seq -> args[0]
191+ var gotSeq Seq
192+ var gotArg uint32
193+ var gotKind uint16
194+ delivered := 0
195+ h := Handler{
196+ OnReply: func(r *Reply) { gotSeq, gotArg, gotKind = r.Seq, r.Args[0], r.Kind; delivered++ },
197+ OnError: func(e *XError) { gotSeq, gotArg, gotKind = e.Seq, e.Args[0], e.Kind; delivered++ },
198+ }
199+ wire := Seq(0) // last sequence sent by the "server"
200+ for i, op := range ops {
201+ step := Seq(op >> 2)
202+ jump := step // replies may leap past the ring and the 16-bit wrap
203+ if step >= 56 {
204+ jump = (step - 55) * 700
205+ }
206+ switch op & 3 {
207+ case 0: // Expect wire+1+step*13
208+ seq := wire + 1 + step*13
209+ err := c.Expect(seq, kindGeom, uint32(i), 0, 0, 0)
210+ clash := false
211+ for s := range model {
212+ if s != seq && s&ringMask == seq&ringMask {
213+ clash = true
214+ }
215+ }
216+ if clash != errors.Is(err, ErrRingFull) || (err != nil && !clash) {
217+ t.Fatalf("Expect(%d) = %v, clash %v", seq, err, clash)
218+ }
219+ if err == nil {
220+ model[seq] = uint32(i)
221+ }
222+ default: // a reply, error or event for wire+step
223+ seq := wire + jump
224+ if seq == 0 {
225+ seq = 1
226+ }
227+ var p []byte
228+ switch op & 3 {
229+ case 1:
230+ p = pktReply(uint16(seq), 0)
231+ case 2:
232+ p = pktError(3, uint16(seq), 1)
233+ default:
234+ p = pktEvent(22, uint16(seq), 0)
235+ }
236+ delivered = 0
237+ gotKind = 0
238+ feed(t, c, srv, p, &h)
239+ if c.seq != seq {
240+ t.Fatalf("widened %d, want %d", c.seq, seq)
241+ }
242+ arg, reg := model[seq]
243+ switch op & 3 {
244+ case 1:
245+ if reg != (delivered == 1) || reg && (gotSeq != seq || gotArg != arg || gotKind != kindGeom) {
246+ t.Fatalf("reply %d: reg %v delivered %d", seq, reg, delivered)
247+ }
248+ case 2:
249+ if delivered != 1 || gotSeq != seq || reg != (gotKind == kindGeom) || reg && gotArg != arg {
250+ t.Fatalf("error %d: reg %v kind %d", seq, reg, gotKind)
251+ }
252+ }
253+ if op&3 != 3 {
254+ delete(model, seq)
255+ }
256+ for s := range model {
257+ if s < seq {
258+ delete(model, s)
259+ }
260+ }
261+ wire = seq
262+ }
263+ if c.live != len(model) || c.live > ringSize {
264+ t.Fatalf("live %d, model %d", c.live, len(model))
265+ }
266+ if c.Stats().MaxRingUse > ringSize {
267+ t.Fatal(c.Stats())
268+ }
269+ }
270+ })
271+}

Comments0 open / 0

no comments yet.

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
l
merge request: the list of files (j k move, enter opens, esc back)
M
merge request: the merge menu; there 1–3 pick the mode, D deletes the branch, enter merges
R
merge request: mark the head read
esc
close a dialog, leave a field

Branches 1

Tags 1

↑↓ moveenter openesc close