From 326a3ee583eb906314d364c7bfc9b91c112a49d3 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 # Conflicts: # eth/tracers/firehose.go # eth/tracers/internal/tracetest/firehose/firehose_test.go --- eth/tracers/firehose.go | 104 +++++++---------------- eth/tracers/firehose_concurrency.md | 21 ++--- eth/tracers/firehose_concurrency_test.go | 25 +----- 3 files changed, 46 insertions(+), 104 deletions(-) diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index d09392fb04..d6e985a305 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,13 +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 - blockOutputQueue chan *blockOutput - flushDone sync.WaitGroup - flushOutputDone sync.WaitGroup - closeOnce sync.Once } const FirehoseProtocolVersion = "3.0" @@ -260,38 +256,11 @@ func NewFirehose(config *FirehoseConfig) *Firehose { } if config.ConcurrentBlockFlushing { - log.Info("Concurrent block flushing enabled: starting 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++ { - firehose.flushDone.Add(1) - go firehose.blockPrintWorker() - } - - // Output channel to order the flushing linearly - firehose.flushOutputDone.Add(1) - go func() { - defer firehose.flushOutputDone.Done() - - expected := uint64(0) - buffer := make(map[uint64][]byte) - - for result := range firehose.blockOutputQueue { - buffer[result.blockNum] = result.data - - for { - data, ok := buffer[expected] - if !ok { - break - } - firehose.flushToFirehose(data) - delete(buffer, expected) - expected++ - } - } - }() + log.Info("Firehose concurrent block flushing enabled, starting block " + + "print worker goroutine") + firehose.blockPrintQueue = make(chan *blockPrintJob, 100) + firehose.blockFlushDone.Add(1) + go firehose.blockPrintWorker() } return firehose @@ -660,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() } } @@ -1857,6 +1825,9 @@ func (f *Firehose) isChainOneOf(chainIDs ...*big.Int) bool { } func isChainIDOneOf(actualChainID *big.Int, chainIDs ...*big.Int) bool { + if actualChainID == nil { + return false + } for _, chainID := range chainIDs { if actualChainID.Cmp(chainID) == 0 { return true @@ -1885,13 +1856,14 @@ func (f *Firehose) panicInvalidState(msg string, callerSkip int) string { // printBlockToFirehose is a helper function to print a block to Firehose protocl format. func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *FinalityStatus) { - var buf bytes.Buffer - marshalled, err := proto.Marshal(block) if err != nil { panic(fmt.Errorf("failed to marshal block: %w", err)) } + // TODO: If multiple goroutine, this becomes shared resource + f.outputBuffer.Reset() + previousHash := block.PreviousID() previousNum := 0 if block.Number > 0 { @@ -1910,9 +1882,9 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina } // **Important* The final space in the Sprintf template is mandatory! - buf.WriteString(fmt.Sprintf("FIRE BLOCK %d %s %d %s %d %d ", block.Number, hex.EncodeToString(block.Hash), previousNum, previousHash, libNum, block.MustTime().UnixNano())) + f.outputBuffer.WriteString(fmt.Sprintf("FIRE BLOCK %d %s %d %s %d %d ", block.Number, hex.EncodeToString(block.Hash), previousNum, previousHash, libNum, block.MustTime().UnixNano())) - encoder := base64.NewEncoder(base64.StdEncoding, &buf) + encoder := base64.NewEncoder(base64.StdEncoding, f.outputBuffer) if _, err = encoder.Write(marshalled); err != nil { panic(fmt.Errorf("write to encoder should have been infaillible: %w", err)) } @@ -1921,16 +1893,9 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina panic(fmt.Errorf("closing encoder should have been infaillible: %w", err)) } - buf.WriteString("\n") + f.outputBuffer.WriteString("\n") - if f.concurrentBlockFlushing { - f.blockOutputQueue <- &blockOutput{ - blockNum: block.Number, - data: buf.Bytes(), - } - } else { - f.flushToFirehose(buf.Bytes()) - } + f.flushToFirehose(f.outputBuffer.Bytes()) } // printToFirehose is an easy way to print to Firehose format, it essentially @@ -2871,25 +2836,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() - - close(f.blockOutputQueue) - f.flushOutputDone.Wait() + f.blockFlushDone.Wait() }) } } - -type blockOutput struct { - blockNum uint64 - data []byte -} 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 f7d2dd369a..f50bbc5697 100644 --- a/eth/tracers/firehose_concurrency_test.go +++ b/eth/tracers/firehose_concurrency_test.go @@ -94,7 +94,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 { @@ -139,24 +141,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) - } -}