bench-forgejo/vendor/github.com/couchbase/gomemcached/mc_req.go

483 lines
14 KiB
Go
Raw Normal View History

2019-02-05 22:22:51 +05:30
package gomemcached
import (
"encoding/binary"
"fmt"
"io"
)
// The maximum reasonable body length to expect.
// Anything larger than this will result in an error.
// The current limit, 20MB, is the size limit supported by ep-engine.
var MaxBodyLen = int(20 * 1024 * 1024)
// MCRequest is memcached Request
type MCRequest struct {
// The command being issued
Opcode CommandCode
// The CAS (if applicable, or 0)
Cas uint64
// An opaque value to be returned with this request
Opaque uint32
// The vbucket to which this command belongs
VBucket uint16
// Command extras, key, and body
Extras, Key, Body, ExtMeta []byte
// Datatype identifier
DataType uint8
// len() calls are expensive - cache this in case for collection
Keylen int
// Flexible Framing Extras
FramingExtras []FrameInfo
// Stored length of incoming framing extras
FramingElen int
2019-02-05 22:22:51 +05:30
}
// Size gives the number of bytes this request requires.
func (req *MCRequest) Size() int {
return HDR_LEN + len(req.Extras) + len(req.Key) + len(req.Body) + len(req.ExtMeta) + req.FramingElen
2019-02-05 22:22:51 +05:30
}
// A debugging string representation of this request
func (req MCRequest) String() string {
return fmt.Sprintf("{MCRequest opcode=%s, bodylen=%d, key='%s'}",
req.Opcode, len(req.Body), req.Key)
}
func (req *MCRequest) fillRegularHeaderBytes(data []byte) int {
// Byte/ 0 | 1 | 2 | 3 |
// / | | | |
// |0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|
// +---------------+---------------+---------------+---------------+
// 0| Magic | Opcode | Key length |
// +---------------+---------------+---------------+---------------+
// 4| Extras length | Data type | vbucket id |
// +---------------+---------------+---------------+---------------+
// 8| Total body length |
// +---------------+---------------+---------------+---------------+
// 12| Opaque |
// +---------------+---------------+---------------+---------------+
// 16| CAS |
// | |
// +---------------+---------------+---------------+---------------+
// Total 24 bytes
2019-02-05 22:22:51 +05:30
pos := 0
data[pos] = REQ_MAGIC
pos++
data[pos] = byte(req.Opcode)
pos++
binary.BigEndian.PutUint16(data[pos:pos+2],
uint16(len(req.Key)))
pos += 2
// 4
data[pos] = byte(len(req.Extras))
pos++
// Data type
if req.DataType != 0 {
data[pos] = byte(req.DataType)
}
pos++
binary.BigEndian.PutUint16(data[pos:pos+2], req.VBucket)
pos += 2
// 8
binary.BigEndian.PutUint32(data[pos:pos+4],
uint32(len(req.Body)+len(req.Key)+len(req.Extras)+len(req.ExtMeta)))
pos += 4
// 12
binary.BigEndian.PutUint32(data[pos:pos+4], req.Opaque)
pos += 4
// 16
if req.Cas != 0 {
binary.BigEndian.PutUint64(data[pos:pos+8], req.Cas)
}
pos += 8
if len(req.Extras) > 0 {
copy(data[pos:pos+len(req.Extras)], req.Extras)
pos += len(req.Extras)
}
if len(req.Key) > 0 {
copy(data[pos:pos+len(req.Key)], req.Key)
pos += len(req.Key)
}
return pos
}
// Returns pos and if trailing by half byte
func (req *MCRequest) fillFlexHeaderBytes(data []byte) (int, bool) {
// Byte/ 0 | 1 | 2 | 3 |
// / | | | |
// |0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|0 1 2 3 4 5 6 7|
// +---------------+---------------+---------------+---------------+
// 0| Magic (0x08) | Opcode | Framing extras| Key Length |
// +---------------+---------------+---------------+---------------+
// 4| Extras length | Data type | vbucket id |
// +---------------+---------------+---------------+---------------+
// 8| Total body length |
// +---------------+---------------+---------------+---------------+
// 12| Opaque |
// +---------------+---------------+---------------+---------------+
// 16| CAS |
// | |
// +---------------+---------------+---------------+---------------+
// Total 24 bytes
data[0] = FLEX_MAGIC
data[1] = byte(req.Opcode)
data[2] = byte(req.FramingElen)
data[3] = byte(req.Keylen)
elen := len(req.Extras)
data[4] = byte(elen)
if req.DataType != 0 {
data[5] = byte(req.DataType)
}
binary.BigEndian.PutUint16(data[6:8], req.VBucket)
binary.BigEndian.PutUint32(data[8:12],
uint32(len(req.Body)+req.Keylen+elen+len(req.ExtMeta)+req.FramingElen))
binary.BigEndian.PutUint32(data[12:16], req.Opaque)
if req.Cas != 0 {
binary.BigEndian.PutUint64(data[16:24], req.Cas)
}
pos := HDR_LEN
// Add framing infos
var framingExtras []byte
var outputBytes []byte
var mergeModeSrc []byte
var frameBytes int
var halfByteMode bool
var mergeMode bool
for _, frameInfo := range req.FramingExtras {
if !mergeMode {
outputBytes, halfByteMode = frameInfo.Bytes()
if !halfByteMode {
framingExtras = append(framingExtras, outputBytes...)
frameBytes += len(outputBytes)
} else {
mergeMode = true
mergeModeSrc = outputBytes
}
} else {
outputBytes, halfByteMode = frameInfo.Bytes()
outputBytes := ShiftByteSliceRight4Bits(outputBytes)
if halfByteMode {
// Previous halfbyte merge with this halfbyte will result in a complete byte
mergeMode = false
outputBytes = Merge2HalfByteSlices(mergeModeSrc, outputBytes)
framingExtras = append(framingExtras, outputBytes...)
frameBytes += len(outputBytes)
} else {
// Merge half byte with a non-half byte will result in a combined half-byte that will
// become the source for the next iteration
mergeModeSrc = Merge2HalfByteSlices(mergeModeSrc, outputBytes)
}
}
}
if mergeMode {
// Commit the temporary merge area into framingExtras
framingExtras = append(framingExtras, mergeModeSrc...)
frameBytes += len(mergeModeSrc)
}
copy(data[pos:pos+frameBytes], framingExtras)
pos += frameBytes
// Add Extras
if len(req.Extras) > 0 {
if mergeMode {
outputBytes = ShiftByteSliceRight4Bits(req.Extras)
data = Merge2HalfByteSlices(data, outputBytes)
} else {
copy(data[pos:pos+elen], req.Extras)
}
pos += elen
}
// Add keys
if req.Keylen > 0 {
if mergeMode {
outputBytes = ShiftByteSliceRight4Bits(req.Key)
data = Merge2HalfByteSlices(data, outputBytes)
} else {
copy(data[pos:pos+req.Keylen], req.Key)
}
pos += req.Keylen
}
return pos, mergeMode
}
func (req *MCRequest) FillHeaderBytes(data []byte) (int, bool) {
if req.FramingElen == 0 {
return req.fillRegularHeaderBytes(data), false
} else {
return req.fillFlexHeaderBytes(data)
}
}
2019-02-05 22:22:51 +05:30
// HeaderBytes will return the wire representation of the request header
// (with the extras and key).
func (req *MCRequest) HeaderBytes() []byte {
data := make([]byte, HDR_LEN+len(req.Extras)+len(req.Key)+req.FramingElen)
2019-02-05 22:22:51 +05:30
req.FillHeaderBytes(data)
2019-02-05 22:22:51 +05:30
return data
}
// Bytes will return the wire representation of this request.
func (req *MCRequest) Bytes() []byte {
data := make([]byte, req.Size())
pos, halfByteMode := req.FillHeaderBytes(data)
// TODO - the halfByteMode should be revisited for a more efficient
// way of doing things
2019-02-05 22:22:51 +05:30
if len(req.Body) > 0 {
if halfByteMode {
shifted := ShiftByteSliceRight4Bits(req.Body)
data = Merge2HalfByteSlices(data, shifted)
} else {
copy(data[pos:pos+len(req.Body)], req.Body)
}
2019-02-05 22:22:51 +05:30
}
if len(req.ExtMeta) > 0 {
if halfByteMode {
shifted := ShiftByteSliceRight4Bits(req.ExtMeta)
data = Merge2HalfByteSlices(data, shifted)
} else {
copy(data[pos+len(req.Body):pos+len(req.Body)+len(req.ExtMeta)], req.ExtMeta)
}
2019-02-05 22:22:51 +05:30
}
return data
}
// Transmit will send this request message across a writer.
func (req *MCRequest) Transmit(w io.Writer) (n int, err error) {
if len(req.Body) < 128 {
n, err = w.Write(req.Bytes())
} else {
n, err = w.Write(req.HeaderBytes())
if err == nil {
m := 0
m, err = w.Write(req.Body)
n += m
}
}
return
}
func (req *MCRequest) receiveHeaderCommon(hdrBytes []byte) (elen, totalBodyLen int) {
elen = int(hdrBytes[4])
2019-02-05 22:22:51 +05:30
// Data type at 5
req.DataType = uint8(hdrBytes[5])
req.Opcode = CommandCode(hdrBytes[1])
// Vbucket at 6:7
req.VBucket = binary.BigEndian.Uint16(hdrBytes[6:])
totalBodyLen = int(binary.BigEndian.Uint32(hdrBytes[8:]))
2019-02-05 22:22:51 +05:30
req.Opaque = binary.BigEndian.Uint32(hdrBytes[12:])
req.Cas = binary.BigEndian.Uint64(hdrBytes[16:])
return
}
2019-02-05 22:22:51 +05:30
func (req *MCRequest) receiveRegHeader(hdrBytes []byte) (elen, totalBodyLen int) {
elen, totalBodyLen = req.receiveHeaderCommon(hdrBytes)
req.Keylen = int(binary.BigEndian.Uint16(hdrBytes[2:]))
return
}
func (req *MCRequest) receiveFlexibleFramingHeader(hdrBytes []byte) (elen, totalBodyLen, framingElen int) {
elen, totalBodyLen = req.receiveHeaderCommon(hdrBytes)
// For flexible framing header, key length is a single byte at byte index 3
req.Keylen = int(binary.BigEndian.Uint16(hdrBytes[2:]) & 0x0ff)
// Flexible framing lengh is a single byte at index 2
framingElen = int(binary.BigEndian.Uint16(hdrBytes[2:]) >> 8)
req.FramingElen = framingElen
return
}
func (req *MCRequest) populateRegularBody(r io.Reader, totalBodyLen, elen int) (int, error) {
var m int
var err error
2019-02-05 22:22:51 +05:30
if totalBodyLen > 0 {
buf := make([]byte, totalBodyLen)
m, err = io.ReadFull(r, buf)
2019-02-05 22:22:51 +05:30
if err == nil {
if req.Opcode >= TAP_MUTATION &&
req.Opcode <= TAP_CHECKPOINT_END &&
len(buf) > 1 {
// In these commands there is "engine private"
// data at the end of the extras. The first 2
// bytes of extra data give its length.
elen += int(binary.BigEndian.Uint16(buf))
}
req.Extras = buf[0:elen]
req.Key = buf[elen : req.Keylen+elen]
2019-02-05 22:22:51 +05:30
// get the length of extended metadata
extMetaLen := 0
if elen > 29 {
extMetaLen = int(binary.BigEndian.Uint16(req.Extras[28:30]))
}
bodyLen := totalBodyLen - req.Keylen - elen - extMetaLen
2019-02-05 22:22:51 +05:30
if bodyLen > MaxBodyLen {
return m, fmt.Errorf("%d is too big (max %d)",
2019-02-05 22:22:51 +05:30
bodyLen, MaxBodyLen)
}
req.Body = buf[req.Keylen+elen : req.Keylen+elen+bodyLen]
req.ExtMeta = buf[req.Keylen+elen+bodyLen:]
}
}
return m, err
}
func (req *MCRequest) populateFlexBody(r io.Reader, totalBodyLen, elen, framingElen int) (int, error) {
var m int
var err error
if totalBodyLen > 0 {
buf := make([]byte, totalBodyLen)
m, err = io.ReadFull(r, buf)
if err != nil {
return m, err
}
err = req.populateFlexBodyInternal(buf, totalBodyLen, elen, framingElen)
}
return m, err
}
func (req *MCRequest) populateFlexBodyInternal(buf []byte, totalBodyLen, elen, framingElen int) error {
var halfByteOffset bool
var err error
if framingElen > 0 {
var objs []FrameInfo
objs, err, halfByteOffset = parseFrameInfoObjects(buf, framingElen)
if err != nil {
return err
2019-02-05 22:22:51 +05:30
}
req.FramingExtras = objs
}
err = req.populateFlexBodyAfterFrames(buf, totalBodyLen, elen, framingElen, halfByteOffset)
if err != nil {
return err
}
return nil
}
func (req *MCRequest) populateFlexBodyAfterFrames(buf []byte, totalBodyLen, elen, framingElen int, halfByteOffset bool) error {
var idxCursor int = framingElen
if req.Opcode >= TAP_MUTATION && req.Opcode <= TAP_CHECKPOINT_END && len(buf[idxCursor:]) > 1 {
// In these commands there is "engine private"
// data at the end of the extras. The first 2
// bytes of extra data give its length.
if !halfByteOffset {
elen += int(binary.BigEndian.Uint16(buf[idxCursor:]))
} else {
// 0 1 2 3 4 .... 19 20 21 22 ... 32
// ^-----^ ^-------^ ^------------^
// offset data do not care
var buffer uint32 = binary.BigEndian.Uint32(buf[idxCursor:])
buffer &= 0xffff000
elen += int(buffer >> 12)
}
}
// Get the extras
if !halfByteOffset {
req.Extras = buf[idxCursor : idxCursor+elen]
} else {
preShift := buf[idxCursor : idxCursor+elen+1]
req.Extras = ShiftByteSliceLeft4Bits(preShift)
}
idxCursor += elen
// Get the Key
if !halfByteOffset {
req.Key = buf[idxCursor : idxCursor+req.Keylen]
} else {
preShift := buf[idxCursor : idxCursor+req.Keylen+1]
req.Key = ShiftByteSliceLeft4Bits(preShift)
}
idxCursor += req.Keylen
// get the length of extended metadata
extMetaLen := 0
if elen > 29 {
extMetaLen = int(binary.BigEndian.Uint16(req.Extras[28:30]))
}
idxCursor += extMetaLen
bodyLen := totalBodyLen - req.Keylen - elen - extMetaLen - framingElen
if bodyLen > MaxBodyLen {
return fmt.Errorf("%d is too big (max %d)",
bodyLen, MaxBodyLen)
}
if !halfByteOffset {
req.Body = buf[idxCursor : idxCursor+bodyLen]
idxCursor += bodyLen
} else {
preShift := buf[idxCursor : idxCursor+bodyLen+1]
req.Body = ShiftByteSliceLeft4Bits(preShift)
idxCursor += bodyLen
}
if extMetaLen > 0 {
if !halfByteOffset {
req.ExtMeta = buf[idxCursor:]
} else {
preShift := buf[idxCursor:]
req.ExtMeta = ShiftByteSliceLeft4Bits(preShift)
}
}
return nil
}
// Receive will fill this MCRequest with the data from a reader.
func (req *MCRequest) Receive(r io.Reader, hdrBytes []byte) (int, 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 {
fmt.Printf("Err %v\n", err)
return n, err
}
switch hdrBytes[0] {
case RES_MAGIC:
fallthrough
case REQ_MAGIC:
elen, totalBodyLen := req.receiveRegHeader(hdrBytes)
bodyRead, err := req.populateRegularBody(r, totalBodyLen, elen)
return n + bodyRead, err
case FLEX_MAGIC:
elen, totalBodyLen, framingElen := req.receiveFlexibleFramingHeader(hdrBytes)
bodyRead, err := req.populateFlexBody(r, totalBodyLen, elen, framingElen)
return n + bodyRead, err
default:
return n, fmt.Errorf("bad magic: 0x%02x", hdrBytes[0])
2019-02-05 22:22:51 +05:30
}
}