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 diff --git a/routing/missioncontrol_store.go b/routing/missioncontrol_store.go index dd60d9346..e149e8545 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. @@ -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 } 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) + } + }) } }