Merge pull request #14 from streamingfast/david-zhou/worker-queue

DEV1: Implementation of one goroutine for concurrent flushing

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,
 
 		OnTxStart: tracer.OnTxStart,
 		OnTxEnd:   tracer.OnTxEnd,
@@ -133,6 +134,7 @@ func NewTracingHooksFromFirehose(tracer *Firehose) *tracing.Hooks {
 
 type FirehoseConfig struct {
 	ApplyBackwardCompatibility *bool `json:"applyBackwardCompatibility"`
+	ConcurrentBlockFlushing    bool  `json:"concurrentBlockFlushing"`
 
 	// Only used for testing, only possible through JSON configuration
 	private *privateFirehoseConfig
@@ -153,6 +155,7 @@ func (c *FirehoseConfig) LogKeyValues() []any {
 
 	return []any{
 		"config.applyBackwardCompatibility", applyBackwardCompatibility,
+		"config.concurrentBlockFlushing", c.ConcurrentBlockFlushing,
 	}
 }
 
@@ -173,6 +176,7 @@ type Firehose struct {
 	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
@@ -182,6 +186,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
@@ -194,6 +199,8 @@ type Firehose struct {
 	blockReorderOrdinalSnapshot uint64
 	blockReorderOrdinalOnce     sync.Once
 	blockIsGenesis              bool
+	blockPrintQueue             chan *blockPrintJob
+	blockFlushDone              sync.WaitGroup
 
 	// Transaction state
 	evm                  *tracing.VMContext
@@ -250,6 +257,7 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
 		hasher:                     crypto.NewKeccakState(),
 		tracerID:                   "global",
 		applyBackwardCompatibility: config.ApplyBackwardCompatibility,
+		concurrentBlockFlushing:    config.ConcurrentBlockFlushing,
 
 		// Block state
 		blockOrdinal:        &Ordinal{},
@@ -272,6 +280,14 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
 		}
 	}
 
+	if config.ConcurrentBlockFlushing {
+		log.Info("Firehose concurrent block flushing enabled, starting block " +
+			"print worker goroutine")
+		firehose.blockPrintQueue = make(chan *blockPrintJob, 100)
+		firehose.blockFlushDone.Add(1)
+		go firehose.blockPrintWorker()
+	}
+
 	return firehose
 }
 
@@ -508,7 +524,18 @@ func (f *Firehose) OnBlockEnd(err error) {
 		}
 
 		f.ensureInBlockAndNotInTrx()
+
+		// 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)
+		}
+
 	} 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)
@@ -626,6 +653,13 @@ func (f *Firehose) reorderCallOrdinals(call *pbeth.Call, ordinalBase uint64) (or
 	return call.EndOrdinal
 }
 
+func (f *Firehose) OnClose() {
+	if f.concurrentBlockFlushing {
+		log.Info("Firehose closing, flushing queued blocks to standard output")
+		f.CloseBlockPrintQueue()
+	}
+}
+
 func (f *Firehose) OnSystemCallStart() {
 	firehoseInfo("system call start")
 	f.ensureInBlockAndNotInTrx()
@@ -1877,6 +1911,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()
@@ -2844,3 +2879,27 @@ func (m Memory) GetPtr(offset, size int64) []byte {
 	reminder := m[min(offset, int64(len(m))):]
 	return append(reminder, make([]byte, int(size)-len(reminder))...)
 }
+
+type blockPrintJob struct {
+	block    *pbeth.Block
+	finality *FinalityStatus
+}
+
+func (f *Firehose) blockPrintWorker() {
+	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.closeChannels.Do(func() {
+			close(f.blockPrintQueue)
+			f.blockFlushDone.Wait()
+		})
+	}
+}

eth/tracers/firehose_concurrency.md

@@ -0,0 +1,144 @@
+# 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":<true|false>}' \
+    --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`.
+
+    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
+
+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 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.

eth/tracers/firehose_concurrency_test.go

@@ -0,0 +1,146 @@
+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_SingleBlock(t *testing.T) {
+
+	f := NewFirehose(&FirehoseConfig{
+		ConcurrentBlockFlushing:    true,
+		ApplyBackwardCompatibility: ptr(false),
+		private: &privateFirehoseConfig{
+			FlushToTestBuffer: true,
+		},
+	})
+
+	f.OnBlockchainInit(params.AllEthashProtocolChanges)
+
+	blockNumbers := []uint64{0}
+
+	for i, blockNum := range blockNumbers {
+		f.OnBlockStart(blockEvent(blockNum))
+
+		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)
+	}
+
+	f.OnClose()
+
+	lines := strings.Split(strings.TrimSpace(f.InternalTestingBuffer().String()), "\n")
+	require.Len(t, lines, 2)
+
+	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)
+	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) {
+
+	const blockCount = 100
+	const baseBlockNum = 1000
+
+	f := NewFirehose(&FirehoseConfig{
+		ConcurrentBlockFlushing:    true,
+		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.OnClose()
+
+	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
+	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 {
+		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
+}

eth/tracers/internal/tracetest/firehose/firehose_test.go

@@ -1,6 +1,8 @@
 package firehose_test
 
 import (
+	"fmt"
+	"github.com/ethereum/go-ethereum/eth/tracers"
 	"math/big"
 	"path/filepath"
 	"strings"
@@ -39,25 +41,36 @@ func TestFirehosePrestate(t *testing.T) {
 		"./testdata/TestFirehosePrestate/keccak256_memory_out_of_bounds",
 	}
 
+	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) {
-				tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model)
+				t.Run(fmt.Sprintf("%s/%s/%s", model, name, concurrencyLabel), func(t *testing.T) {
+					config := &tracers.FirehoseConfig{
+						ConcurrentBlockFlushing: concurrent,
+					}
+
+					tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model, config)
 					defer onClose()
 
 					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.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,9 +200,19 @@ func TestFirehose_SystemCalls(t *testing.T) {
 func testBlockTracesCorrectly(t *testing.T, genesisSpec *core.Genesis, engine consensus.Engine, blocks []*types.Block, goldenDir string) {
 	t.Helper()
 
+	for _, concurrent := range []bool{true, false} {
+		concurrencyLabel := "sequential"
+		if concurrent {
+			concurrencyLabel = "concurrent"
+		}
+
 		for _, model := range tracingModels {
-		t.Run(string(model), func(t *testing.T) {
-			tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model)
+			t.Run(fmt.Sprintf("%s/%s", model, concurrencyLabel), func(t *testing.T) {
+				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)
@@ -204,7 +227,10 @@ 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))
 			})
 		}
 	}
+}

eth/tracers/internal/tracetest/firehose/helpers_test.go

@@ -29,16 +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, config *tracers.FirehoseConfig) (*tracers.Firehose, *tracing.Hooks, func()) {
 	t.Helper()
 
 	tracer, err := tracers.NewFirehoseFromRawJSON([]byte(fmt.Sprintf(`{
+		"concurrentBlockFlushing": %t,
 		"_private": {
 			"flushToTestBuffer": true,
 			"ignoreGenesisBlock": true,
 			"forcedBackwardCompatibility": %t
 		}
-	}`, model == tracingModelFirehose2_3)))
+	}`, config.ConcurrentBlockFlushing, model == tracingModelFirehose2_3)))
 	require.NoError(t, err)
 
 	hooks := tracers.NewTracingHooksFromFirehose(tracer)