Merge pull request #6136 from bhandras/forall_etcd

etcd: add ForAll to speed up graph operations

docs/etcd.md

@@ -39,10 +39,8 @@ To start your local etcd instance for testing run:
 ```
 
 The large `max-txn-ops` and `max-request-bytes` values are currently required in
-case of running LND with the full graph in etcd. Upcoming versions will split
-the database to local and replicated parts and only essential parts will remain
-in the replicated database, removing the requirement for these additional 
-settings. These parameters have been tested to work with testnet LND.
+case of running LND with the full graph in etcd. These parameters have been
+tested to work with testnet LND.
 
 ## Configuring LND to run on etcd
 

docs/release-notes/release-notes-0.15.0.md

@@ -92,6 +92,11 @@
   change, it allows encrypted failure messages to be returned to the sender.
   Additionally it is possible to signal a malformed htlc.
 
+## Database
+
+* [Add ForAll implementation for etcd to speed up
+  graph cache at startup](https://github.com/lightningnetwork/lnd/pull/6136)
+
 ## Documentation
 
 * Improved instructions on [how to build lnd for mobile](https://github.com/lightningnetwork/lnd/pull/6085).
@@ -139,6 +144,7 @@ gRPC performance metrics (latency to process `GetInfo`, etc)](https://github.com
 # Contributors (Alphabetical Order)
 
 * 3nprob
+* Andras Banki-Horvath
 * Andreas Schjønhaug
 * asvdf
 * bitromortac

go.mod

@@ -44,7 +44,7 @@ require (
 	github.com/lightningnetwork/lnd/cert v1.1.0
 	github.com/lightningnetwork/lnd/clock v1.1.0
 	github.com/lightningnetwork/lnd/healthcheck v1.2.0
-	github.com/lightningnetwork/lnd/kvdb v1.3.0
+	github.com/lightningnetwork/lnd/kvdb v1.3.1
 	github.com/lightningnetwork/lnd/queue v1.1.0
 	github.com/lightningnetwork/lnd/ticker v1.1.0
 	github.com/lightningnetwork/lnd/tlv v1.0.1

kvdb/etcd/config.go

@@ -61,6 +61,7 @@ func (c *Config) CloneWithSubNamespace(subNamespace string) *Config {
 		KeyFile:            c.KeyFile,
 		InsecureSkipVerify: c.InsecureSkipVerify,
 		CollectStats:       c.CollectStats,
+		MaxMsgSize:         c.MaxMsgSize,
 		SingleWriter:       c.SingleWriter,
 	}
 }
@@ -81,6 +82,7 @@ func (c *Config) CloneWithSingleWriter() *Config {
 		KeyFile:            c.KeyFile,
 		InsecureSkipVerify: c.InsecureSkipVerify,
 		CollectStats:       c.CollectStats,
+		MaxMsgSize:         c.MaxMsgSize,
 		SingleWriter:       true,
 	}
 }

kvdb/etcd/embed.go

@@ -68,7 +68,7 @@ func NewEmbeddedEtcdInstance(path string, clientPort, peerPort uint16,
 	cfg.Dir = path
 
 	// To ensure that we can submit large transactions.
-	cfg.MaxTxnOps = 8192
+	cfg.MaxTxnOps = 16384
 	cfg.MaxRequestBytes = 16384 * 1024
 	cfg.Logger = "zap"
 	if logFile != "" {
@@ -109,6 +109,7 @@ func NewEmbeddedEtcdInstance(path string, clientPort, peerPort uint16,
 		Host:               "http://" + clientURL,
 		InsecureSkipVerify: true,
 		Namespace:          defaultNamespace,
+		MaxMsgSize:         int(cfg.MaxRequestBytes),
 	}
 
 	return connConfig, func() {

kvdb/etcd/readwrite_bucket.go

@@ -17,6 +17,9 @@ type readWriteBucket struct {
 	// appropriate prefix to prefix the key.
 	id []byte
 
+	// key is the bucket key.
+	key []byte
+
 	// tx holds the parent transaction.
 	tx *readWriteTx
 }
@@ -25,8 +28,9 @@ type readWriteBucket struct {
 // and bucket id.
 func newReadWriteBucket(tx *readWriteTx, key, id []byte) *readWriteBucket {
 	return &readWriteBucket{
-		id: id,
-		tx: tx,
+		id:  id,
+		key: key,
+		tx:  tx,
 	}
 }
 
@@ -66,6 +70,31 @@ func (b *readWriteBucket) ForEach(cb func(k, v []byte) error) error {
 	return nil
 }
 
+// ForAll is an optimized version of ForEach for the case when we know we will
+// fetch all (or almost all) items.
+//
+// NOTE: ForAll differs from ForEach in that no additional queries can
+// be executed within the callback.
+func (b *readWriteBucket) ForAll(cb func(k, v []byte) error) error {
+	// When we opened this bucket, we fetched the bucket key using the STM
+	// which put a revision "lock" in the read set. We can leverage this
+	// by incrementing the revision on the bucket, making any transaction
+	// retry that'd touch this same bucket. This way we can safely read all
+	// keys from the bucket and not cache them in the STM.
+	// To increment the bucket's revision, we simply put in the bucket key
+	// value again (which is idempotent if the bucket has just been created).
+	b.tx.stm.Put(string(b.key), string(b.id))
+
+	// TODO(bhandras): page size should be configurable in ForAll.
+	return b.tx.stm.FetchRangePaginatedRaw(
+		string(b.id), 1000,
+		func(kv KV) error {
+			key, val := getKeyVal(&kv)
+			return cb(key, val)
+		},
+	)
+}
+
 // Get returns the value for the given key. Returns nil if the key does
 // not exist in this bucket.
 func (b *readWriteBucket) Get(key []byte) []byte {
@@ -406,9 +435,3 @@ func (b *readWriteBucket) Prefetch(paths ...[]string) {
 
 	b.tx.stm.Prefetch(flattenMap(keys), flattenMap(ranges))
 }
-
-// ForAll is an optimized version of ForEach with the limitation that no
-// additional queries can be executed within the callback.
-func (b *readWriteBucket) ForAll(cb func(k, v []byte) error) error {
-	return b.ForEach(cb)
-}

kvdb/etcd/stm.go

@@ -81,6 +81,11 @@ type STM interface {
 	// Prefetch prefetches the passed keys and prefixes. For prefixes it'll
 	// fetch the whole range.
 	Prefetch(keys []string, prefix []string)
+
+	// FetchRangePaginatedRaw will fetch the range with the passed prefix up
+	// to the passed limit per page.
+	FetchRangePaginatedRaw(prefix string, limit int64,
+		cb func(kv KV) error) error
 }
 
 // CommitError is used to check if there was an error
@@ -589,6 +594,52 @@ func (ws writeSet) puts() []v3.Op {
 	return puts
 }
 
+// FetchRangePaginatedRaw will fetch the range with the passed prefix up to the
+// passed limit per page.
+func (s *stm) FetchRangePaginatedRaw(prefix string, limit int64,
+	cb func(kv KV) error) error {
+
+	s.callCount++
+
+	opts := []v3.OpOption{
+		v3.WithSort(v3.SortByKey, v3.SortAscend),
+		v3.WithRange(v3.GetPrefixRangeEnd(prefix)),
+		v3.WithLimit(limit),
+	}
+
+	key := prefix
+	for {
+		resp, err := s.client.Get(
+			s.options.ctx, key, append(opts, s.getOpts...)...,
+		)
+		if err != nil {
+			return DatabaseError{
+				msg: "stm.fetch() failed",
+				err: err,
+			}
+		}
+
+		// Fill the read set with key/values returned.
+		for _, kv := range resp.Kvs {
+			err := cb(KV{string(kv.Key), string(kv.Value)})
+
+			if err != nil {
+				return err
+			}
+		}
+
+		// We've reached the range end.
+		if !resp.More {
+			break
+		}
+
+		// Continue from the page end + "\x00".
+		key = string(resp.Kvs[len(resp.Kvs)-1].Key) + "\x00"
+	}
+
+	return nil
+}
+
 // fetch is a helper to fetch key/value given options. If a value is returned
 // then fetch will try to fix the STM's snapshot revision (if not already set).
 // We'll also cache the returned key/value in the read set.