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

Misson Control Store: Improve performance

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{}
+	done chan struct{}
-	wg      sync.WaitGroup
+	wg   sync.WaitGroup
-	db      kvdb.Backend
+	db   kvdb.Backend
-	queueMx sync.Mutex
+
+	// 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()
+	b.queueCond.L.Lock()
-	defer b.queueMx.Unlock()
+	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()
+	b.queueCond.L.Lock()
-	defer b.queueMx.Unlock()
 	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
+		newKeys    map[string]struct{}
-		keysMap 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))
+			storeCount++
-			keysMap[string(k)] = struct{}{}
 		}
 
 		// Prune oldest entries.
-		for {
+		for _, key := range delKeys {
-			if b.maxRecords == 0 || keys.Len() <= b.maxRecords {
-				break
-			}
-
-			front := keys.Front()
-			key := front.Value.(string)
-
 			if err := bucket.Delete([]byte(key)); err != nil {
 				return err
 			}
-
+			pruneCount++
-			keys.Remove(front)
-			delete(keysMap, key)
 		}
 
 		return nil
 	}, func() {
-		keys = list.New()
+		storeCount, pruneCount = 0, 0
-		keys.PushBackList(b.keys)
-
-		keysMap = make(map[string]struct{})
-		for k := range b.keysMap {
-			keysMap[k] = struct{}{}
-		}
 	})
 
 	if err != nil {
 		return err
 	}
 
-	b.keys = keys
+	log.Debugf("Stored mission control results: %d added, %d deleted",
-	b.keysMap = keysMap
+		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
 }

routing/missioncontrol_store_test.go

