mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-25 06:06:44 +00:00
Merge branch 'firehose-fh3.0' into release/geth-1.15.11-fh3.0
This commit is contained in:
Matthieu Vachon
commit
03454c844b
7 changed files with 705 additions and 45 deletions
|
|
@ -9,6 +9,7 @@ import (
|
|||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"math/big"
|
||||
"os"
|
||||
"regexp"
|
||||
|
|
@ -78,6 +79,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 +135,7 @@ func NewTracingHooksFromFirehose(tracer *Firehose) *tracing.Hooks {
|
|||
|
||||
type FirehoseConfig struct {
|
||||
ApplyBackwardCompatibility *bool `json:"applyBackwardCompatibility"`
|
||||
ConcurrentBlockFlushing int `json:"concurrentBlockFlushing"`
|
||||
|
||||
// Only used for testing, only possible through JSON configuration
|
||||
private *privateFirehoseConfig
|
||||
|
|
@ -153,6 +156,7 @@ func (c *FirehoseConfig) LogKeyValues() []any {
|
|||
|
||||
return []any{
|
||||
"config.applyBackwardCompatibility", applyBackwardCompatibility,
|
||||
"config.concurrentBlockFlushing", c.ConcurrentBlockFlushing,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -173,6 +177,8 @@ 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
|
||||
concurrentFlushQueue *ConcurrentFlushQueue
|
||||
concurrentFlushBufferSize int
|
||||
// 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 +188,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 int
|
||||
|
||||
// Block state
|
||||
block *pbeth.Block
|
||||
|
|
@ -250,6 +257,8 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
|
|||
hasher: crypto.NewKeccakState(),
|
||||
tracerID: "global",
|
||||
applyBackwardCompatibility: config.ApplyBackwardCompatibility,
|
||||
concurrentBlockFlushing: config.ConcurrentBlockFlushing,
|
||||
concurrentFlushBufferSize: 100,
|
||||
|
||||
// Block state
|
||||
blockOrdinal: &Ordinal{},
|
||||
|
|
@ -365,6 +374,16 @@ func (f *Firehose) OnBlockchainInit(chainConfig *params.ChainConfig) {
|
|||
applyBackwardCompatibilityLogSuffix = " (disabled)"
|
||||
}
|
||||
|
||||
if f.config.ConcurrentBlockFlushing > 0 {
|
||||
log.Info("Firehose concurrent block flushing enabled, starting goroutine")
|
||||
f.concurrentFlushQueue = NewConcurrentFlushQueue(
|
||||
f.concurrentFlushBufferSize,
|
||||
f.printBlockToFirehose,
|
||||
f.flushToFirehose,
|
||||
)
|
||||
f.concurrentFlushQueue.Start(f.config.ConcurrentBlockFlushing)
|
||||
}
|
||||
|
||||
log.Info("Firehose tracer initialized",
|
||||
"chain_id", chainConfig.ChainID,
|
||||
"apply_backward_compatibility", fmt.Sprintf("%t%s", *f.applyBackwardCompatibility, applyBackwardCompatibilityLogSuffix),
|
||||
|
|
@ -508,7 +527,14 @@ func (f *Firehose) OnBlockEnd(err error) {
|
|||
}
|
||||
|
||||
f.ensureInBlockAndNotInTrx()
|
||||
|
||||
// Flush block to firehose and optionally use goroutine
|
||||
if f.concurrentBlockFlushing > 0 {
|
||||
f.concurrentFlushQueue.Enqueue(f.block, f.blockFinality)
|
||||
} 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 +652,13 @@ func (f *Firehose) reorderCallOrdinals(call *pbeth.Call, ordinalBase uint64) (or
|
|||
return call.EndOrdinal
|
||||
}
|
||||
|
||||
func (f *Firehose) OnClose() {
|
||||
if f.concurrentFlushQueue != nil {
|
||||
log.Info("Firehose closing, flushing queued blocks to standard output")
|
||||
f.concurrentFlushQueue.CloseChannels()
|
||||
}
|
||||
}
|
||||
|
||||
func (f *Firehose) OnSystemCallStart() {
|
||||
firehoseInfo("system call start")
|
||||
f.ensureInBlockAndNotInTrx()
|
||||
|
|
@ -1875,13 +1908,18 @@ func (f *Firehose) panicInvalidState(msg string, callerSkip int) string {
|
|||
|
||||
// printBlockToFirehose is a helper function to print a block to Firehose protocl format.
|
||||
func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *FinalityStatus) {
|
||||
|
||||
headerSize := 225 // FIRE BLOCK:11 <blockNum:20> <blockHash:64> <prevNum:20> <prevHash:64> <libNum:20> <timestamp:20>
|
||||
base64Size := math.Ceil(float64(proto.Size(block)) * 8 / 6)
|
||||
bufferSize := headerSize + int(base64Size)
|
||||
buf := bytes.NewBuffer(make([]byte, 0, bufferSize))
|
||||
|
||||
marshalled, err := proto.Marshal(block)
|
||||
|
||||
if err != nil {
|
||||
panic(fmt.Errorf("failed to marshal block: %w", err))
|
||||
}
|
||||
|
||||
f.outputBuffer.Reset()
|
||||
|
||||
previousHash := block.PreviousID()
|
||||
previousNum := 0
|
||||
if block.Number > 0 {
|
||||
|
|
@ -1900,9 +1938,9 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina
|
|||
}
|
||||
|
||||
// **Important* The final space in the Sprintf template is mandatory!
|
||||
f.outputBuffer.WriteString(fmt.Sprintf("FIRE BLOCK %d %s %d %s %d %d ", block.Number, hex.EncodeToString(block.Hash), previousNum, previousHash, libNum, block.MustTime().UnixNano()))
|
||||
buf.WriteString(fmt.Sprintf("FIRE BLOCK %d %s %d %s %d %d ", block.Number, hex.EncodeToString(block.Hash), previousNum, previousHash, libNum, block.MustTime().UnixNano()))
|
||||
|
||||
encoder := base64.NewEncoder(base64.StdEncoding, f.outputBuffer)
|
||||
encoder := base64.NewEncoder(base64.StdEncoding, buf)
|
||||
if _, err = encoder.Write(marshalled); err != nil {
|
||||
panic(fmt.Errorf("write to encoder should have been infaillible: %w", err))
|
||||
}
|
||||
|
|
@ -1911,9 +1949,16 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina
|
|||
panic(fmt.Errorf("closing encoder should have been infaillible: %w", err))
|
||||
}
|
||||
|
||||
f.outputBuffer.WriteString("\n")
|
||||
buf.WriteString("\n")
|
||||
|
||||
f.flushToFirehose(f.outputBuffer.Bytes())
|
||||
if f.concurrentBlockFlushing > 0 {
|
||||
f.concurrentFlushQueue.outputQueue <- &outputJob{
|
||||
blockNum: block.Number,
|
||||
data: buf.Bytes(),
|
||||
}
|
||||
} else {
|
||||
f.flushToFirehose(buf.Bytes())
|
||||
}
|
||||
}
|
||||
|
||||
// printToFirehose is an easy way to print to Firehose format, it essentially
|
||||
|
|
|
|||
103
eth/tracers/firehose_concurrency.go
Normal file
103
eth/tracers/firehose_concurrency.go
Normal file
|
|
@ -0,0 +1,103 @@
|
|||
package tracers
|
||||
|
||||
import (
|
||||
pbeth "github.com/streamingfast/firehose-ethereum/types/pb/sf/ethereum/type/v2"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type blockPrintJob struct {
|
||||
block *pbeth.Block
|
||||
finality *FinalityStatus
|
||||
}
|
||||
|
||||
type outputJob struct {
|
||||
blockNum uint64
|
||||
data []byte
|
||||
}
|
||||
|
||||
type ConcurrentFlushQueue struct {
|
||||
bufferSize int
|
||||
|
||||
startSignal chan uint64
|
||||
jobQueue chan *blockPrintJob
|
||||
outputQueue chan *outputJob
|
||||
printBlockFunc func(block *pbeth.Block, finality *FinalityStatus)
|
||||
outputFunc func([]byte)
|
||||
|
||||
jobWG sync.WaitGroup
|
||||
outputWG sync.WaitGroup
|
||||
closeOnce sync.Once
|
||||
}
|
||||
|
||||
func NewConcurrentFlushQueue(bufferSize int, printBlockFunc func(*pbeth.Block, *FinalityStatus), outputFunc func([]byte)) *ConcurrentFlushQueue {
|
||||
return &ConcurrentFlushQueue{
|
||||
startSignal: make(chan uint64, 1),
|
||||
jobQueue: make(chan *blockPrintJob, bufferSize),
|
||||
outputQueue: make(chan *outputJob, bufferSize),
|
||||
outputFunc: outputFunc,
|
||||
bufferSize: bufferSize,
|
||||
printBlockFunc: printBlockFunc,
|
||||
}
|
||||
}
|
||||
|
||||
func (q *ConcurrentFlushQueue) Start(concurrency int) {
|
||||
for i := 0; i < concurrency; i++ {
|
||||
q.jobWG.Add(1)
|
||||
go q.worker()
|
||||
}
|
||||
|
||||
q.outputWG.Add(1)
|
||||
go q.outputOrderer()
|
||||
}
|
||||
|
||||
func (q *ConcurrentFlushQueue) Enqueue(block *pbeth.Block, finality *FinalityStatus) {
|
||||
select {
|
||||
case q.startSignal <- block.Number:
|
||||
default:
|
||||
}
|
||||
|
||||
q.jobQueue <- &blockPrintJob{
|
||||
block: block,
|
||||
finality: finality,
|
||||
}
|
||||
}
|
||||
|
||||
// CloseChannels signals 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 (q *ConcurrentFlushQueue) CloseChannels() {
|
||||
q.closeOnce.Do(func() {
|
||||
close(q.jobQueue)
|
||||
q.jobWG.Wait()
|
||||
close(q.outputQueue)
|
||||
q.outputWG.Wait()
|
||||
})
|
||||
}
|
||||
|
||||
// Instantiates a worker that listens for jobs
|
||||
func (q *ConcurrentFlushQueue) worker() {
|
||||
defer q.jobWG.Done()
|
||||
for job := range q.jobQueue {
|
||||
q.printBlockFunc(job.block, job.finality)
|
||||
}
|
||||
}
|
||||
|
||||
// Channel ensuring that blocks are linearly flushed out in order
|
||||
func (q *ConcurrentFlushQueue) outputOrderer() {
|
||||
defer q.outputWG.Done()
|
||||
buffer := make(map[uint64][]byte)
|
||||
next := <-q.startSignal
|
||||
|
||||
for job := range q.outputQueue {
|
||||
buffer[job.blockNum] = job.data
|
||||
for {
|
||||
data, ok := buffer[next]
|
||||
if !ok {
|
||||
break
|
||||
}
|
||||
q.outputFunc(data)
|
||||
delete(buffer, next)
|
||||
next++
|
||||
}
|
||||
}
|
||||
}
|
||||
144
eth/tracers/firehose_concurrency.md
Normal file
144
eth/tracers/firehose_concurrency.md
Normal file
|
|
@ -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.
|
||||
199
eth/tracers/firehose_concurrency_2.md
Normal file
199
eth/tracers/firehose_concurrency_2.md
Normal file
|
|
@ -0,0 +1,199 @@
|
|||
# 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":<number_of_workers>}' \
|
||||
--synctarget=<last_block_hash> \
|
||||
--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
|
||||
|
||||
Geth version used is 1.15.10
|
||||
|
||||
### 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.
|
||||
146
eth/tracers/firehose_concurrency_test.go
Normal file
146
eth/tracers/firehose_concurrency_test.go
Normal file
|
|
@ -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: 1,
|
||||
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 = 0
|
||||
|
||||
f := NewFirehose(&FirehoseConfig{
|
||||
ConcurrentBlockFlushing: 1,
|
||||
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
|
||||
}
|
||||
|
|
@ -1,6 +1,8 @@
|
|||
package firehose_test
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/ethereum/go-ethereum/eth/tracers"
|
||||
"math/big"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
|
@ -39,25 +41,34 @@ func TestFirehosePrestate(t *testing.T) {
|
|||
"./testdata/TestFirehosePrestate/keccak256_memory_out_of_bounds",
|
||||
}
|
||||
|
||||
for _, concurrent := range []int{0, 1} {
|
||||
for _, folder := range testFolders {
|
||||
name := filepath.Base(folder)
|
||||
concurrencyLabel := "sequential"
|
||||
if concurrent == 1 {
|
||||
concurrencyLabel = "concurrent"
|
||||
}
|
||||
|
||||
for _, model := range tracingModels {
|
||||
t.Run(string(model)+"/"+name, func(t *testing.T) {
|
||||
tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model)
|
||||
defer onClose()
|
||||
t.Run(fmt.Sprintf("%s/%s/%s", model, name, concurrencyLabel), func(t *testing.T) {
|
||||
config := &tracers.FirehoseConfig{
|
||||
ConcurrentBlockFlushing: concurrent,
|
||||
}
|
||||
|
||||
tracer, tracingHooks, _ := newFirehoseTestTracer(t, model, config)
|
||||
|
||||
runPrestateBlock(t, filepath.Join(folder, "prestate.json"), tracingHooks)
|
||||
|
||||
tracer.OnClose()
|
||||
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,10 +198,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 []int{0, 1} {
|
||||
concurrencyLabel := "sequential"
|
||||
if concurrent == 1 {
|
||||
concurrencyLabel = "concurrent"
|
||||
}
|
||||
|
||||
for _, model := range tracingModels {
|
||||
t.Run(string(model), func(t *testing.T) {
|
||||
tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model)
|
||||
defer onClose()
|
||||
t.Run(fmt.Sprintf("%s/%s", model, concurrencyLabel), func(t *testing.T) {
|
||||
config := &tracers.FirehoseConfig{
|
||||
ConcurrentBlockFlushing: concurrent,
|
||||
}
|
||||
|
||||
tracer, tracingHooks, _ := newFirehoseTestTracer(t, model, config)
|
||||
|
||||
chain, err := core.NewBlockChain(rawdb.NewMemoryDatabase(), nil, genesisSpec, nil, engine, vm.Config{Tracer: tracingHooks}, nil)
|
||||
require.NoError(t, err, "failed to create tester chain")
|
||||
|
|
@ -204,7 +224,9 @@ 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.OnClose()
|
||||
assertBlockEquals(t, tracer, filepath.Join("testdata", goldenDir, string(model)), len(blocks))
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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": %d,
|
||||
"_private": {
|
||||
"flushToTestBuffer": true,
|
||||
"ignoreGenesisBlock": true,
|
||||
"forcedBackwardCompatibility": %t
|
||||
}
|
||||
}`, model == tracingModelFirehose2_3)))
|
||||
}`, config.ConcurrentBlockFlushing, model == tracingModelFirehose2_3)))
|
||||
require.NoError(t, err)
|
||||
|
||||
hooks := tracers.NewTracingHooksFromFirehose(tracer)
|
||||
|
|
|
|||
Loading…
Reference in a new issue