Merge pull request #5642 from guggero/in-memory-graph

In-memory graph cache for faster pathfinding

htlcswitch/circuit_map.go

@@ -199,9 +199,16 @@ type circuitMap struct {
 // parameterize an instance of circuitMap.
 type CircuitMapConfig struct {
 	// DB provides the persistent storage engine for the circuit map.
-	// TODO(conner): create abstraction to allow for the substitution of
-	// other persistence engines.
-	DB *channeldb.DB
+	DB kvdb.Backend
+
+	// FetchAllOpenChannels is a function that fetches all currently open
+	// channels from the channel database.
+	FetchAllOpenChannels func() ([]*channeldb.OpenChannel, error)
+
+	// FetchClosedChannels is a function that fetches all closed channels
+	// from the channel database.
+	FetchClosedChannels func(
+		pendingOnly bool) ([]*channeldb.ChannelCloseSummary, error)
 
 	// ExtractErrorEncrypter derives the shared secret used to encrypt
 	// errors from the obfuscator's ephemeral public key.
@@ -296,7 +303,7 @@ func (cm *circuitMap) cleanClosedChannels() error {
 
 	// Find closed channels and cache their ShortChannelIDs into a map.
 	// This map will be used for looking up relative circuits and keystones.
-	closedChannels, err := cm.cfg.DB.FetchClosedChannels(false)
+	closedChannels, err := cm.cfg.FetchClosedChannels(false)
 	if err != nil {
 		return err
 	}
@@ -629,7 +636,7 @@ func (cm *circuitMap) decodeCircuit(v []byte) (*PaymentCircuit, error) {
 // channels. Therefore, it must be called before any links are created to avoid
 // interfering with normal operation.
 func (cm *circuitMap) trimAllOpenCircuits() error {
-	activeChannels, err := cm.cfg.DB.FetchAllOpenChannels()
+	activeChannels, err := cm.cfg.FetchAllOpenChannels()
 	if err != nil {
 		return err
 	}
@@ -860,7 +867,7 @@ func (cm *circuitMap) CommitCircuits(circuits ...*PaymentCircuit) (
 	// Write the entire batch of circuits to the persistent circuit bucket
 	// using bolt's Batch write. This method must be called from multiple,
 	// distinct goroutines to have any impact on performance.
-	err := kvdb.Batch(cm.cfg.DB.Backend, func(tx kvdb.RwTx) error {
+	err := kvdb.Batch(cm.cfg.DB, func(tx kvdb.RwTx) error {
 		circuitBkt := tx.ReadWriteBucket(circuitAddKey)
 		if circuitBkt == nil {
 			return ErrCorruptedCircuitMap
@@ -1091,7 +1098,7 @@ func (cm *circuitMap) DeleteCircuits(inKeys ...CircuitKey) error {
 	}
 	cm.mtx.Unlock()
 
-	err := kvdb.Batch(cm.cfg.DB.Backend, func(tx kvdb.RwTx) error {
+	err := kvdb.Batch(cm.cfg.DB, func(tx kvdb.RwTx) error {
 		for _, circuit := range removedCircuits {
 			// If this htlc made it to an outgoing link, load the
 			// keystone bucket from which we will remove the

htlcswitch/circuit_test.go

@@ -103,8 +103,11 @@ func newCircuitMap(t *testing.T) (*htlcswitch.CircuitMapConfig,
 
 	onionProcessor := newOnionProcessor(t)
 
+	db := makeCircuitDB(t, "")
 	circuitMapCfg := &htlcswitch.CircuitMapConfig{
-		DB:                    makeCircuitDB(t, ""),
+		DB:                    db,
+		FetchAllOpenChannels:  db.ChannelStateDB().FetchAllOpenChannels,
+		FetchClosedChannels:   db.ChannelStateDB().FetchClosedChannels,
 		ExtractErrorEncrypter: onionProcessor.ExtractErrorEncrypter,
 	}
 
@@ -634,13 +637,17 @@ func makeCircuitDB(t *testing.T, path string) *channeldb.DB {
 func restartCircuitMap(t *testing.T, cfg *htlcswitch.CircuitMapConfig) (
 	*htlcswitch.CircuitMapConfig, htlcswitch.CircuitMap) {
 
-	// Record the current temp path and close current db.
-	dbPath := cfg.DB.Path()
+	// Record the current temp path and close current db. We know we have
+	// a full channeldb.DB here since we created it just above.
+	dbPath := cfg.DB.(*channeldb.DB).Path()
 	cfg.DB.Close()
 
 	// Reinitialize circuit map with same db path.
+	db := makeCircuitDB(t, dbPath)
 	cfg2 := &htlcswitch.CircuitMapConfig{
-		DB:                    makeCircuitDB(t, dbPath),
+		DB:                    db,
+		FetchAllOpenChannels:  db.ChannelStateDB().FetchAllOpenChannels,
+		FetchClosedChannels:   db.ChannelStateDB().FetchClosedChannels,
 		ExtractErrorEncrypter: cfg.ExtractErrorEncrypter,
 	}
 	cm2, err := htlcswitch.NewCircuitMap(cfg2)

htlcswitch/link_test.go

@@ -1938,7 +1938,7 @@ func newSingleLinkTestHarness(chanAmt, chanReserve btcutil.Amount) (
 
 	pCache := newMockPreimageCache()
 
-	aliceDb := aliceLc.channel.State().Db
+	aliceDb := aliceLc.channel.State().Db.GetParentDB()
 	aliceSwitch, err := initSwitchWithDB(testStartingHeight, aliceDb)
 	if err != nil {
 		return nil, nil, nil, nil, nil, nil, err
@@ -4438,7 +4438,7 @@ func (h *persistentLinkHarness) restartLink(
 		pCache = newMockPreimageCache()
 	)
 
-	aliceDb := aliceChannel.State().Db
+	aliceDb := aliceChannel.State().Db.GetParentDB()
 	aliceSwitch := h.coreLink.cfg.Switch
 	if restartSwitch {
 		var err error

htlcswitch/mock.go

@@ -170,8 +170,10 @@ func initSwitchWithDB(startingHeight uint32, db *channeldb.DB) (*Switch, error)
 	}
 
 	cfg := Config{
-		DB:             db,
-		SwitchPackager: channeldb.NewSwitchPackager(),
+		DB:                   db,
+		FetchAllOpenChannels: db.ChannelStateDB().FetchAllOpenChannels,
+		FetchClosedChannels:  db.ChannelStateDB().FetchClosedChannels,
+		SwitchPackager:       channeldb.NewSwitchPackager(),
 		FwdingLog: &mockForwardingLog{
 			events: make(map[time.Time]channeldb.ForwardingEvent),
 		},

htlcswitch/payment_result.go

@@ -83,7 +83,7 @@ func deserializeNetworkResult(r io.Reader) (*networkResult, error) {
 // is back. The Switch will checkpoint any received result to the store, and
 // the store will keep results and notify the callers about them.
 type networkResultStore struct {
-	db *channeldb.DB
+	backend kvdb.Backend
 
 	// results is a map from paymentIDs to channels where subscribers to
 	// payment results will be notified.
@@ -96,9 +96,9 @@ type networkResultStore struct {
 	paymentIDMtx *multimutex.Mutex
 }
 
-func newNetworkResultStore(db *channeldb.DB) *networkResultStore {
+func newNetworkResultStore(db kvdb.Backend) *networkResultStore {
 	return &networkResultStore{
-		db:           db,
+		backend:      db,
 		results:      make(map[uint64][]chan *networkResult),
 		paymentIDMtx: multimutex.NewMutex(),
 	}
@@ -126,7 +126,7 @@ func (store *networkResultStore) storeResult(paymentID uint64,
 	var paymentIDBytes [8]byte
 	binary.BigEndian.PutUint64(paymentIDBytes[:], paymentID)
 
-	err := kvdb.Batch(store.db.Backend, func(tx kvdb.RwTx) error {
+	err := kvdb.Batch(store.backend, func(tx kvdb.RwTx) error {
 		networkResults, err := tx.CreateTopLevelBucket(
 			networkResultStoreBucketKey,
 		)
@@ -171,7 +171,7 @@ func (store *networkResultStore) subscribeResult(paymentID uint64) (
 		resultChan = make(chan *networkResult, 1)
 	)
 
-	err := kvdb.View(store.db, func(tx kvdb.RTx) error {
+	err := kvdb.View(store.backend, func(tx kvdb.RTx) error {
 		var err error
 		result, err = fetchResult(tx, paymentID)
 		switch {
@@ -219,7 +219,7 @@ func (store *networkResultStore) getResult(pid uint64) (
 	*networkResult, error) {
 
 	var result *networkResult
-	err := kvdb.View(store.db, func(tx kvdb.RTx) error {
+	err := kvdb.View(store.backend, func(tx kvdb.RTx) error {
 		var err error
 		result, err = fetchResult(tx, pid)
 		return err
@@ -260,7 +260,7 @@ func fetchResult(tx kvdb.RTx, pid uint64) (*networkResult, error) {
 // concurrently while this process is ongoing, as its result might end up being
 // deleted.
 func (store *networkResultStore) cleanStore(keep map[uint64]struct{}) error {
-	return kvdb.Update(store.db.Backend, func(tx kvdb.RwTx) error {
+	return kvdb.Update(store.backend, func(tx kvdb.RwTx) error {
 		networkResults, err := tx.CreateTopLevelBucket(
 			networkResultStoreBucketKey,
 		)

htlcswitch/switch.go

@@ -130,9 +130,18 @@ type Config struct {
 	// subsystem.
 	LocalChannelClose func(pubKey []byte, request *ChanClose)
 
-	// DB is the channeldb instance that will be used to back the switch's
+	// DB is the database backend that will be used to back the switch's
 	// persistent circuit map.
-	DB *channeldb.DB
+	DB kvdb.Backend
+
+	// FetchAllOpenChannels is a function that fetches all currently open
+	// channels from the channel database.
+	FetchAllOpenChannels func() ([]*channeldb.OpenChannel, error)
+
+	// FetchClosedChannels is a function that fetches all closed channels
+	// from the channel database.
+	FetchClosedChannels func(
+		pendingOnly bool) ([]*channeldb.ChannelCloseSummary, error)
 
 	// SwitchPackager provides access to the forwarding packages of all
 	// active channels. This gives the switch the ability to read arbitrary
@@ -294,6 +303,8 @@ type Switch struct {
 func New(cfg Config, currentHeight uint32) (*Switch, error) {
 	circuitMap, err := NewCircuitMap(&CircuitMapConfig{
 		DB:                    cfg.DB,
+		FetchAllOpenChannels:  cfg.FetchAllOpenChannels,
+		FetchClosedChannels:   cfg.FetchClosedChannels,
 		ExtractErrorEncrypter: cfg.ExtractErrorEncrypter,
 	})
 	if err != nil {
@@ -1455,7 +1466,7 @@ func (s *Switch) closeCircuit(pkt *htlcPacket) (*PaymentCircuit, error) {
 // we're the originator of the payment, so the link stops attempting to
 // re-broadcast.
 func (s *Switch) ackSettleFail(settleFailRefs ...channeldb.SettleFailRef) error {
-	return kvdb.Batch(s.cfg.DB.Backend, func(tx kvdb.RwTx) error {
+	return kvdb.Batch(s.cfg.DB, func(tx kvdb.RwTx) error {
 		return s.cfg.SwitchPackager.AckSettleFails(tx, settleFailRefs...)
 	})
 }
@@ -1859,7 +1870,7 @@ func (s *Switch) Start() error {
 // forwarding packages and reforwards any Settle or Fail HTLCs found. This is
 // used to resurrect the switch's mailboxes after a restart.
 func (s *Switch) reforwardResponses() error {
-	openChannels, err := s.cfg.DB.FetchAllOpenChannels()
+	openChannels, err := s.cfg.FetchAllOpenChannels()
 	if err != nil {
 		return err
 	}
@@ -2122,6 +2133,17 @@ func (s *Switch) getLink(chanID lnwire.ChannelID) (ChannelLink, error) {
 	return link, nil
 }
 
+// GetLinkByShortID attempts to return the link which possesses the target short
+// channel ID.
+func (s *Switch) GetLinkByShortID(chanID lnwire.ShortChannelID) (ChannelLink,
+	error) {
+
+	s.indexMtx.RLock()
+	defer s.indexMtx.RUnlock()
+
+	return s.getLinkByShortID(chanID)
+}
+
 // getLinkByShortID attempts to return the link which possesses the target
 // short channel ID.
 //

htlcswitch/test_utils.go

@@ -306,7 +306,7 @@ func createTestChannel(alicePrivKey, bobPrivKey []byte,
 		LocalCommitment:         aliceCommit,
 		RemoteCommitment:        aliceCommit,
 		ShortChannelID:          chanID,
-		Db:                      dbAlice,
+		Db:                      dbAlice.ChannelStateDB(),
 		Packager:                channeldb.NewChannelPackager(chanID),
 		FundingTxn:              channels.TestFundingTx,
 	}
@@ -325,7 +325,7 @@ func createTestChannel(alicePrivKey, bobPrivKey []byte,
 		LocalCommitment:         bobCommit,
 		RemoteCommitment:        bobCommit,
 		ShortChannelID:          chanID,
-		Db:                      dbBob,
+		Db:                      dbBob.ChannelStateDB(),
 		Packager:                channeldb.NewChannelPackager(chanID),
 	}
 
@@ -384,7 +384,8 @@ func createTestChannel(alicePrivKey, bobPrivKey []byte,
 	}
 
 	restoreAlice := func() (*lnwallet.LightningChannel, error) {
-		aliceStoredChannels, err := dbAlice.FetchOpenChannels(aliceKeyPub)
+		aliceStoredChannels, err := dbAlice.ChannelStateDB().
+			FetchOpenChannels(aliceKeyPub)
 		switch err {
 		case nil:
 		case kvdb.ErrDatabaseNotOpen:
@@ -394,7 +395,8 @@ func createTestChannel(alicePrivKey, bobPrivKey []byte,
 					"db: %v", err)
 			}
 
-			aliceStoredChannels, err = dbAlice.FetchOpenChannels(aliceKeyPub)
+			aliceStoredChannels, err = dbAlice.ChannelStateDB().
+				FetchOpenChannels(aliceKeyPub)
 			if err != nil {
 				return nil, errors.Errorf("unable to fetch alice "+
 					"channel: %v", err)
@@ -428,7 +430,8 @@ func createTestChannel(alicePrivKey, bobPrivKey []byte,
 	}
 
 	restoreBob := func() (*lnwallet.LightningChannel, error) {
-		bobStoredChannels, err := dbBob.FetchOpenChannels(bobKeyPub)
+		bobStoredChannels, err := dbBob.ChannelStateDB().
+			FetchOpenChannels(bobKeyPub)
 		switch err {
 		case nil:
 		case kvdb.ErrDatabaseNotOpen:
@@ -438,7 +441,8 @@ func createTestChannel(alicePrivKey, bobPrivKey []byte,
 					"db: %v", err)
 			}
 
-			bobStoredChannels, err = dbBob.FetchOpenChannels(bobKeyPub)
+			bobStoredChannels, err = dbBob.ChannelStateDB().
+				FetchOpenChannels(bobKeyPub)
 			if err != nil {
 				return nil, errors.Errorf("unable to fetch bob "+
 					"channel: %v", err)
@@ -950,9 +954,9 @@ func newThreeHopNetwork(t testing.TB, aliceChannel, firstBobChannel,
 	secondBobChannel, carolChannel *lnwallet.LightningChannel,
 	startingHeight uint32, opts ...serverOption) *threeHopNetwork {
 
-	aliceDb := aliceChannel.State().Db
-	bobDb := firstBobChannel.State().Db
-	carolDb := carolChannel.State().Db
+	aliceDb := aliceChannel.State().Db.GetParentDB()
+	bobDb := firstBobChannel.State().Db.GetParentDB()
+	carolDb := carolChannel.State().Db.GetParentDB()
 
 	hopNetwork := newHopNetwork()
 
@@ -1201,8 +1205,8 @@ func newTwoHopNetwork(t testing.TB,
 	aliceChannel, bobChannel *lnwallet.LightningChannel,
 	startingHeight uint32) *twoHopNetwork {
 
-	aliceDb := aliceChannel.State().Db
-	bobDb := bobChannel.State().Db
+	aliceDb := aliceChannel.State().Db.GetParentDB()
+	bobDb := bobChannel.State().Db.GetParentDB()
 
 	hopNetwork := newHopNetwork()