Use a *mapTombstoneReader instead of map

We need to recalculate the sorted ref list everytime we make a
Tombstones() call. This avoids that.

Signed-off-by: Goutham Veeramachaneni <cs14btech11014@iith.ac.in>

block.go

@@ -163,8 +163,7 @@ type persistedBlock struct {
 	indexr *indexReader
 
 	// For tombstones.
-	stones     []uint32
-	tombstones map[uint32][]trange
+	tombstones *mapTombstoneReader
 }
 
 func newPersistedBlock(dir string) (*persistedBlock, error) {
@@ -186,6 +185,7 @@ func newPersistedBlock(dir string) (*persistedBlock, error) {
 	if err != nil {
 		return nil, err
 	}
+
 	ts := make(map[uint32][]trange)
 	for tr.Next() {
 		s := tr.At()
@@ -198,7 +198,7 @@ func newPersistedBlock(dir string) (*persistedBlock, error) {
 		chunkr: cr,
 		indexr: ir,
 
-		tombstones: ts,
+		tombstones: newMapTombstoneReader(ts),
 	}
 	return pb, nil
 }
@@ -229,7 +229,7 @@ func (pb *persistedBlock) Dir() string         { return pb.dir }
 func (pb *persistedBlock) Index() IndexReader  { return pb.indexr }
 func (pb *persistedBlock) Chunks() ChunkReader { return pb.chunkr }
 func (pb *persistedBlock) Tombstones() TombstoneReader {
-	return newMapTombstoneReader(pb.tombstones)
+	return pb.tombstones.Copy()
 }
 func (pb *persistedBlock) Meta() BlockMeta { return pb.meta }
 
@@ -270,7 +270,7 @@ Outer:
 	}
 
 	// Merge the current and new tombstones.
-	tr := newMapTombstoneReader(pb.tombstones)
+	tr := pb.tombstones.Copy()
 	str := newSimpleTombstoneReader(vPostings, []trange{{mint, maxt}})
 	tombreader := newMergedTombstoneReader(tr, str)
 

head.go

@@ -66,7 +66,7 @@ type HeadBlock struct {
 	values   map[string]stringset // label names to possible values
 	postings *memPostings         // postings lists for terms
 
-	tombstones map[uint32][]trange
+	tombstones *mapTombstoneReader
 
 	meta BlockMeta
 }
@@ -120,7 +120,7 @@ func OpenHeadBlock(dir string, l log.Logger, wal WAL) (*HeadBlock, error) {
 		values:     map[string]stringset{},
 		postings:   &memPostings{m: make(map[term][]uint32)},
 		meta:       *meta,
-		tombstones: make(map[uint32][]trange),
+		tombstones: emptyTombstoneReader,
 	}
 	return h, h.init()
 }
@@ -158,7 +158,8 @@ func (h *HeadBlock) init() error {
 
 	for tr.Next() {
 		s := tr.At()
-		h.tombstones[s.ref] = s.ranges
+		h.tombstones.refs = append(h.tombstones.refs, s.ref)
+		h.tombstones.stones[s.ref] = s.ranges
 	}
 	return errors.Wrap(err, "tombstones reader iteration")
 }
@@ -229,23 +230,19 @@ func (h *HeadBlock) Chunks() ChunkReader { return &headChunkReader{h} }
 
 // Tombstones implements headBlock.
 func (h *HeadBlock) Tombstones() TombstoneReader {
-	return newMapTombstoneReader(h.tombstones)
+	return h.tombstones.Copy()
 }
 
 // Delete implements headBlock.
 func (h *HeadBlock) Delete(mint int64, maxt int64, ms ...labels.Matcher) error {
-	h.mtx.RLock()
+	h.mtx.Lock() // We are modifying the tombstones here.
+	defer h.mtx.Unlock()
 
 	ir := h.Index()
 
 	pr := newPostingsReader(ir)
 	p, absent := pr.Select(ms...)
 
-	h.mtx.RUnlock()
-
-	h.mtx.Lock() // We are modifying the tombstones here.
-	defer h.mtx.Unlock()
-
 Outer:
 	for p.Next() {
 		ref := p.At()
@@ -256,14 +253,14 @@ Outer:
 			}
 		}
 
-		h.tombstones[ref] = addNewInterval(h.tombstones[ref], trange{mint, maxt})
+		h.tombstones.stones[ref] = addNewInterval(h.tombstones.stones[ref], trange{mint, maxt})
 	}
 
 	if p.Err() != nil {
 		return p.Err()
 	}
 
-	return writeTombstoneFile(h.dir, newMapTombstoneReader(h.tombstones))
+	return writeTombstoneFile(h.dir, newMapTombstoneReader(h.tombstones.stones))
 }
 
 // Querier implements Queryable and headBlock.

tombstones.go

@@ -218,7 +218,6 @@ type mapTombstoneReader struct {
 	stones map[uint32][]trange
 }
 
-// TODO(gouthamve): Take pre-sorted refs.
 func newMapTombstoneReader(ts map[uint32][]trange) *mapTombstoneReader {
 	refs := make([]uint32, 0, len(ts))
 	for k := range ts {