@@ -1,13 +1,14 @@
 package routing
 
 import (
+	"fmt"
 	"os"
-	"reflect"
 	"testing"
 	"time"
 
-	"github.com/davecgh/go-spew/spew"
+	"github.com/btcsuite/btcwallet/walletdb"
 	"github.com/lightningnetwork/lnd/kvdb"
+	"github.com/lightningnetwork/lnd/lntest/wait"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/routing/route"
 	"github.com/stretchr/testify/require"
@@ -15,18 +16,33 @@ import (
 
 const testMaxRecords = 2
 
-// TestMissionControlStore tests the recording of payment failure events
+var (
-// in mission control. It tests encoding and decoding of differing lnwire
+	// mcStoreTestRoute is a test route for the mission control store tests.
-// failures (FailIncorrectDetails and FailMppTimeout), pruning of results
+	mcStoreTestRoute = route.Route{
-// and idempotent writes.
+		SourcePubKey: route.Vertex{1},
-func TestMissionControlStore(t *testing.T) {
+		Hops: []*route.Hop{
+			{
+				PubKeyBytes:   route.Vertex{2},
+				LegacyPayload: true,
+			},
+		},
+	}
+)
+
+// mcStoreTestHarness is the harness for a MissonControlStore test.
+type mcStoreTestHarness struct {
+	db    walletdb.DB
+	store *missionControlStore
+}
+
+// newMCStoreTestHarness initializes a test mission control store.
+func newMCStoreTestHarness(t testing.TB, maxRecords int,
+	flushInterval time.Duration) mcStoreTestHarness {
 	// Set time zone explicitly to keep test deterministic.
 	time.Local = time.UTC
 
 	file, err := os.CreateTemp("", "*.db")
-	if err != nil {
+	require.NoError(t, err)
-		t.Fatal(err)
-	}
 
 	dbPath := file.Name()
 	t.Cleanup(func() {
@@ -37,40 +53,33 @@ func TestMissionControlStore(t *testing.T) {
 	db, err := kvdb.Create(
 		kvdb.BoltBackendName, dbPath, true, kvdb.DefaultDBTimeout,
 	)
-	if err != nil {
+	require.NoError(t, err)
-		t.Fatal(err)
-	}
 	t.Cleanup(func() {
 		require.NoError(t, db.Close())
 	})
 
-	store, err := newMissionControlStore(db, testMaxRecords, time.Second)
+	store, err := newMissionControlStore(db, maxRecords, flushInterval)
-	if err != nil {
+	require.NoError(t, err)
-		t.Fatal(err)
+
-	}
+	return mcStoreTestHarness{db: db, store: store}
+}
+
+// TestMissionControlStore tests the recording of payment failure events
+// in mission control. It tests encoding and decoding of differing lnwire
+// failures (FailIncorrectDetails and FailMppTimeout), pruning of results
+// and idempotent writes.
+func TestMissionControlStore(t *testing.T) {
+	h := newMCStoreTestHarness(t, testMaxRecords, time.Second)
+	db, store := h.db, h.store
 
 	results, err := store.fetchAll()
-	if err != nil {
+	require.NoError(t, err)
-		t.Fatal(err)
+	require.Len(t, results, 0)
-	}
-	if len(results) != 0 {
-		t.Fatal("expected no results")
-	}
-
-	testRoute := route.Route{
-		SourcePubKey: route.Vertex{1},
-		Hops: []*route.Hop{
-			{
-				PubKeyBytes:   route.Vertex{2},
-				LegacyPayload: true,
-			},
-		},
-	}
 
 	failureSourceIdx := 1
 
 	result1 := paymentResult{
-		route:            &testRoute,
+		route:            &mcStoreTestRoute,
 		failure:          lnwire.NewFailIncorrectDetails(100, 1000),
 		failureSourceIdx: &failureSourceIdx,
 		id:               99,
@@ -94,30 +103,16 @@ func TestMissionControlStore(t *testing.T) {
 	require.NoError(t, store.storeResults())
 
 	results, err = store.fetchAll()
-	if err != nil {
+	require.NoError(t, err)
-		t.Fatal(err)
+	require.Len(t, results, 2)
-	}
-	require.Equal(t, 2, len(results))
-
-	if len(results) != 2 {
-		t.Fatal("expected two results")
-	}
 
 	// Check that results are stored in chronological order.
-	if !reflect.DeepEqual(&result1, results[0]) {
+	require.Equal(t, &result1, results[0])
-		t.Fatalf("the results differ: %v vs %v", spew.Sdump(&result1),
+	require.Equal(t, &result2, results[1])
-			spew.Sdump(results[0]))
-	}
-	if !reflect.DeepEqual(&result2, results[1]) {
-		t.Fatalf("the results differ: %v vs %v", spew.Sdump(&result2),
-			spew.Sdump(results[1]))
-	}
 
 	// Recreate store to test pruning.
 	store, err = newMissionControlStore(db, testMaxRecords, time.Second)
-	if err != nil {
+	require.NoError(t, err)
-		t.Fatal(err)
-	}
 
 	// Add a newer result which failed due to mpp timeout.
 	result3 := result1
@@ -131,20 +126,170 @@ func TestMissionControlStore(t *testing.T) {
 
 	// Check that results are pruned.
 	results, err = store.fetchAll()
-	if err != nil {
+	require.NoError(t, err)
-		t.Fatal(err)
+	require.Len(t, results, 2)
-	}
+
-	require.Equal(t, 2, len(results))
+	require.Equal(t, &result2, results[0])
-	if len(results) != 2 {
+	require.Equal(t, &result3, results[1])
-		t.Fatal("expected two results")
+}
+
+// TestMissionControlStoreFlushing asserts the periodic flushing of the store
+// works correctly.
+func TestMissionControlStoreFlushing(t *testing.T) {
+	const flushInterval = 500 * time.Millisecond
+
+	h := newMCStoreTestHarness(t, testMaxRecords, flushInterval)
+	db, store := h.db, h.store
+
+	var (
+		failureSourceIdx = 1
+		failureDetails   = lnwire.NewFailIncorrectDetails(100, 1000)
+		lastID           uint64
+	)
+	nextResult := func() *paymentResult {
+		lastID += 1
+		return &paymentResult{
+			route:            &mcStoreTestRoute,
+			failure:          failureDetails,
+			failureSourceIdx: &failureSourceIdx,
+			id:               lastID,
+			timeReply:        testTime,
+			timeFwd:          testTime.Add(-time.Minute),
+		}
 	}
 
-	if !reflect.DeepEqual(&result2, results[0]) {
+	// Helper to assert the number of results is correct.
-		t.Fatalf("the results differ: %v vs %v", spew.Sdump(&result2),
+	assertResults := func(wantCount int) {
-			spew.Sdump(results[0]))
+		t.Helper()
+		err := wait.NoError(func() error {
+			results, err := store.fetchAll()
+			if err != nil {
+				return err
+			}
+			if wantCount != len(results) {
+				return fmt.Errorf("wrong nb of results: got "+
+					"%d, want %d", len(results), wantCount)
+			}
+			if len(results) == 0 {
+				return nil
+			}
+			gotLastID := results[len(results)-1].id
+			if len(results) > 0 && gotLastID != lastID {
+				return fmt.Errorf("wrong id for last item: "+
+					"got %d, want %d", gotLastID, lastID)
+			}
+
+			return nil
+		}, flushInterval*5)
+		require.NoError(t, err)
 	}
-	if !reflect.DeepEqual(&result3, results[1]) {
+
-		t.Fatalf("the results differ: %v vs %v", spew.Sdump(&result3),
+	// Run the store.
-			spew.Sdump(results[1]))
+	store.run()
+	time.Sleep(flushInterval)
+
+	// Wait for the flush interval. There should be no records.
+	assertResults(0)
+
+	// Store a result and check immediately. There still shouldn't be
+	// any results stored (flush interval has not elapsed).
+	store.AddResult(nextResult())
+	assertResults(0)
+
+	// Assert that eventually the result is stored after being flushed.
+	assertResults(1)
+
+	// Store enough results that fill the max number of results.
+	for i := 0; i < testMaxRecords; i++ {
+		store.AddResult(nextResult())
+	}
+	assertResults(testMaxRecords)
+
+	// Finally, stop the store to recreate it.
+	store.stop()
+
+	// Recreate store.
+	store, err := newMissionControlStore(db, testMaxRecords, flushInterval)
+	require.NoError(t, err)
+	store.run()
+	defer store.stop()
+	time.Sleep(flushInterval)
+	assertResults(testMaxRecords)
+
+	// Fill the store with results again.
+	for i := 0; i < testMaxRecords; i++ {
+		store.AddResult(nextResult())
+	}
+	assertResults(testMaxRecords)
+}
+
+// BenchmarkMissionControlStoreFlushing benchmarks the periodic storage of data
+// from the mission control store when additional results are added between
+// runs.
+func BenchmarkMissionControlStoreFlushing(b *testing.B) {
+	var (
+		failureSourceIdx = 1
+		failureDetails   = lnwire.NewFailIncorrectDetails(100, 1000)
+		testTimeFwd      = testTime.Add(-time.Minute)
+
+		tests = []int{0, 1, 10, 100, 250, 500}
+	)
+
+	const testMaxRecords = 1000
+
+	for _, tc := range tests {
+		tc := tc
+		name := fmt.Sprintf("%v additional results", tc)
+		b.Run(name, func(b *testing.B) {
+			h := newMCStoreTestHarness(
+				b, testMaxRecords, time.Second,
+			)
+			store := h.store
+
+			// Fill the store.
+			var lastID uint64
+			for i := 0; i < testMaxRecords; i++ {
+				lastID++
+				result := &paymentResult{
+					route:            &mcStoreTestRoute,
+					failure:          failureDetails,
+					failureSourceIdx: &failureSourceIdx,
+					id:               lastID,
+					timeReply:        testTime,
+					timeFwd:          testTimeFwd,
+				}
+				store.AddResult(result)
+			}
+
+			// Do the first flush.
+			err := store.storeResults()
+			require.NoError(b, err)
+
+			// Create the additional results.
+			results := make([]*paymentResult, tc)
+			for i := 0; i < len(results); i++ {
+				results[i] = &paymentResult{
+					route:            &mcStoreTestRoute,
+					failure:          failureDetails,
+					failureSourceIdx: &failureSourceIdx,
+					timeReply:        testTime,
+					timeFwd:          testTimeFwd,
+				}
+			}
+
+			// Run the actual benchmark.
+			b.ResetTimer()
+			b.ReportAllocs()
+
+			for i := 0; i < b.N; i++ {
+				for j := 0; j < len(results); j++ {
+					lastID++
+					results[j].id = lastID
+					store.AddResult(results[j])
+				}
+				err := store.storeResults()
+				require.NoError(b, err)
+			}
+		})
 	}
 }