statediff: Use a worker pool

cmd/geth/config.go

@@ -191,8 +191,17 @@ func makeFullNode(ctx *cli.Context) (*node.Node, ethapi.Backend) {
 			} else {
 				utils.Fatalf("Must specify client name for statediff DB output")
 			}
+		} else {
+			if ctx.GlobalBool(utils.StateDiffWritingFlag.Name) {
+				utils.Fatalf("Must pass DB parameters if enabling statediff write loop")
+			}
+		}
+		params := statediff.ServiceParams{
+			DBParams:        dbParams,
+			EnableWriteLoop: ctx.GlobalBool(utils.StateDiffWritingFlag.Name),
+			NumWorkers:      ctx.GlobalUint(utils.StateDiffWorkersFlag.Name),
 		}
-		utils.RegisterStateDiffService(stack, backend, dbParams, ctx.GlobalBool(utils.StateDiffWritingFlag.Name))
+		utils.RegisterStateDiffService(stack, backend, params)
 	}
 
 	// Configure GraphQL if requested

cmd/geth/main.go

@@ -162,6 +162,7 @@ var (
 		utils.StateDiffDBNodeIDFlag,
 		utils.StateDiffDBClientNameFlag,
 		utils.StateDiffWritingFlag,
+		utils.StateDiffWorkersFlag,
 		configFileFlag,
 	}
 

cmd/geth/usage.go

@@ -242,6 +242,7 @@ var AppHelpFlagGroups = []flags.FlagGroup{
 			utils.StateDiffDBNodeIDFlag,
 			utils.StateDiffDBClientNameFlag,
 			utils.StateDiffWritingFlag,
+			utils.StateDiffWorkersFlag,
 		},
 	},
 	{

cmd/utils/flags.go

@@ -746,6 +746,10 @@ var (
 		Name:  "statediff.writing",
 		Usage: "Activates progressive writing of state diffs to database as new block are synced",
 	}
+	StateDiffWorkersFlag = cli.UintFlag{
+		Name:  "statediff.workers",
+		Usage: "Number of concurrent workers to use during statediff processing (0 = 1)",
+	}
 )
 
 // MakeDataDir retrieves the currently requested data directory, terminating
@@ -1744,9 +1748,8 @@ func RegisterGraphQLService(stack *node.Node, backend ethapi.Backend, cfg node.C
 }
 
 // RegisterStateDiffService configures and registers a service to stream state diff data over RPC
-// dbParams are: Postgres connection URI, Node ID, client name
-func RegisterStateDiffService(stack *node.Node, ethServ *eth.Ethereum, dbParams *statediff.DBParams, startWriteLoop bool) {
-	if err := statediff.New(stack, ethServ, dbParams, startWriteLoop); err != nil {
+func RegisterStateDiffService(stack *node.Node, ethServ *eth.Ethereum, params statediff.ServiceParams) {
+	if err := statediff.New(stack, ethServ, params); err != nil {
 		Fatalf("Failed to register the Statediff service: %v", err)
 	}
 }

statediff/indexer/metrics.go

@@ -8,17 +8,17 @@ import (
 )
 
 const (
-	indexerNamespace = "indexer"
+	namespace = "statediff"
 )
 
 // Build a fully qualified metric name
 func metricName(subsystem, name string) string {
 	if name == "" {
 		return ""
 	}
-	parts := []string{indexerNamespace, name}
+	parts := []string{namespace, name}
 	if subsystem != "" {
-		parts = []string{indexerNamespace, subsystem, name}
+		parts = []string{namespace, subsystem, name}
 	}
 	// Prometheus uses _ but geth metrics uses / and replaces
 	return strings.Join(parts, "/")
@@ -57,7 +57,7 @@ func RegisterIndexerMetrics(reg metrics.Registry) indexerMetricsHandles {
 		tTxAndRecProcessing:       metrics.NewTimer(),
 		tStateStoreCodeProcessing: metrics.NewTimer(),
 	}
-	subsys := "" // todo
+	subsys := "indexer"
 	reg.Register(metricName(subsys, "blocks"), ctx.blocks)
 	reg.Register(metricName(subsys, "transactions"), ctx.transactions)
 	reg.Register(metricName(subsys, "receipts"), ctx.receipts)

statediff/metrics.go

@@ -0,0 +1,54 @@
+package statediff
+
+import (
+	"strings"
+
+	"github.com/ethereum/go-ethereum/metrics"
+)
+
+const (
+	namespace = "statediff"
+)
+
+// Build a fully qualified metric name
+func metricName(subsystem, name string) string {
+	if name == "" {
+		return ""
+	}
+	parts := []string{namespace, name}
+	if subsystem != "" {
+		parts = []string{namespace, subsystem, name}
+	}
+	// Prometheus uses _ but geth metrics uses / and replaces
+	return strings.Join(parts, "/")
+}
+
+type statediffMetricsHandles struct {
+	// Height of latest synced by core.BlockChain
+	// FIXME
+	lastSyncHeight metrics.Gauge
+	// Height of the latest block received from chainEvent channel
+	lastEventHeight metrics.Gauge
+	// Height of latest state diff
+	lastStatediffHeight metrics.Gauge
+	// Current length of chainEvent channels
+	serviceLoopChannelLen metrics.Gauge
+	writeLoopChannelLen   metrics.Gauge
+}
+
+func RegisterStatediffMetrics(reg metrics.Registry) statediffMetricsHandles {
+	ctx := statediffMetricsHandles{
+		lastSyncHeight:        metrics.NewGauge(),
+		lastEventHeight:       metrics.NewGauge(),
+		lastStatediffHeight:   metrics.NewGauge(),
+		serviceLoopChannelLen: metrics.NewGauge(),
+		writeLoopChannelLen:   metrics.NewGauge(),
+	}
+	subsys := "service"
+	reg.Register(metricName(subsys, "last_sync_height"), ctx.lastSyncHeight)
+	reg.Register(metricName(subsys, "last_event_height"), ctx.lastEventHeight)
+	reg.Register(metricName(subsys, "last_statediff_height"), ctx.lastStatediffHeight)
+	reg.Register(metricName(subsys, "service_loop_channel_len"), ctx.serviceLoopChannelLen)
+	reg.Register(metricName(subsys, "write_loop_channel_len"), ctx.writeLoopChannelLen)
+	return ctx
+}