From 526ca333b6b2b07228ec8ae4396741b91849baaa Mon Sep 17 00:00:00 2001 From: David Zhou Date: Fri, 9 May 2025 09:34:43 -0400 Subject: [PATCH 01/18] DEV: Implementation of concurrency in firehose --- eth/tracers/firehose.go | 39 +++++++++++++++++++++++- eth/tracers/firehose_concurrency_test.go | 9 ++++++ 2 files changed, 47 insertions(+), 1 deletion(-) create mode 100644 eth/tracers/firehose_concurrency_test.go diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index 9d30c30f21..7ca1b01e12 100644 --- a/eth/tracers/firehose.go +++ b/eth/tracers/firehose.go @@ -186,6 +186,10 @@ type Firehose struct { // Testing state, only used in tests and private configs testingBuffer *bytes.Buffer testingIgnoreGenesisBlock bool + + // Worker queuefor blockprinting + blockPrintQueue chan *blockPrintJob + workerWg sync.WaitGroup } const FirehoseProtocolVersion = "3.0" @@ -238,6 +242,8 @@ func NewFirehose(config *FirehoseConfig) *Firehose { callStack: NewCallStack(), deferredCallState: NewDeferredCallState(), latestCallEnterSuicided: false, + + blockPrintQueue: make(chan *blockPrintJob, 100), } if config.private != nil { @@ -247,6 +253,13 @@ func NewFirehose(config *FirehoseConfig) *Firehose { } } + // Worker goroutines + numWorkers := 1 + firehose.workerWg.Add(numWorkers) + for i := 0; i < numWorkers; i++ { + go firehose.blockPrintWorker() + } + return firehose } @@ -483,7 +496,13 @@ func (f *Firehose) OnBlockEnd(err error) { } f.ensureInBlockAndNotInTrx() - f.printBlockToFirehose(f.block, f.blockFinality) + + job := &blockPrintJob{ + block: f.block, + finality: f.blockFinality, + } + f.blockPrintQueue <- job + } else { // An error occurred, could have happen in transaction/call context, we must not check if in trx/call, only check in block f.ensureInBlock(0) @@ -2791,3 +2810,21 @@ func (m Memory) GetPtr(offset, size int64) []byte { reminder := m[offset:] return append(reminder, make([]byte, int(size)-len(reminder))...) } + +type blockPrintJob struct { + block *pbeth.Block + finality *FinalityStatus +} + +func (f *Firehose) blockPrintWorker() { + defer f.workerWg.Done() + + for job := range f.blockPrintQueue { + f.printBlockToFirehose(job.block, job.finality) + } +} + +func (f *Firehose) Shutdown() { + close(f.blockPrintQueue) + f.workerWg.Wait() +} diff --git a/eth/tracers/firehose_concurrency_test.go b/eth/tracers/firehose_concurrency_test.go new file mode 100644 index 0000000000..ace16035f4 --- /dev/null +++ b/eth/tracers/firehose_concurrency_test.go @@ -0,0 +1,9 @@ +package tracers + +import ( + "testing" +) + +func TestFirehoseBlockPrintingOrder(t *testing.T) { + +} From fe10930ea986d2d718d96d3115fe57872a89dbeb Mon Sep 17 00:00:00 2001 From: David Zhou Date: Fri, 9 May 2025 11:06:56 -0400 Subject: [PATCH 02/18] TEST: Basic test file for block flushing --- eth/tracers/firehose.go | 12 +++++----- eth/tracers/firehose_concurrency_test.go | 28 +++++++++++++++++++++++- 2 files changed, 33 insertions(+), 7 deletions(-) diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index 7ca1b01e12..a1be26f862 100644 --- a/eth/tracers/firehose.go +++ b/eth/tracers/firehose.go @@ -496,12 +496,12 @@ func (f *Firehose) OnBlockEnd(err error) { } f.ensureInBlockAndNotInTrx() - - job := &blockPrintJob{ - block: f.block, - finality: f.blockFinality, - } - f.blockPrintQueue <- job + f.printBlockToFirehose(f.block, f.blockFinality) + //job := &blockPrintJob{ + // block: f.block, + // finality: f.blockFinality, + //} + //f.blockPrintQueue <- job } else { // An error occurred, could have happen in transaction/call context, we must not check if in trx/call, only check in block diff --git a/eth/tracers/firehose_concurrency_test.go b/eth/tracers/firehose_concurrency_test.go index ace16035f4..04338baf12 100644 --- a/eth/tracers/firehose_concurrency_test.go +++ b/eth/tracers/firehose_concurrency_test.go @@ -1,9 +1,35 @@ package tracers import ( + "encoding/hex" + "github.com/ethereum/go-ethereum/core/vm" + "github.com/ethereum/go-ethereum/params" + "github.com/stretchr/testify/require" "testing" ) -func TestFirehoseBlockPrintingOrder(t *testing.T) { +func TestFirehose_BlockPrintsToFirehose(t *testing.T) { + f := NewFirehose(&FirehoseConfig{ + ApplyBackwardCompatibility: ptr(false), + private: &privateFirehoseConfig{ + FlushToTestBuffer: true, + }, + }) + f.OnBlockchainInit(params.AllEthashProtocolChanges) + + f.OnBlockStart(blockEvent(123)) + blockHash := hex.EncodeToString(f.block.Hash) // Store the block hash before it gets reset + f.onTxStart(txEvent(), hex2Hash("ABCD"), from, to) + f.OnCallEnter(0, byte(vm.CALL), from, to, nil, 0, nil) + f.OnBalanceChange(from, b(100), b(50), 0) + f.OnCallExit(0, nil, 0, nil, false) + f.OnTxEnd(txReceiptEvent(0), nil) + f.OnBlockEnd(nil) + + output := f.InternalTestingBuffer().String() + + require.Contains(t, output, "FIRE BLOCK", "expected FIRE BLOCK output not found") + require.Contains(t, output, "123", "expected block number not found in output") + require.Contains(t, output, blockHash, "expected block hash not found in output") } From 6634465edba7fc2b0ec49f21d8cf541bc0b93018 Mon Sep 17 00:00:00 2001 From: David Zhou Date: Fri, 9 May 2025 11:34:01 -0400 Subject: [PATCH 03/18] TEST: Blocks print to firehose in order --- eth/tracers/firehose_concurrency_test.go | 155 +++++++++++++++++++++-- 1 file changed, 144 insertions(+), 11 deletions(-) diff --git a/eth/tracers/firehose_concurrency_test.go b/eth/tracers/firehose_concurrency_test.go index 04338baf12..ee108f385e 100644 --- a/eth/tracers/firehose_concurrency_test.go +++ b/eth/tracers/firehose_concurrency_test.go @@ -2,13 +2,17 @@ package tracers import ( "encoding/hex" + "fmt" "github.com/ethereum/go-ethereum/core/vm" "github.com/ethereum/go-ethereum/params" "github.com/stretchr/testify/require" + "regexp" + "strconv" + "strings" "testing" ) -func TestFirehose_BlockPrintsToFirehose(t *testing.T) { +func TestFirehose_BlockPrintsToFirehoseInOrder(t *testing.T) { f := NewFirehose(&FirehoseConfig{ ApplyBackwardCompatibility: ptr(false), private: &privateFirehoseConfig{ @@ -18,18 +22,147 @@ func TestFirehose_BlockPrintsToFirehose(t *testing.T) { f.OnBlockchainInit(params.AllEthashProtocolChanges) - f.OnBlockStart(blockEvent(123)) - blockHash := hex.EncodeToString(f.block.Hash) // Store the block hash before it gets reset - f.onTxStart(txEvent(), hex2Hash("ABCD"), from, to) - f.OnCallEnter(0, byte(vm.CALL), from, to, nil, 0, nil) - f.OnBalanceChange(from, b(100), b(50), 0) - f.OnCallExit(0, nil, 0, nil, false) - f.OnTxEnd(txReceiptEvent(0), nil) - f.OnBlockEnd(nil) + blockNumbers := []uint64{123, 124, 125} + blockHashes := make([]string, 3) + + for i, blockNum := range blockNumbers { + f.OnBlockStart(blockEvent(blockNum)) + blockHashes[i] = hex.EncodeToString(f.block.Hash) // Store the hash before it gets reset + + f.onTxStart(txEvent(), hex2Hash(fmt.Sprintf("ABCD%d", i)), from, to) + f.OnCallEnter(0, byte(vm.CALL), from, to, nil, 0, nil) + f.OnBalanceChange(from, b(100), b(50), 0) + f.OnCallExit(0, nil, 0, nil, false) + f.OnTxEnd(txReceiptEvent(0), nil) + + f.OnBlockEnd(nil) + } output := f.InternalTestingBuffer().String() require.Contains(t, output, "FIRE BLOCK", "expected FIRE BLOCK output not found") - require.Contains(t, output, "123", "expected block number not found in output") - require.Contains(t, output, blockHash, "expected block hash not found in output") + + for i, blockNum := range blockNumbers { + require.Contains(t, output, fmt.Sprintf("%d", blockNum), + "expected block number %d not found in output", blockNum) + require.Contains(t, output, blockHashes[i], + "expected block hash for block %d not found in output", blockNum) + } + + blockNumIndex123 := strings.Index(output, "123") + blockNumIndex124 := strings.Index(output, "124") + blockNumIndex125 := strings.Index(output, "125") + + require.True(t, blockNumIndex123 < blockNumIndex124, + "Block 123 should appear before block 124 in output") + require.True(t, blockNumIndex124 < blockNumIndex125, + "Block 124 should appear before block 125 in output") +} + +func TestFirehose_BlocksPrintToFirehoseInOrder(t *testing.T) { + + const blockCount = 100 + const baseBlockNum = 1000 + + f := NewFirehose(&FirehoseConfig{ + ApplyBackwardCompatibility: ptr(false), + private: &privateFirehoseConfig{ + FlushToTestBuffer: true, + }, + }) + + f.OnBlockchainInit(params.AllEthashProtocolChanges) + + blockHashes := make(map[uint64]string, blockCount) + + for i := 0; i < blockCount; i++ { + blockNum := uint64(baseBlockNum + i) + + f.OnBlockStart(blockEvent(blockNum)) + blockHashes[blockNum] = hex.EncodeToString(f.block.Hash) // Store hash before block reset + + f.onTxStart(txEvent(), hex2Hash(fmt.Sprintf("TX%d", i)), from, to) + f.OnCallEnter(0, byte(vm.CALL), from, to, nil, 0, nil) + f.OnBalanceChange(from, b(100), b(50), 0) + f.OnCallExit(0, nil, 0, nil, false) + f.OnTxEnd(txReceiptEvent(0), nil) + + f.OnBlockEnd(nil) + } + + f.Shutdown() + + output := f.InternalTestingBuffer().String() + extractedBlocks := extractBlocksFromOutput(t, output) + + // Verify block count + require.Equal(t, blockCount, len(extractedBlocks), + "Expected %d blocks in output, found %d", blockCount, len(extractedBlocks)) + + // Verify blocks in order + verifyBlockSequence(t, extractedBlocks, baseBlockNum) + + // Verify block hashes + for _, block := range extractedBlocks { + expectedHash, exists := blockHashes[block.number] + require.True(t, exists, "Block %d not found in tracked blocks", block.number) + require.Equal(t, expectedHash, block.hash, + "Hash mismatch for block %d", block.number) + } +} + +type extractedBlock struct { + number uint64 + hash string +} + +func extractBlocksFromOutput(t *testing.T, output string) []extractedBlock { + t.Helper() + + // Regex to extract the block number and hash from the FIRE BLOCK line + blockInfoRegex := regexp.MustCompile(`FIRE BLOCK (\d+) ([0-9a-fA-F]+)`) + + lines := strings.Split(output, "\n") + var blocks []extractedBlock + + for _, line := range lines { + if strings.HasPrefix(line, "FIRE BLOCK") { + matches := blockInfoRegex.FindStringSubmatch(line) + if len(matches) == 3 { + blockNumStr := matches[1] + blockHash := matches[2] + + blockNum, err := strconv.ParseUint(blockNumStr, 10, 64) + require.NoError(t, err, "failed to parse block number: %s", blockNumStr) + + blocks = append(blocks, extractedBlock{ + number: blockNum, + hash: blockHash, + }) + } + } + } + + 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) + } } From a465709da4acae5b2e1a4212cc09ba4fc95de0ea Mon Sep 17 00:00:00 2001 From: David Zhou Date: Fri, 9 May 2025 13:05:40 -0400 Subject: [PATCH 04/18] TEST: All unit tests for concurrency pass --- eth/tracers/firehose.go | 12 ++++++------ eth/tracers/firehose_concurrency_test.go | 6 ++++-- 2 files changed, 10 insertions(+), 8 deletions(-) diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index a1be26f862..a278983eb5 100644 --- a/eth/tracers/firehose.go +++ b/eth/tracers/firehose.go @@ -495,13 +495,13 @@ func (f *Firehose) OnBlockEnd(err error) { f.fixOrdinalsForEndOfBlockChanges() } + // f.printBlockToFirehose(f.block, f.blockFinality) f.ensureInBlockAndNotInTrx() - f.printBlockToFirehose(f.block, f.blockFinality) - //job := &blockPrintJob{ - // block: f.block, - // finality: f.blockFinality, - //} - //f.blockPrintQueue <- job + job := &blockPrintJob{ + block: f.block, + finality: f.blockFinality, + } + f.blockPrintQueue <- job } else { // An error occurred, could have happen in transaction/call context, we must not check if in trx/call, only check in block diff --git a/eth/tracers/firehose_concurrency_test.go b/eth/tracers/firehose_concurrency_test.go index ee108f385e..2bc3220d41 100644 --- a/eth/tracers/firehose_concurrency_test.go +++ b/eth/tracers/firehose_concurrency_test.go @@ -12,7 +12,7 @@ import ( "testing" ) -func TestFirehose_BlockPrintsToFirehoseInOrder(t *testing.T) { +func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) { f := NewFirehose(&FirehoseConfig{ ApplyBackwardCompatibility: ptr(false), private: &privateFirehoseConfig{ @@ -38,6 +38,8 @@ func TestFirehose_BlockPrintsToFirehoseInOrder(t *testing.T) { f.OnBlockEnd(nil) } + f.Shutdown() + output := f.InternalTestingBuffer().String() require.Contains(t, output, "FIRE BLOCK", "expected FIRE BLOCK output not found") @@ -59,7 +61,7 @@ func TestFirehose_BlockPrintsToFirehoseInOrder(t *testing.T) { "Block 124 should appear before block 125 in output") } -func TestFirehose_BlocksPrintToFirehoseInOrder(t *testing.T) { +func TestFirehose_BlocksPrintToFirehose_MultipleBlocksInOrder(t *testing.T) { const blockCount = 100 const baseBlockNum = 1000 From 2ef4ea641c95c4a5865aa941748dfcfc34be928c Mon Sep 17 00:00:00 2001 From: David Zhou Date: Fri, 9 May 2025 15:23:23 -0400 Subject: [PATCH 05/18] TEST: Fixed tests --- eth/tracers/firehose.go | 3 +++ eth/tracers/firehose_concurrency.md | 0 eth/tracers/firehose_concurrency_test.go | 29 +++++++++++------------- 3 files changed, 16 insertions(+), 16 deletions(-) create mode 100644 eth/tracers/firehose_concurrency.md diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index a278983eb5..655f5852ee 100644 --- a/eth/tracers/firehose.go +++ b/eth/tracers/firehose.go @@ -78,6 +78,7 @@ func NewTracingHooksFromFirehose(tracer *Firehose) *tracing.Hooks { OnBlockStart: tracer.OnBlockStart, OnBlockEnd: tracer.OnBlockEnd, OnSkippedBlock: tracer.OnSkippedBlock, + // TODO: OnClose OnTxStart: tracer.OnTxStart, OnTxEnd: tracer.OnTxEnd, @@ -108,6 +109,8 @@ 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 private *privateFirehoseConfig diff --git a/eth/tracers/firehose_concurrency.md b/eth/tracers/firehose_concurrency.md new file mode 100644 index 0000000000..e69de29bb2 diff --git a/eth/tracers/firehose_concurrency_test.go b/eth/tracers/firehose_concurrency_test.go index 2bc3220d41..d1619da7f8 100644 --- a/eth/tracers/firehose_concurrency_test.go +++ b/eth/tracers/firehose_concurrency_test.go @@ -13,6 +13,7 @@ import ( ) func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) { + f := NewFirehose(&FirehoseConfig{ ApplyBackwardCompatibility: ptr(false), private: &privateFirehoseConfig{ @@ -23,11 +24,9 @@ func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) { f.OnBlockchainInit(params.AllEthashProtocolChanges) blockNumbers := []uint64{123, 124, 125} - blockHashes := make([]string, 3) for i, blockNum := range blockNumbers { f.OnBlockStart(blockEvent(blockNum)) - blockHashes[i] = hex.EncodeToString(f.block.Hash) // Store the hash before it gets reset f.onTxStart(txEvent(), hex2Hash(fmt.Sprintf("ABCD%d", i)), from, to) f.OnCallEnter(0, byte(vm.CALL), from, to, nil, 0, nil) @@ -42,23 +41,21 @@ func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) { output := f.InternalTestingBuffer().String() - require.Contains(t, output, "FIRE BLOCK", "expected FIRE BLOCK output not found") + outNumber := make([]string, 0) + for i, line := range strings.Split(output, "\n") { + if i == 0 { + require.Equal(t, "FIRE INIT 3.0 geth 1.15.10", line) + continue + } - for i, blockNum := range blockNumbers { - require.Contains(t, output, fmt.Sprintf("%d", blockNum), - "expected block number %d not found in output", blockNum) - require.Contains(t, output, blockHashes[i], - "expected block hash for block %d not found in output", blockNum) + fields := strings.SplitN(line, " ", 4) + if len(fields) >= 3 { + outNumber = append(outNumber, fields[2]) + } } - blockNumIndex123 := strings.Index(output, "123") - blockNumIndex124 := strings.Index(output, "124") - blockNumIndex125 := strings.Index(output, "125") - - require.True(t, blockNumIndex123 < blockNumIndex124, - "Block 123 should appear before block 124 in output") - require.True(t, blockNumIndex124 < blockNumIndex125, - "Block 124 should appear before block 125 in output") + require.Contains(t, output, "FIRE BLOCK", "expected FIRE BLOCK output not found") + require.Equal(t, []string{"123", "124", "125"}, outNumber) } func TestFirehose_BlocksPrintToFirehose_MultipleBlocksInOrder(t *testing.T) { From 9bd147e3f59777d264ed1e0f152c55ba547e9cb9 Mon Sep 17 00:00:00 2001 From: David Zhou Date: Fri, 9 May 2025 16:20:57 -0400 Subject: [PATCH 06/18] DEV: Implemented onClose() --- eth/tracers/firehose.go | 26 +++++++++++++----------- eth/tracers/firehose_concurrency_test.go | 7 ++++--- 2 files changed, 18 insertions(+), 15 deletions(-) diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index 655f5852ee..6ea929bc47 100644 --- a/eth/tracers/firehose.go +++ b/eth/tracers/firehose.go @@ -78,6 +78,7 @@ func NewTracingHooksFromFirehose(tracer *Firehose) *tracing.Hooks { OnBlockStart: tracer.OnBlockStart, OnBlockEnd: tracer.OnBlockEnd, OnSkippedBlock: tracer.OnSkippedBlock, + OnClose: tracer.OnClose, // TODO: OnClose OnTxStart: tracer.OnTxStart, @@ -498,13 +499,13 @@ func (f *Firehose) OnBlockEnd(err error) { f.fixOrdinalsForEndOfBlockChanges() } - // f.printBlockToFirehose(f.block, f.blockFinality) - f.ensureInBlockAndNotInTrx() - job := &blockPrintJob{ - block: f.block, - finality: f.blockFinality, - } - f.blockPrintQueue <- job + f.printBlockToFirehose(f.block, f.blockFinality) + //f.ensureInBlockAndNotInTrx() + //job := &blockPrintJob{ + // block: f.block, + // finality: f.blockFinality, + //} + //f.blockPrintQueue <- job } else { // An error occurred, could have happen in transaction/call context, we must not check if in trx/call, only check in block @@ -623,6 +624,12 @@ func (f *Firehose) reorderCallOrdinals(call *pbeth.Call, ordinalBase uint64) (or return call.EndOrdinal } +func (f *Firehose) OnClose() { + log.Info("Firehose closing: waiting for worker goroutines to finish and shutting down channels") + close(f.blockPrintQueue) + f.workerWg.Wait() +} + func (f *Firehose) OnSystemCallStart() { firehoseInfo("system call start") f.ensureInBlockAndNotInTrx() @@ -2826,8 +2833,3 @@ func (f *Firehose) blockPrintWorker() { f.printBlockToFirehose(job.block, job.finality) } } - -func (f *Firehose) Shutdown() { - close(f.blockPrintQueue) - f.workerWg.Wait() -} diff --git a/eth/tracers/firehose_concurrency_test.go b/eth/tracers/firehose_concurrency_test.go index d1619da7f8..d69291e858 100644 --- a/eth/tracers/firehose_concurrency_test.go +++ b/eth/tracers/firehose_concurrency_test.go @@ -37,7 +37,7 @@ func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) { f.OnBlockEnd(nil) } - f.Shutdown() + f.OnClose() output := f.InternalTestingBuffer().String() @@ -50,11 +50,12 @@ func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) { fields := strings.SplitN(line, " ", 4) if len(fields) >= 3 { + require.Equal(t, "FIRE", fields[0]) + require.Equal(t, "BLOCK", fields[1]) outNumber = append(outNumber, fields[2]) } } - require.Contains(t, output, "FIRE BLOCK", "expected FIRE BLOCK output not found") require.Equal(t, []string{"123", "124", "125"}, outNumber) } @@ -89,7 +90,7 @@ func TestFirehose_BlocksPrintToFirehose_MultipleBlocksInOrder(t *testing.T) { f.OnBlockEnd(nil) } - f.Shutdown() + f.OnClose() output := f.InternalTestingBuffer().String() extractedBlocks := extractBlocksFromOutput(t, output) From bd85c1897005cbb5ace1fc20f9ba506b11ec996f Mon Sep 17 00:00:00 2001 From: David Zhou Date: Fri, 9 May 2025 16:56:15 -0400 Subject: [PATCH 07/18] TEST: Fixed race issue in existing test files caused by integration of goroutine. All tests passing --- eth/tracers/firehose.go | 39 ++++++++++--------- .../tracetest/firehose/firehose_test.go | 4 ++ 2 files changed, 24 insertions(+), 19 deletions(-) diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index 6ea929bc47..2fb2e4a009 100644 --- a/eth/tracers/firehose.go +++ b/eth/tracers/firehose.go @@ -79,7 +79,6 @@ func NewTracingHooksFromFirehose(tracer *Firehose) *tracing.Hooks { OnBlockEnd: tracer.OnBlockEnd, OnSkippedBlock: tracer.OnSkippedBlock, OnClose: tracer.OnClose, - // TODO: OnClose OnTxStart: tracer.OnTxStart, OnTxEnd: tracer.OnTxEnd, @@ -193,7 +192,8 @@ type Firehose struct { // Worker queuefor blockprinting blockPrintQueue chan *blockPrintJob - workerWg sync.WaitGroup + flushDone sync.WaitGroup + closeOnce sync.Once } const FirehoseProtocolVersion = "3.0" @@ -257,12 +257,8 @@ func NewFirehose(config *FirehoseConfig) *Firehose { } } - // Worker goroutines - numWorkers := 1 - firehose.workerWg.Add(numWorkers) - for i := 0; i < numWorkers; i++ { - go firehose.blockPrintWorker() - } + firehose.flushDone.Add(1) + go firehose.blockPrintWorker() return firehose } @@ -499,13 +495,13 @@ func (f *Firehose) OnBlockEnd(err error) { f.fixOrdinalsForEndOfBlockChanges() } - f.printBlockToFirehose(f.block, f.blockFinality) - //f.ensureInBlockAndNotInTrx() - //job := &blockPrintJob{ - // block: f.block, - // finality: f.blockFinality, - //} - //f.blockPrintQueue <- job + f.ensureInBlockAndNotInTrx() + // f.printBlockToFirehose(f.block, f.blockFinality) + job := &blockPrintJob{ + block: f.block, + finality: f.blockFinality, + } + f.blockPrintQueue <- job } else { // An error occurred, could have happen in transaction/call context, we must not check if in trx/call, only check in block @@ -626,8 +622,7 @@ 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") - close(f.blockPrintQueue) - f.workerWg.Wait() + f.CloseBlockPrintQueue() } func (f *Firehose) OnSystemCallStart() { @@ -2827,9 +2822,15 @@ type blockPrintJob struct { } func (f *Firehose) blockPrintWorker() { - defer f.workerWg.Done() - + defer f.flushDone.Done() for job := range f.blockPrintQueue { f.printBlockToFirehose(job.block, job.finality) } } + +func (f *Firehose) CloseBlockPrintQueue() { + f.closeOnce.Do(func() { + close(f.blockPrintQueue) + f.flushDone.Wait() + }) +} diff --git a/eth/tracers/internal/tracetest/firehose/firehose_test.go b/eth/tracers/internal/tracetest/firehose/firehose_test.go index b226241996..cd959260cc 100644 --- a/eth/tracers/internal/tracetest/firehose/firehose_test.go +++ b/eth/tracers/internal/tracetest/firehose/firehose_test.go @@ -48,6 +48,8 @@ func TestFirehosePrestate(t *testing.T) { runPrestateBlock(t, filepath.Join(folder, "prestate.json"), tracingHooks) + tracer.CloseBlockPrintQueue() + genesisLine, blockLines, unknownLines := readTracerFirehoseLines(t, tracer) require.Len(t, unknownLines, 0, "Lines:\n%s", strings.Join(slicesMap(unknownLines, func(l unknownLine) string { return "- '" + string(l) + "'" }), "\n")) require.NotNil(t, genesisLine) @@ -203,6 +205,8 @@ func testBlockTracesCorrectly(t *testing.T, genesisSpec *core.Genesis, engine co n, err := chain.InsertChain(blocks) require.NoError(t, err, "failed to insert chain block %d", n) + tracer.CloseBlockPrintQueue() + assertBlockEquals(t, tracer, filepath.Join("testdata", goldenDir, string(model)), len(blocks)) }) } From e8ddd052849695f7c889c99963d68ffdf4e5a747 Mon Sep 17 00:00:00 2001 From: David Zhou Date: Mon, 12 May 2025 09:36:55 -0400 Subject: [PATCH 08/18] DEV: Added configuration --- eth/tracers/firehose.go | 51 ++++++++++++------- .../tracetest/firehose/firehose_test.go | 1 - .../tracetest/firehose/helpers_test.go | 1 + 3 files changed, 34 insertions(+), 19 deletions(-) diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index 2fb2e4a009..6905da0c9f 100644 --- a/eth/tracers/firehose.go +++ b/eth/tracers/firehose.go @@ -109,8 +109,7 @@ func NewTracingHooksFromFirehose(tracer *Firehose) *tracing.Hooks { type FirehoseConfig struct { ApplyBackwardCompatibility *bool `json:"applyBackwardCompatibility"` - // TODO: Add LOGGING on init - ConcurrentBlockFlushing bool `json:"concurrentBlockFlushing"` + ConcurrentBlockFlushing bool `json:"concurrentBlockFlushing"` // Only used for testing, only possible through JSON configuration private *privateFirehoseConfig @@ -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 @@ -190,7 +191,7 @@ type Firehose struct { testingBuffer *bytes.Buffer testingIgnoreGenesisBlock bool - // Worker queuefor blockprinting + // Worker queue for block printing blockPrintQueue chan *blockPrintJob flushDone sync.WaitGroup closeOnce sync.Once @@ -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 { } } - firehose.flushDone.Add(1) - go firehose.blockPrintWorker() + 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) - job := &blockPrintJob{ - block: f.block, - finality: 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) } - f.blockPrintQueue <- job } else { // An error occurred, could have happen in transaction/call context, we must not check if in trx/call, only check in block @@ -621,8 +630,11 @@ 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") - f.CloseBlockPrintQueue() + 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() { @@ -2829,8 +2841,11 @@ func (f *Firehose) blockPrintWorker() { } func (f *Firehose) CloseBlockPrintQueue() { - f.closeOnce.Do(func() { - close(f.blockPrintQueue) - f.flushDone.Wait() - }) + if f.concurrentBlockFlushing { + f.closeOnce.Do(func() { + log.Info("Closing channel: flushing the remaining blocks to firehose") + close(f.blockPrintQueue) + f.flushDone.Wait() + }) + } } diff --git a/eth/tracers/internal/tracetest/firehose/firehose_test.go b/eth/tracers/internal/tracetest/firehose/firehose_test.go index cd959260cc..36600317b9 100644 --- a/eth/tracers/internal/tracetest/firehose/firehose_test.go +++ b/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) }) } - } } diff --git a/eth/tracers/internal/tracetest/firehose/helpers_test.go b/eth/tracers/internal/tracetest/firehose/helpers_test.go index 98d6cbba90..b7263dc9ae 100644 --- a/eth/tracers/internal/tracetest/firehose/helpers_test.go +++ b/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, From 8d0ce228dbfe0483fdee268e1c5a3872f514e400 Mon Sep 17 00:00:00 2001 From: David Zhou Date: Mon, 12 May 2025 11:11:41 -0400 Subject: [PATCH 09/18] TEST: Testing concurrent and not concurrent --- eth/tracers/firehose.go | 4 +- .../tracetest/firehose/firehose_test.go | 64 ++++++++++--------- .../tracetest/firehose/helpers_test.go | 6 +- 3 files changed, 39 insertions(+), 35 deletions(-) diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index 6905da0c9f..19aeea5a19 100644 --- a/eth/tracers/firehose.go +++ b/eth/tracers/firehose.go @@ -632,7 +632,7 @@ func (f *Firehose) reorderCallOrdinals(call *pbeth.Call, ordinalBase uint64) (or func (f *Firehose) OnClose() { log.Info("Firehose closing...") if f.concurrentBlockFlushing { - log.Info("waiting for worker goroutines to finish and shutting down channels") + log.Info("Closing channel: flushing the remaining blocks to firehose") f.CloseBlockPrintQueue() } } @@ -1860,6 +1860,7 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina panic(fmt.Errorf("failed to marshal block: %w", err)) } + // TODO: If multiple goroutine, this becomes shared resource f.outputBuffer.Reset() previousHash := block.PreviousID() @@ -2843,7 +2844,6 @@ 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() }) diff --git a/eth/tracers/internal/tracetest/firehose/firehose_test.go b/eth/tracers/internal/tracetest/firehose/firehose_test.go index 36600317b9..e004105a06 100644 --- a/eth/tracers/internal/tracetest/firehose/firehose_test.go +++ b/eth/tracers/internal/tracetest/firehose/firehose_test.go @@ -38,26 +38,28 @@ func TestFirehosePrestate(t *testing.T) { "./testdata/TestFirehosePrestate/extra_account_creations", } - for _, folder := range testFolders { - name := filepath.Base(folder) + // TODO: have goroutine config on AND off + for _, concurrent := range []bool{true, false} { + for _, folder := range testFolders { + name := filepath.Base(folder) - for _, model := range tracingModels { - t.Run(string(model)+"/"+name, func(t *testing.T) { - tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model) - defer onClose() + for _, model := range tracingModels { + t.Run(string(model)+"/"+name, func(t *testing.T) { + tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model, concurrent) + defer onClose() - runPrestateBlock(t, filepath.Join(folder, "prestate.json"), tracingHooks) + runPrestateBlock(t, filepath.Join(folder, "prestate.json"), tracingHooks) - tracer.CloseBlockPrintQueue() + tracer.CloseBlockPrintQueue() - genesisLine, blockLines, unknownLines := readTracerFirehoseLines(t, tracer) - require.Len(t, unknownLines, 0, "Lines:\n%s", strings.Join(slicesMap(unknownLines, func(l unknownLine) string { return "- '" + string(l) + "'" }), "\n")) - require.NotNil(t, genesisLine) - blockLines.assertOnlyBlockEquals(t, filepath.Join(folder, string(model)), 1) - }) + genesisLine, blockLines, unknownLines := readTracerFirehoseLines(t, tracer) + require.Len(t, unknownLines, 0, "Lines:\n%s", strings.Join(slicesMap(unknownLines, func(l unknownLine) string { return "- '" + string(l) + "'" }), "\n")) + require.NotNil(t, genesisLine) + blockLines.assertOnlyBlockEquals(t, filepath.Join(folder, string(model)), 1) + }) + } } } - } func TestFirehose_EIP7702(t *testing.T) { // Copied from ./core/blockchain_test.go#L4180 (TestEIP7702) @@ -187,26 +189,28 @@ 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 _, model := range tracingModels { - t.Run(string(model), func(t *testing.T) { - tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model) - defer onClose() + for _, concurrent := range []bool{true, false} { + for _, model := range tracingModels { + t.Run(string(model), func(t *testing.T) { + tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model, concurrent) + defer onClose() - chain, err := core.NewBlockChain(rawdb.NewMemoryDatabase(), nil, genesisSpec, nil, engine, vm.Config{Tracer: tracingHooks}, nil) - require.NoError(t, err, "failed to create tester chain") + chain, err := core.NewBlockChain(rawdb.NewMemoryDatabase(), nil, genesisSpec, nil, engine, vm.Config{Tracer: tracingHooks}, nil) + require.NoError(t, err, "failed to create tester chain") - chain.SetBlockValidatorAndProcessorForTesting( - ignoreValidateStateValidator{core.NewBlockValidator(genesisSpec.Config, chain)}, - core.NewStateProcessor(genesisSpec.Config, chain.HeaderChain()), - ) + chain.SetBlockValidatorAndProcessorForTesting( + ignoreValidateStateValidator{core.NewBlockValidator(genesisSpec.Config, chain)}, + core.NewStateProcessor(genesisSpec.Config, chain.HeaderChain()), + ) - defer chain.Stop() - n, err := chain.InsertChain(blocks) - require.NoError(t, err, "failed to insert chain block %d", n) + defer chain.Stop() + n, err := chain.InsertChain(blocks) + require.NoError(t, err, "failed to insert chain block %d", n) - tracer.CloseBlockPrintQueue() + tracer.CloseBlockPrintQueue() - assertBlockEquals(t, tracer, filepath.Join("testdata", goldenDir, string(model)), len(blocks)) - }) + assertBlockEquals(t, tracer, filepath.Join("testdata", goldenDir, string(model)), len(blocks)) + }) + } } } diff --git a/eth/tracers/internal/tracetest/firehose/helpers_test.go b/eth/tracers/internal/tracetest/firehose/helpers_test.go index b7263dc9ae..0541d37748 100644 --- a/eth/tracers/internal/tracetest/firehose/helpers_test.go +++ b/eth/tracers/internal/tracetest/firehose/helpers_test.go @@ -29,17 +29,17 @@ type firehoseInitLine struct { type firehoseBlockLines []firehoseBlockLine -func newFirehoseTestTracer(t *testing.T, model tracingModel) (*tracers.Firehose, *tracing.Hooks, func()) { +func newFirehoseTestTracer(t *testing.T, model tracingModel, concurrentBlockFlushing bool) (*tracers.Firehose, *tracing.Hooks, func()) { t.Helper() tracer, err := tracers.NewFirehoseFromRawJSON([]byte(fmt.Sprintf(`{ - "concurrentBlockFlushing": true, + "concurrentBlockFlushing": %t, "_private": { "flushToTestBuffer": true, "ignoreGenesisBlock": true, "forcedBackwardCompatibility": %t } - }`, model == tracingModelFirehose2_3))) + }`, concurrentBlockFlushing, model == tracingModelFirehose2_3))) require.NoError(t, err) hooks := tracers.NewTracingHooksFromFirehose(tracer) From a11993b1dc2964c3e24470ec64a9935dc25ef4bc Mon Sep 17 00:00:00 2001 From: David Zhou Date: Mon, 12 May 2025 16:16:06 -0400 Subject: [PATCH 10/18] DOC: Report on benchmarking results --- eth/tracers/firehose_concurrency.md | 142 ++++++++++++++++++++++++++++ 1 file changed, 142 insertions(+) diff --git a/eth/tracers/firehose_concurrency.md b/eth/tracers/firehose_concurrency.md index e69de29bb2..cb395750c2 100644 --- a/eth/tracers/firehose_concurrency.md +++ b/eth/tracers/firehose_concurrency.md @@ -0,0 +1,142 @@ +# Report: Concurrent Block Flushing in Tracer Processing System + +## Section 1: Introduction + +Currently, the tracer processing system operates in a linear fashion, requiring the complete processing of a block's code before proceeding to the next one. As a result, computationally expensive operations—such as `proto.Marshal` and base64 encoding, which are essential for flushing a block to the firehose—can introduce significant delays. To mitigate this bottleneck, a potential solution is to leverage concurrency by introducing goroutines. These heavy operations can be offloaded to a separate channel, enabling asynchronous processing. This approach allows the main execution flow to begin processing subsequent blocks while previous ones are being flushed, thereby improving overall throughput and reducing latency. The goal of this report is to outline the solution and provide benchmarking results. + +## Section 2: Background + +`proto.Marshal`, part of Protocol Buffers (protobuf), serializes structured data into a compact binary format. While efficient in output size, the process of traversing and encoding complex data structures—especially those with deeply nested or repeated fields—can be CPU-intensive. Additionally, memory allocations during marshaling can introduce further overhead. + +Similarly, base64 encoding, which transforms binary data into an ASCII string format for transmission or storage, involves non-trivial byte-wise transformations and increases the data size. This added computational cost becomes significant when processing large blocks or high-throughput workloads. + +Together, these operations introduce latency in a linear processing pipeline. + +## Section 3: Method + +### 3.1 Proposed Solution + +A critical point in the tracer processing system is the `OnBlockEnd` hook, which invokes the `printBlockToFirehose` method. This method includes computationally expensive operations such as `proto.Marshal` and base64 encoding, which are necessary to serialize and flush the block data to the firehose. + +To address this, the current solution introduces a worker queue mechanism. Specifically, a single goroutine backed by a channel is used to enqueue and process `printBlockToFirehose` tasks asynchronously. This decouples the expensive flush operations from the main block processing path, allowing the tracer to begin handling the next block immediately after `OnBlockEnd` is invoked. This behavior is controlled by the `FirehoseConfig.ConcurrencyBlockFlushing` flag: when set to `true`, the asynchronous flushing mode is enabled; when set to `false`, the system falls back to the default linear execution of `printBlockToFirehose`. + +As part of future work, this model can be extended from a single worker goroutine to multiple concurrent workers. This could further improve throughput by increasing parallelism. However, such an enhancement must address critical challenges, including proper synchronization of shared resources like `output.Buffer` and maintaining the strict block ordering requirement—i.e., block N must be flushed before block N+1 to preserve data consistency. + +### 3.2 Validation and Performance Metrics + +The implementation was validated at multiple levels to ensure correctness, configurability, and performance improvements of the concurrent block flushing mechanism. + +1. **Unit-Level Validation** + + To confirm the functional correctness of the concurrent flushing implementation, a unit test was written that creates and processes 1,000 blocks, flushing each to an `InternalTestingBuffer`. The results were then compared against expected outputs to verify equivalence. The test confirms that the output produced by the concurrent mechanism matches that of the original linear method, thereby validating correctness at the unit level. +2. **Integration Testing on Battlefield-Ethereum** + + To verify integration within the battlefield-ethereum environment, the feature was exposed via a new configuration flag: `CONCURRENT_BLOCK_FLUSHING`. This flag determines whether the system uses the default sequential method or the new concurrent implementation. The system can be toggled between these modes with the following commands: + + Original (linear flushing): + + ```bash + ./scripts/run_firehose_geth_dev.sh 3.0 prague + ``` + + Concurrent flushing enabled: + + ```bash + CONCURRENT_BLOCK_FLUSHING=true ./scripts/run_firehose_geth_dev.sh 3.0 prague + ``` + + Behavioral differences were observed through log output. In the concurrent mode, log lines such as: + + ``` + "Closing channel: flushing the remaining blocks to firehose" + ``` + + appear when the program is interrupted (e.g., via Ctrl + C), indicating that the concurrent flushing logic and cleanup path are active. These lines are absent in the linear configuration, confirming that the switch is functioning as intended. + + Furthermore, when running the integration test suite using: + + ```bash + pnpm test:fh3.0:geth-dev + ``` + + all tests passed successfully (64 passing), indicating that the concurrent implementation does not introduce regressions in battlefield compatibility. +3. **Performance Benchmarking** + + To quantify performance differences, benchmarking was conducted using firehose-ethereum. The following command was used for both the baseline and concurrent configurations: + + ```bash + time geth --vmtrace=firehose \ + --vmtrace.jsonconfig='{"concurrentBlockFlushing":}' \ + --synctarget=0x7ae82cb3e60f13272a59319a4b617022228227258e18e0c5e7404236d773d2a3 \ + --syncmode=full --holesky --datadir=./geth --db.engine=pebble \ + --state.scheme=path --port=30305 --authrpc.jwtsecret=jwt.txt \ + --authrpc.addr=0.0.0.0 --authrpc.port=9551 --authrpc.vhosts="*" \ + --http --http.addr=0.0.0.0 --http.api=eth,net,web3 --http.port=9545 \ + --http.vhosts="*" --port=40303 --ws.port=9546 --ipcpath=/tmp/geth.ipc > /dev/null + ``` + + The benchmark was conducted in two phases: + + * With `concurrentBlockFlushing: false`, the node was synced to block 10,000, the data directory (`./geth`) was removed, and then resynced up to block 100,000. + * The same steps were repeated with `concurrentBlockFlushing: true`. + + The `time` command outputs wall-clock time and system/user CPU usage upon completion, providing a baseline for comparing performance between the linear and concurrent implementations. This methodology enables a controlled, reproducible environment for evaluating the effectiveness of the concurrent block flushing feature. + +## Section 4: Analysis + +### 4.1 Results + +The following outlines the results for metric three: + +**Until Block 10000** + +**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 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 3: 68.60s user 16.37s system 48% cpu 2:54.22 total (Not sure what happened here) + +**Until Block 100000** + +**No concurrency** + +364.19s user 172.14s system 34% cpu 26:13.33 total + +**Concurrency** + +358.42s user 171.21s system 35% cpu 25:11.97 total + +**Table 1: Result comparison with and without concurrency** + +| | No Concurrency | Concurrency | +| :-------------- | :------------- | :----------------- | +| **Block 10 000** | | | +| user | 70.02s | 68.72s | +| system | 15.43s | 15.49s | +| cpu | 54.33% | 52.33% | +| total | 2:36.73 | 2:40.22 (because of last run) | +| **Block 100 000**| | | +| user | 364.19s | 358.42s | +| system | 172.14s | 171.21s | +| cpu | 34% | 35% | +| total | 26:13.33 | 25:11.97 | + +### 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. + +User seems slightly lower, whereas system and cpu are relatively the same. + +## Section 5: Conclusion + +The implementation of a single goroutine seems to lead to a decrease in the total time by a factor of 0.6 millisecond per block. From fb57716d17ef546c7cb3fdf2ef24b8bd10026156 Mon Sep 17 00:00:00 2001 From: David Zhou Date: Tue, 13 May 2025 10:07:55 -0400 Subject: [PATCH 11/18] TEST: Added config to test --- eth/tracers/firehose_concurrency_test.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/eth/tracers/firehose_concurrency_test.go b/eth/tracers/firehose_concurrency_test.go index d69291e858..691fa00931 100644 --- a/eth/tracers/firehose_concurrency_test.go +++ b/eth/tracers/firehose_concurrency_test.go @@ -15,6 +15,7 @@ import ( func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) { f := NewFirehose(&FirehoseConfig{ + ConcurrentBlockFlushing: true, ApplyBackwardCompatibility: ptr(false), private: &privateFirehoseConfig{ FlushToTestBuffer: true, @@ -65,6 +66,7 @@ func TestFirehose_BlocksPrintToFirehose_MultipleBlocksInOrder(t *testing.T) { const baseBlockNum = 1000 f := NewFirehose(&FirehoseConfig{ + ConcurrentBlockFlushing: true, ApplyBackwardCompatibility: ptr(false), private: &privateFirehoseConfig{ FlushToTestBuffer: true, From 7eaa0cb83b16f996bf34c48eec3752ef91f38892 Mon Sep 17 00:00:00 2001 From: David Zhou Date: Tue, 13 May 2025 15:01:56 -0400 Subject: [PATCH 12/18] 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) From 42aa31c86709eda020190b625a5a147afeded42e Mon Sep 17 00:00:00 2001 From: David Zhou Date: Tue, 13 May 2025 15:24:48 -0400 Subject: [PATCH 13/18] PR: Part 2 --- eth/tracers/firehose_concurrency_test.go | 26 +++++++------------ .../tracetest/firehose/firehose_test.go | 19 +++++++++++--- 2 files changed, 24 insertions(+), 21 deletions(-) diff --git a/eth/tracers/firehose_concurrency_test.go b/eth/tracers/firehose_concurrency_test.go index 78ae9bc95e..70188a6d40 100644 --- a/eth/tracers/firehose_concurrency_test.go +++ b/eth/tracers/firehose_concurrency_test.go @@ -24,7 +24,7 @@ func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) { f.OnBlockchainInit(params.AllEthashProtocolChanges) - blockNumbers := []uint64{123, 124, 125} + blockNumbers := []uint64{0} for i, blockNum := range blockNumbers { f.OnBlockStart(blockEvent(blockNum)) @@ -40,24 +40,16 @@ func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) { f.OnClose() - output := f.InternalTestingBuffer().String() + lines := strings.Split(strings.TrimSpace(f.InternalTestingBuffer().String()), "\n") + require.Len(t, lines, 2) - outNumber := make([]string, 0) - for i, line := range strings.Split(output, "\n") { - if i == 0 { - require.Equal(t, "FIRE INIT 3.0 geth 1.15.10", line) - continue - } + require.Equal(t, "FIRE INIT 3.0 geth 1.15.10", lines[0]) - fields := strings.SplitN(line, " ", 4) - if len(fields) >= 3 { - require.Equal(t, "FIRE", fields[0]) - require.Equal(t, "BLOCK", fields[1]) - outNumber = append(outNumber, fields[2]) - } - } - - require.Equal(t, []string{"123", "124", "125"}, outNumber) + fields := strings.SplitN(lines[1], " ", 4) + require.GreaterOrEqual(t, len(fields), 3) + require.Equal(t, "FIRE", fields[0]) + require.Equal(t, "BLOCK", fields[1]) + require.Equal(t, "0", fields[2]) } func TestFirehose_BlocksPrintToFirehose_MultipleBlocksInOrder(t *testing.T) { diff --git a/eth/tracers/internal/tracetest/firehose/firehose_test.go b/eth/tracers/internal/tracetest/firehose/firehose_test.go index 59959805b4..bbc1d1ed5d 100644 --- a/eth/tracers/internal/tracetest/firehose/firehose_test.go +++ b/eth/tracers/internal/tracetest/firehose/firehose_test.go @@ -1,6 +1,7 @@ package firehose_test import ( + "fmt" "math/big" "path/filepath" "strings" @@ -37,13 +38,17 @@ func TestFirehosePrestate(t *testing.T) { "./testdata/TestFirehosePrestate/suicide_double_withdraw", "./testdata/TestFirehosePrestate/extra_account_creations", } - + for _, concurrent := range []bool{true, false} { for _, folder := range testFolders { name := filepath.Base(folder) + concurrencyLabel := "sequential" + if concurrent { + concurrencyLabel = "concurrent" + } for _, model := range tracingModels { - t.Run(string(model)+"/"+name, func(t *testing.T) { + t.Run(fmt.Sprintf("%s/%s/%s", model, name, concurrencyLabel), func(t *testing.T) { tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model, concurrent) defer onClose() @@ -52,7 +57,8 @@ func TestFirehosePrestate(t *testing.T) { tracer.CloseBlockPrintQueue() genesisLine, blockLines, unknownLines := readTracerFirehoseLines(t, tracer) - require.Len(t, unknownLines, 0, "Lines:\n%s", strings.Join(slicesMap(unknownLines, func(l unknownLine) string { return "- '" + string(l) + "'" }), "\n")) + require.Len(t, unknownLines, 0, "Lines:\n%s", strings.Join( + slicesMap(unknownLines, func(l unknownLine) string { return "- '" + string(l) + "'" }), "\n")) require.NotNil(t, genesisLine) blockLines.assertOnlyBlockEquals(t, filepath.Join(folder, string(model)), 1) }) @@ -189,8 +195,13 @@ func testBlockTracesCorrectly(t *testing.T, genesisSpec *core.Genesis, engine co t.Helper() for _, concurrent := range []bool{true, false} { + concurrencyLabel := "sequential" + if concurrent { + concurrencyLabel = "concurrent" + } + for _, model := range tracingModels { - t.Run(string(model), func(t *testing.T) { + t.Run(fmt.Sprintf("%s/%s", model, concurrencyLabel), func(t *testing.T) { tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model, concurrent) defer onClose() From 0d26c9a9b760012d747b7daf2d2cfdc2ca6eb275 Mon Sep 17 00:00:00 2001 From: David Zhou Date: Tue, 13 May 2025 15:57:18 -0400 Subject: [PATCH 14/18] PR: Part 3 --- .../internal/tracetest/firehose/firehose_test.go | 13 +++++++++++-- .../internal/tracetest/firehose/helpers_test.go | 4 ++-- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/eth/tracers/internal/tracetest/firehose/firehose_test.go b/eth/tracers/internal/tracetest/firehose/firehose_test.go index bbc1d1ed5d..1d7d3fd248 100644 --- a/eth/tracers/internal/tracetest/firehose/firehose_test.go +++ b/eth/tracers/internal/tracetest/firehose/firehose_test.go @@ -2,6 +2,7 @@ package firehose_test import ( "fmt" + "github.com/ethereum/go-ethereum/eth/tracers" "math/big" "path/filepath" "strings" @@ -49,7 +50,11 @@ func TestFirehosePrestate(t *testing.T) { for _, model := range tracingModels { t.Run(fmt.Sprintf("%s/%s/%s", model, name, concurrencyLabel), func(t *testing.T) { - tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model, concurrent) + config := &tracers.FirehoseConfig{ + ConcurrentBlockFlushing: concurrent, + } + + tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model, config) defer onClose() runPrestateBlock(t, filepath.Join(folder, "prestate.json"), tracingHooks) @@ -202,7 +207,11 @@ func testBlockTracesCorrectly(t *testing.T, genesisSpec *core.Genesis, engine co for _, model := range tracingModels { t.Run(fmt.Sprintf("%s/%s", model, concurrencyLabel), func(t *testing.T) { - tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model, concurrent) + config := &tracers.FirehoseConfig{ + ConcurrentBlockFlushing: concurrent, + } + + tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model, config) defer onClose() chain, err := core.NewBlockChain(rawdb.NewMemoryDatabase(), nil, genesisSpec, nil, engine, vm.Config{Tracer: tracingHooks}, nil) diff --git a/eth/tracers/internal/tracetest/firehose/helpers_test.go b/eth/tracers/internal/tracetest/firehose/helpers_test.go index 0541d37748..e7f4a4017e 100644 --- a/eth/tracers/internal/tracetest/firehose/helpers_test.go +++ b/eth/tracers/internal/tracetest/firehose/helpers_test.go @@ -29,7 +29,7 @@ type firehoseInitLine struct { type firehoseBlockLines []firehoseBlockLine -func newFirehoseTestTracer(t *testing.T, model tracingModel, concurrentBlockFlushing bool) (*tracers.Firehose, *tracing.Hooks, func()) { +func newFirehoseTestTracer(t *testing.T, model tracingModel, config *tracers.FirehoseConfig) (*tracers.Firehose, *tracing.Hooks, func()) { t.Helper() tracer, err := tracers.NewFirehoseFromRawJSON([]byte(fmt.Sprintf(`{ @@ -39,7 +39,7 @@ func newFirehoseTestTracer(t *testing.T, model tracingModel, concurrentBlockFlus "ignoreGenesisBlock": true, "forcedBackwardCompatibility": %t } - }`, concurrentBlockFlushing, model == tracingModelFirehose2_3))) + }`, config.ConcurrentBlockFlushing, model == tracingModelFirehose2_3))) require.NoError(t, err) hooks := tracers.NewTracingHooksFromFirehose(tracer) From eefb9b9830a9e5ecb197bf98fe2c3029eddec215 Mon Sep 17 00:00:00 2001 From: David Zhou Date: Wed, 14 May 2025 10:42:39 -0400 Subject: [PATCH 15/18] DOC: Buffer specification in report --- eth/tracers/firehose_concurrency.md | 1 + 1 file changed, 1 insertion(+) diff --git a/eth/tracers/firehose_concurrency.md b/eth/tracers/firehose_concurrency.md index a47734595b..3c9f82c936 100644 --- a/eth/tracers/firehose_concurrency.md +++ b/eth/tracers/firehose_concurrency.md @@ -80,6 +80,7 @@ The implementation was validated at multiple levels to ensure correctness, confi * With `concurrentBlockFlushing: false`, the node was synced to block 10,000, the data directory (`./geth`) was removed, and then resynced up to block 100,000. * The same steps were repeated with `concurrentBlockFlushing: true`. + Note: A channel with a buffer of 100 was created to allow the tasks to queue without blocking the producer. \ The `time` command outputs wall-clock time and system/user CPU usage upon completion, providing a baseline for comparing performance between the linear and concurrent implementations. This methodology enables a controlled, reproducible environment for evaluating the effectiveness of the concurrent block flushing feature. ## Section 4: Analysis From c3199f0f60dc33ad3ca03eb7d12de018604fbd0c Mon Sep 17 00:00:00 2001 From: David Zhou Date: Wed, 14 May 2025 16:51:07 -0400 Subject: [PATCH 16/18] DOC: Second report --- eth/tracers/firehose_concurrency_2.md | 197 ++++++++++++++++++++++++++ 1 file changed, 197 insertions(+) create mode 100644 eth/tracers/firehose_concurrency_2.md diff --git a/eth/tracers/firehose_concurrency_2.md b/eth/tracers/firehose_concurrency_2.md new file mode 100644 index 0000000000..fc09e73bd8 --- /dev/null +++ b/eth/tracers/firehose_concurrency_2.md @@ -0,0 +1,197 @@ +# Report 2: Concurrent Block Flushing in Tracer Processing System + +## Section 1: Introduction + +Currently, the tracer processing system operates in a linear fashion, requiring the complete processing of a block's code before proceeding to the next one. As a result, computationally expensive operations—such as `proto.Marshal` and base64 encoding, which are essential for flushing a block to the firehose—can introduce significant delays. To mitigate this bottleneck, a potential solution is to leverage concurrency by introducing goroutines. These heavy operations can be offloaded to a separate channel, enabling asynchronous processing. This approach allows the main execution flow to begin processing subsequent blocks while previous ones are being flushed, thereby improving overall throughput and reducing latency. The goal of this report is to outline the solution and provide benchmarking results. + +## Section 2: Background + +`proto.Marshal`, part of Protocol Buffers (protobuf), serializes structured data into a compact binary format. While efficient in output size, the process of traversing and encoding complex data structures—especially those with deeply nested or repeated fields—can be CPU-intensive. Additionally, memory allocations during marshaling can introduce further overhead. + +Similarly, base64 encoding, which transforms binary data into an ASCII string format for transmission or storage, involves non-trivial byte-wise transformations and increases the data size. This added computational cost becomes significant when processing large blocks or high-throughput workloads. + +Together, these operations introduce latency in a linear processing pipeline. + +## Section 3: Method + +### 3.1 Proposed Solution + +A critical point in the tracer processing system is the `OnBlockEnd` hook, which invokes the `printBlockToFirehose` method. This method includes computationally expensive operations such as `proto.Marshal` and base64 encoding, which are necessary to serialize and flush the block data to the firehose. + + +To address this, the proposed solution introduces a concurrency mechanism. Specifically, the user specifies a number of worker goroutines that will be concurrently working through the `FirehoseConfig.ConcurrencyBlockFlushing` configuration. There are three layers to this process: + +1. A worker queue channel with a buffer of 100 is created. All `printBlockToFirehose` tasks are enqueued while waiting for a worker to take and process it. +2. A number of workers specified by the configuration will be taking and processing these tasks asynchronously until the data is in a byte format and ready to be sent to stdout. The goroutine will then send that data, along with the block number, to a second channel. +3. A final channel is created to store pairs of (block number, [byte]). This channel allows linear flushing by only allowing the current expected block number to be flushed, while storing the remaining blocks until it is their turn. To achieve this, a block number is stored globally the first time `OnBlockEnd` is called. Therefore, the channel will know the first block number expected to be flushed and will increment the expected number by one each time. + +### 3.2 Validation and Performance Metrics + +The implementation was validated at multiple levels to ensure correctness, configurability, and performance improvements of the concurrent block flushing mechanism. + +1. **Unit-Level Validation** + + To confirm the functional correctness of the concurrent flushing implementation, a unit test was written that creates and processes 1,000 blocks, flushing each to an `InternalTestingBuffer`. The results were then compared against expected outputs to verify equivalence. The test confirms that the output produced by the concurrent mechanism matches that of the original linear method, thereby validating correctness at the unit level. +2. **Integration Testing on Battlefield-Ethereum** + + To verify integration within the battlefield-ethereum environment, the feature was exposed via a new configuration flag: `CONCURRENT_BLOCK_FLUSHING`. This flag determines the number of workers that will be processing tasks concurrently. \ + + Original (linear flushing): + + ```bash + ./scripts/run_firehose_geth_dev.sh 3.0 prague + ``` + + Concurrent flushing enabled: + + ```bash + CONCURRENT_BLOCK_FLUSHING=1 ./scripts/run_firehose_geth_dev.sh 3.0 prague + ``` + + Behavioral differences were observed through log output. In the concurrent mode, log lines such as: + + ``` + "Firehose closing, flushing queued blocks to standard output" + ``` + + appear when the program is interrupted (e.g., via Ctrl + C), indicating that the concurrent flushing logic and cleanup path are active. These lines are absent in the linear configuration, confirming that the switch is functioning as intended. + + Furthermore, when running the integration test suite using: + + ```bash + pnpm test:fh3.0:geth-dev + ``` + + all tests passed successfully (64 passing), indicating that the concurrent implementation does not introduce regressions in battlefield compatibility. +3. **Performance Benchmarking** + + To quantify performance differences, benchmarking was conducted using firehose-ethereum. The following command was used for both the baseline and concurrent configurations: + + ```bash + time geth --vmtrace=firehose \ + --vmtrace.jsonconfig='{"concurrentBlockFlushing":}' \ + --synctarget= \ + --syncmode=full --holesky --datadir=./geth --db.engine=pebble \ + --state.scheme=path --port=30305 --authrpc.jwtsecret=jwt.txt \ + --authrpc.addr=0.0.0.0 --authrpc.port=9551 --authrpc.vhosts="*" \ + --http --http.addr=0.0.0.0 --http.api=eth,net,web3 --http.port=9545 \ + --http.vhosts="*" --port=40303 --ws.port=9546 --ipcpath=/tmp/geth.ipc > /dev/null + ``` + + The benchmark was conducted in four phases: + + * With `concurrentBlockFlushing: 0`, the node was synced to block 10,000, the data directory (`./geth`) was removed, and then resynced up to block 100,000. + * The same steps were repeated with `concurrentBlockFlushing: 10`. + * The same steps were repeated with `concurrentBlockFlushing: 100`. + * The same steps were repeated with `concurrentBlockFlushing: 1000`. + + The `time` command outputs wall-clock time and system/user CPU usage upon completion, providing a baseline for comparing performance between the linear and concurrent implementations. This methodology enables a controlled, reproducible environment for evaluating the effectiveness of the concurrent block flushing feature. + +## 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: + +### Until Block 10,000 + +#### No Concurrency + +- Run 1: 84.43s user, 15.27s system, 61% CPU, 2:41.39 total +- Run 2: 86.76s user, 15.93s system, 64% CPU, 2:40.36 total +- Run 3: 78.30s user, 14.62s system, 55% CPU, 2:47.51 total +- Run 4: 78.29s user, 13.72s system, 55% CPU, 2:44.45 total +- Run 5: 83.29s user, 13.37s system, 59% CPU, 2:43.47 total + +#### Concurrency (10 Goroutines) + +- Run 1: 68.14s user, 13.85s system, 51% CPU, 2:39.27 total +- Run 2: 70.81s user, 14.10s system, 52% CPU, 2:40.37 total +- Run 3: 64.79s user, 13.65s system, 49% CPU, 2:39.22 total +- Run 4: 64.34s user, 15.18s system, 48% CPU, 2:43.25 total +- Run 5: 70.21s user, 15.58s system, 51% CPU, 2:47.34 total + +#### Concurrency (100 Goroutines) + +- Run 1: 71.19s user, 16.06s system, 53% CPU, 2:42.77 total +- Run 2: 66.74s user, 12.12s system, 51% CPU, 2:34.47 total +- Run 3: 72.53s user, 16.11s system, 56% CPU, 2:37.50 total +- Run 4: 70.99s user, 14.02s system, 56% CPU, 2:29.46 total +- Run 5: 65.86s user, 13.66s system, 54% CPU, 2:24.97 total + +#### Concurrency (1000 Goroutines) + +- Run 1: 68.44s user, 14.65s system, 54% CPU, 2:33.83 total +- Run 2: 72.79s user, 15.06s system, 55% CPU, 2:38.76 total +- Run 3: 65.60s user, 13.19s system, 53% CPU, 2:28.36 total +- Run 4: 71.82s user, 15.05s system, 54% CPU, 2:38.10 total +- Run 5: 71.36s user, 14.87s system, 56% CPU, 2:33.32 total + +--- + +### Until Block 100,000 + +#### No Concurrency + +- 331.03s user, 165.76s system, 32% CPU, 25:28.66 total + +#### Concurrency (10 Goroutines) + +- 383.44s user, 183.39s system, 37% CPU, 25:22.84 total + +#### Concurrency (100 Goroutines) + +- 399.57s user, 175.51s system, 37% CPU, 25:28.03 total + +#### Concurrency (1000 Goroutines) + +- 346.83s user, 165.96s system, 34% CPU, 25:02.78 total + +--- + +### Table 1: Result Comparison with and Without Concurrency + +#### Block 10,000 Summary + +| Goroutines | User Time | System Time | CPU | Total Time | +|------------|-----------|-------------|-------|------------| +| 0 | 82.214s | 14.582s | 58.8% | 2:43.44 | +| 10 | 67.666s | 14.270s | 50.2% | 2:41.89 | +| 100 | 69.462s | 14.394s | 54.0% | 2:33.83 | +| 1000 | 69.990s | 14.564s | 54.4% | 2:34.47 | + +#### Block 100,000 Summary + +| Goroutines | User Time | System Time | CPU | Total Time | +|------------|-----------|-------------|--------|-------------| +| 0 | 331.03s | 165.76s | 32.0% | 25:28.66 | +| 10 | 383.44s | 183.39s | 37.0% | 25:22.84 | +| 100 | 399.57s | 175.51s | 37.0% | 25:28.03 | +| 1000 | 346.83s | 165.96s | 34.0% | 25:02.78 | + +--- + +### 4.2 Discussion + +The four parameters analyzed are **user**, **system**, **CPU**, and **total**. + +- **System** and **CPU** times do not vary significantly with the implementation of goroutines. +- **User** and **Total** time are generally lower with the use of concurrency for 10,000 blocks. +- However, performance improvement for 100,000 blocks is less conclusive. + +This could be due to: +- Single-run variance +- Varying system conditions during execution +- Larger processing overhead over a long duration + +More runs and broader statistical analysis may be required to solidify findings. + +## Section 5: Conclusion + +The introduction of concurrency to the block flushing process demonstrates performance improvements for smaller workloads (up to 10,000 blocks), particularly in reducing user and total processing time. While results for larger workloads (100,000 blocks) show less consistent gains. From 609843a24e5e52538982929b24a421d974d7dd1f Mon Sep 17 00:00:00 2001 From: David Zhou Date: Thu, 15 May 2025 09:03:14 -0400 Subject: [PATCH 17/18] DEL: Report 2 --- eth/tracers/firehose_concurrency_2.md | 197 -------------------------- 1 file changed, 197 deletions(-) delete mode 100644 eth/tracers/firehose_concurrency_2.md diff --git a/eth/tracers/firehose_concurrency_2.md b/eth/tracers/firehose_concurrency_2.md deleted file mode 100644 index fc09e73bd8..0000000000 --- a/eth/tracers/firehose_concurrency_2.md +++ /dev/null @@ -1,197 +0,0 @@ -# Report 2: Concurrent Block Flushing in Tracer Processing System - -## Section 1: Introduction - -Currently, the tracer processing system operates in a linear fashion, requiring the complete processing of a block's code before proceeding to the next one. As a result, computationally expensive operations—such as `proto.Marshal` and base64 encoding, which are essential for flushing a block to the firehose—can introduce significant delays. To mitigate this bottleneck, a potential solution is to leverage concurrency by introducing goroutines. These heavy operations can be offloaded to a separate channel, enabling asynchronous processing. This approach allows the main execution flow to begin processing subsequent blocks while previous ones are being flushed, thereby improving overall throughput and reducing latency. The goal of this report is to outline the solution and provide benchmarking results. - -## Section 2: Background - -`proto.Marshal`, part of Protocol Buffers (protobuf), serializes structured data into a compact binary format. While efficient in output size, the process of traversing and encoding complex data structures—especially those with deeply nested or repeated fields—can be CPU-intensive. Additionally, memory allocations during marshaling can introduce further overhead. - -Similarly, base64 encoding, which transforms binary data into an ASCII string format for transmission or storage, involves non-trivial byte-wise transformations and increases the data size. This added computational cost becomes significant when processing large blocks or high-throughput workloads. - -Together, these operations introduce latency in a linear processing pipeline. - -## Section 3: Method - -### 3.1 Proposed Solution - -A critical point in the tracer processing system is the `OnBlockEnd` hook, which invokes the `printBlockToFirehose` method. This method includes computationally expensive operations such as `proto.Marshal` and base64 encoding, which are necessary to serialize and flush the block data to the firehose. - - -To address this, the proposed solution introduces a concurrency mechanism. Specifically, the user specifies a number of worker goroutines that will be concurrently working through the `FirehoseConfig.ConcurrencyBlockFlushing` configuration. There are three layers to this process: - -1. A worker queue channel with a buffer of 100 is created. All `printBlockToFirehose` tasks are enqueued while waiting for a worker to take and process it. -2. A number of workers specified by the configuration will be taking and processing these tasks asynchronously until the data is in a byte format and ready to be sent to stdout. The goroutine will then send that data, along with the block number, to a second channel. -3. A final channel is created to store pairs of (block number, [byte]). This channel allows linear flushing by only allowing the current expected block number to be flushed, while storing the remaining blocks until it is their turn. To achieve this, a block number is stored globally the first time `OnBlockEnd` is called. Therefore, the channel will know the first block number expected to be flushed and will increment the expected number by one each time. - -### 3.2 Validation and Performance Metrics - -The implementation was validated at multiple levels to ensure correctness, configurability, and performance improvements of the concurrent block flushing mechanism. - -1. **Unit-Level Validation** - - To confirm the functional correctness of the concurrent flushing implementation, a unit test was written that creates and processes 1,000 blocks, flushing each to an `InternalTestingBuffer`. The results were then compared against expected outputs to verify equivalence. The test confirms that the output produced by the concurrent mechanism matches that of the original linear method, thereby validating correctness at the unit level. -2. **Integration Testing on Battlefield-Ethereum** - - To verify integration within the battlefield-ethereum environment, the feature was exposed via a new configuration flag: `CONCURRENT_BLOCK_FLUSHING`. This flag determines the number of workers that will be processing tasks concurrently. \ - - Original (linear flushing): - - ```bash - ./scripts/run_firehose_geth_dev.sh 3.0 prague - ``` - - Concurrent flushing enabled: - - ```bash - CONCURRENT_BLOCK_FLUSHING=1 ./scripts/run_firehose_geth_dev.sh 3.0 prague - ``` - - Behavioral differences were observed through log output. In the concurrent mode, log lines such as: - - ``` - "Firehose closing, flushing queued blocks to standard output" - ``` - - appear when the program is interrupted (e.g., via Ctrl + C), indicating that the concurrent flushing logic and cleanup path are active. These lines are absent in the linear configuration, confirming that the switch is functioning as intended. - - Furthermore, when running the integration test suite using: - - ```bash - pnpm test:fh3.0:geth-dev - ``` - - all tests passed successfully (64 passing), indicating that the concurrent implementation does not introduce regressions in battlefield compatibility. -3. **Performance Benchmarking** - - To quantify performance differences, benchmarking was conducted using firehose-ethereum. The following command was used for both the baseline and concurrent configurations: - - ```bash - time geth --vmtrace=firehose \ - --vmtrace.jsonconfig='{"concurrentBlockFlushing":}' \ - --synctarget= \ - --syncmode=full --holesky --datadir=./geth --db.engine=pebble \ - --state.scheme=path --port=30305 --authrpc.jwtsecret=jwt.txt \ - --authrpc.addr=0.0.0.0 --authrpc.port=9551 --authrpc.vhosts="*" \ - --http --http.addr=0.0.0.0 --http.api=eth,net,web3 --http.port=9545 \ - --http.vhosts="*" --port=40303 --ws.port=9546 --ipcpath=/tmp/geth.ipc > /dev/null - ``` - - The benchmark was conducted in four phases: - - * With `concurrentBlockFlushing: 0`, the node was synced to block 10,000, the data directory (`./geth`) was removed, and then resynced up to block 100,000. - * The same steps were repeated with `concurrentBlockFlushing: 10`. - * The same steps were repeated with `concurrentBlockFlushing: 100`. - * The same steps were repeated with `concurrentBlockFlushing: 1000`. - - The `time` command outputs wall-clock time and system/user CPU usage upon completion, providing a baseline for comparing performance between the linear and concurrent implementations. This methodology enables a controlled, reproducible environment for evaluating the effectiveness of the concurrent block flushing feature. - -## 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: - -### Until Block 10,000 - -#### No Concurrency - -- Run 1: 84.43s user, 15.27s system, 61% CPU, 2:41.39 total -- Run 2: 86.76s user, 15.93s system, 64% CPU, 2:40.36 total -- Run 3: 78.30s user, 14.62s system, 55% CPU, 2:47.51 total -- Run 4: 78.29s user, 13.72s system, 55% CPU, 2:44.45 total -- Run 5: 83.29s user, 13.37s system, 59% CPU, 2:43.47 total - -#### Concurrency (10 Goroutines) - -- Run 1: 68.14s user, 13.85s system, 51% CPU, 2:39.27 total -- Run 2: 70.81s user, 14.10s system, 52% CPU, 2:40.37 total -- Run 3: 64.79s user, 13.65s system, 49% CPU, 2:39.22 total -- Run 4: 64.34s user, 15.18s system, 48% CPU, 2:43.25 total -- Run 5: 70.21s user, 15.58s system, 51% CPU, 2:47.34 total - -#### Concurrency (100 Goroutines) - -- Run 1: 71.19s user, 16.06s system, 53% CPU, 2:42.77 total -- Run 2: 66.74s user, 12.12s system, 51% CPU, 2:34.47 total -- Run 3: 72.53s user, 16.11s system, 56% CPU, 2:37.50 total -- Run 4: 70.99s user, 14.02s system, 56% CPU, 2:29.46 total -- Run 5: 65.86s user, 13.66s system, 54% CPU, 2:24.97 total - -#### Concurrency (1000 Goroutines) - -- Run 1: 68.44s user, 14.65s system, 54% CPU, 2:33.83 total -- Run 2: 72.79s user, 15.06s system, 55% CPU, 2:38.76 total -- Run 3: 65.60s user, 13.19s system, 53% CPU, 2:28.36 total -- Run 4: 71.82s user, 15.05s system, 54% CPU, 2:38.10 total -- Run 5: 71.36s user, 14.87s system, 56% CPU, 2:33.32 total - ---- - -### Until Block 100,000 - -#### No Concurrency - -- 331.03s user, 165.76s system, 32% CPU, 25:28.66 total - -#### Concurrency (10 Goroutines) - -- 383.44s user, 183.39s system, 37% CPU, 25:22.84 total - -#### Concurrency (100 Goroutines) - -- 399.57s user, 175.51s system, 37% CPU, 25:28.03 total - -#### Concurrency (1000 Goroutines) - -- 346.83s user, 165.96s system, 34% CPU, 25:02.78 total - ---- - -### Table 1: Result Comparison with and Without Concurrency - -#### Block 10,000 Summary - -| Goroutines | User Time | System Time | CPU | Total Time | -|------------|-----------|-------------|-------|------------| -| 0 | 82.214s | 14.582s | 58.8% | 2:43.44 | -| 10 | 67.666s | 14.270s | 50.2% | 2:41.89 | -| 100 | 69.462s | 14.394s | 54.0% | 2:33.83 | -| 1000 | 69.990s | 14.564s | 54.4% | 2:34.47 | - -#### Block 100,000 Summary - -| Goroutines | User Time | System Time | CPU | Total Time | -|------------|-----------|-------------|--------|-------------| -| 0 | 331.03s | 165.76s | 32.0% | 25:28.66 | -| 10 | 383.44s | 183.39s | 37.0% | 25:22.84 | -| 100 | 399.57s | 175.51s | 37.0% | 25:28.03 | -| 1000 | 346.83s | 165.96s | 34.0% | 25:02.78 | - ---- - -### 4.2 Discussion - -The four parameters analyzed are **user**, **system**, **CPU**, and **total**. - -- **System** and **CPU** times do not vary significantly with the implementation of goroutines. -- **User** and **Total** time are generally lower with the use of concurrency for 10,000 blocks. -- However, performance improvement for 100,000 blocks is less conclusive. - -This could be due to: -- Single-run variance -- Varying system conditions during execution -- Larger processing overhead over a long duration - -More runs and broader statistical analysis may be required to solidify findings. - -## Section 5: Conclusion - -The introduction of concurrency to the block flushing process demonstrates performance improvements for smaller workloads (up to 10,000 blocks), particularly in reducing user and total processing time. While results for larger workloads (100,000 blocks) show less consistent gains. From 3c9a6ec0b53b529a421d1e5c15c39ba0f843ad8c Mon Sep 17 00:00:00 2001 From: David Zhou Date: Fri, 16 May 2025 12:44:42 -0400 Subject: [PATCH 18/18] TEST: Fix test to not restrict geth version --- eth/tracers/firehose_concurrency_test.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/eth/tracers/firehose_concurrency_test.go b/eth/tracers/firehose_concurrency_test.go index 70188a6d40..8b865e9ec4 100644 --- a/eth/tracers/firehose_concurrency_test.go +++ b/eth/tracers/firehose_concurrency_test.go @@ -43,7 +43,10 @@ func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) { lines := strings.Split(strings.TrimSpace(f.InternalTestingBuffer().String()), "\n") require.Len(t, lines, 2) - require.Equal(t, "FIRE INIT 3.0 geth 1.15.10", lines[0]) + fieldsInit := strings.SplitN(lines[0], " ", 3) + require.Equal(t, "FIRE", fieldsInit[0]) + require.Equal(t, "INIT", fieldsInit[1]) + require.Contains(t, fieldsInit[2], "geth") fields := strings.SplitN(lines[1], " ", 4) require.GreaterOrEqual(t, len(fields), 3)