msgpack

Documentation

Index

• func DecodeLogEntryFast(b []byte) (schema.LogEntry, error)
• func DecodeLogMetadataFast(b []byte) (schema.LogMetadata, error)
• func EncodeLogEntryFast(b []byte, entry schema.LogEntry) ([]byte, error)
• func EncodeLogMetadataFast(b []byte, entry schema.LogMetadata) ([]byte, error)
• type ByteDecoderStream
  • func NewByteDecoderStream(b []byte) ByteDecoderStream
• type ByteStream
• type DecodeLogEntryRemainingToken
• type Decoder
  • func NewDecoder(opts DecodingOptions) *Decoder
  • func (dec *Decoder) DecodeIndexEntry(bytesPool pool.BytesPool) (schema.IndexEntry, error)
  • func (dec *Decoder) DecodeIndexInfo() (schema.IndexInfo, error)
  • func (dec *Decoder) DecodeIndexSummary() (schema.IndexSummary, IndexSummaryToken, error)
  • func (dec *Decoder) DecodeLogEntry() (schema.LogEntry, error)
  • func (dec *Decoder) DecodeLogEntryRemaining(token DecodeLogEntryRemainingToken, index uint64) (schema.LogEntry, error)
  • func (dec *Decoder) DecodeLogEntryUniqueIndex() (DecodeLogEntryRemainingToken, uint64, error)
  • func (dec *Decoder) DecodeLogInfo() (schema.LogInfo, error)
  • func (dec *Decoder) DecodeLogMetadata() (schema.LogMetadata, error)
  • func (dec *Decoder) Reset(stream DecoderStream)
• type DecoderStream
• type DecodingOptions
  • func NewDecodingOptions() DecodingOptions
• type Encoder
  • func NewEncoder() *Encoder
  • func (enc *Encoder) Bytes() []byte
  • func (enc *Encoder) EncodeIndexEntry(entry schema.IndexEntry) error
  • func (enc *Encoder) EncodeIndexInfo(info schema.IndexInfo) error
  • func (enc *Encoder) EncodeIndexSummary(summary schema.IndexSummary) error
  • func (enc *Encoder) EncodeLogEntry(entry schema.LogEntry) error
  • func (enc *Encoder) EncodeLogInfo(info schema.LogInfo) error
  • func (enc *Encoder) EncodeLogMetadata(entry schema.LogMetadata) error
  • func (enc *Encoder) Reset()
• type IndexSummaryToken
  • func NewIndexSummaryToken(idStartOffset, idLength uint32) IndexSummaryToken
  • func (m IndexSummaryToken) ID(buf []byte) []byte
  • func (m IndexSummaryToken) IndexOffset(buf []byte, stream ByteDecoderStream, msgpackDecoder *msgpack.Decoder) (int64, error)

Functions

func DecodeLogEntryFast

func DecodeLogEntryFast(b []byte) (schema.LogEntry, error)

DecodeLogEntryFast decodes a commit log entry with no buffering and using optimized helper functions that bypass the msgpack decoding library by manually inlining the equivalent code.

The reason we had to bypass the msgpack decoding library is that during perf testing we found that this function was spending most of its time setting up stack frames for function calls. While the overhead of a function call in Golang is small, when every helper function does nothing more than read a few bytes from an in-memory array the function call overhead begins to dominate, especially when each call to this function results in dozens of such helper function calls.

Manually inlining the msgpack decoding results in a lot of code duplication for this one path, but we pay the price because this codepath is one of the primary bottlenecks influencing how fast we can bootstrap M3DB from the commitlog. As a result, almost any performance gains that can be had in this function are worth it.

Before modifying this function, please run the BenchmarkLogEntryDecodeFast benchmark.

Also note that there are extensive prop tests for this function in the encoder_decoder_prop_test.go file which verify its correctness, as well as its resilience to arbitrary data corruption and truncation.

func DecodeLogMetadataFast

func DecodeLogMetadataFast(b []byte) (schema.LogMetadata, error)

DecodeLogMetadataFast is the same as DecodeLogEntryFast except for the metadata entries instead of the data entries.

func EncodeLogEntryFast

func EncodeLogEntryFast(b []byte, entry schema.LogEntry) ([]byte, error)

