diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index e77507a477..de1da76c17 100644 --- a/eth/tracers/firehose.go +++ b/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 { - log.Info("Firehose concurrent block flushing enabled, starting block " + - "print worker goroutine") - const numWorkers = 10 // TODO: This becomes a parameter + if config.ConcurrentBlockFlushing > 0 { + log.Info("Firehose concurrent block flushing enabled, starting", config.ConcurrentBlockFlushing, + "block print worker goroutine") + 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() diff --git a/eth/tracers/firehose_concurrency_test.go b/eth/tracers/firehose_concurrency_test.go index f50bbc5697..6c356cf6a1 100644 --- a/eth/tracers/firehose_concurrency_test.go +++ b/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, diff --git a/eth/tracers/internal/tracetest/firehose/firehose_test.go b/eth/tracers/internal/tracetest/firehose/firehose_test.go index 1d7d3fd248..d4b0874018 100644 --- a/eth/tracers/internal/tracetest/firehose/firehose_test.go +++ b/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" } diff --git a/eth/tracers/internal/tracetest/firehose/helpers_test.go b/eth/tracers/internal/tracetest/firehose/helpers_test.go index e7f4a4017e..0dde5e7058 100644 --- a/eth/tracers/internal/tracetest/firehose/helpers_test.go +++ b/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,