DEV: Config

eth/tracers/firehose.go

@@ -109,7 +109,7 @@ func NewTracingHooksFromFirehose(tracer *Firehose) *tracing.Hooks {
 
 type FirehoseConfig struct {
 	ApplyBackwardCompatibility *bool `json:"applyBackwardCompatibility"`
-	ConcurrentBlockFlushing    bool  `json:"concurrentBlockFlushing"`
+	ConcurrentBlockFlushing    int   `json:"concurrentBlockFlushing"`
 
 	// Only used for testing, only possible through JSON configuration
 	private *privateFirehoseConfig
@@ -161,7 +161,7 @@ type Firehose struct {
 	// here. If not set in the config, then we inspect `OnBlockchainInit` the chain config to determine
 	// if it's a network for which we must reproduce the legacy bugs.
 	applyBackwardCompatibility *bool
-	concurrentBlockFlushing    bool
+	concurrentBlockFlushing    int
 
 	// Block state
 	block                       *pbeth.Block
@@ -264,13 +264,13 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
 		}
 	}
 
-	if config.ConcurrentBlockFlushing {
+	if config.ConcurrentBlockFlushing > 0 {
-		log.Info("Firehose concurrent block flushing enabled, starting block " +
+		log.Info("Firehose concurrent block flushing enabled, starting", config.ConcurrentBlockFlushing,
-			"print worker goroutine")
+			"block print worker goroutine")
-		const numWorkers = 10                                     // TODO: This becomes a parameter
+
 		firehose.blockPrintQueue = make(chan *blockPrintJob, 100) // TODO: Optimal buffer size tbd
 		firehose.blockOutputQueue = make(chan *blockOutput, 100)
-		for i := 0; i < numWorkers; i++ {
+		for i := 0; i < config.ConcurrentBlockFlushing; i++ {
 			firehose.blockFlushDone.Add(1)
 			go firehose.blockPrintWorker()
 		}
@@ -545,7 +545,7 @@ func (f *Firehose) OnBlockEnd(err error) {
 		f.ensureInBlockAndNotInTrx()
 
 		// Flush block to firehose and optionally use goroutine
-		if f.concurrentBlockFlushing {
+		if f.concurrentBlockFlushing > 0 {
 			f.startBlockCond.L.Lock()
 			if !f.startBlockInitialized {
 				f.startBlockNum = f.block.Number
@@ -681,7 +681,7 @@ func (f *Firehose) reorderCallOrdinals(call *pbeth.Call, ordinalBase uint64) (or
 }
 
 func (f *Firehose) OnClose() {
-	if f.concurrentBlockFlushing {
+	if f.concurrentBlockFlushing > 0 {
 		log.Info("Firehose closing, flushing queued blocks to standard output")
 		f.CloseBlockPrintQueue()
 	}
@@ -1946,7 +1946,7 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina
 
 	buf.WriteString("\n")
 
-	if f.concurrentBlockFlushing {
+	if f.concurrentBlockFlushing > 0 {
 		f.blockOutputQueue <- &blockOutput{
 			blockNum: block.Number,
 			data:     buf.Bytes(),
@@ -2904,7 +2904,7 @@ func (f *Firehose) blockPrintWorker() {
 // It blocks until all concurrent block flushing operations are completed, ensuring a clean
 // shutdown of the printing pipeline.
 func (f *Firehose) CloseBlockPrintQueue() {
-	if f.concurrentBlockFlushing {
+	if f.concurrentBlockFlushing > 0 {
 		f.closeChannels.Do(func() {
 			close(f.blockPrintQueue)
 			f.blockFlushDone.Wait()

eth/tracers/firehose_concurrency_test.go

@@ -15,7 +15,7 @@ import (
 func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) {
 
 	f := NewFirehose(&FirehoseConfig{
-		ConcurrentBlockFlushing:    true,
+		ConcurrentBlockFlushing:    1,
 		ApplyBackwardCompatibility: ptr(false),
 		private: &privateFirehoseConfig{
 			FlushToTestBuffer: true,
@@ -58,7 +58,7 @@ func TestFirehose_BlocksPrintToFirehose_MultipleBlocksInOrder(t *testing.T) {
 	const baseBlockNum = 0
 
 	f := NewFirehose(&FirehoseConfig{
-		ConcurrentBlockFlushing:    true,
+		ConcurrentBlockFlushing:    1,
 		ApplyBackwardCompatibility: ptr(false),
 		private: &privateFirehoseConfig{
 			FlushToTestBuffer: true,

eth/tracers/internal/tracetest/firehose/firehose_test.go

@@ -40,11 +40,11 @@ func TestFirehosePrestate(t *testing.T) {
 		"./testdata/TestFirehosePrestate/extra_account_creations",
 	}
 
-	for _, concurrent := range []bool{true, false} {
+	for _, concurrent := range []int{0, 1} {
 		for _, folder := range testFolders {
 			name := filepath.Base(folder)
 			concurrencyLabel := "sequential"
-			if concurrent {
+			if concurrent == 1 {
 				concurrencyLabel = "concurrent"
 			}
 
@@ -199,9 +199,9 @@ func TestFirehose_SystemCalls(t *testing.T) {
 func testBlockTracesCorrectly(t *testing.T, genesisSpec *core.Genesis, engine consensus.Engine, blocks []*types.Block, goldenDir string) {
 	t.Helper()
 
-	for _, concurrent := range []bool{true, false} {
+	for _, concurrent := range []int{0, 1} {
 		concurrencyLabel := "sequential"
-		if concurrent {
+		if concurrent == 1 {
 			concurrencyLabel = "concurrent"
 		}
 

eth/tracers/internal/tracetest/firehose/helpers_test.go

@@ -33,7 +33,7 @@ func newFirehoseTestTracer(t *testing.T, model tracingModel, config *tracers.Fir
 	t.Helper()
 
 	tracer, err := tracers.NewFirehoseFromRawJSON([]byte(fmt.Sprintf(`{
-		"concurrentBlockFlushing": %t,
+		"concurrentBlockFlushing": %d,
 		"_private": {
 			"flushToTestBuffer": true,
 			"ignoreGenesisBlock": true,