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

Fix1

Browse source
This commit is contained in:
David Zhou 2025-05-22 16:23:56 -04:00
parent b1057965c9
commit 45c4e00a0f
1 changed files with 10 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
eth/tracers/firehose.go
View file

@ -222,9 +222,7 @@ type Firehose struct {
flushOrderedOutputQueue chan *blockOutput flushOrderedOutputQueue chan *blockOutput
flushJobWG sync.WaitGroup flushJobWG sync.WaitGroup
flushOrderedOutputWG sync.WaitGroup flushOrderedOutputWG sync.WaitGroup
flushStarted bool flushStartSignal chan uint64
flushStartBlockNum uint64
flushStartCond *sync.Cond
flushBufferSize int flushBufferSize int
} }
@ -280,7 +278,7 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
deferredCallState: NewDeferredCallState(), deferredCallState: NewDeferredCallState(),
latestCallEnterSuicided: false, latestCallEnterSuicided: false,
flushStartCond: sync.NewCond(&sync.Mutex{}), flushStartSignal: make(chan uint64, 1),
flushBufferSize: 100, // TODO: Optimal buffer size tbd flushBufferSize: 100, // TODO: Optimal buffer size tbd
} }
@ -308,13 +306,8 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
buffer := make(map[uint64][]byte) buffer := make(map[uint64][]byte)
firehose.flushStartCond.L.Lock() // Blocks until tracer sends first block number to flush
for !firehose.flushStarted { nextExpected := <-firehose.flushStartSignal
firehose.flushStartCond.Wait()
}
nextExpected := firehose.flushStartBlockNum
firehose.flushStartCond.L.Unlock()
for result := range firehose.flushOrderedOutputQueue { for result := range firehose.flushOrderedOutputQueue {
buffer[result.blockNum] = result.data buffer[result.blockNum] = result.data
@ -572,13 +565,10 @@ func (f *Firehose) OnBlockEnd(err error) {
// Flush block to firehose and optionally use goroutine // Flush block to firehose and optionally use goroutine
if f.concurrentBlockFlushing > 0 { if f.concurrentBlockFlushing > 0 {
f.flushStartCond.L.Lock() select {
if !f.flushStarted { case f.flushStartSignal <- f.block.Number:
f.flushStartBlockNum = f.block.Number default:
f.flushStarted = true
f.flushStartCond.Broadcast()
} }
f.flushStartCond.L.Unlock()
job := &blockPrintJob{ job := &blockPrintJob{
block: f.block, block: f.block,
@ -1962,7 +1952,7 @@ func (f *Firehose) panicInvalidState(msg string, callerSkip int) string {
// printBlockToFirehose is a helper function to print a block to Firehose protocl format. // printBlockToFirehose is a helper function to print a block to Firehose protocl format.
func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *FinalityStatus) { func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *FinalityStatus) {
var buf bytes.Buffer buf := bytes.NewBuffer(make([]byte, 0, 128*1024)) // 128 KB
marshalled, err := proto.Marshal(block) marshalled, err := proto.Marshal(block)
if err != nil { if err != nil {
@ -1989,7 +1979,7 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina
// **Important* The final space in the Sprintf template is mandatory! // **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())) 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, &buf) encoder := base64.NewEncoder(base64.StdEncoding, buf)
if _, err = encoder.Write(marshalled); err != nil { if _, err = encoder.Write(marshalled); err != nil {
panic(fmt.Errorf("write to encoder should have been infaillible: %w", err)) panic(fmt.Errorf("write to encoder should have been infaillible: %w", err))
} }