From 6a27bc29ba1b848bf9433cc68b6edb44c54c5fb3 Mon Sep 17 00:00:00 2001 From: Matheus Degiovani Date: Wed, 13 Mar 2024 12:54:10 -0300 Subject: [PATCH 1/5] missioncontrolstore: add additional tests and benchmarks These will be useful in the next commits. --- routing/missioncontrol_store_test.go | 275 ++++++++++++++++++++------- 1 file changed, 210 insertions(+), 65 deletions(-) diff --git a/routing/missioncontrol_store_test.go b/routing/missioncontrol_store_test.go index 81d1da4b3..34b925a3e 100644 --- a/routing/missioncontrol_store_test.go +++ b/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 -// 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) { +var ( + // mcStoreTestRoute is a test route for the mission control store tests. + mcStoreTestRoute = route.Route{ + SourcePubKey: route.Vertex{1}, + 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 { - t.Fatal(err) - } + require.NoError(t, 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 { - t.Fatal(err) - } + require.NoError(t, err) t.Cleanup(func() { require.NoError(t, db.Close()) }) - store, err := newMissionControlStore(db, testMaxRecords, time.Second) - if err != nil { - t.Fatal(err) - } + store, err := newMissionControlStore(db, maxRecords, flushInterval) + require.NoError(t, 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 { - t.Fatal(err) - } - 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, - }, - }, - } + require.NoError(t, err) + require.Len(t, results, 0) 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 { - t.Fatal(err) - } - require.Equal(t, 2, len(results)) - - if len(results) != 2 { - t.Fatal("expected two results") - } + require.NoError(t, err) + require.Len(t, results, 2) // Check that results are stored in chronological order. - if !reflect.DeepEqual(&result1, results[0]) { - t.Fatalf("the results differ: %v vs %v", spew.Sdump(&result1), - 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])) - } + require.Equal(t, &result1, results[0]) + require.Equal(t, &result2, results[1]) // Recreate store to test pruning. store, err = newMissionControlStore(db, testMaxRecords, time.Second) - if err != nil { - t.Fatal(err) - } + require.NoError(t, 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 { - t.Fatal(err) - } - require.Equal(t, 2, len(results)) - if len(results) != 2 { - t.Fatal("expected two results") + require.NoError(t, err) + require.Len(t, results, 2) + + require.Equal(t, &result2, results[0]) + require.Equal(t, &result3, results[1]) +} + +// 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]) { - t.Fatalf("the results differ: %v vs %v", spew.Sdump(&result2), - spew.Sdump(results[0])) + // Helper to assert the number of results is correct. + assertResults := func(wantCount int) { + 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), - spew.Sdump(results[1])) + + // Run the store. + 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) + } + }) } } From 637ac85d1d8914bff14366e14fd8b54e03b5e861 Mon Sep 17 00:00:00 2001 From: Matheus Degiovani Date: Wed, 13 Mar 2024 12:46:42 -0300 Subject: [PATCH 2/5] missioncontrolstore: skip work when there are no new entries This modifies the mission control store to avoid doing any work when no new payment result entries are in the queue to be processed. The mission control store maintains keeps the latest N (in production: 1000) entries in its DB, evicting older entries when new ones are added. Currently, its implementation is somewhat less performant than it could be. This commit adds an early return to the storeResults function to avoid doing any DB or memory operations when its outstanding queue is empty, improving the performance during quiescent periods of the LN node's execution. --- routing/missioncontrol_store.go | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/routing/missioncontrol_store.go b/routing/missioncontrol_store.go index dd60d9346..cfe0a2cf0 100644 --- a/routing/missioncontrol_store.go +++ b/routing/missioncontrol_store.go @@ -96,6 +96,8 @@ 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, @@ -302,14 +304,24 @@ func (b *missionControlStore) run() { // storeResults stores all accumulated results. func (b *missionControlStore) storeResults() error { + // We copy a reference to the queue and clear the original queue to be + // able to release the lock. b.queueMx.Lock() l := b.queue + + if l.Len() == 0 { + b.queueMx.Unlock() + + return nil + } b.queue = list.New() b.queueMx.Unlock() var ( - keys *list.List - keysMap map[string]struct{} + keys *list.List + keysMap map[string]struct{} + storeCount int + pruneCount int ) err := kvdb.Update(b.db, func(tx kvdb.RwTx) error { @@ -337,6 +349,7 @@ func (b *missionControlStore) storeResults() error { keys.PushBack(string(k)) keysMap[string(k)] = struct{}{} + storeCount++ } // Prune oldest entries. @@ -354,6 +367,7 @@ func (b *missionControlStore) storeResults() error { keys.Remove(front) delete(keysMap, key) + pruneCount++ } return nil @@ -365,12 +379,17 @@ func (b *missionControlStore) storeResults() error { for k := range b.keysMap { keysMap[k] = struct{}{} } + + storeCount, pruneCount = 0, 0 }) if err != nil { return err } + log.Debugf("Stored mission control results: %d added, %d deleted", + storeCount, pruneCount) + b.keys = keys b.keysMap = keysMap From 9059a655db20580b5f411e921b9ab6d03ce3ee51 Mon Sep 17 00:00:00 2001 From: Matheus Degiovani Date: Wed, 13 Mar 2024 12:55:28 -0300 Subject: [PATCH 3/5] missioncontrolstore: avoid ticker when there is no work to be done. This modifies the mission control store to avoid running the one second ticker for flushing data when there is no work to be done. This improves performance of a quiscent LN node by avoiding a one second interval busy loop that does nothing when there are no payments flowing through the node. --- routing/missioncontrol_store.go | 69 ++++++++++++++++++++++++++------- 1 file changed, 55 insertions(+), 14 deletions(-) diff --git a/routing/missioncontrol_store.go b/routing/missioncontrol_store.go index cfe0a2cf0..975380af6 100644 --- a/routing/missioncontrol_store.go +++ b/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. @@ -101,6 +104,7 @@ func newMissionControlStore(db kvdb.Backend, maxRecords int, return &missionControlStore{ done: make(chan struct{}), db: db, + queueCond: sync.NewCond(&sync.Mutex{}), queue: list.New(), keys: keys, keysMap: keysMap, @@ -111,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 { @@ -267,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() } @@ -283,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 } } @@ -306,16 +347,16 @@ func (b *missionControlStore) run() { func (b *missionControlStore) storeResults() error { // We copy a reference to the queue and clear the original queue to be // able to release the lock. - b.queueMx.Lock() + b.queueCond.L.Lock() l := b.queue if l.Len() == 0 { - b.queueMx.Unlock() + b.queueCond.L.Unlock() return nil } b.queue = list.New() - b.queueMx.Unlock() + b.queueCond.L.Unlock() var ( keys *list.List From 0c7a173354464080a9bca7428dd2bfaaf2327580 Mon Sep 17 00:00:00 2001 From: Matheus Degiovani Date: Wed, 13 Mar 2024 18:29:04 -0300 Subject: [PATCH 4/5] missioncontrolstore: remove duplication of in-memory data This removes duplication of in-memory data during the periodic flushing stage of the mission control store. The existing code entirely duplicates the in-memory cache of the store, which is very wasteful when only a few additional results are being rotated into the store. This has a significant performance penalty specially for wallets that remain online for a long time with a low volume of payments. The worst case scenario are wallets that see at most 1 new payment a second, where the entire in-memory cache is recreated every second. This commit improves the situation by determining what will be the actual changes that need to be committed before initiating the db transaction and only keeping track of these to update the in-memory cache if the db tx is successful. --- routing/missioncontrol_store.go | 108 ++++++++++++++++++++++---------- 1 file changed, 75 insertions(+), 33 deletions(-) diff --git a/routing/missioncontrol_store.go b/routing/missioncontrol_store.go index 975380af6..e149e8545 100644 --- a/routing/missioncontrol_store.go +++ b/routing/missioncontrol_store.go @@ -359,68 +359,98 @@ func (b *missionControlStore) storeResults() error { 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 }) @@ -431,8 +461,20 @@ func (b *missionControlStore) storeResults() error { log.Debugf("Stored mission control results: %d added, %d deleted", storeCount, pruneCount) - b.keys = keys - b.keysMap = keysMap + // 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 } From f39edaa19a26f4e487848ebac9797e14d9ba62c6 Mon Sep 17 00:00:00 2001 From: Matheus Degiovani Date: Thu, 13 Jun 2024 13:48:33 -0300 Subject: [PATCH 5/5] docs: update release notes --- docs/release-notes/release-notes-0.18.1.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/release-notes/release-notes-0.18.1.md b/docs/release-notes/release-notes-0.18.1.md index 92ce40a67..1d6608407 100644 --- a/docs/release-notes/release-notes-0.18.1.md +++ b/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