1package xturbo
2
3import (
4 "encoding/binary"
5 "errors"
6)
7
8// ErrRingFull reports more than ringSize registered requests outstanding:
9// the Expect slot of seq still belongs to an earlier request that has not
10// completed. Treat it as a bug: Sync, then retry.
11var ErrRingFull = errors.New("xturbo: reply ring full")
12
13var errExpectUnsent = errors.New("xturbo: Expect of a sequence number not yet sent")
14
15var errReservedKind = errors.New("xturbo: reply kinds below 16 are reserved")
16
17const (
18 ringSize = 1024
19 ringMask = ringSize - 1
20)
21
22// Reply kinds below kindUser are reserved by root and consumed internally;
23// they never reach Handler.OnReply.
24const (
25 kindNone uint16 = iota // unregistered
26 kindDiscard // consume and drop (resync GetInputFocus)
27 kindInternAtom // InternAtoms: atom into dst
28 kindQueryExtension // connect: fill the ExtTable
29 kindXIDRange // XC-MISC GetXIDRange refill
30 kindBigreqEnable // BIG-REQUESTS Enable: Buf limit
31 kindUser uint16 = 16
32)
33
34// slot is one Expect registration; seq 0 means free.
35type slot struct {
36 seq Seq
37 kind uint16
38 args [4]uint32
39}
40
41// Stats are connection counters, for diagnostics.
42type Stats struct {
43 DroppedReplies uint64 // replies with no Expect slot
44 UnhandledErrors uint64 // errors dispatched with a nil Handler.OnError
45 UnknownEvents uint64 // events decoded as KindUnknown
46 Resyncs uint64 // GetInputFocus appended by Flush to re-sync sequences
47 MaxRingUse int // most Expect slots outstanding at once
48}
49
50// Stats returns the connection counters.
51func (c *Conn) Stats() Stats { return c.stats }
52
53// Expect registers the request seq so its reply, or its error, is delivered
54// with kind and args a0..a3 (Reply.Kind/Args, XError.Kind/Args). Call it
55// before the next Dispatch. kind must be 16 or more; lower kinds are reserved.
56// A reply without a registration is dropped (Stats.DroppedReplies); a
57// registered request that completes without reply or error frees its slot
58// silently once a later sequence number arrives. seq 0 (a failed request) and
59// requests already completed are ignored. A seq that was never sent (at or
60// beyond Buf.Next) is a caller bug and fails. ErrRingFull: more than 1024
61// registered requests outstanding.
62func (c *Conn) Expect(seq Seq, kind uint16, a0, a1, a2, a3 uint32) error {
63 if kind < kindUser {
64 return errReservedKind
65 }
66 if seq >= c.buf.Next() {
67 return errExpectUnsent
68 }
69 return c.expect(seq, kind, [4]uint32{a0, a1, a2, a3})
70}
71
72func (c *Conn) expect(seq Seq, kind uint16, args [4]uint32) error {
73 if seq == 0 || seq < c.seq {
74 return nil
75 }
76 s := &c.ring[seq&ringMask]
77 switch s.seq {
78 case seq:
79 case 0:
80 c.live++
81 c.stats.MaxRingUse = max(c.stats.MaxRingUse, c.live)
82 default:
83 return ErrRingFull
84 }
85 *s = slot{seq: seq, kind: kind, args: args}
86 return nil
87}
88
89// retire frees the slots of requests in [prev, full): a packet with sequence
90// full proves they completed. The slot of full itself stays, its reply or
91// error may still follow its events.
92func (c *Conn) retire(prev, full Seq) {
93 if c.live == 0 || full <= prev {
94 return
95 }
96 if full-prev >= ringSize {
97 for i := range c.ring {
98 if s := c.ring[i].seq; s != 0 && s < full {
99 c.ring[i] = slot{}
100 c.live--
101 }
102 }
103 return
104 }
105 for seq := max(prev, 1); seq < full; seq++ {
106 if s := &c.ring[seq&ringMask]; s.seq == seq {
107 *s = slot{}
108 c.live--
109 }
110 }
111}
112
113// take frees and returns the slot of seq, or ok=false when none is registered.
114func (c *Conn) take(seq Seq) (kind uint16, args [4]uint32, ok bool) {
115 s := &c.ring[seq&ringMask]
116 if s.seq != seq || seq == 0 {
117 return 0, [4]uint32{}, false
118 }
119 kind, args = s.kind, s.args
120 *s = slot{}
121 c.live--
122 return kind, args, true
123}
124
125// consumeReply handles a reply registered with a reserved kind. kindDiscard
126// and malformed replies are dropped.
127func (c *Conn) consumeReply(r *Reply) {
128 if len(r.Raw) < packetHeadSize {
129 return
130 }
131 switch r.Kind {
132 case kindInternAtom:
133 if i := int(r.Args[0]); i < len(c.atoms) {
134 c.atoms[i] = binary.LittleEndian.Uint32(r.Raw[8:])
135 }
136 case kindQueryExtension:
137 id := ExtID(r.Args[0])
138 if r.Raw[8] == 0 || id == ExtCore || id >= extCount {
139 return
140 }
141 c.ext.Set(id, r.Raw[9], r.Raw[10], r.Raw[11])
142 if id == ExtBigRequests && r.Raw[9] != 0 {
143 // Enable needs the major opcode: queue it from here so one
144 // Sync covers the whole bootstrap.
145 if _, seq := c.buf.Request(r.Raw[9], 0, 0); seq != 0 {
146 _ = c.expect(seq, kindBigreqEnable, [4]uint32{})
147 }
148 }
149 case kindBigreqEnable:
150 // A limit below the core one would be a server bug; keep the core one.
151 if words := binary.LittleEndian.Uint32(r.Raw[8:]); words > coreMaxReq/4 {
152 c.buf.setMaxReqWords(words)
153 }
154 case kindXIDRange:
155 c.gotXIDRange(binary.LittleEndian.Uint32(r.Raw[8:]), binary.LittleEndian.Uint32(r.Raw[12:]))
156 }
157}
158
159// consumeError handles an error whose request was registered with a reserved
160// kind (1..15) and reports whether it consumed it; an unconsumed error is
161// delivered to OnError. Every reserved kind is root's own request, so all are
162// consumed: the first is kept for the blocking call (Sync, InternAtoms) that
163// issued the request, which returns it as *XError; a GetXIDRange error (maybe
164// from PrefetchXIDs) is kept for NewXID.
165func (c *Conn) consumeError(e *XError) bool {
166 if e.Kind == kindXIDRange {
167 // Possibly a prefetch: kept apart from the error of a Sync.
168 c.xidNone, c.xidFailed, c.xidErr = true, true, *e
169 return true
170 }
171 if !c.syncFailed {
172 c.syncFailed = true
173 c.syncErr = *e
174 }
175 return true
176}