From 7eaa0cb83b16f996bf34c48eec3752ef91f38892 Mon Sep 17 00:00:00 2001 From: David Zhou Date: Tue, 13 May 2025 15:01:56 -0400 Subject: [PATCH] PR: PR comments addressed Part 1 --- eth/tracers/firehose.go | 39 ++++++++++--------- eth/tracers/firehose_concurrency.md | 21 +++++----- eth/tracers/firehose_concurrency_test.go | 25 ++---------- .../tracetest/firehose/firehose_test.go | 3 +- 4 files changed, 35 insertions(+), 53 deletions(-) diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index 19aeea5a19..27d1aea327 100644 --- a/eth/tracers/firehose.go +++ b/eth/tracers/firehose.go @@ -144,13 +144,14 @@ func (c *FirehoseConfig) ForcedBackwardCompatibility() bool { type Firehose struct { // Global state - outputBuffer *bytes.Buffer - initSent *atomic.Bool - config *FirehoseConfig - chainConfig *params.ChainConfig - hasher crypto.KeccakState // Keccak256 hasher instance shared across tracer needs (non-concurrent safe) - hasherBuf common.Hash // Keccak256 hasher result array shared across tracer needs (non-concurrent safe) - tracerID string + outputBuffer *bytes.Buffer + initSent *atomic.Bool + config *FirehoseConfig + chainConfig *params.ChainConfig + hasher crypto.KeccakState // Keccak256 hasher instance shared across tracer needs (non-concurrent safe) + hasherBuf common.Hash // Keccak256 hasher result array shared across tracer needs (non-concurrent safe) + tracerID string + closeChannels sync.Once // The FirehoseTracer is used in multiple chains, some for which were produced using a legacy version // of the whole tracing infrastructure. This legacy version had many small bugs here and there that // we must "reproduce" on some chain to ensure that the FirehoseTracer produces the same output @@ -173,6 +174,8 @@ type Firehose struct { blockReorderOrdinalSnapshot uint64 blockReorderOrdinalOnce sync.Once blockIsGenesis bool + blockPrintQueue chan *blockPrintJob + blockFlushDone sync.WaitGroup // Transaction state evm *tracing.VMContext @@ -190,11 +193,6 @@ type Firehose struct { // Testing state, only used in tests and private configs testingBuffer *bytes.Buffer testingIgnoreGenesisBlock bool - - // Worker queue for block printing - blockPrintQueue chan *blockPrintJob - flushDone sync.WaitGroup - closeOnce sync.Once } const FirehoseProtocolVersion = "3.0" @@ -258,9 +256,10 @@ func NewFirehose(config *FirehoseConfig) *Firehose { } if config.ConcurrentBlockFlushing { - log.Info("Concurrent block flushing enabled: starting goroutine...") + log.Info("Firehose concurrent block flushing enabled, starting block " + + "print worker goroutine") firehose.blockPrintQueue = make(chan *blockPrintJob, 100) - firehose.flushDone.Add(1) + firehose.blockFlushDone.Add(1) go firehose.blockPrintWorker() } @@ -630,9 +629,8 @@ func (f *Firehose) reorderCallOrdinals(call *pbeth.Call, ordinalBase uint64) (or } func (f *Firehose) OnClose() { - log.Info("Firehose closing...") if f.concurrentBlockFlushing { - log.Info("Closing channel: flushing the remaining blocks to firehose") + log.Info("Firehose closing, flushing queued blocks to standard output") f.CloseBlockPrintQueue() } } @@ -2835,17 +2833,20 @@ type blockPrintJob struct { } func (f *Firehose) blockPrintWorker() { - defer f.flushDone.Done() + defer f.blockFlushDone.Done() for job := range f.blockPrintQueue { f.printBlockToFirehose(job.block, job.finality) } } +// CloseBlockPrintQueue signals block printing goroutines to shut down and waits for them. +// 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 { - f.closeOnce.Do(func() { + f.closeChannels.Do(func() { close(f.blockPrintQueue) - f.flushDone.Wait() + f.blockFlushDone.Wait() }) } } diff --git a/eth/tracers/firehose_concurrency.md b/eth/tracers/firehose_concurrency.md index cb395750c2..a47734595b 100644 --- a/eth/tracers/firehose_concurrency.md +++ b/eth/tracers/firehose_concurrency.md @@ -84,6 +84,12 @@ The implementation was validated at multiple levels to ensure correctness, confi ## Section 4: Analysis +The specifications of the operating system used for testing are as follows: + +Model: Macbook Air \ +Processor: Apple M1 chip \ +Memory: 8 GB + ### 4.1 Results The following outlines the results for metric three: @@ -92,18 +98,14 @@ The following outlines the results for metric three: **No concurrency** -Run 1: 71.49s user 15.56s system 56% cpu 2:34.28 total - -Run 2: 69.77s user 15.60s system 54% cpu 2:37.73 total - +Run 1: 71.49s user 15.56s system 56% cpu 2:34.28 total\ +Run 2: 69.77s user 15.60s system 54% cpu 2:37.73 total\ Run 3: 68.81s user 15.12s system 53% cpu 2:38.18 total **Concurrency** -Run 1: 69.61s user 14.97s system 55% cpu 2:32.85 total - -Run 2: 67.94s user 15.13s system 54% cpu 2:33.59 total - +Run 1: 69.61s user 14.97s system 55% cpu 2:32.85 total\ +Run 2: 67.94s user 15.13s system 54% cpu 2:33.59 total\ Run 3: 68.60s user 16.37s system 48% cpu 2:54.22 total (Not sure what happened here) **Until Block 100000** @@ -133,8 +135,7 @@ Run 3: 68.60s user 16.37s system 48% cpu 2:54.22 total (Not sure what happened h ### 4.2 Discussion -Run 3 with concurrency seems to be an outlier. Without it, the general trend would be that every block saves around 0.0006 second, or 0.6 millisecond. - +Run 3 with concurrency seems to be an outlier. Without it, the general trend would be that every block saves around 0.0006 second, or 0.6 millisecond. \ User seems slightly lower, whereas system and cpu are relatively the same. ## Section 5: Conclusion diff --git a/eth/tracers/firehose_concurrency_test.go b/eth/tracers/firehose_concurrency_test.go index 691fa00931..78ae9bc95e 100644 --- a/eth/tracers/firehose_concurrency_test.go +++ b/eth/tracers/firehose_concurrency_test.go @@ -102,7 +102,9 @@ func TestFirehose_BlocksPrintToFirehose_MultipleBlocksInOrder(t *testing.T) { "Expected %d blocks in output, found %d", blockCount, len(extractedBlocks)) // Verify blocks in order - verifyBlockSequence(t, extractedBlocks, baseBlockNum) + for i, block := range extractedBlocks { + require.Equal(t, baseBlockNum+uint64(i), block.number, "Blocks out of order at position %d", i) + } // Verify block hashes for _, block := range extractedBlocks { @@ -147,24 +149,3 @@ func extractBlocksFromOutput(t *testing.T, output string) []extractedBlock { return blocks } - -func verifyBlockSequence(t *testing.T, blocks []extractedBlock, baseBlockNum uint64) { - t.Helper() - - // First block should be the base block number - require.Equal(t, baseBlockNum, blocks[0].number, - "First block should be %d, got %d", baseBlockNum, blocks[0].number) - - // Last block should be base + count - 1 - expectedLast := baseBlockNum + uint64(len(blocks)) - 1 - require.Equal(t, expectedLast, blocks[len(blocks)-1].number, - "Last block should be %d, got %d", expectedLast, blocks[len(blocks)-1].number) - - // Verify sequence - for i := 0; i < len(blocks)-1; i++ { - current := blocks[i].number - next := blocks[i+1].number - require.Equal(t, current+1, next, - "Blocks out of order at position %d: %d followed by %d", i, current, next) - } -} diff --git a/eth/tracers/internal/tracetest/firehose/firehose_test.go b/eth/tracers/internal/tracetest/firehose/firehose_test.go index e004105a06..59959805b4 100644 --- a/eth/tracers/internal/tracetest/firehose/firehose_test.go +++ b/eth/tracers/internal/tracetest/firehose/firehose_test.go @@ -37,8 +37,7 @@ func TestFirehosePrestate(t *testing.T) { "./testdata/TestFirehosePrestate/suicide_double_withdraw", "./testdata/TestFirehosePrestate/extra_account_creations", } - - // TODO: have goroutine config on AND off + for _, concurrent := range []bool{true, false} { for _, folder := range testFolders { name := filepath.Base(folder)