EncodeLogEntryFast encodes a commit log entry with no buffering and using optimized helper functions that bypass the msgpack encoding library by manually inlining the equivalent code.

The reason we had to bypass the msgpack encoding library is that during perf testing we found that this function was spending most of its time setting up stack frames for function calls. While the overhead of a function call in Golang is small, when every helper function does nothing more than write a few bytes to an in-memory array the function call overhead begins to dominate, especially when each call to this function results in dozens of such helper function calls.

Manually inlining the msgpack encoding results in a lot of code duplication for this one path, but we pay the price because this is the most frequently called function in M3DB and it indirectly applies back-pressure on every other part of the system via the commitlog queue. As a result, almost any performance gains that can be had in this function are worth it.

Before modifying this function, please run the BenchmarkLogEntryEncoderFast benchmark as a small degradation in this functions performance can have a substantial impact on M3DB.

Also note that there are extensive prop tests for this function in the encoder_decoder_prop_test.go file which verify its correctness.

func EncodeLogMetadataFast

func EncodeLogMetadataFast(b []byte, entry schema.LogMetadata) ([]byte, error)

EncodeLogMetadataFast is the same as EncodeLogEntryFast except for the metadata entries instead of the data entries.

Types

type ByteDecoderStream

type ByteDecoderStream interface {
	DecoderStream
	ByteStream
}

ByteDecoderStream is an additional interface that some decoder streams can implement if they are backed by a byte slice.

func NewByteDecoderStream

func NewByteDecoderStream(b []byte) ByteDecoderStream

NewByteDecoderStream creates a new decoder stream from a bytes ref.

type ByteStream

type ByteStream interface {
	// Bytes returns the ref to the bytes provided when Reset(...) is
	// called. To get the current position into the byte slice use:
	// len(s.Bytes()) - s.Remaining()
	Bytes() []byte

	// Remaining returns the remaining bytes in the stream.
	Remaining() int64

	// Reset resets the decoder stream for decoding a new byte slice.
	Reset(b []byte)

	// Skip progresses the reader by a certain amount of bytes, useful
	// when taking a ref to some of the bytes and progressing the reader
	// itself.
	Skip(length int64) error

	// Offset returns the current offset in the byte stream.
	Offset() int
}

ByteStream is the interface that contains the additional methods which can be implemented by streams that are backed by byte slices.

type DecodeLogEntryRemainingToken

type DecodeLogEntryRemainingToken struct {
	// contains filtered or unexported fields
}

DecodeLogEntryRemainingToken contains all the information that DecodeLogEntryRemaining requires to continue decoding a log entry after a call to DecodeLogEntryUniqueIndex.

type Decoder

type Decoder struct {
	// contains filtered or unexported fields
}

Decoder decodes persisted msgpack-encoded data

func NewDecoder

func NewDecoder(opts DecodingOptions) *Decoder

NewDecoder creates a new decoder

func (*Decoder) DecodeIndexEntry

func (dec *Decoder) DecodeIndexEntry(bytesPool pool.BytesPool) (schema.IndexEntry, error)

DecodeIndexEntry decodes index entry

func (*Decoder) DecodeIndexInfo

func (dec *Decoder) DecodeIndexInfo() (schema.IndexInfo, error)

DecodeIndexInfo decodes the index info

func (*Decoder) DecodeIndexSummary

func (dec *Decoder) DecodeIndexSummary() (
	schema.IndexSummary, IndexSummaryToken, error)

DecodeIndexSummary decodes index summary

func (*Decoder) DecodeLogEntry

func (dec *Decoder) DecodeLogEntry() (schema.LogEntry, error)

DecodeLogEntry decodes commit log entry

func (*Decoder) DecodeLogEntryRemaining

func (dec *Decoder) DecodeLogEntryRemaining(token DecodeLogEntryRemainingToken, index uint64) (schema.LogEntry, error)

DecodeLogEntryRemaining can only be called after DecodeLogEntryUniqueIndex, and it returns a complete schema.LogEntry.

func (*Decoder) DecodeLogEntryUniqueIndex

func (dec *Decoder) DecodeLogEntryUniqueIndex() (DecodeLogEntryRemainingToken, uint64, error)

DecodeLogEntryUniqueIndex decodes a log entry as much as is required to return the series unique index. Call DecodeLogEntryRemaining afterwards to decode the remaining fields.

