1package xturbo
2
3import (
4 "errors"
5 "os"
6 "strings"
7 "testing"
8 "time"
9
10 "golang.org/x/sys/unix"
11)
12
13const shortDeadline = 150 * time.Millisecond
14
15// within runs f and fails unless it returns an error wrapping
16// os.ErrDeadlineExceeded after about d (not much later).
17func within(t *testing.T, d time.Duration, f func() error) {
18 t.Helper()
19 start := time.Now()
20 err := f()
21 took := time.Since(start)
22 if !errors.Is(err, os.ErrDeadlineExceeded) {
23 t.Fatalf("err %v, want a deadline error", err)
24 }
25 if took < d || took > d+2*time.Second {
26 t.Fatalf("returned after %v, deadline %v", took, d)
27 }
28}
29
30// A server that accepts the connection but never answers the setup request.
31func TestNewConnSetupDeadline(t *testing.T) {
32 shortTimeouts(t, shortDeadline)
33 fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_CLOEXEC, 0)
34 if err != nil {
35 t.Fatal(err)
36 }
37 defer unix.Close(fds[1])
38 within(t, shortDeadline, func() error {
39 c, err := NewConn(fds[0])
40 if c != nil {
41 t.Fatal("conn")
42 }
43 return err
44 })
45 if _, err := unix.FcntlInt(uintptr(fds[0]), unix.F_GETFD, 0); err != unix.EBADF {
46 t.Fatal("fd not closed")
47 }
48}
49
50// A server that completes the setup, then hangs during the bootstrap.
51func TestNewConnBootstrapDeadline(t *testing.T) {
52 shortTimeouts(t, shortDeadline)
53 sc := &script{}
54 sc.silent.Store(true)
55 fd, _ := startScript(t, sc)
56 within(t, shortDeadline, func() error { _, err := NewConn(fd); return err })
57}
58
59func TestSyncDeadline(t *testing.T) {
60 sc := &script{exts: fullExts()}
61 c, _ := dialScript(t, sc)
62 sc.silent.Store(true)
63 c.timeout = shortDeadline
64 within(t, shortDeadline, func() error { return c.Sync(nil) })
65 within(t, shortDeadline, func() error { return c.InternAtoms([]string{"A"}, false, make([]uint32, 1)) })
66 c.xids = xidRange{}
67 within(t, shortDeadline, func() error { _, err := c.NewXID(); return err })
68}
69
70// A server flooding events but never answering the fence still times out.
71func TestSyncDeadlineFlood(t *testing.T) {
72 sc := &script{}
73 c, srv := dialScript(t, sc)
74 c.timeout = shortDeadline
75 sc.silent.Store(true)
76 // Events carry the last request the server processed: the bootstrap's.
77 ev := pktEvent(19, uint16(c.seq), 0)
78 stop := make(chan struct{})
79 done := make(chan struct{})
80 go func() {
81 defer close(done)
82 for {
83 select {
84 case <-stop:
85 return
86 default:
87 // Written raw: the silent server goroutine owns srv.seq.
88 if _, err := srv.f.Write(ev); err != nil {
89 return
90 }
91 }
92 }
93 }()
94 // Closing the client unblocks a writer stuck on a full socket buffer.
95 defer func() { close(stop); _ = c.Close(); <-done }()
96 within(t, shortDeadline, func() error { return c.Sync(nil) })
97}
98
99// A GetXIDRange answer whose last ID overflows the mask (into other clients'
100// IDs) is rejected.
101func TestXIDRangeBounds(t *testing.T) {
102 c, _ := dialScript(t, &script{})
103 // Recorded setup: base 0x00200000, mask 0x001fffff (inc 1).
104 for _, tc := range []struct {
105 start, count uint32
106 ok bool
107 }{
108 {0x00200000, 0xffffffff, false},
109 {0x00200000, 0x00200001, false},
110 {0x00200000, 0x00200000, true},
111 {0x003fffff, 1, true},
112 {0x003fffff, 2, false},
113 {0x00400000, 1, false}, // other client's base
114 {0, 1, false},
115 {0x00200040, 0, false},
116 } {
117 if got := c.validXIDRange(tc.start, tc.count); got != tc.ok {
118 t.Errorf("{%#x, %#x}: %v", tc.start, tc.count, got)
119 }
120 }
121 // A mask with inc > 1: the start must be aligned.
122 c.xidMask = 0x001ffff0
123 if c.validXIDRange(0x00200008, 1) || !c.validXIDRange(0x00200010, 2) || c.validXIDRange(0x001ffff0|0x00200000, 2) {
124 t.Fatal("inc 16")
125 }
126}
127
128func TestNewXIDRejectsOverflowingRange(t *testing.T) {
129 sc := &script{exts: fullExts(), xidRanges: [][2]uint32{{0x00200000, 0xffffffff}}}
130 c, _ := dialScript(t, sc)
131 c.xids = xidRange{}
132 if _, err := c.NewXID(); !errors.Is(err, errXIDsExhausted) {
133 t.Fatal(err)
134 }
135}
136
137// XIDsLow + PrefetchXIDs let a callback allocate past the end of the current
138// range without blocking.
139func TestPrefetchXIDsFromCallback(t *testing.T) {
140 sc := &script{exts: fullExts(), xidRanges: [][2]uint32{{0x00200100, 3}}}
141 sc.hook = func(s *scriptConn, major, _ uint8, _ uint16, _ []byte) bool {
142 if major == 8 {
143 s.event(19, 0)
144 return true
145 }
146 return false
147 }
148 c, srv := dialScript(t, sc)
149 c.xids = xidRange{next: 0x10, left: 2}
150 if !c.XIDsLow() {
151 t.Fatal("not low")
152 }
153 var got []uint32
154 var errs []error
155 h := Handler{OnEvent: func(*Event) {
156 // Inside a callback: the prefetch is queued, no wait.
157 if c.XIDsLow() {
158 if err := c.PrefetchXIDs(); err != nil {
159 errs = append(errs, err)
160 }
161 }
162 }}
163 c.Buf().Request(8, 0, 4)
164 if err := c.Sync(&h); err != nil {
165 t.Fatal(err)
166 }
167 if len(errs) != 0 || c.xidPend.left != 3 || c.XIDsLow() {
168 t.Fatal(errs, c.xidPend, c.XIDsLow())
169 }
170 // Second event: allocate 5 ids inside the callback, crossing ranges.
171 h.OnEvent = func(*Event) {
172 for range 5 {
173 id, err := c.NewXID()
174 if err != nil {
175 errs = append(errs, err)
176 return
177 }
178 got = append(got, id)
179 }
180 // Now everything is used; NewXID cannot block here.
181 if _, err := c.NewXID(); !errors.Is(err, errReentrantDispatch) {
182 errs = append(errs, err)
183 }
184 }
185 c.Buf().Request(8, 0, 4)
186 if err := c.Sync(&h); err != nil {
187 t.Fatal(err)
188 }
189 want := []uint32{0x00200010, 0x00200011, 0x00200100, 0x00200101, 0x00200102}
190 if len(errs) != 0 || len(got) != 5 {
191 t.Fatal(errs, got)
192 }
193 for i := range want {
194 if got[i] != want[i] {
195 t.Fatalf("%#x", got)
196 }
197 }
198 var ranges int
199 for _, r := range srv.requests() {
200 if r.major == 136 && r.minor == minorGetXIDRange {
201 ranges++
202 }
203 }
204 if ranges != 1 {
205 t.Fatal("GetXIDRange requests", ranges)
206 }
207}
208
209func TestPrefetchXIDsStates(t *testing.T) {
210 sc := &script{exts: fullExts(), xidRanges: [][2]uint32{{0, 1}}}
211 c, srv := dialScript(t, sc)
212 if c.XIDsLow() {
213 t.Fatal("fresh setup range reported low")
214 }
215 c.xids = xidRange{next: 0x10, left: 1}
216 before := len(srv.requests())
217 // Twice: one request only while in flight.
218 first, second := c.PrefetchXIDs(), c.PrefetchXIDs()
219 if first != nil || second != nil || c.XIDsLow() {
220 t.Fatal("prefetch", first, second)
221 }
222 if err := c.Sync(nil); err != nil {
223 t.Fatal(err)
224 }
225 if n := len(srv.requests()) - before; n != 2 { // GetXIDRange + fence
226 t.Fatal(n)
227 }
228 // The server had none: not low (nothing to gain), current id still works,
229 // then NewXID's blocking refill reports exhaustion.
230 if c.XIDsLow() {
231 t.Fatal("low after a refused prefetch")
232 }
233 if _, err := c.NewXID(); err != nil {
234 t.Fatal(err)
235 }
236 if _, err := c.NewXID(); !errors.Is(err, errXIDsExhausted) {
237 t.Fatal(err)
238 }
239 // Without XC-MISC.
240 c2, _ := dialScript(t, &script{})
241 c2.xids = xidRange{}
242 if c2.XIDsLow() || !errors.Is(c2.PrefetchXIDs(), ErrExtMissing) {
243 t.Fatal("no XC-MISC")
244 }
245}
246
247// A blocking refill waits for a prefetch already in flight instead of sending
248// a second request; an X error on it is returned by NewXID, not OnError.
249func TestRefillWaitsForPrefetch(t *testing.T) {
250 sc := &script{exts: fullExts(), xidRanges: [][2]uint32{{0x00200200, 4}}}
251 c, srv := dialScript(t, sc)
252 c.xids = xidRange{}
253 if err := c.PrefetchXIDs(); err != nil {
254 t.Fatal(err)
255 }
256 id, err := c.NewXID()
257 if err != nil || id != 0x00200200 {
258 t.Fatalf("%#x %v", id, err)
259 }
260 var ranges int
261 for _, r := range srv.requests() {
262 if r.major == 136 && r.minor == minorGetXIDRange {
263 ranges++
264 }
265 }
266 if ranges != 1 {
267 t.Fatal(ranges)
268 }
269
270 esc := &script{exts: fullExts()}
271 esc.hook = func(s *scriptConn, major, minor uint8, _ uint16, _ []byte) bool {
272 if major == 136 && minor == minorGetXIDRange {
273 s.xerror(11, major, 0)
274 return true
275 }
276 return false
277 }
278 ec, _ := dialScript(t, esc)
279 var userErrs int
280 h := Handler{OnError: func(*XError) { userErrs++ }}
281 ec.xids = xidRange{}
282 if err := ec.PrefetchXIDs(); err != nil {
283 t.Fatal(err)
284 }
285 if err := ec.Sync(&h); err != nil {
286 t.Fatal("a prefetch error leaked into Sync:", err)
287 }
288 var xe *XError
289 if _, err := ec.NewXID(); !errors.As(err, &xe) || xe.Kind != kindXIDRange || userErrs != 0 {
290 t.Fatal(err, userErrs)
291 }
292}
293
294// Steady state: XIDsLow does not allocate.
295func TestAllocXIDsLow(t *testing.T) {
296 c, _ := dialScript(t, &script{exts: fullExts()})
297 if n := testing.AllocsPerRun(1000, func() { _ = c.XIDsLow() }); n != 0 {
298 t.Fatal(n)
299 }
300}
301
302// An over-long name is a caller bug: ErrBadLength, nothing queued, the Buf
303// still usable.
304func TestInternAtomsNameTooLong(t *testing.T) {
305 c, _ := dialScript(t, &script{})
306 next := c.Buf().Next()
307 err := c.InternAtoms([]string{"OK", strings.Repeat("x", 0x10000)}, false, make([]uint32, 2))
308 if !errors.Is(err, ErrBadLength) || c.Buf().Err() != nil || c.Buf().Next() != next {
309 t.Fatal(err, c.Buf().Err())
310 }
311 dst := make([]uint32, 1)
312 if err := c.InternAtoms([]string{"OK"}, false, dst); err != nil || dst[0] == 0 {
313 t.Fatal(err)
314 }
315}
316
317// 0xffff requests queued without a Flush reach ErrSeqOverflow (the auto-resync
318// fence needs a Flush to be appended); with a Flush every so often the fence
319// keeps the stream going far past 0xffff.
320func TestAutoResyncVsSeqOverflow(t *testing.T) {
321 c, _ := dialScript(t, &script{})
322 b := c.Buf()
323 for b.Err() == nil {
324 b.Request(8, 0, 4)
325 }
326 if !errors.Is(b.Err(), ErrSeqOverflow) || b.Next()-b.acked != maxSeqAhead {
327 t.Fatal(b.Err(), b.Next()-b.acked)
328 }
329 if c.Stats().Resyncs != 0 {
330 t.Fatal(c.Stats())
331 }
332
333 c2, _ := dialScript(t, &script{})
334 b2 := c2.Buf()
335 for i := range 3 * 0x10000 {
336 b2.Request(8, 0, 4)
337 if i%4096 == 0 {
338 if err := flushAll(c2); err != nil {
339 t.Fatal(err)
340 }
341 if err := c2.Dispatch(nil); err != nil {
342 t.Fatal(err)
343 }
344 // Give the fence's reply time to come back before overflow.
345 if b2.Next()-b2.acked > 0xe000 {
346 dispatchFor(t, c2, func() bool { return b2.Next()-b2.acked < 0x9000 })
347 }
348 }
349 }
350 if b2.Err() != nil {
351 t.Fatal(b2.Err())
352 }
353 if err := c2.Sync(nil); err != nil {
354 t.Fatal(err)
355 }
356 if st := c2.Stats(); st.Resyncs < 3 || st.DroppedReplies != 0 {
357 t.Fatalf("%+v", st)
358 }
359}