DEV: Added configuration

eth/tracers/firehose.go

@@ -109,7 +109,6 @@ func NewTracingHooksFromFirehose(tracer *Firehose) *tracing.Hooks {
 
 type FirehoseConfig struct {
 	ApplyBackwardCompatibility *bool `json:"applyBackwardCompatibility"`
-	// TODO: Add LOGGING on init
 	ConcurrentBlockFlushing    bool  `json:"concurrentBlockFlushing"`
 
 	// Only used for testing, only possible through JSON configuration
@@ -131,6 +130,7 @@ func (c *FirehoseConfig) LogKeyValues() []any {
 
 	return []any{
 		"config.applyBackwardCompatibility", applyBackwardCompatibility,
+		"config.concurrentBlockFlushing", c.ConcurrentBlockFlushing,
 	}
 }
 
@@ -160,6 +160,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
 
 	// Block state
 	block                       *pbeth.Block
@@ -233,6 +234,7 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
 		hasher:                     crypto.NewKeccakState(),
 		tracerID:                   "global",
 		applyBackwardCompatibility: config.ApplyBackwardCompatibility,
+		concurrentBlockFlushing:    config.ConcurrentBlockFlushing,
 
 		// Block state
 		blockOrdinal:        &Ordinal{},
@@ -246,8 +248,6 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
 		callStack:               NewCallStack(),
 		deferredCallState:       NewDeferredCallState(),
 		latestCallEnterSuicided: false,
-
-		blockPrintQueue: make(chan *blockPrintJob, 100),
 	}
 
 	if config.private != nil {
@@ -257,8 +257,12 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
 		}
 	}
 
+	if config.ConcurrentBlockFlushing {
+		log.Info("Concurrent block flushing enabled: starting goroutine...")
+		firehose.blockPrintQueue = make(chan *blockPrintJob, 100)
 		firehose.flushDone.Add(1)
 		go firehose.blockPrintWorker()
+	}
 
 	return firehose
 }
@@ -496,12 +500,17 @@ func (f *Firehose) OnBlockEnd(err error) {
 		}
 
 		f.ensureInBlockAndNotInTrx()
-		// f.printBlockToFirehose(f.block, f.blockFinality)
+
+		// Flush block to firehose and optionally use goroutine
+		if f.concurrentBlockFlushing {
 			job := &blockPrintJob{
 				block:    f.block,
 				finality: f.blockFinality,
 			}
 			f.blockPrintQueue <- job
+		} else {
+			f.printBlockToFirehose(f.block, f.blockFinality)
+		}
 
 	} else {
 		// An error occurred, could have happen in transaction/call context, we must not check if in trx/call, only check in block
@@ -621,9 +630,12 @@ func (f *Firehose) reorderCallOrdinals(call *pbeth.Call, ordinalBase uint64) (or
 }
 
 func (f *Firehose) OnClose() {
-	log.Info("Firehose closing: waiting for worker goroutines to finish and shutting down channels")
+	log.Info("Firehose closing...")
+	if f.concurrentBlockFlushing {
+		log.Info("waiting for worker goroutines to finish and shutting down channels")
 		f.CloseBlockPrintQueue()
 	}
+}
 
 func (f *Firehose) OnSystemCallStart() {
 	firehoseInfo("system call start")
@@ -2829,8 +2841,11 @@ func (f *Firehose) blockPrintWorker() {
 }
 
 func (f *Firehose) CloseBlockPrintQueue() {
+	if f.concurrentBlockFlushing {
 		f.closeOnce.Do(func() {
+			log.Info("Closing channel: flushing the remaining blocks to firehose")
 			close(f.blockPrintQueue)
 			f.flushDone.Wait()
 		})
 	}
+}

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

@@ -56,7 +56,6 @@ func TestFirehosePrestate(t *testing.T) {
 				blockLines.assertOnlyBlockEquals(t, filepath.Join(folder, string(model)), 1)
 			})
 		}
-
 	}
 
 }

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

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