-
1
-
2
-
3
-
4
-
5
-
6
-
7
-
8
-
9
-
10
-
11
-
12
-
13
-
14
-
15
-
16
-
17
-
18
-
19
-
20
-
21
-
22
-
23
-
24
-
25
-
26
-
27
-
28
-
29
-
30
-
31
-
32
-
33
-
34
-
35
-
36
-
37
-
38
-
39
-
40
-
41
-
42
-
43
-
44
-
45
-
46
-
47
-
48
-
49
-
50
-
51
-
52
-
53
-
54
-
55
-
56
-
57
-
58
-
59
-
60
-
61
-
62
-
63
-
64
-
65
-
66
-
67
-
68
-
69
-
70
-
71
-
72
-
73
-
74
-
75
-
76
-
77
-
78
-
79
-
80
-
81
-
82
-
83
-
84
-
85
-
86
-
87
-
88
-
89
-
90
-
91
-
92
-
93
-
94
-
95
-
96
-
97
-
98
-
99
-
100
-
101
-
102
-
103
-
104
-
105
-
106
-
107
-
108
-
109
-
110
-
111
-
112
-
113
-
114
-
115
-
116
-
117
-
118
-
119
-
120
-
121
-
122
-
123
-
124
-
125
-
126
-
127
-
128
-
129
-
130
-
131
-
132
-
133
-
134
-
135
-
136
-
137
-
138
-
139
-
140
-
141
-
142
-
143
-
144
-
145
-
146
-
147
-
148
-
149
-
150
-
151
-
152
-
153
-
154
-
155
-
156
-
157
-
158
-
159
-
160
-
161
-
162
-
163
-
164
-
165
-
166
-
167
-
168
-
169
-
170
-
171
-
172
-
173
-
174
-
175
-
176
-
177
-
178
-
179
-
180
-
181
-
182
-
183
-
184
-
185
-
186
-
187
-
188
-
189
-
190
-
191
-
192
-
193
-
194
-
195
-
196
-
197
-
198
-
199
-
200
-
201
-
202
-
203
-
204
-
205
-
206
-
207
-
208
-
209
-
210
-
211
-
212
-
213
-
214
-
215
-
216
-
217
-
218
-
219
-
220
-
221
-
222
-
223
-
224
-
225
-
226
-
227
-
228
-
229
-
230
-
231
-
232
-
233
-
234
-
235
-
236
-
237
-
238
-
239
-
240
-
241
-
242
-
243
-
244
-
245
-
246
-
247
-
248
-
249
-
250
-
251
-
252
-
253
-
254
-
255
-
256
-
257
-
258
-
259
-
260
-
261
-
262
-
263
-
264
-
265
-
266
-
267
-
268
-
269
-
270
-
271
-
272
-
273
-
274
-
275
-
276
-
277
-
278
-
279
-
280
package gomemcached
import (
"encoding/binary"
"fmt"
"io"
"sync"
)
// MCResponse is memcached response
type MCResponse struct {
// The command opcode of the command that sent the request
Opcode CommandCode
// The status of the response
Status Status
// The opaque sent in the request
Opaque uint32
// The CAS identifier (if applicable)
Cas uint64
// Extras, key, and body for this response
Extras, Key, Body []byte
// If true, this represents a fatal condition and we should hang up
Fatal bool
// Datatype identifier
DataType uint8
}
// A debugging string representation of this response
func (res MCResponse) String() string {
return fmt.Sprintf("{MCResponse status=%v keylen=%d, extralen=%d, bodylen=%d}",
res.Status, len(res.Key), len(res.Extras), len(res.Body))
}
// Response as an error.
func (res *MCResponse) Error() string {
return fmt.Sprintf("MCResponse status=%v, opcode=%v, opaque=%v, msg: %s",
res.Status, res.Opcode, res.Opaque, string(res.Body))
}
func errStatus(e error) Status {
status := UNKNOWN_STATUS
if res, ok := e.(*MCResponse); ok {
status = res.Status
}
return status
}
// IsNotFound is true if this error represents a "not found" response.
func IsNotFound(e error) bool {
return errStatus(e) == KEY_ENOENT
}
// IsFatal is false if this error isn't believed to be fatal to a connection.
func IsFatal(e error) bool {
if e == nil {
return false
}
_, ok := isFatal[errStatus(e)]
if ok {
return true
}
return false
}
// Size is number of bytes this response consumes on the wire.
func (res *MCResponse) Size() int {
return HDR_LEN + len(res.Extras) + len(res.Key) + len(res.Body)
}
func (res *MCResponse) fillHeaderBytes(data []byte) int {
pos := 0
data[pos] = RES_MAGIC
pos++
data[pos] = byte(res.Opcode)
pos++
binary.BigEndian.PutUint16(data[pos:pos+2],
uint16(len(res.Key)))
pos += 2
// 4
data[pos] = byte(len(res.Extras))
pos++
// Data type
if res.DataType != 0 {
data[pos] = byte(res.DataType)
} else {
data[pos] = 0
}
pos++
binary.BigEndian.PutUint16(data[pos:pos+2], uint16(res.Status))
pos += 2
// 8
binary.BigEndian.PutUint32(data[pos:pos+4],
uint32(len(res.Body)+len(res.Key)+len(res.Extras)))
pos += 4
// 12
binary.BigEndian.PutUint32(data[pos:pos+4], res.Opaque)
pos += 4
// 16
binary.BigEndian.PutUint64(data[pos:pos+8], res.Cas)
pos += 8
if len(res.Extras) > 0 {
copy(data[pos:pos+len(res.Extras)], res.Extras)
pos += len(res.Extras)
}
if len(res.Key) > 0 {
copy(data[pos:pos+len(res.Key)], res.Key)
pos += len(res.Key)
}
return pos
}
// HeaderBytes will get just the header bytes for this response.
func (res *MCResponse) HeaderBytes() []byte {
data := make([]byte, HDR_LEN+len(res.Extras)+len(res.Key))
res.fillHeaderBytes(data)
return data
}
// Bytes will return the actual bytes transmitted for this response.
func (res *MCResponse) Bytes() []byte {
data := make([]byte, res.Size())
pos := res.fillHeaderBytes(data)
copy(data[pos:pos+len(res.Body)], res.Body)
return data
}
// Transmit will send this response message across a writer.
func (res *MCResponse) Transmit(w io.Writer) (n int, err error) {
if len(res.Body) < 128 {
n, err = w.Write(res.Bytes())
} else {
n, err = w.Write(res.HeaderBytes())
if err == nil {
m := 0
m, err = w.Write(res.Body)
m += n
}
}
return
}
// Receive will fill this MCResponse with the data from this reader.
func (res *MCResponse) Receive(r io.Reader, hdrBytes []byte) (n int, err error) {
return res.ReceiveWithBuf(r, hdrBytes, nil)
}
// ReceiveWithBuf takes an optional pre-allocated []byte buf which
// will be used if its capacity is large enough, otherwise a new
// []byte slice is allocated.
func (res *MCResponse) ReceiveWithBuf(r io.Reader, hdrBytes, buf []byte) (n int, err error) {
if len(hdrBytes) < HDR_LEN {
hdrBytes = []byte{
0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0, 0, 0, 0, 0, 0}
}
n, err = io.ReadFull(r, hdrBytes)
if err != nil {
return n, err
}
if hdrBytes[0] != RES_MAGIC && hdrBytes[0] != REQ_MAGIC {
return n, fmt.Errorf("bad magic: 0x%02x", hdrBytes[0])
}
klen := int(binary.BigEndian.Uint16(hdrBytes[2:4]))
elen := int(hdrBytes[4])
res.Opcode = CommandCode(hdrBytes[1])
res.DataType = uint8(hdrBytes[5])
res.Status = Status(binary.BigEndian.Uint16(hdrBytes[6:8]))
res.Opaque = binary.BigEndian.Uint32(hdrBytes[12:16])
res.Cas = binary.BigEndian.Uint64(hdrBytes[16:24])
bodyLen := int(binary.BigEndian.Uint32(hdrBytes[8:12])) - (klen + elen)
//defer function to debug the panic seen with MB-15557
defer func() {
if e := recover(); e != nil {
err = fmt.Errorf(`Panic in Receive. Response %v \n
key len %v extra len %v bodylen %v`, res, klen, elen, bodyLen)
}
}()
bufNeed := klen + elen + bodyLen
if buf != nil && cap(buf) >= bufNeed {
buf = buf[0:bufNeed]
} else {
buf = make([]byte, bufNeed)
}
m, err := io.ReadFull(r, buf)
if err == nil {
res.Extras = buf[0:elen]
res.Key = buf[elen : klen+elen]
res.Body = buf[klen+elen:]
}
return n + m, err
}
type MCResponsePool struct {
pool *sync.Pool
}
func NewMCResponsePool() *MCResponsePool {
rv := &MCResponsePool{
pool: &sync.Pool{
New: func() interface{} {
return &MCResponse{}
},
},
}
return rv
}
func (this *MCResponsePool) Get() *MCResponse {
return this.pool.Get().(*MCResponse)
}
func (this *MCResponsePool) Put(r *MCResponse) {
if r == nil {
return
}
r.Extras = nil
r.Key = nil
r.Body = nil
r.Fatal = false
this.pool.Put(r)
}
type StringMCResponsePool struct {
pool *sync.Pool
size int
}
func NewStringMCResponsePool(size int) *StringMCResponsePool {
rv := &StringMCResponsePool{
pool: &sync.Pool{
New: func() interface{} {
return make(map[string]*MCResponse, size)
},
},
size: size,
}
return rv
}
func (this *StringMCResponsePool) Get() map[string]*MCResponse {
return this.pool.Get().(map[string]*MCResponse)
}
func (this *StringMCResponsePool) Put(m map[string]*MCResponse) {
if m == nil || len(m) > 2*this.size {
return
}
for k := range m {
m[k] = nil
delete(m, k)
}
this.pool.Put(m)
}