Use a double linked list to track active reads

Author: brian-brazil

block.go

@@ -134,17 +134,18 @@ type BlockReader interface {
 type Appendable interface {
 	// Appender returns a new Appender against an underlying store.
 	Appender(writeId, cleanupWriteIdsBelow uint64) Appender
-
-	// Busy returns whether there are any currently active appenders.
-	Busy() bool
 }
 
 type IsolationState struct {
 	// We will ignore all writes above the max, or that are incomplete.
 	maxWriteId       uint64
 	incompleteWrites map[uint64]struct{}
-	readId           uint64
+	lowWaterMark     uint64 // Lowest of incompleteWrites/maxWriteId.
 	db               *DB
+
+	// Doubly linked list of active reads.
+	next *IsolationState
+	prev *IsolationState
 }
 
 // BlockMeta provides meta information about a block.

db.go

@@ -123,11 +123,10 @@ type DB struct {
 	// Which writes are currently in progress.
 	writesOpen map[uint64]struct{}
 	// Mutex for accessing readLastId.
+	// If taking both writeMtx and readMtx, take writeMtx first.
 	readMtx sync.Mutex
-	// Each isolated read is given an internal id.
-	readLastId uint64
-	// All current in use isolationStates.
-	readsOpen map[uint64]*IsolationState
+	// All current in use isolationStates. This is a doubly-linked list.
+	readsOpen *IsolationState
 }
 
 type dbMetrics struct {
@@ -206,12 +205,16 @@ func Open(dir string, l log.Logger, r prometheus.Registerer, opts *Options) (db
 		return nil, err
 	}
 
+	head := &IsolationState{}
+	head.next = head
+	head.prev = head
+
 	db = &DB{
 		dir:                dir,
 		logger:             l,
 		opts:               opts,
 		writesOpen:         map[uint64]struct{}{},
-		readsOpen:          map[uint64]*IsolationState{},
+		readsOpen:          head,
 		compactc:           make(chan struct{}, 1),
 		donec:              make(chan struct{}),
 		stopc:              make(chan struct{}),
@@ -748,25 +751,15 @@ func rangeForTimestamp(t int64, width int64) (mint, maxt int64) {
 // readLowWatermark returns the writeId below which
 // we no longer need to track which writes were from
 // which writeId.
-// TODO: Optimise this, needs to be O(1).
 func (db *DB) readLowWatermark() uint64 {
-	db.writeMtx.Lock()
-	id := db.writeLastId
-	db.writeMtx.Unlock()
-
+	db.writeMtx.Lock() // Take writeMtx first.
+	defer db.writeMtx.Unlock()
 	db.readMtx.Lock()
-	for _, isolation := range db.readsOpen {
-		if isolation.maxWriteId < id {
-			id = isolation.maxWriteId
-		}
-		for i := range isolation.incompleteWrites {
-			if i < id {
-				id = i
-			}
-		}
+	defer db.readMtx.Unlock()
+	if db.readsOpen.prev == db.readsOpen {
+		return db.writeLastId
 	}
-	db.readMtx.Unlock()
-	return id
+	return db.readsOpen.prev.lowWaterMark
 }
 
 // Delete implements deletion of metrics. It only has atomicity guarantees on a per-block basis.

querier.go

@@ -61,28 +61,34 @@ type querier struct {
 // IsolationState returns an objet used to control isolation
 // between a query and writes. Must be closed when complete.
 func (s *DB) IsolationState() *IsolationState {
-	s.writeMtx.Lock()
+	s.writeMtx.Lock() // Take write mutex before read mutex.
+	defer s.writeMtx.Unlock()
 	isolation := &IsolationState{
 		maxWriteId:       s.writeLastId,
+		lowWaterMark:     s.writeLastId,
 		incompleteWrites: make(map[uint64]struct{}, len(s.writesOpen)),
 		db:               s,
 	}
 	for k, _ := range s.writesOpen {
 		isolation.incompleteWrites[k] = struct{}{}
+		if k < isolation.lowWaterMark {
+			isolation.lowWaterMark = k
+		}
 	}
-	s.writeMtx.Unlock()
 
 	s.readMtx.Lock()
-	isolation.readId = s.readLastId
-	s.readLastId++
-	s.readsOpen[isolation.readId] = isolation
-	s.readMtx.Unlock()
+	defer s.readMtx.Unlock()
+	isolation.prev = s.readsOpen
+	isolation.next = s.readsOpen.next
+	s.readsOpen.next.prev = isolation
+	s.readsOpen.next = isolation
 	return isolation
 }
 
 func (i *IsolationState) Close() {
 	i.db.readMtx.Lock()
-	delete(i.db.readsOpen, i.readId)
+	i.next.prev = i.prev
+	i.prev.next = i.next
 	i.db.readMtx.Unlock()
 }