Make state sync request size configurable (#614)

* Make state sync request size configurable

* finish passing the parameter

* set default to 256

plugin/evm/config.go

@@ -55,7 +55,8 @@ const (
 	// time assumptions:
 	// - normal bootstrap processing time: ~14 blocks / second
 	// - state sync time: ~6 hrs.
-	defaultStateSyncMinBlocks = 300_000
+	defaultStateSyncMinBlocks   = 300_000
+	defaultStateSyncRequestSize = 256 // the number of key/values to ask peers for per request
 )
 
 var (
@@ -186,6 +187,7 @@ type Config struct {
 	StateSyncIDs             string `json:"state-sync-ids"`
 	StateSyncCommitInterval  uint64 `json:"state-sync-commit-interval"`
 	StateSyncMinBlocks       uint64 `json:"state-sync-min-blocks"`
+	StateSyncRequestSize     uint16 `json:"state-sync-request-size"`
 
 	// SkipUpgradeCheck disables checking that upgrades must take place before the last
 	// accepted block. Skipping this check is useful when a node operator does not update
@@ -263,6 +265,7 @@ func (c *Config) SetDefaults() {
 	c.StateSyncServerTrieCache = defaultStateSyncServerTrieCache
 	c.StateSyncCommitInterval = defaultSyncableCommitInterval
 	c.StateSyncMinBlocks = defaultStateSyncMinBlocks
+	c.StateSyncRequestSize = defaultStateSyncRequestSize
 	c.AllowUnprotectedTxHashes = defaultAllowUnprotectedTxHashes
 	c.AcceptedCacheSize = defaultAcceptedCacheSize
 }

plugin/evm/syncervm_client.go

@@ -42,7 +42,8 @@ type stateSyncClientConfig struct {
 	// Specifies the number of blocks behind the latest state summary that the chain must be
 	// in order to prefer performing state sync over falling back to the normal bootstrapping
 	// algorithm.
-	stateSyncMinBlocks uint64
+	stateSyncMinBlocks   uint64
+	stateSyncRequestSize uint16 // number of key/value pairs to ask peers for per request
 
 	lastAcceptedHeight uint64
 
@@ -283,6 +284,7 @@ func (client *stateSyncerClient) syncStateTrie(ctx context.Context) error {
 		DB:                       client.chaindb,
 		MaxOutstandingCodeHashes: statesync.DefaultMaxOutstandingCodeHashes,
 		NumCodeFetchingWorkers:   statesync.DefaultNumCodeFetchingWorkers,
+		RequestSize:              client.stateSyncRequestSize,
 	})
 	if err != nil {
 		return err

plugin/evm/vm.go

@@ -505,15 +505,16 @@ func (vm *VM) initializeStateSyncClient(lastAcceptedHeight uint64) error {
 				BlockParser:      vm,
 			},
 		),
-		enabled:            vm.config.StateSyncEnabled,
-		skipResume:         vm.config.StateSyncSkipResume,
-		stateSyncMinBlocks: vm.config.StateSyncMinBlocks,
-		lastAcceptedHeight: lastAcceptedHeight, // TODO clean up how this is passed around
-		chaindb:            vm.chaindb,
-		metadataDB:         vm.metadataDB,
-		acceptedBlockDB:    vm.acceptedBlockDB,
-		db:                 vm.db,
-		toEngine:           vm.toEngine,
+		enabled:              vm.config.StateSyncEnabled,
+		skipResume:           vm.config.StateSyncSkipResume,
+		stateSyncMinBlocks:   vm.config.StateSyncMinBlocks,
+		stateSyncRequestSize: vm.config.StateSyncRequestSize,
+		lastAcceptedHeight:   lastAcceptedHeight, // TODO clean up how this is passed around
+		chaindb:              vm.chaindb,
+		metadataDB:           vm.metadataDB,
+		acceptedBlockDB:      vm.acceptedBlockDB,
+		db:                   vm.db,
+		toEngine:             vm.toEngine,
 	})
 
 	// If StateSync is disabled, clear any ongoing summary so that we will not attempt to resume

sync/client/client_test.go

@@ -799,7 +799,7 @@ func TestGetLeafsRetries(t *testing.T) {
 		Root:  root,
 		Start: bytes.Repeat([]byte{0x00}, common.HashLength),
 		End:   bytes.Repeat([]byte{0xff}, common.HashLength),
-		Limit: defaultLeafRequestLimit,
+		Limit: 1024,
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())

sync/client/leaf_syncer.go

@@ -20,8 +20,6 @@ var (
 	errFailedToFetchLeafs = errors.New("failed to fetch leafs")
 )
 
-const defaultLeafRequestLimit = 1024
-
 // LeafSyncTask represents a complete task to be completed by the leaf syncer.
 // Note: each LeafSyncTask is processed on its own goroutine and there will
 // not be concurrent calls to the callback methods. Implementations should return
@@ -38,9 +36,10 @@ type LeafSyncTask interface {
 }
 
 type CallbackLeafSyncer struct {
-	client LeafClient
-	done   chan error
-	tasks  <-chan LeafSyncTask
+	client      LeafClient
+	done        chan error
+	tasks       <-chan LeafSyncTask
+	requestSize uint16
 }
 
 type LeafClient interface {
@@ -50,11 +49,12 @@ type LeafClient interface {
 }
 
 // NewCallbackLeafSyncer creates a new syncer object to perform leaf sync of tries.
-func NewCallbackLeafSyncer(client LeafClient, tasks <-chan LeafSyncTask) *CallbackLeafSyncer {
+func NewCallbackLeafSyncer(client LeafClient, tasks <-chan LeafSyncTask, requestSize uint16) *CallbackLeafSyncer {
 	return &CallbackLeafSyncer{
-		client: client,
-		done:   make(chan error),
-		tasks:  tasks,
+		client:      client,
+		done:        make(chan error),
+		tasks:       tasks,
+		requestSize: requestSize,
 	}
 }
 
@@ -100,7 +100,7 @@ func (c *CallbackLeafSyncer) syncTask(ctx context.Context, task LeafSyncTask) er
 			Root:    root,
 			Account: task.Account(),
 			Start:   start,
-			Limit:   defaultLeafRequestLimit,
+			Limit:   c.requestSize,
 		})
 		if err != nil {
 			return fmt.Errorf("%s: %w", errFailedToFetchLeafs, err)

sync/statesync/state_syncer.go

@@ -28,8 +28,9 @@ type StateSyncerConfig struct {
 	Client                   syncclient.Client
 	DB                       ethdb.Database
 	BatchSize                int
-	MaxOutstandingCodeHashes int // Maximum number of code hashes in the code syncer queue
-	NumCodeFetchingWorkers   int // Number of code syncing threads
+	MaxOutstandingCodeHashes int    // Maximum number of code hashes in the code syncer queue
+	NumCodeFetchingWorkers   int    // Number of code syncing threads
+	RequestSize              uint16 // Number of leafs to request from a peer at a time
 }
 
 // stateSync keeps the state of the entire state sync operation.
@@ -82,7 +83,7 @@ func NewStateSyncer(config *StateSyncerConfig) (*stateSync, error) {
 		mainTrieDone: make(chan struct{}),
 		done:         make(chan error, 1),
 	}
-	ss.syncer = syncclient.NewCallbackLeafSyncer(config.Client, ss.segments)
+	ss.syncer = syncclient.NewCallbackLeafSyncer(config.Client, ss.segments, config.RequestSize)
 	ss.codeSyncer = newCodeSyncer(CodeSyncerConfig{
 		DB:                       config.DB,
 		Client:                   config.Client,

sync/statesync/sync_test.go

@@ -62,6 +62,7 @@ func testSync(t *testing.T, test syncTest) {
 		BatchSize:                1000, // Use a lower batch size in order to get test coverage of batches being written early.
 		NumCodeFetchingWorkers:   DefaultNumCodeFetchingWorkers,
 		MaxOutstandingCodeHashes: DefaultMaxOutstandingCodeHashes,
+		RequestSize:              1024,
 	})
 	if err != nil {
 		t.Fatal(err)