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

TEST: Fixed race issue in existing test files caused by integration of goroutine. All tests passing

Browse source
This commit is contained in:
David Zhou 2025-05-09 16:56:15 -04:00
parent 7520990fa6
commit 5bd391008a
2 changed files with 24 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
eth/tracers/firehose.go
View file

@ -79,7 +79,6 @@ func NewTracingHooksFromFirehose(tracer *Firehose) *tracing.Hooks {
OnBlockEnd: tracer.OnBlockEnd, OnBlockEnd: tracer.OnBlockEnd,
OnSkippedBlock: tracer.OnSkippedBlock, OnSkippedBlock: tracer.OnSkippedBlock,
OnClose: tracer.OnClose, OnClose: tracer.OnClose,
// TODO: OnClose
OnTxStart: tracer.OnTxStart, OnTxStart: tracer.OnTxStart,
OnTxEnd: tracer.OnTxEnd, OnTxEnd: tracer.OnTxEnd,
@ -193,7 +192,8 @@ type Firehose struct {
// Worker queuefor blockprinting // Worker queuefor blockprinting
blockPrintQueue chan *blockPrintJob blockPrintQueue chan *blockPrintJob
workerWg sync.WaitGroup flushDone sync.WaitGroup
closeOnce sync.Once
} }
const FirehoseProtocolVersion = "3.0" const FirehoseProtocolVersion = "3.0"
@ -257,12 +257,8 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
} }
} }
// Worker goroutines firehose.flushDone.Add(1)
numWorkers := 1 go firehose.blockPrintWorker()
firehose.workerWg.Add(numWorkers)
for i := 0; i < numWorkers; i++ {
go firehose.blockPrintWorker()
}
return firehose return firehose
} }
@ -499,13 +495,13 @@ func (f *Firehose) OnBlockEnd(err error) {
f.fixOrdinalsForEndOfBlockChanges() f.fixOrdinalsForEndOfBlockChanges()
} }
f.printBlockToFirehose(f.block, f.blockFinality) f.ensureInBlockAndNotInTrx()
//f.ensureInBlockAndNotInTrx() // f.printBlockToFirehose(f.block, f.blockFinality)
//job := &blockPrintJob{ job := &blockPrintJob{
// block: f.block, block: f.block,
// finality: f.blockFinality, finality: f.blockFinality,
//} }
//f.blockPrintQueue <- job f.blockPrintQueue <- job
} else { } else {
// An error occurred, could have happen in transaction/call context, we must not check if in trx/call, only check in block // 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() { func (f *Firehose) OnClose() {
log.Info("Firehose closing: waiting for worker goroutines to finish and shutting down channels") log.Info("Firehose closing: waiting for worker goroutines to finish and shutting down channels")
close(f.blockPrintQueue) f.CloseBlockPrintQueue()
f.workerWg.Wait()
} }
func (f *Firehose) OnSystemCallStart() { func (f *Firehose) OnSystemCallStart() {
@ -2827,9 +2822,15 @@ type blockPrintJob struct {
} }
func (f *Firehose) blockPrintWorker() { func (f *Firehose) blockPrintWorker() {
defer f.workerWg.Done() defer f.flushDone.Done()
for job := range f.blockPrintQueue { for job := range f.blockPrintQueue {
f.printBlockToFirehose(job.block, job.finality) f.printBlockToFirehose(job.block, job.finality)
} }
} }
func (f *Firehose) CloseBlockPrintQueue() {
f.closeOnce.Do(func() {
close(f.blockPrintQueue)
f.flushDone.Wait()
})
}

4
eth/tracers/internal/tracetest/firehose/firehose_test.go
View file

@ -48,6 +48,8 @@ func TestFirehosePrestate(t *testing.T) {
runPrestateBlock(t, filepath.Join(folder, "prestate.json"), tracingHooks) runPrestateBlock(t, filepath.Join(folder, "prestate.json"), tracingHooks)
tracer.CloseBlockPrintQueue()
genesisLine, blockLines, unknownLines := readTracerFirehoseLines(t, tracer) 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) require.NotNil(t, genesisLine)
@ -203,6 +205,8 @@ func testBlockTracesCorrectly(t *testing.T, genesisSpec *core.Genesis, engine co
n, err := chain.InsertChain(blocks) n, err := chain.InsertChain(blocks)
require.NoError(t, err, "failed to insert chain block %d", n) require.NoError(t, err, "failed to insert chain block %d", n)
tracer.CloseBlockPrintQueue()
assertBlockEquals(t, tracer, filepath.Join("testdata", goldenDir, string(model)), len(blocks)) assertBlockEquals(t, tracer, filepath.Join("testdata", goldenDir, string(model)), len(blocks))
}) })
} }