mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-26 06:36:43 +00:00
PR: PR comments addressed Part 1
# Conflicts: # eth/tracers/firehose.go # eth/tracers/internal/tracetest/firehose/firehose_test.go
This commit is contained in:
David Zhou
parent
a053ac76fd
commit
326a3ee583
3 changed files with 46 additions and 104 deletions
|
|
@ -144,13 +144,14 @@ func (c *FirehoseConfig) ForcedBackwardCompatibility() bool {
|
|||
|
||||
type Firehose struct {
|
||||
// Global state
|
||||
outputBuffer *bytes.Buffer
|
||||
initSent *atomic.Bool
|
||||
config *FirehoseConfig
|
||||
chainConfig *params.ChainConfig
|
||||
hasher crypto.KeccakState // Keccak256 hasher instance shared across tracer needs (non-concurrent safe)
|
||||
hasherBuf common.Hash // Keccak256 hasher result array shared across tracer needs (non-concurrent safe)
|
||||
tracerID string
|
||||
outputBuffer *bytes.Buffer
|
||||
initSent *atomic.Bool
|
||||
config *FirehoseConfig
|
||||
chainConfig *params.ChainConfig
|
||||
hasher crypto.KeccakState // Keccak256 hasher instance shared across tracer needs (non-concurrent safe)
|
||||
hasherBuf common.Hash // Keccak256 hasher result array shared across tracer needs (non-concurrent safe)
|
||||
tracerID string
|
||||
closeChannels sync.Once
|
||||
// The FirehoseTracer is used in multiple chains, some for which were produced using a legacy version
|
||||
// of the whole tracing infrastructure. This legacy version had many small bugs here and there that
|
||||
// we must "reproduce" on some chain to ensure that the FirehoseTracer produces the same output
|
||||
|
|
@ -173,6 +174,8 @@ type Firehose struct {
|
|||
blockReorderOrdinalSnapshot uint64
|
||||
blockReorderOrdinalOnce sync.Once
|
||||
blockIsGenesis bool
|
||||
blockPrintQueue chan *blockPrintJob
|
||||
blockFlushDone sync.WaitGroup
|
||||
|
||||
// Transaction state
|
||||
evm *tracing.VMContext
|
||||
|
|
@ -190,13 +193,6 @@ type Firehose struct {
|
|||
// Testing state, only used in tests and private configs
|
||||
testingBuffer *bytes.Buffer
|
||||
testingIgnoreGenesisBlock bool
|
||||
|
||||
// Worker queue for block printing
|
||||
blockPrintQueue chan *blockPrintJob
|
||||
blockOutputQueue chan *blockOutput
|
||||
flushDone sync.WaitGroup
|
||||
flushOutputDone sync.WaitGroup
|
||||
closeOnce sync.Once
|
||||
}
|
||||
|
||||
const FirehoseProtocolVersion = "3.0"
|
||||
|
|
@ -260,38 +256,11 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
|
|||
}
|
||||
|
||||
if config.ConcurrentBlockFlushing {
|
||||
log.Info("Concurrent block flushing enabled: starting goroutine...")
|
||||
const numWorkers = 10 // TODO: This becomes a parameter
|
||||
firehose.blockPrintQueue = make(chan *blockPrintJob, 100) // TODO: Optimal buffer size tbd
|
||||
firehose.blockOutputQueue = make(chan *blockOutput, 100)
|
||||
|
||||
for i := 0; i < numWorkers; i++ {
|
||||
firehose.flushDone.Add(1)
|
||||
go firehose.blockPrintWorker()
|
||||
}
|
||||
|
||||
// Output channel to order the flushing linearly
|
||||
firehose.flushOutputDone.Add(1)
|
||||
go func() {
|
||||
defer firehose.flushOutputDone.Done()
|
||||
|
||||
expected := uint64(0)
|
||||
buffer := make(map[uint64][]byte)
|
||||
|
||||
for result := range firehose.blockOutputQueue {
|
||||
buffer[result.blockNum] = result.data
|
||||
|
||||
for {
|
||||
data, ok := buffer[expected]
|
||||
if !ok {
|
||||
break
|
||||
}
|
||||
firehose.flushToFirehose(data)
|
||||
delete(buffer, expected)
|
||||
expected++
|
||||
}
|
||||
}
|
||||
}()
|
||||
log.Info("Firehose concurrent block flushing enabled, starting block " +
|
||||
"print worker goroutine")
|
||||
firehose.blockPrintQueue = make(chan *blockPrintJob, 100)
|
||||
firehose.blockFlushDone.Add(1)
|
||||
go firehose.blockPrintWorker()
|
||||
}
|
||||
|
||||
return firehose
|
||||
|
|
@ -660,9 +629,8 @@ func (f *Firehose) reorderCallOrdinals(call *pbeth.Call, ordinalBase uint64) (or
|
|||
}
|
||||
|
||||
func (f *Firehose) OnClose() {
|
||||
log.Info("Firehose closing...")
|
||||
if f.concurrentBlockFlushing {
|
||||
log.Info("Closing channel: flushing the remaining blocks to firehose")
|
||||
log.Info("Firehose closing, flushing queued blocks to standard output")
|
||||
f.CloseBlockPrintQueue()
|
||||
}
|
||||
}
|
||||
|
|
@ -1857,6 +1825,9 @@ func (f *Firehose) isChainOneOf(chainIDs ...*big.Int) bool {
|
|||
}
|
||||
|
||||
func isChainIDOneOf(actualChainID *big.Int, chainIDs ...*big.Int) bool {
|
||||
if actualChainID == nil {
|
||||
return false
|
||||
}
|
||||
for _, chainID := range chainIDs {
|
||||
if actualChainID.Cmp(chainID) == 0 {
|
||||
return true
|
||||
|
|
@ -1885,13 +1856,14 @@ func (f *Firehose) panicInvalidState(msg string, callerSkip int) string {
|
|||
|
||||
// printBlockToFirehose is a helper function to print a block to Firehose protocl format.
|
||||
func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *FinalityStatus) {
|
||||
var buf bytes.Buffer
|
||||
|
||||
marshalled, err := proto.Marshal(block)
|
||||
if err != nil {
|
||||
panic(fmt.Errorf("failed to marshal block: %w", err))
|
||||
}
|
||||
|
||||
// TODO: If multiple goroutine, this becomes shared resource
|
||||
f.outputBuffer.Reset()
|
||||
|
||||
previousHash := block.PreviousID()
|
||||
previousNum := 0
|
||||
if block.Number > 0 {
|
||||
|
|
@ -1910,9 +1882,9 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina
|
|||
}
|
||||
|
||||
// **Important* The final space in the Sprintf template is mandatory!
|
||||
buf.WriteString(fmt.Sprintf("FIRE BLOCK %d %s %d %s %d %d ", block.Number, hex.EncodeToString(block.Hash), previousNum, previousHash, libNum, block.MustTime().UnixNano()))
|
||||
f.outputBuffer.WriteString(fmt.Sprintf("FIRE BLOCK %d %s %d %s %d %d ", block.Number, hex.EncodeToString(block.Hash), previousNum, previousHash, libNum, block.MustTime().UnixNano()))
|
||||
|
||||
encoder := base64.NewEncoder(base64.StdEncoding, &buf)
|
||||
encoder := base64.NewEncoder(base64.StdEncoding, f.outputBuffer)
|
||||
if _, err = encoder.Write(marshalled); err != nil {
|
||||
panic(fmt.Errorf("write to encoder should have been infaillible: %w", err))
|
||||
}
|
||||
|
|
@ -1921,16 +1893,9 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina
|
|||
panic(fmt.Errorf("closing encoder should have been infaillible: %w", err))
|
||||
}
|
||||
|
||||
buf.WriteString("\n")
|
||||
f.outputBuffer.WriteString("\n")
|
||||
|
||||
if f.concurrentBlockFlushing {
|
||||
f.blockOutputQueue <- &blockOutput{
|
||||
blockNum: block.Number,
|
||||
data: buf.Bytes(),
|
||||
}
|
||||
} else {
|
||||
f.flushToFirehose(buf.Bytes())
|
||||
}
|
||||
f.flushToFirehose(f.outputBuffer.Bytes())
|
||||
}
|
||||
|
||||
// printToFirehose is an easy way to print to Firehose format, it essentially
|
||||
|
|
@ -2871,25 +2836,20 @@ type blockPrintJob struct {
|
|||
}
|
||||
|
||||
func (f *Firehose) blockPrintWorker() {
|
||||
defer f.flushDone.Done()
|
||||
defer f.blockFlushDone.Done()
|
||||
for job := range f.blockPrintQueue {
|
||||
f.printBlockToFirehose(job.block, job.finality)
|
||||
}
|
||||
}
|
||||
|
||||
// CloseBlockPrintQueue signals block printing goroutines to shut down and waits for them.
|
||||
// It blocks until all concurrent block flushing operations are completed, ensuring a clean
|
||||
// shutdown of the printing pipeline.
|
||||
func (f *Firehose) CloseBlockPrintQueue() {
|
||||
if f.concurrentBlockFlushing {
|
||||
f.closeOnce.Do(func() {
|
||||
f.closeChannels.Do(func() {
|
||||
close(f.blockPrintQueue)
|
||||
f.flushDone.Wait()
|
||||
|
||||
close(f.blockOutputQueue)
|
||||
f.flushOutputDone.Wait()
|
||||
f.blockFlushDone.Wait()
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type blockOutput struct {
|
||||
blockNum uint64
|
||||
data []byte
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -94,7 +94,9 @@ func TestFirehose_BlocksPrintToFirehose_MultipleBlocksInOrder(t *testing.T) {
|
|||
"Expected %d blocks in output, found %d", blockCount, len(extractedBlocks))
|
||||
|
||||
// Verify blocks in order
|
||||
verifyBlockSequence(t, extractedBlocks, baseBlockNum)
|
||||
for i, block := range extractedBlocks {
|
||||
require.Equal(t, baseBlockNum+uint64(i), block.number, "Blocks out of order at position %d", i)
|
||||
}
|
||||
|
||||
// Verify block hashes
|
||||
for _, block := range extractedBlocks {
|
||||
|
|
@ -139,24 +141,3 @@ func extractBlocksFromOutput(t *testing.T, output string) []extractedBlock {
|
|||
|
||||
return blocks
|
||||
}
|
||||
|
||||
func verifyBlockSequence(t *testing.T, blocks []extractedBlock, baseBlockNum uint64) {
|
||||
t.Helper()
|
||||
|
||||
// First block should be the base block number
|
||||
require.Equal(t, baseBlockNum, blocks[0].number,
|
||||
"First block should be %d, got %d", baseBlockNum, blocks[0].number)
|
||||
|
||||
// Last block should be base + count - 1
|
||||
expectedLast := baseBlockNum + uint64(len(blocks)) - 1
|
||||
require.Equal(t, expectedLast, blocks[len(blocks)-1].number,
|
||||
"Last block should be %d, got %d", expectedLast, blocks[len(blocks)-1].number)
|
||||
|
||||
// Verify sequence
|
||||
for i := 0; i < len(blocks)-1; i++ {
|
||||
current := blocks[i].number
|
||||
next := blocks[i+1].number
|
||||
require.Equal(t, current+1, next,
|
||||
"Blocks out of order at position %d: %d followed by %d", i, current, next)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue