Productization of the simple dependency manager

liran-funaro · liran-funaro · commit 860d5de80181 · 2025-07-06T12:51:43.000+03:00
Signed-off-by: Liran Funaro &lt;liran.funaro@gmail.com&gt;
diff --git a/.gitignore b/.gitignore
@@ -17,6 +17,9 @@ archbin
 # Output of the go coverage tool, specifically when used with LiteIDE
 *.out
 
+# Profile output.
+*.prof
+
 # Output of the build and experiment runs
 out/
 
diff --git a/Makefile b/Makefile
@@ -151,7 +151,7 @@ bench-loadgen: FORCE
 
 # Run dependency detector benchmarks with added op/sec column.
 bench-dep: FORCE
-	$(go_cmd) test ./service/coordinator/dependencygraph/... -bench "BenchmarkDependencyGraph.*" -run="^$$" | awk -f scripts/bench-tx-per-sec.awk
+	$(go_cmd) test ./service/coordinator/dependencygraph/... -timeout 60m -bench "BenchmarkDependencyGraph.*" -run="^$$" | awk -f scripts/bench-tx-per-sec.awk
 
 # Run dependency detector benchmarks with added op/sec column.
 bench-preparer: FORCE
diff --git a/cmd/config/app_config_test.go b/cmd/config/app_config_test.go
@@ -118,9 +118,8 @@ func TestReadConfigCoordinator(t *testing.T) {
 			Server:     makeServer("localhost", 9001),
 			Monitoring: makeMonitoring("localhost", 2119),
 			DependencyGraphConfig: &coordinator.DependencyGraphConfig{
-				NumOfLocalDepConstructors:       1,
-				WaitingTxsLimit:                 100_000,
-				NumOfWorkersForGlobalDepManager: 1,
+				NumOfLocalDepConstructors: 1,
+				WaitingTxsLimit:           100_000,
 			},
 			ChannelBufferSizePerGoroutine: 10,
 		},
@@ -137,9 +136,8 @@ func TestReadConfigCoordinator(t *testing.T) {
 				Endpoints: []*connection.Endpoint{makeEndpoint("validator-persister", 6001)},
 			},
 			DependencyGraphConfig: &coordinator.DependencyGraphConfig{
-				NumOfLocalDepConstructors:       1,
-				WaitingTxsLimit:                 10_000,
-				NumOfWorkersForGlobalDepManager: 1,
+				NumOfLocalDepConstructors: 1,
+				WaitingTxsLimit:           100_000,
 			},
 			ChannelBufferSizePerGoroutine: 10,
 		},
diff --git a/cmd/config/samples/coordinator.yaml b/cmd/config/samples/coordinator.yaml
@@ -17,8 +17,7 @@ validator-committer:
 
 dependency-graph:
   num-of-local-dep-constructors: 1
-  waiting-txs-limit: 10_000
-  num-of-workers-for-global-dep-manager: 1
+  waiting-txs-limit: 100_000
   per-channel-buffer-size-per-goroutine: 10
 
 logging:
diff --git a/cmd/config/templates/coordinator.yaml b/cmd/config/templates/coordinator.yaml
@@ -21,8 +21,7 @@ validator-committer:
 
 dependency-graph:
   num-of-local-dep-constructors: 1
-  waiting-txs-limit: 10_000
-  num-of-workers-for-global-dep-manager: 1
+  waiting-txs-limit: 100_000
   # Add monitoring configuration here if applicable
 per-channel-buffer-size-per-goroutine: 10
 
diff --git a/cmd/config/viper.go b/cmd/config/viper.go
@@ -20,7 +20,6 @@ func NewViperWithCoordinatorDefaults() *viper.Viper {
 	v := NewViperWithServiceDefault(9001, 2119)
 	v.SetDefault("dependency-graph.num-of-local-dep-constructors", 1)
 	v.SetDefault("dependency-graph.waiting-txs-limit", 100_000)
-	v.SetDefault("dependency-graph.num-of-workers-for-global-dep-manager", 1)
 	v.SetDefault("per-channel-buffer-size-per-goroutine", 10)
 	return v
 }
diff --git a/loadgen/client_test.go b/loadgen/client_test.go
@@ -150,9 +150,8 @@ func TestLoadGenForCoordinator(t *testing.T) {
 				VerifierConfig:           *test.ServerToClientConfig(sigVerServer.Configs...),
 				ValidatorCommitterConfig: *test.ServerToClientConfig(vcServer.Configs...),
 				DependencyGraphConfig: &coordinator.DependencyGraphConfig{
-					NumOfLocalDepConstructors:       1,
-					WaitingTxsLimit:                 10_000,
-					NumOfWorkersForGlobalDepManager: 1,
+					NumOfLocalDepConstructors: 1,
+					WaitingTxsLimit:           100_000,
 				},
 				ChannelBufferSizePerGoroutine: 10,
 			}
diff --git a/service/coordinator/config.go b/service/coordinator/config.go
@@ -24,8 +24,7 @@ type (
 
 	// DependencyGraphConfig is the configuration for dependency graph manager. It contains resource limits.
 	DependencyGraphConfig struct {
-		NumOfLocalDepConstructors       int `mapstructure:"num-of-local-dep-constructors"`
-		WaitingTxsLimit                 int `mapstructure:"waiting-txs-limit"`
-		NumOfWorkersForGlobalDepManager int `mapstructure:"num-of-workers-for-global-dep-manager"`
+		NumOfLocalDepConstructors int `mapstructure:"num-of-local-dep-constructors"`
+		WaitingTxsLimit           int `mapstructure:"waiting-txs-limit"`
 	}
 )
diff --git a/service/coordinator/coordinator.go b/service/coordinator/coordinator.go
@@ -131,7 +131,7 @@ func NewCoordinatorService(c *Config) *Service {
 	metrics := newPerformanceMetrics()
 
 	depMgr := dependencygraph.NewManager(
-		&dependencygraph.Config{
+		&dependencygraph.Parameters{
 			IncomingTxs:               queues.coordinatorToDepGraphTxs,
 			OutgoingDepFreeTxsNode:    queues.depGraphToSigVerifierFreeTxs,
 			IncomingValidatedTxsNode:  queues.vcServiceToDepGraphValidatedTxs,
diff --git a/service/coordinator/coordinator_test.go b/service/coordinator/coordinator_test.go
@@ -75,9 +75,8 @@ func newCoordinatorTestEnv(t *testing.T, tConfig *testConfig) *coordinatorTestEn
 		VerifierConfig:           *test.ServerToClientConfig(svServers.Configs...),
 		ValidatorCommitterConfig: *test.ServerToClientConfig(vcServerConfigs...),
 		DependencyGraphConfig: &DependencyGraphConfig{
-			NumOfLocalDepConstructors:       3,
-			WaitingTxsLimit:                 10,
-			NumOfWorkersForGlobalDepManager: 3,
+			NumOfLocalDepConstructors: 3,
+			WaitingTxsLimit:           10,
 		},
 		ChannelBufferSizePerGoroutine: 2000,
 		Monitoring: monitoring.Config{
diff --git a/service/coordinator/dependencygraph/local_dependency_constructor_test.go b/service/coordinator/dependencygraph/local_dependency_constructor_test.go
@@ -8,6 +8,7 @@ package dependencygraph
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
@@ -299,7 +300,7 @@ func TestLocalDependencyConstructorWithOrder(t *testing.T) {
 func makeTestKeys(_ *testing.T, numKeys int) [][]byte {
 	keys := make([][]byte, numKeys)
 	for i := range numKeys {
-		keys[i] = []byte{byte(i)}
+		keys[i] = fmt.Appendf(nil, "%d", i)
 	}
 	return keys
 }
diff --git a/service/coordinator/dependencygraph/manager.go b/service/coordinator/dependencygraph/manager.go
@@ -12,56 +12,32 @@ import (
 
 	"golang.org/x/sync/errgroup"
 
-	"github.com/hyperledger/fabric-x-committer/utils/monitoring"
 	"github.com/hyperledger/fabric-x-committer/utils/monitoring/promutil"
 )
 
-type (
-	// Manager is the main component of the dependency graph module.
-	// It is responsible for managing the local dependency constructor
-	// and the global dependency manager.
-	Manager struct {
-		localDepConstructor         *localDependencyConstructor
-		globalDepManager            *globalDependencyManager
-		config                      *Config
-		outgoingTxsNodeWithLocalDep chan *transactionNodeBatch
-		metrics                     *perfMetrics
-	}
-
-	// Config holds the configuration for the dependency graph manager.
-	Config struct {
-		// IncomingTxs is the channel for dependency manager to receive
-		// incoming transactions.
-		IncomingTxs <-chan *TransactionBatch
-		// OutgoingDepFreeTxsNode is the channel dependency manager to send
-		// dependency free transactions for validation and commit.
-		OutgoingDepFreeTxsNode chan<- TxNodeBatch
-		// IncomingValidatedTxsNode is the channel for dependency manager
-		// to receive validated transactions.
-		IncomingValidatedTxsNode <-chan TxNodeBatch
-		// NumOfLocalDepConstructors defines the number of local
-		// dependency constructors.
-		NumOfLocalDepConstructors int
-		// WaitingTxsLimit defines the maximum number of transactions
-		// that can be waiting at the dependency manager.
-		WaitingTxsLimit int
-		// PrometheusMetricsProvider is the provider for Prometheus metrics.
-		PrometheusMetricsProvider *monitoring.Provider
-	}
-)
+// Manager is the main component of the dependency graph module.
+// It is responsible for managing the local dependency constructor
+// and the global dependency manager.
+type Manager struct {
+	localDepConstructor         *localDependencyConstructor
+	globalDepManager            *globalDependencyManager
+	parameters                  *Parameters
+	outgoingTxsNodeWithLocalDep chan *transactionNodeBatch
+	metrics                     *perfMetrics
+}
 
 // NewManager creates a new dependency graph manager.
-func NewManager(c *Config) *Manager {
-	metrics := newPerformanceMetrics(c.PrometheusMetricsProvider)
+func NewManager(p *Parameters) *Manager {
+	metrics := newPerformanceMetrics(p.PrometheusMetricsProvider)
 
-	outgoingTxsNodeWithLocalDep := make(chan *transactionNodeBatch, cap(c.IncomingTxs))
-	ldp := newLocalDependencyConstructor(c.IncomingTxs, outgoingTxsNodeWithLocalDep, metrics)
+	outgoingTxsNodeWithLocalDep := make(chan *transactionNodeBatch, cap(p.IncomingTxs))
+	ldp := newLocalDependencyConstructor(p.IncomingTxs, outgoingTxsNodeWithLocalDep, metrics)
 
 	gdConf := &globalDepConfig{
 		incomingTxsNode:        outgoingTxsNodeWithLocalDep,
-		outgoingDepFreeTxsNode: c.OutgoingDepFreeTxsNode,
-		validatedTxsNode:       c.IncomingValidatedTxsNode,
-		waitingTxsLimit:        c.WaitingTxsLimit,
+		outgoingDepFreeTxsNode: p.OutgoingDepFreeTxsNode,
+		validatedTxsNode:       p.IncomingValidatedTxsNode,
+		waitingTxsLimit:        p.WaitingTxsLimit,
 		metrics:                metrics,
 	}
 
@@ -70,7 +46,7 @@ func NewManager(c *Config) *Manager {
 	return &Manager{
 		localDepConstructor:         ldp,
 		globalDepManager:            gdp,
-		config:                      c,
+		parameters:                  p,
 		outgoingTxsNodeWithLocalDep: outgoingTxsNodeWithLocalDep,
 		metrics:                     metrics,
 	}
@@ -87,7 +63,7 @@ func (m *Manager) Run(ctx context.Context) {
 	})
 
 	g.Go(func() error {
-		m.localDepConstructor.run(gCtx, m.config.NumOfLocalDepConstructors)
+		m.localDepConstructor.run(gCtx, m.parameters.NumOfLocalDepConstructors)
 		return nil
 	})
 
diff --git a/service/coordinator/dependencygraph/manager_test.go b/service/coordinator/dependencygraph/manager_test.go
diff --git a/service/coordinator/dependencygraph/parameters.go b/service/coordinator/dependencygraph/parameters.go
diff --git a/service/coordinator/dependencygraph/simple.go b/service/coordinator/dependencygraph/simple.go
diff --git a/service/coordinator/dependencygraph/transaction_node.go b/service/coordinator/dependencygraph/transaction_node.go
diff --git a/service/coordinator/validator_committer_manager_test.go b/service/coordinator/validator_committer_manager_test.go
diff --git a/utils/signature/sigtest/signer.go b/utils/signature/sigtest/signer.go

Original file line number	Diff line number	Diff line change
`@@ -20,7 +20,6 @@ func NewViperWithCoordinatorDefaults() *viper.Viper {`
`20`	`20`	`v := NewViperWithServiceDefault(9001, 2119)`
`21`	`21`	`v.SetDefault("dependency-graph.num-of-local-dep-constructors", 1)`
`22`	`22`	`v.SetDefault("dependency-graph.waiting-txs-limit", 100_000)`
`23`		`- v.SetDefault("dependency-graph.num-of-workers-for-global-dep-manager", 1)`
`24`	`23`	`v.SetDefault("per-channel-buffer-size-per-goroutine", 10)`
`25`	`24`	`return v`
`26`	`25`	`}`
Original file line number	Diff line number	Diff line change
`@@ -24,8 +24,7 @@ type (`
`24`	`24`
`25`	`25`	`// DependencyGraphConfig is the configuration for dependency graph manager. It contains resource limits.`
`26`	`26`	`DependencyGraphConfig struct {`
`27`		- NumOfLocalDepConstructors int `mapstructure:"num-of-local-dep-constructors"`
`28`		- WaitingTxsLimit int `mapstructure:"waiting-txs-limit"`
`29`		- NumOfWorkersForGlobalDepManager int `mapstructure:"num-of-workers-for-global-dep-manager"`
	`27`	+ NumOfLocalDepConstructors int `mapstructure:"num-of-local-dep-constructors"`
	`28`	+ WaitingTxsLimit int `mapstructure:"waiting-txs-limit"`
`30`	`29`	`}`
`31`	`30`	`)`