forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-26 06:36:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

PR: PR comments addressed Part 1

Browse source
This commit is contained in:
David Zhou 2025-05-13 15:01:56 -04:00
parent fb57716d17
commit 7eaa0cb83b
4 changed files with 35 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
eth/tracers/firehose.go
View file

@ -144,13 +144,14 @@ func (c *FirehoseConfig) ForcedBackwardCompatibility() bool {
type Firehose struct { type Firehose struct {
// Global state // Global state
outputBuffer *bytes.Buffer outputBuffer *bytes.Buffer
initSent *atomic.Bool initSent *atomic.Bool
config *FirehoseConfig config *FirehoseConfig
chainConfig *params.ChainConfig chainConfig *params.ChainConfig
hasher crypto.KeccakState // Keccak256 hasher instance shared across tracer needs (non-concurrent safe) 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) hasherBuf common.Hash // Keccak256 hasher result array shared across tracer needs (non-concurrent safe)
tracerID string tracerID string
closeChannels sync.Once
// The FirehoseTracer is used in multiple chains, some for which were produced using a legacy version // 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 // 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 // we must "reproduce" on some chain to ensure that the FirehoseTracer produces the same output
@ -173,6 +174,8 @@ type Firehose struct {
blockReorderOrdinalSnapshot uint64 blockReorderOrdinalSnapshot uint64
blockReorderOrdinalOnce sync.Once blockReorderOrdinalOnce sync.Once
blockIsGenesis bool blockIsGenesis bool
blockPrintQueue chan *blockPrintJob
blockFlushDone sync.WaitGroup
// Transaction state // Transaction state
evm *tracing.VMContext evm *tracing.VMContext
@ -190,11 +193,6 @@ type Firehose struct {
// Testing state, only used in tests and private configs // Testing state, only used in tests and private configs
testingBuffer *bytes.Buffer testingBuffer *bytes.Buffer
testingIgnoreGenesisBlock bool testingIgnoreGenesisBlock bool
// Worker queue for block printing
blockPrintQueue chan *blockPrintJob
flushDone sync.WaitGroup
closeOnce sync.Once
} }
const FirehoseProtocolVersion = "3.0" const FirehoseProtocolVersion = "3.0"
@ -258,9 +256,10 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
} }
if config.ConcurrentBlockFlushing { 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.blockPrintQueue = make(chan *blockPrintJob, 100)
firehose.flushDone.Add(1) firehose.blockFlushDone.Add(1)
go firehose.blockPrintWorker() go firehose.blockPrintWorker()
} }
@ -630,9 +629,8 @@ func (f *Firehose) reorderCallOrdinals(call *pbeth.Call, ordinalBase uint64) (or
} }
func (f *Firehose) OnClose() { func (f *Firehose) OnClose() {
log.Info("Firehose closing...")
if f.concurrentBlockFlushing { 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() f.CloseBlockPrintQueue()
} }
} }
@ -2835,17 +2833,20 @@ type blockPrintJob struct {
} }
func (f *Firehose) blockPrintWorker() { func (f *Firehose) blockPrintWorker() {
defer f.flushDone.Done() defer f.blockFlushDone.Done()
for job := range f.blockPrintQueue { for job := range f.blockPrintQueue {
f.printBlockToFirehose(job.block, job.finality) 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() { func (f *Firehose) CloseBlockPrintQueue() {
if f.concurrentBlockFlushing { if f.concurrentBlockFlushing {
f.closeOnce.Do(func() { f.closeChannels.Do(func() {
close(f.blockPrintQueue) close(f.blockPrintQueue)
f.flushDone.Wait() f.blockFlushDone.Wait()
}) })
} }
} }

21
eth/tracers/firehose_concurrency.md
View file

@ -84,6 +84,12 @@ The implementation was validated at multiple levels to ensure correctness, confi
## Section 4: Analysis ## 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 ### 4.1 Results
The following outlines the results for metric three: The following outlines the results for metric three:
@ -92,18 +98,14 @@ The following outlines the results for metric three:
**No concurrency** **No concurrency**
Run 1: 71.49s user 15.56s system 56% cpu 2:34.28 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 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 Run 3: 68.81s user 15.12s system 53% cpu 2:38.18 total
**Concurrency** **Concurrency**
Run 1: 69.61s user 14.97s system 55% cpu 2:32.85 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 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) Run 3: 68.60s user 16.37s system 48% cpu 2:54.22 total (Not sure what happened here)
**Until Block 100000** **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 ### 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. User seems slightly lower, whereas system and cpu are relatively the same.
## Section 5: Conclusion ## Section 5: Conclusion

25
eth/tracers/firehose_concurrency_test.go
View file

@ -102,7 +102,9 @@ func TestFirehose_BlocksPrintToFirehose_MultipleBlocksInOrder(t *testing.T) {
"Expected %d blocks in output, found %d", blockCount, len(extractedBlocks)) "Expected %d blocks in output, found %d", blockCount, len(extractedBlocks))
// Verify blocks in order // 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 // Verify block hashes
for _, block := range extractedBlocks { for _, block := range extractedBlocks {
@ -147,24 +149,3 @@ func extractBlocksFromOutput(t *testing.T, output string) []extractedBlock {
return blocks 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)
}
}

1
eth/tracers/internal/tracetest/firehose/firehose_test.go
View file

@ -38,7 +38,6 @@ func TestFirehosePrestate(t *testing.T) {
"./testdata/TestFirehosePrestate/extra_account_creations", "./testdata/TestFirehosePrestate/extra_account_creations",
} }
// TODO: have goroutine config on AND off
for _, concurrent := range []bool{true, false} { for _, concurrent := range []bool{true, false} {
for _, folder := range testFolders { for _, folder := range testFolders {
name := filepath.Base(folder) name := filepath.Base(folder)