1package xturbo
2
3import (
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.
15const MaxPacket = 16 << 20
16
17// ErrPacketTooLarge reports a server packet longer than MaxPacket.
18var ErrPacketTooLarge = errors.New("xturbo: packet too large")
19
20// errReentrantDispatch is returned by Dispatch, and by the blocking calls
21// built on it (Sync, InternAtoms, a NewXID refill), when called from a Handler
22// callback. It is unexported: it reports a programming error, not a condition
23// callers handle. The outer Dispatch is unaffected.
24var errReentrantDispatch = errors.New("xturbo: Dispatch called from a Handler callback")
25
26const (
27 readBufSize = 64 << 10
28 packetHeadSize = 32
29)
30
31// Handler receives decoded packets from Dispatch, in wire order, on the
32// goroutine calling Dispatch. OnReply sees only replies registered with
33// Expect; OnError sees every error (Kind 0 when unregistered). A nil field
34// drops that packet class (nil OnError counts Stats.UnhandledErrors). Fill it
35// once at init; the pointers passed to callbacks are owned by the Conn and are
36// reused for the next packet, so callbacks must not retain them, nor the
37// views they hold (Reply.Raw, Event.Tail). Callbacks may queue requests and
38// call Expect but must not call Dispatch, Sync, InternAtoms, nor NewXID when
39// it needs a refill: those return an error there and change nothing.
40type Handler struct {
41 OnEvent func(ev *Event)
42 OnReply func(r *Reply)
43 OnError func(e *XError)
44}
45
46// Reply is one reply packet to a request registered with Expect. Seq is
47// widened; Kind and Args are the registration's. Raw is the whole reply,
48// 32+4*length bytes, valid only until the callback returns.
49type Reply struct {
50 Seq Seq
51 Kind uint16
52 Args [4]uint32
53 Raw []byte
54}
55
56var noHandler Handler
57
58// Dispatch reads what the server has sent without blocking and calls h for
59// every complete packet, in wire order. It returns nil once the socket would
60// block or after a bounded amount of input (about one read ring, so a flooding
61// server cannot starve the caller's event loop): call it again while the fd is
62// still readable, and Flush requests queued by callbacks in between. It
63// returns io.EOF when the server closed the connection and ErrPacketTooLarge
64// on a packet beyond MaxPacket. Read errors are sticky and also fail the Buf.
65// Dispatch never flushes. A nil h drops every packet; a non-nil h is
66// remembered for the packets Sync(nil), InternAtoms and NewXID read while they
67// block. Errors on the requests root issues itself (reserved kinds) are
68// handled internally and never reach OnError.
69func (c *Conn) Dispatch(h *Handler) error {
70 if c.dispatching {
71 return errReentrantDispatch
72 }
73 c.dispatching = true
74 defer func() { c.dispatching = false }()
75 if c.fd < 0 {
76 return ErrClosed
77 }
78 if c.rerr != nil {
79 return c.rerr
80 }
81 if h == nil {
82 h = &noHandler
83 } else {
84 c.h = h
85 }
86 budget := len(c.rb)
87 c.rdIdle = false
88 read := false
89 for {
90 if err := c.drain(h); err != nil {
91 return c.failRead(err)
92 }
93 if c.fd < 0 { // a callback closed the Conn
94 return ErrClosed
95 }
96 if budget <= 0 {
97 return nil
98 }
99 n, err := unix.Read(c.fd, c.rb[c.rend:])
100 switch {
101 case err == unix.EINTR:
102 continue
103 case err == unix.EAGAIN:
104 c.rdIdle = !read
105 return nil
106 case err == unix.ECONNRESET:
107 return c.failRead(io.EOF)
108 case err != nil:
109 return c.failRead(os.NewSyscallError("read", err))
110 case n == 0:
111 return c.failRead(io.EOF)
112 }
113 c.rend += n
114 budget -= n
115 read = true
116 }
117}
118
119func (c *Conn) failRead(err error) error {
120 c.rerr = err
121 if err == io.EOF {
122 c.buf.Fail(ErrClosed)
123 } else {
124 c.buf.Fail(err)
125 }
126 return err
127}
128
129// drain handles every complete packet in the ring, then makes room for the
130// next one (compacting, growing up to maxPacket).
131func (c *Conn) drain(h *Handler) error {
132 for c.fd >= 0 {
133 avail := c.rend - c.rpos
134 need := packetHeadSize
135 if avail >= packetHeadSize {
136 size, ok := packetSize(c.rb[c.rpos:], c.maxPacket)
137 if !ok {
138 return ErrPacketTooLarge
139 }
140 need = size
141 if avail >= need {
142 end := c.rpos + need
143 p := c.rb[c.rpos:end:end]
144 c.packet(h, p)
145 if poison {
146 fillPoison(p)
147 }
148 c.rpos = end
149 continue
150 }
151 }
152 c.makeRoom(need)
153 break
154 }
155 return nil
156}
157
158// packetSize returns the length of the packet starting at p (len(p) >= 32):
159// 32 for errors and events, 32+4*length for replies and XGE events.
160func packetSize(p []byte, limit int) (int, bool) {
161 n := uint64(packetHeadSize)
162 if p[0] == 1 || p[0]&0x7f == codeGenericEvent {
163 n += 4 * uint64(binary.LittleEndian.Uint32(p[4:]))
164 }
165 if n > uint64(limit) {
166 return 0, false
167 }
168 return int(n), true
169}
170
171// makeRoom moves the unconsumed bytes to the start of the ring and grows it
172// by doubling until need bytes fit. need <= maxPacket.
173func (c *Conn) makeRoom(need int) {
174 if c.rpos > 0 {
175 k := copy(c.rb, c.rb[c.rpos:c.rend])
176 if poison {
177 fillPoison(c.rb[k:c.rend])
178 }
179 c.rpos, c.rend = 0, k
180 }
181 if need <= len(c.rb) {
182 return
183 }
184 size := max(len(c.rb), packetHeadSize)
185 for size < need {
186 size *= 2
187 }
188 grown := make([]byte, min(size, c.maxPacket))
189 copy(grown, c.rb[:c.rend])
190 if poison {
191 fillPoison(c.rb)
192 }
193 c.rb = grown
194}
195
196// packet decodes one complete packet and calls h.
197func (c *Conn) packet(h *Handler, p []byte) {
198 switch p[0] {
199 case 0:
200 decodeError(p, &c.ext, &c.xerr)
201 c.xerr.Seq = c.widen(c.xerr.Seq)
202 c.xerr.Kind, c.xerr.Args, _ = c.take(c.xerr.Seq)
203 switch {
204 case c.xerr.Kind != kindNone && c.xerr.Kind < kindUser:
205 if !c.consumeError(&c.xerr) && h.OnError != nil {
206 h.OnError(&c.xerr)
207 }
208 case h.OnError != nil:
209 h.OnError(&c.xerr)
210 default:
211 c.stats.UnhandledErrors++
212 }
213 case 1:
214 seq := c.widen(Seq(binary.LittleEndian.Uint16(p[2:])))
215 kind, args, ok := c.take(seq)
216 c.reply = Reply{Seq: seq, Kind: kind, Args: args, Raw: p}
217 switch {
218 case !ok:
219 c.stats.DroppedReplies++
220 case kind < kindUser:
221 c.consumeReply(&c.reply)
222 case h.OnReply != nil:
223 h.OnReply(&c.reply)
224 }
225 c.reply.Raw = nil // do not retain the ring
226 default:
227 ev, _, ok := DecodeEvent(p, &c.ext)
228 if !ok { // a sent event claiming code 0 or 1
229 ev = Event{Kind: KindUnknown, SendEvent: true, Seq: Seq(binary.LittleEndian.Uint16(p[2:]))}
230 copy(ev.Raw[:], p)
231 }
232 if p[0]&0x7f == codeKeymapNotify {
233 ev.Seq = c.seq
234 } else {
235 ev.Seq = c.widen(ev.Seq)
236 }
237 if ev.Kind == KindUnknown {
238 c.stats.UnknownEvents++
239 }
240 c.ev = ev
241 if h.OnEvent != nil {
242 h.OnEvent(&c.ev)
243 }
244 c.ev.Tail = nil // do not retain the ring
245 }
246}
247
248// widen turns a 16-bit wire sequence into the full Seq, relative to the last
249// one seen, acknowledges it to the Buf and retires the Expect slots of the
250// requests it proves complete.
251func (c *Conn) widen(wire Seq) Seq {
252 prev := c.seq
253 full := prev&^0xffff | wire&0xffff
254 if full < prev {
255 full += 0x10000
256 }
257 c.seq = full
258 c.buf.ack(full)
259 c.retire(prev, full)
260 return full
261}