func (*Decoder) DecodeLogInfo

func (dec *Decoder) DecodeLogInfo() (schema.LogInfo, error)

DecodeLogInfo decodes commit log info

func (*Decoder) DecodeLogMetadata

func (dec *Decoder) DecodeLogMetadata() (schema.LogMetadata, error)

DecodeLogMetadata decodes commit log metadata

func (*Decoder) Reset

func (dec *Decoder) Reset(stream DecoderStream)

Reset resets the data stream to decode from

type DecoderStream

type DecoderStream interface {
	io.Reader

	// ReadByte reads the next byte.
	ReadByte() (byte, error)

	// UnreadByte unreads the last read byte or returns error if none read
	// yet. Only a single byte can be unread at a time, a consecutive call
	// to UnreadByte will result in an error.
	UnreadByte() error
}

DecoderStream is a data stream that is read by the decoder, it takes both a reader and the underlying backing bytes. This is constructed this way since the decoder needs access to the backing bytes when taking refs directly for decoding byte slices without allocating bytes itself but also needs to progress the reader (for instance when a reader is a ReaderWithDigest that is calculating a digest as its being read).

type DecodingOptions

type DecodingOptions interface {
	// SetAllocDecodedBytes sets whether we allocate new space when decoding
	// a byte slice
	SetAllocDecodedBytes(value bool) DecodingOptions

	// AllocDecodedBytes determines whether we allocate new space when decoding
	// a byte slice
	AllocDecodedBytes() bool
}

DecodingOptions provide a set of options for decoding data

func NewDecodingOptions

func NewDecodingOptions() DecodingOptions

NewDecodingOptions creates a new set of decoding options

type Encoder

type Encoder struct {
	// contains filtered or unexported fields
}

Encoder encodes data in msgpack format for persistence.

func NewEncoder

func NewEncoder() *Encoder

NewEncoder creates a new encoder.

func (*Encoder) Bytes

func (enc *Encoder) Bytes() []byte

Bytes returns the encoded bytes.

func (*Encoder) EncodeIndexEntry

func (enc *Encoder) EncodeIndexEntry(entry schema.IndexEntry) error

EncodeIndexEntry encodes index entry.

func (*Encoder) EncodeIndexInfo

func (enc *Encoder) EncodeIndexInfo(info schema.IndexInfo) error

EncodeIndexInfo encodes index info.

func (*Encoder) EncodeIndexSummary

func (enc *Encoder) EncodeIndexSummary(summary schema.IndexSummary) error

EncodeIndexSummary encodes index summary.

func (*Encoder) EncodeLogEntry

func (enc *Encoder) EncodeLogEntry(entry schema.LogEntry) error

EncodeLogEntry encodes commit log entry.

func (*Encoder) EncodeLogInfo

func (enc *Encoder) EncodeLogInfo(info schema.LogInfo) error

EncodeLogInfo encodes commit log info.

func (*Encoder) EncodeLogMetadata

func (enc *Encoder) EncodeLogMetadata(entry schema.LogMetadata) error

EncodeLogMetadata encodes commit log metadata

func (*Encoder) Reset

func (enc *Encoder) Reset()

Reset resets the buffer.

type IndexSummaryToken

type IndexSummaryToken struct {
	// contains filtered or unexported fields
}

IndexSummaryToken can be used, along with the summaries file buffer, to quickly retrieve the ID or index file offset of an index summary entry. It's structured such that the ID and Index file offset can be retrieved quickly, while only requiring 64-bits of memory per entry.

func NewIndexSummaryToken

func NewIndexSummaryToken(idStartOffset, idLength uint32) IndexSummaryToken

NewIndexSummaryToken returns a new IndexSummaryToken

func (IndexSummaryToken) ID

func (m IndexSummaryToken) ID(buf []byte) []byte

ID returns the ID that the metadata corresponds to

func (IndexSummaryToken) IndexOffset

func (m IndexSummaryToken) IndexOffset(
	buf []byte, stream ByteDecoderStream, msgpackDecoder *msgpack.Decoder) (int64, error)

IndexOffset returns the offset in the index file for the series that the metadata corresponds to. The buf, stream, and decoder arguments are passed in so that the IndexSummaryToken struct can be kept small, as well as so that the caller can have control over re-use and allocations.