Refactoring of initial sync (#5096)

* implements blocks queue

* refactors updateCounter method

* fixes deadlock on stop w/o start

* refactors updateSchedulerState

* more tests on schduler

* parseFetchResponse tests

* wraps up tests for blocks queue

* eod commit

* fixes data race in round robin

* revamps fetcher

* fixes race conditions + livelocks + deadlocks

* less verbose output

* fixes data race, by isolating critical sections

* minor refactoring: resolves blocking calls

* implements init-sync queue

* udpate fetch/send buffers in blocks fetcher

* blockState enum-like type alias

* refactors common code into releaseTicket()

* better gc

* linter

* minor fix to round robin

* moves original round robin into its own package

* adds enableInitSyncQueue flag

* fixes issue with init-sync service selection

* Update beacon-chain/sync/initial-sync/round_robin.go

Co-Authored-By: terence tsao <[email protected]>

* initsyncv1 -> initsyncold

* adds span

Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>
Co-authored-by: Raul Jordan <[email protected]>
Co-authored-by: terence tsao <[email protected]>

beacon-chain/node/BUILD.bazel

@@ -24,6 +24,7 @@ go_library(
         "//beacon-chain/state/stategen:go_default_library",
         "//beacon-chain/sync:go_default_library",
         "//beacon-chain/sync/initial-sync:go_default_library",
+        "//beacon-chain/sync/initial-sync-old:go_default_library",
         "//shared:go_default_library",
         "//shared/cmd:go_default_library",
         "//shared/debug:go_default_library",

beacon-chain/node/node.go

@@ -34,6 +34,7 @@ import (
 	"github.com/prysmaticlabs/prysm/beacon-chain/state/stategen"
 	prysmsync "github.com/prysmaticlabs/prysm/beacon-chain/sync"
 	initialsync "github.com/prysmaticlabs/prysm/beacon-chain/sync/initial-sync"
+	initialsyncold "github.com/prysmaticlabs/prysm/beacon-chain/sync/initial-sync-old"
 	"github.com/prysmaticlabs/prysm/shared"
 	"github.com/prysmaticlabs/prysm/shared/cmd"
 	"github.com/prysmaticlabs/prysm/shared/debug"
@@ -404,9 +405,19 @@ func (b *BeaconNode) registerSyncService(ctx *cli.Context) error {
 		return err
 	}
 
-	var initSync *initialsync.Service
-	if err := b.services.FetchService(&initSync); err != nil {
-		return err
+	var initSync prysmsync.Checker
+	if cfg := featureconfig.Get(); cfg.EnableInitSyncQueue {
+		var initSyncTmp *initialsync.Service
+		if err := b.services.FetchService(&initSyncTmp); err != nil {
+			return err
+		}
+		initSync = initSyncTmp
+	} else {
+		var initSyncTmp *initialsyncold.Service
+		if err := b.services.FetchService(&initSyncTmp); err != nil {
+			return err
+		}
+		initSync = initSyncTmp
 	}
 
 	rs := prysmsync.NewRegularSync(&prysmsync.Config{
@@ -431,16 +442,25 @@ func (b *BeaconNode) registerInitialSyncService(ctx *cli.Context) error {
 		return err
 	}
 
-	is := initialsync.NewInitialSync(&initialsync.Config{
+	if cfg := featureconfig.Get(); cfg.EnableInitSyncQueue {
+		is := initialsync.NewInitialSync(&initialsync.Config{
+			DB:            b.db,
+			Chain:         chainService,
+			P2P:           b.fetchP2P(ctx),
+			StateNotifier: b,
+			BlockNotifier: b,
+		})
+		return b.services.RegisterService(is)
+	}
+
+	is := initialsyncold.NewInitialSync(&initialsyncold.Config{
 		DB:            b.db,
 		Chain:         chainService,
 		P2P:           b.fetchP2P(ctx),
 		StateNotifier: b,
 		BlockNotifier: b,
 	})
-
 	return b.services.RegisterService(is)
-
 }
 
 func (b *BeaconNode) registerRPCService(ctx *cli.Context) error {
@@ -454,9 +474,19 @@ func (b *BeaconNode) registerRPCService(ctx *cli.Context) error {
 		return err
 	}
 
-	var syncService *initialsync.Service
-	if err := b.services.FetchService(&syncService); err != nil {
-		return err
+	var syncService prysmsync.Checker
+	if cfg := featureconfig.Get(); cfg.EnableInitSyncQueue {
+		var initSyncTmp *initialsync.Service
+		if err := b.services.FetchService(&initSyncTmp); err != nil {
+			return err
+		}
+		syncService = initSyncTmp
+	} else {
+		var initSyncTmp *initialsyncold.Service
+		if err := b.services.FetchService(&initSyncTmp); err != nil {
+			return err
+		}
+		syncService = initSyncTmp
 	}
 
 	genesisValidators := ctx.GlobalUint64(flags.InteropNumValidatorsFlag.Name)

beacon-chain/sync/initial-sync-old/BUILD.bazel

@@ -0,0 +1,63 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "log.go",
+        "round_robin.go",
+        "service.go",
+    ],
+    importpath = "github.com/prysmaticlabs/prysm/beacon-chain/sync/initial-sync-old",
+    visibility = ["//beacon-chain:__subpackages__"],
+    deps = [
+        "//beacon-chain/blockchain:go_default_library",
+        "//beacon-chain/core/feed:go_default_library",
+        "//beacon-chain/core/feed/block:go_default_library",
+        "//beacon-chain/core/feed/state:go_default_library",
+        "//beacon-chain/core/helpers:go_default_library",
+        "//beacon-chain/db:go_default_library",
+        "//beacon-chain/flags:go_default_library",
+        "//beacon-chain/p2p:go_default_library",
+        "//beacon-chain/sync:go_default_library",
+        "//proto/beacon/p2p/v1:go_default_library",
+        "//shared:go_default_library",
+        "//shared/bytesutil:go_default_library",
+        "//shared/featureconfig:go_default_library",
+        "//shared/mathutil:go_default_library",
+        "//shared/params:go_default_library",
+        "//shared/roughtime:go_default_library",
+        "@com_github_kevinms_leakybucket_go//:go_default_library",
+        "@com_github_libp2p_go_libp2p_core//peer:go_default_library",
+        "@com_github_paulbellamy_ratecounter//:go_default_library",
+        "@com_github_pkg_errors//:go_default_library",
+        "@com_github_prysmaticlabs_ethereumapis//eth/v1alpha1:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["round_robin_test.go"],
+    embed = [":go_default_library"],
+    race = "on",
+    tags = ["race_on"],
+    deps = [
+        "//beacon-chain/blockchain/testing:go_default_library",
+        "//beacon-chain/core/helpers:go_default_library",
+        "//beacon-chain/db/testing:go_default_library",
+        "//beacon-chain/p2p/peers:go_default_library",
+        "//beacon-chain/p2p/testing:go_default_library",
+        "//beacon-chain/state:go_default_library",
+        "//beacon-chain/sync:go_default_library",
+        "//proto/beacon/p2p/v1:go_default_library",
+        "//shared/hashutil:go_default_library",
+        "//shared/params:go_default_library",
+        "//shared/roughtime:go_default_library",
+        "//shared/sliceutil:go_default_library",
+        "@com_github_kevinms_leakybucket_go//:go_default_library",
+        "@com_github_libp2p_go_libp2p_core//network:go_default_library",
+        "@com_github_prysmaticlabs_ethereumapis//eth/v1alpha1:go_default_library",
+        "@com_github_prysmaticlabs_go_ssz//:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
+    ],
+)

beacon-chain/sync/initial-sync-old/log.go

@@ -0,0 +1,7 @@
+package initialsyncold
+
+import (
+	"github.com/sirupsen/logrus"
+)
+
+var log = logrus.WithField("prefix", "initial-sync")