Avoid chunk allocations and refactor compactions

block.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/oklog/ulid"
 	"github.com/pkg/errors"
+	"github.com/prometheus/tsdb/chunks"
 	"github.com/prometheus/tsdb/labels"
 )
 
@@ -112,7 +113,7 @@ type BlockStats struct {
 type BlockMetaCompaction struct {
 	// Maximum number of compaction cycles any source block has
 	// gone through.
-	Generation int `json:"generation"`
+	Level int `json:"level"`
 	// ULIDs of all source head blocks that went into the block.
 	Sources []ulid.ULID `json:"sources,omitempty"`
 }
@@ -181,13 +182,13 @@ type persistedBlock struct {
 	tombstones tombstoneReader
 }
 
-func newPersistedBlock(dir string) (*persistedBlock, error) {
+func newPersistedBlock(dir string, pool chunks.Pool) (*persistedBlock, error) {
 	meta, err := readMetaFile(dir)
 	if err != nil {
 		return nil, err
 	}
 
-	cr, err := newChunkReader(chunkDir(dir))
+	cr, err := newChunkReader(chunkDir(dir), pool)
 	if err != nil {
 		return nil, err
 	}
@@ -252,7 +253,7 @@ func (pb *persistedBlock) Delete(mint, maxt int64, ms ...labels.Matcher) error {
 	stones := map[uint32]intervals{}
 
 	var lset labels.Labels
-	var chks []*ChunkMeta
+	var chks []ChunkMeta
 
 Outer:
 	for p.Next() {

chunks.go

@@ -100,7 +100,7 @@ type ChunkWriter interface {
 	// must be populated.
 	// After returning successfully, the Ref fields in the ChunkMetas
 	// are set and can be used to retrieve the chunks from the written data.
-	WriteChunks(chunks ...*ChunkMeta) error
+	WriteChunks(chunks ...ChunkMeta) error
 
 	// Close writes any required finalization and closes the resources
 	// associated with the underlying writer.
@@ -222,7 +222,7 @@ func (w *chunkWriter) write(b []byte) error {
 	return err
 }
 
-func (w *chunkWriter) WriteChunks(chks ...*ChunkMeta) error {
+func (w *chunkWriter) WriteChunks(chks ...ChunkMeta) error {
 	// Calculate maximum space we need and cut a new segment in case
 	// we don't fit into the current one.
 	maxLen := int64(binary.MaxVarintLen32) // The number of chunks.
@@ -238,23 +238,22 @@ func (w *chunkWriter) WriteChunks(chks ...*ChunkMeta) error {
 		}
 	}
 
-	b := make([]byte, binary.MaxVarintLen32)
-	n := binary.PutUvarint(b, uint64(len(chks)))
-
-	if err := w.write(b[:n]); err != nil {
-		return err
-	}
-	seq := uint64(w.seq()) << 32
+	var (
+		b   = [binary.MaxVarintLen32]byte{}
+		seq = uint64(w.seq()) << 32
+	)
+	for i := range chks {
+		chk := &chks[i]
 
-	for _, chk := range chks {
 		chk.Ref = seq | uint64(w.n)
 
-		n = binary.PutUvarint(b, uint64(len(chk.Chunk.Bytes())))
+		n := binary.PutUvarint(b[:], uint64(len(chk.Chunk.Bytes())))
 
 		if err := w.write(b[:n]); err != nil {
 			return err
 		}
-		if err := w.write([]byte{byte(chk.Chunk.Encoding())}); err != nil {
+		b[0] = byte(chk.Chunk.Encoding())
+		if err := w.write(b[:1]); err != nil {
 			return err
 		}
 		if err := w.write(chk.Chunk.Bytes()); err != nil {
@@ -265,7 +264,7 @@ func (w *chunkWriter) WriteChunks(chks ...*ChunkMeta) error {
 		if err := chk.writeHash(w.crc32); err != nil {
 			return err
 		}
-		if err := w.write(w.crc32.Sum(nil)); err != nil {
+		if err := w.write(w.crc32.Sum(b[:0])); err != nil {
 			return err
 		}
 	}
@@ -298,15 +297,20 @@ type chunkReader struct {
 
 	// Closers for resources behind the byte slices.
 	cs []io.Closer
+
+	pool chunks.Pool
 }
 
 // newChunkReader returns a new chunkReader based on mmaped files found in dir.
-func newChunkReader(dir string) (*chunkReader, error) {
+func newChunkReader(dir string, pool chunks.Pool) (*chunkReader, error) {
 	files, err := sequenceFiles(dir, "")
 	if err != nil {
 		return nil, err
 	}
-	var cr chunkReader
+	if pool == nil {
+		pool = chunks.NewPool()
+	}
+	cr := chunkReader{pool: pool}
 
 	for _, fn := range files {
 		f, err := openMmapFile(fn)
@@ -353,11 +357,6 @@ func (s *chunkReader) Chunk(ref uint64) (chunks.Chunk, error) {
 		return nil, fmt.Errorf("reading chunk length failed")
 	}
 	b = b[n:]
-	enc := chunks.Encoding(b[0])
 
-	c, err := chunks.FromData(enc, b[1:1+l])
-	if err != nil {
-		return nil, err
-	}
-	return c, nil
+	return s.pool.Get(chunks.Encoding(b[0]), b[1:1+l])
 }

chunks/chunk.go

@@ -13,7 +13,12 @@
 
 package chunks
 
-import "fmt"
+import (
+	"fmt"
+	"sync"
+
+	"github.com/pkg/errors"
+)
 
 // Encoding is the identifier for a chunk encoding.
 type Encoding uint8
@@ -63,3 +68,53 @@ type Iterator interface {
 	Err() error
 	Next() bool
 }
+
+type Pool interface {
+	Put(Chunk) error
+	Get(e Encoding, b []byte) (Chunk, error)
+}
+
+// Pool is a memory pool of chunk objects.
+type pool struct {
+	xor sync.Pool
+}
+
+func NewPool() Pool {
+	return &pool{
+		xor: sync.Pool{
+			New: func() interface{} {
+				return &XORChunk{b: &bstream{}}
+			},
+		},
+	}
+}
+
+func (p *pool) Get(e Encoding, b []byte) (Chunk, error) {
+	switch e {
+	case EncXOR:
+		c := p.xor.Get().(*XORChunk)
+		c.b.stream = b
+		c.b.count = 0
+		return c, nil
+	}
+	return nil, errors.Errorf("invalid encoding %q", e)
+}
+
+func (p *pool) Put(c Chunk) error {
+	switch c.Encoding() {
+	case EncXOR:
+		xc, ok := c.(*XORChunk)
+		// This may happen often with wrapped chunks. Nothing we can really do about
+		// it but returning an error would cause a lot of allocations again. Thus,
+		// we just skip it.
+		if !ok {
+			return nil
+		}
+		xc.b.stream = nil
+		xc.b.count = 0
+		p.xor.Put(c)
+	default:
+		return errors.Errorf("invalid encoding %q", c.Encoding())
+	}
+	return nil
+}