buf.go RawHistoryFolder
1package xturbo
2
3import (
4 "encoding/binary"
5 "errors"
6 "io"
7 "os"
8
9 "golang.org/x/sys/unix"
10)
11
12// Seq identifies a request; zero is invalid.
13type Seq uint64
14
15var (
16 ErrWouldBlock = errors.New("xturbo: write would block")
17 ErrClosed = errors.New("xturbo: connection closed")
18 ErrRequestTooLarge = errors.New("xturbo: request too large")
19 ErrExtMissing = errors.New("xturbo: extension missing")
20 ErrBadLength = errors.New("xturbo: bad length")
21 ErrSeqOverflow = errors.New("xturbo: sequence overflow")
22)
23
24const (
25 defaultBufSize = 64 << 10
26 // coreMaxReq is the largest core request in bytes (0xFFFF four-byte units).
27 coreMaxReq = 0xffff * 4
28 // maxSeqAhead bounds unacknowledged requests so 16-bit wire sequence
29 // numbers widen unambiguously.
30 maxSeqAhead = 0xffff
31)
32
33// Buf is the caller-owned request encoder. Conn owns flushing and
34// acknowledgment. It is not safe for concurrent use.
35type Buf struct {
36 Ext *ExtTable
37 b []byte
38 flushed int
39 next, acked Seq
40 maxReq uint32
41 err error
42}
43
44// NewBuf returns an encoder with size bytes preallocated; size <= 0 means 64 KiB.
45func NewBuf(size int) *Buf {
46 if size <= 0 {
47 size = defaultBufSize
48 }
49 return &Buf{b: make([]byte, 0, size), next: 1, maxReq: coreMaxReq}
50}
51
52// Fail records err as the sticky error; the first error wins.
53func (b *Buf) Fail(err error) {
54 if b.err == nil {
55 b.err = err
56 }
57}
58func (b *Buf) Err() error { return b.err }
59
60// Reset drops unflushed bytes and keeps sequence counters. Tests only.
61func (b *Buf) Reset() { b.b = b.b[:0]; b.flushed = 0 }
62
63// Bytes returns the unflushed bytes.
64func (b *Buf) Bytes() []byte { return b.b[b.flushed:] }
65func (b *Buf) Len() int { return len(b.b) - b.flushed }
66
67// Next returns the sequence number the next Request will get.
68func (b *Buf) Next() Seq { return b.next }
69
70// Request reserves a zeroed body of body bytes after the wire header and
71// returns it with the request's sequence number. body must be a multiple of 4.
72// Requests longer than 0xFFFF units get the BIG-REQUESTS 8-byte header.
73// On failure it records a sticky error and returns (nil, 0).
74// The body is valid only until the next Request, flush or Reset on b.
75func (b *Buf) Request(major, data uint8, body int) ([]byte, Seq) {
76 if b.err != nil {
77 return nil, 0
78 }
79 if body < 0 || body&3 != 0 {
80 b.Fail(ErrBadLength)
81 return nil, 0
82 }
83 header := 4
84 if uint64(body)+4 > coreMaxReq {
85 header = 8
86 }
87 if uint64(body)+uint64(header) > uint64(b.maxReq) {
88 b.Fail(ErrRequestTooLarge)
89 return nil, 0
90 }
91 if b.next-b.acked >= maxSeqAhead {
92 b.Fail(ErrSeqOverflow)
93 return nil, 0
94 }
95 total := body + header
96 b.reserve(total)
97 start := len(b.b)
98 b.b = b.b[:start+total]
99 p := b.b[start:]
100 clear(p)
101 p[0] = major
102 p[1] = data
103 if header == 4 {
104 binary.LittleEndian.PutUint16(p[2:], uint16(total/4))
105 } else {
106 binary.LittleEndian.PutUint32(p[4:], uint32(total/4))
107 }
108 seq := b.next
109 b.next++
110 return p[header:], seq
111}
112
113// reserve makes room for n more bytes. Flushed bytes are dropped first; the
114// buffer then doubles until n fits. It never shrinks.
115func (b *Buf) reserve(n int) {
116 if n <= cap(b.b)-len(b.b) {
117 return
118 }
119 pending := len(b.b) - b.flushed
120 if b.flushed > 0 && n <= cap(b.b)-pending {
121 copy(b.b, b.b[b.flushed:])
122 b.b = b.b[:pending]
123 b.flushed = 0
124 return
125 }
126 capacity := max(cap(b.b), 1)
127 for capacity-pending < n {
128 capacity *= 2
129 }
130 grown := make([]byte, pending, capacity)
131 copy(grown, b.b[b.flushed:])
132 b.b = grown
133 b.flushed = 0
134}
135
136// setMaxReqWords sets the request size limit from the BIG-REQUESTS
137// maximum_request_length (four-byte units). Conn calls it after enabling it.
138func (b *Buf) setMaxReqWords(words uint32) {
139 b.maxReq = uint32(min(uint64(words)*4, 0xfffffffc))
140}
141
142// ack records seq as the highest sequence number seen on the read side.
143func (b *Buf) ack(seq Seq) {
144 if seq > b.acked && seq < b.next {
145 b.acked = seq
146 }
147}
148
149// flush writes unflushed bytes to the non-blocking fd. It returns
150// ErrWouldBlock when the socket is full (the remainder stays queued) and nil
151// once everything is written. Any other failure becomes the sticky error.
152// Conn.Flush wraps it.
153func (b *Buf) flush(fd int) error {
154 if b.err != nil {
155 return b.err
156 }
157 for b.flushed < len(b.b) {
158 n, err := unix.Write(fd, b.b[b.flushed:])
159 switch {
160 case err == unix.EINTR:
161 continue
162 case err == unix.EAGAIN:
163 return ErrWouldBlock
164 case err == unix.EPIPE || err == unix.ECONNRESET:
165 b.Fail(ErrClosed)
166 return b.err
167 case err != nil:
168 b.Fail(os.NewSyscallError("write", err))
169 return b.err
170 case n <= 0:
171 b.Fail(io.ErrShortWrite)
172 return b.err
173 }
174 if poison {
175 fillPoison(b.b[b.flushed : b.flushed+n])
176 }
177 b.flushed += n
178 }
179 b.b = b.b[:0]
180 b.flushed = 0
181 return nil
182}