Merge pull request #8549 from matheusdtech/mc-store-perf-improv

Misson Control Store: Improve performance

Author: ellemouton

docs/release-notes/release-notes-0.18.1.md

@@ -74,6 +74,9 @@
 ## Breaking Changes
 ## Performance Improvements
 
+* Mission Control Store [improved performance during DB
+  flushing](https://github.com/lightningnetwork/lnd/pull/8549) stage.
+
 # Technical and Architectural Updates
 ## BOLT Spec Updates
 ## Testing
@@ -85,4 +88,5 @@
 
 * Andras Banki-Horvath
 * Bufo
+* Matheus Degiovani
 * Slyghtning

routing/missioncontrol_store.go

@@ -38,12 +38,15 @@ const (
 // Also changes to mission control parameters can be applied to historical data.
 // Finally, it enables importing raw data from an external source.
 type missionControlStore struct {
-	done    chan struct{}
-	wg      sync.WaitGroup
-	db      kvdb.Backend
-	queueMx sync.Mutex
+	done chan struct{}
+	wg   sync.WaitGroup
+	db   kvdb.Backend
+
+	// queueCond is signalled when items are put into the queue.
+	queueCond *sync.Cond
 
 	// queue stores all pending payment results not yet added to the store.
+	// Access is protected by the queueCond.L mutex.
 	queue *list.List
 
 	// keys holds the stored MC store item keys in the order of storage.
@@ -96,9 +99,12 @@ func newMissionControlStore(db kvdb.Backend, maxRecords int,
 		return nil, err
 	}
 
+	log.Infof("Loaded %d mission control entries", len(keysMap))
+
 	return &missionControlStore{
 		done:          make(chan struct{}),
 		db:            db,
+		queueCond:     sync.NewCond(&sync.Mutex{}),
 		queue:         list.New(),
 		keys:          keys,
 		keysMap:       keysMap,
@@ -109,8 +115,8 @@ func newMissionControlStore(db kvdb.Backend, maxRecords int,
 
 // clear removes all results from the db.
 func (b *missionControlStore) clear() error {
-	b.queueMx.Lock()
-	defer b.queueMx.Unlock()
+	b.queueCond.L.Lock()
+	defer b.queueCond.L.Unlock()
 
 	err := kvdb.Update(b.db, func(tx kvdb.RwTx) error {
 		if err := tx.DeleteTopLevelBucket(resultsKey); err != nil {
@@ -265,14 +271,19 @@ func deserializeResult(k, v []byte) (*paymentResult, error) {
 
 // AddResult adds a new result to the db.
 func (b *missionControlStore) AddResult(rp *paymentResult) {
-	b.queueMx.Lock()
-	defer b.queueMx.Unlock()
+	b.queueCond.L.Lock()
 	b.queue.PushBack(rp)
+	b.queueCond.L.Unlock()
+
+	b.queueCond.Signal()
 }
 
 // stop stops the store ticker goroutine.
 func (b *missionControlStore) stop() {
 	close(b.done)
+
+	b.queueCond.Signal()
+
 	b.wg.Wait()
 }
 
@@ -281,19 +292,51 @@ func (b *missionControlStore) run() {
 	b.wg.Add(1)
 
 	go func() {
-		ticker := time.NewTicker(b.flushInterval)
-		defer ticker.Stop()
 		defer b.wg.Done()
 
+		timer := time.NewTimer(b.flushInterval)
+
+		// Immediately stop the timer. It will be started once new
+		// items are added to the store. As the doc for time.Timer
+		// states, every call to Stop() done on a timer that is not
+		// known to have been fired needs to be checked and the timer's
+		// channel needs to be drained appropriately. This could happen
+		// if the flushInterval is very small (e.g. 1 nanosecond).
+		if !timer.Stop() {
+			<-timer.C
+		}
+
 		for {
+			// Wait for the queue to not be empty.
+			b.queueCond.L.Lock()
+			for b.queue.Front() == nil {
+				b.queueCond.Wait()
+
+				select {
+				case <-b.done:
+					b.queueCond.L.Unlock()
+
+					return
+				default:
+				}
+			}
+			b.queueCond.L.Unlock()
+
+			// Restart the timer.
+			timer.Reset(b.flushInterval)
+
 			select {
-			case <-ticker.C:
+			case <-timer.C:
 				if err := b.storeResults(); err != nil {
 					log.Errorf("Failed to update mission "+
 						"control store: %v", err)
 				}
 
 			case <-b.done:
+				// Release the timer's resources.
+				if !timer.Stop() {
+					<-timer.C
+				}
 				return
 			}
 		}
@@ -302,77 +345,136 @@ func (b *missionControlStore) run() {
 
 // storeResults stores all accumulated results.
 func (b *missionControlStore) storeResults() error {
-	b.queueMx.Lock()
+	// We copy a reference to the queue and clear the original queue to be
+	// able to release the lock.
+	b.queueCond.L.Lock()
 	l := b.queue
+
+	if l.Len() == 0 {
+		b.queueCond.L.Unlock()
+
+		return nil
+	}
 	b.queue = list.New()
-	b.queueMx.Unlock()
+	b.queueCond.L.Unlock()
 
 	var (
-		keys    *list.List
-		keysMap map[string]struct{}
+		newKeys    map[string]struct{}
+		delKeys    []string
+		storeCount int
+		pruneCount int
 	)
 
+	// Create a deduped list of new entries.
+	newKeys = make(map[string]struct{}, l.Len())
+	for e := l.Front(); e != nil; e = e.Next() {
+		pr, ok := e.Value.(*paymentResult)
+		if !ok {
+			return fmt.Errorf("wrong type %T (not *paymentResult)",
+				e.Value)
+		}
+		key := string(getResultKey(pr))
+		if _, ok := b.keysMap[key]; ok {
+			l.Remove(e)
+			continue
+		}
+		if _, ok := newKeys[key]; ok {
+			l.Remove(e)
+			continue
+		}
+		newKeys[key] = struct{}{}
+	}
+
+	// Create a list of entries to delete.
+	toDelete := b.keys.Len() + len(newKeys) - b.maxRecords
+	if b.maxRecords > 0 && toDelete > 0 {
+		delKeys = make([]string, 0, toDelete)
+
+		// Delete as many as needed from old keys.
+		for e := b.keys.Front(); len(delKeys) < toDelete && e != nil; {
+			key, ok := e.Value.(string)
+			if !ok {
+				return fmt.Errorf("wrong type %T (not string)",
+					e.Value)
+			}
+			delKeys = append(delKeys, key)
+			e = e.Next()
+		}
+
+		// If more deletions are needed, simply do not add from the
+		// list of new keys.
+		for e := l.Front(); len(delKeys) < toDelete && e != nil; {
+			toDelete--
+			pr, ok := e.Value.(*paymentResult)
+			if !ok {
+				return fmt.Errorf("wrong type %T (not "+
+					"*paymentResult )", e.Value)
+			}
+			key := string(getResultKey(pr))
+			delete(newKeys, key)
+			l.Remove(e)
+			e = l.Front()
+		}
+	}
+
 	err := kvdb.Update(b.db, func(tx kvdb.RwTx) error {
 		bucket := tx.ReadWriteBucket(resultsKey)
 
 		for e := l.Front(); e != nil; e = e.Next() {
-			pr := e.Value.(*paymentResult)
+			pr, ok := e.Value.(*paymentResult)
+			if !ok {
+				return fmt.Errorf("wrong type %T (not "+
+					"*paymentResult)", e.Value)
+			}
+
 			// Serialize result into key and value byte slices.
 			k, v, err := serializeResult(pr)
 			if err != nil {
 				return err
 			}
 
-			// The store is assumed to be idempotent. It could be
-			// that the same result is added twice and in that case
-			// we don't need to put the value again.
-			if _, ok := keysMap[string(k)]; ok {
-				continue
-			}
-
 			// Put into results bucket.
 			if err := bucket.Put(k, v); err != nil {
 				return err
 			}
 
-			keys.PushBack(string(k))
-			keysMap[string(k)] = struct{}{}
+			storeCount++
 		}
 
 		// Prune oldest entries.
-		for {
-			if b.maxRecords == 0 || keys.Len() <= b.maxRecords {
-				break
-			}
-
-			front := keys.Front()
-			key := front.Value.(string)
-
+		for _, key := range delKeys {
 			if err := bucket.Delete([]byte(key)); err != nil {
 				return err
 			}
-
-			keys.Remove(front)
-			delete(keysMap, key)
+			pruneCount++
 		}
 
 		return nil
 	}, func() {
-		keys = list.New()
-		keys.PushBackList(b.keys)
-
-		keysMap = make(map[string]struct{})
-		for k := range b.keysMap {
-			keysMap[k] = struct{}{}
-		}
+		storeCount, pruneCount = 0, 0
 	})
 
 	if err != nil {
 		return err
 	}
 
-	b.keys = keys
-	b.keysMap = keysMap
+	log.Debugf("Stored mission control results: %d added, %d deleted",
+		storeCount, pruneCount)
+
+	// DB Update was successful, update the in-memory cache.
+	for _, key := range delKeys {
+		delete(b.keysMap, key)
+		b.keys.Remove(b.keys.Front())
+	}
+	for e := l.Front(); e != nil; e = e.Next() {
+		pr, ok := e.Value.(*paymentResult)
+		if !ok {
+			return fmt.Errorf("wrong type %T (not *paymentResult)",
+				e.Value)
+		}
+		key := string(getResultKey(pr))
+		b.keys.PushBack(key)
+	}
 
 	return nil
 }