Node: Multithreaded processor

Author: bruce-riley

node/cmd/guardiand/node.go

@@ -240,6 +240,8 @@ var (
 	chainGovernorEnabled      *bool
 	governorFlowCancelEnabled *bool
 
+	processorWorkerFactor *float64
+
 	ccqEnabled           *bool
 	ccqAllowedRequesters *string
 	ccqP2pPort           *uint
@@ -450,6 +452,7 @@ func init() {
 
 	chainGovernorEnabled = NodeCmd.Flags().Bool("chainGovernorEnabled", false, "Run the chain governor")
 	governorFlowCancelEnabled = NodeCmd.Flags().Bool("governorFlowCancelEnabled", false, "Enable flow cancel on the governor")
+	processorWorkerFactor = NodeCmd.Flags().Float64("processorWorkerFactor", 0.0, "Multiplied by the number of available CPUs on the system to determine the number of workers that the processor uses. 0.0 means single worker")
 
 	ccqEnabled = NodeCmd.Flags().Bool("ccqEnabled", false, "Enable cross chain query support")
 	ccqAllowedRequesters = NodeCmd.Flags().String("ccqAllowedRequesters", "", "Comma separated list of signers allowed to submit cross chain queries")
@@ -1639,7 +1642,7 @@ func runNode(cmd *cobra.Command, args []string) {
 		node.GuardianOptionAdminService(*adminSocketPath, ethRPC, ethContract, rpcMap),
 		node.GuardianOptionP2P(p2pKey, *p2pNetworkID, *p2pBootstrap, *nodeName, *subscribeToVAAs, *disableHeartbeatVerify, *p2pPort, *ccqP2pBootstrap, *ccqP2pPort, *ccqAllowedPeers, *gossipAdvertiseAddress, ibc.GetFeatures),
 		node.GuardianOptionStatusServer(*statusAddr),
-		node.GuardianOptionProcessor(*p2pNetworkID),
+		node.GuardianOptionProcessor(*p2pNetworkID, *processorWorkerFactor),
 	}
 
 	if shouldStart(publicGRPCSocketPath) {

node/pkg/node/node_test.go

@@ -14,6 +14,7 @@ import (
 	"net/http"
 	"os"
 	"regexp"
+	"runtime"
 	"strconv"
 	"strings"
 	"testing"
@@ -196,7 +197,7 @@ func mockGuardianRunnable(t testing.TB, gs []*mockGuardian, mockGuardianIndex ui
 			GuardianOptionPublicWeb(cfg.publicWeb, cfg.publicSocket, "", false, ""),
 			GuardianOptionAdminService(cfg.adminSocket, nil, nil, rpcMap),
 			GuardianOptionStatusServer(fmt.Sprintf("[::]:%d", cfg.statusPort)),
-			GuardianOptionProcessor(networkID),
+			GuardianOptionProcessor(networkID, 3.0/float64(runtime.NumCPU())), // Create three workers.
 		}
 
 		guardianNode := NewGuardianNode(

node/pkg/node/options.go

@@ -575,7 +575,7 @@ func GuardianOptionDatabase(db *db.Database) *GuardianOption {
 
 // GuardianOptionProcessor enables the default processor, which is required to make consensus on messages.
 // Dependencies: db, governor, accountant
-func GuardianOptionProcessor(networkId string) *GuardianOption {
+func GuardianOptionProcessor(networkId string, workerFactor float64) *GuardianOption {
 	return &GuardianOption{
 		name: "processor",
 		// governor and accountant may be set to nil, but that choice needs to be made before the processor is configured
@@ -600,6 +600,7 @@ func GuardianOptionProcessor(networkId string) *GuardianOption {
 				g.acctC.readC,
 				g.gatewayRelayer,
 				networkId,
+				workerFactor,
 			).Run
 
 			return nil

node/pkg/processor/benchmark_test.go

@@ -174,11 +174,10 @@ func createProcessorForTest(b *testing.B, numVAAs int, ctx context.Context, db *
 		gossipAttestationSendC: pd.gossipAttestationSendC,
 		gossipVaaSendC:         pd.gossipVaaSendC,
 		guardianSigner:         ourSigner,
-		gs:                     gs,
 		gst:                    gst,
 		db:                     db,
 		logger:                 logger,
-		state:                  &aggregationState{observationMap{}},
+		state:                  &aggregationState{signatures: observationMap{}},
 		ourAddr:                crypto.PubkeyToAddress(ourSigner.PublicKey()),
 		pythnetVaas:            make(map[string]PythNetVaaEntry),
 		updatedVAAs:            make(map[string]*updateVaaEntry),