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

TEST: Testing concurrent and not concurrent

Browse source
This commit is contained in:
David Zhou 2025-05-12 11:11:41 -04:00
parent e8ddd05284
commit 8d0ce228db
3 changed files with 39 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
eth/tracers/firehose.go
View file

@ -632,7 +632,7 @@ func (f *Firehose) reorderCallOrdinals(call *pbeth.Call, ordinalBase uint64) (or
func (f *Firehose) OnClose() { func (f *Firehose) OnClose() {
log.Info("Firehose closing...") log.Info("Firehose closing...")
if f.concurrentBlockFlushing { if f.concurrentBlockFlushing {
log.Info("waiting for worker goroutines to finish and shutting down channels") log.Info("Closing channel: flushing the remaining blocks to firehose")
f.CloseBlockPrintQueue() f.CloseBlockPrintQueue()
} }
} }
@ -1860,6 +1860,7 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina
panic(fmt.Errorf("failed to marshal block: %w", err)) panic(fmt.Errorf("failed to marshal block: %w", err))
} }
// TODO: If multiple goroutine, this becomes shared resource
f.outputBuffer.Reset() f.outputBuffer.Reset()
previousHash := block.PreviousID() previousHash := block.PreviousID()
@ -2843,7 +2844,6 @@ func (f *Firehose) blockPrintWorker() {
func (f *Firehose) CloseBlockPrintQueue() { func (f *Firehose) CloseBlockPrintQueue() {
if f.concurrentBlockFlushing { if f.concurrentBlockFlushing {
f.closeOnce.Do(func() { f.closeOnce.Do(func() {
log.Info("Closing channel: flushing the remaining blocks to firehose")
close(f.blockPrintQueue) close(f.blockPrintQueue)
f.flushDone.Wait() f.flushDone.Wait()
}) })

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

@ -38,12 +38,14 @@ func TestFirehosePrestate(t *testing.T) {
"./testdata/TestFirehosePrestate/extra_account_creations", "./testdata/TestFirehosePrestate/extra_account_creations",
} }
// TODO: have goroutine config on AND off
for _, concurrent := range []bool{true, false} {
for _, folder := range testFolders { for _, folder := range testFolders {
name := filepath.Base(folder) name := filepath.Base(folder)
for _, model := range tracingModels { for _, model := range tracingModels {
t.Run(string(model)+"/"+name, func(t *testing.T) { t.Run(string(model)+"/"+name, func(t *testing.T) {
tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model) tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model, concurrent)
defer onClose() defer onClose()
runPrestateBlock(t, filepath.Join(folder, "prestate.json"), tracingHooks) runPrestateBlock(t, filepath.Join(folder, "prestate.json"), tracingHooks)
@ -57,7 +59,7 @@ func TestFirehosePrestate(t *testing.T) {
}) })
} }
} }
}
} }
func TestFirehose_EIP7702(t *testing.T) { func TestFirehose_EIP7702(t *testing.T) {
// Copied from ./core/blockchain_test.go#L4180 (TestEIP7702) // Copied from ./core/blockchain_test.go#L4180 (TestEIP7702)
@ -187,9 +189,10 @@ func TestFirehose_SystemCalls(t *testing.T) {
func testBlockTracesCorrectly(t *testing.T, genesisSpec *core.Genesis, engine consensus.Engine, blocks []*types.Block, goldenDir string) { func testBlockTracesCorrectly(t *testing.T, genesisSpec *core.Genesis, engine consensus.Engine, blocks []*types.Block, goldenDir string) {
t.Helper() t.Helper()
for _, concurrent := range []bool{true, false} {
for _, model := range tracingModels { for _, model := range tracingModels {
t.Run(string(model), func(t *testing.T) { t.Run(string(model), func(t *testing.T) {
tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model) tracer, tracingHooks, onClose := newFirehoseTestTracer(t, model, concurrent)
defer onClose() defer onClose()
chain, err := core.NewBlockChain(rawdb.NewMemoryDatabase(), nil, genesisSpec, nil, engine, vm.Config{Tracer: tracingHooks}, nil) chain, err := core.NewBlockChain(rawdb.NewMemoryDatabase(), nil, genesisSpec, nil, engine, vm.Config{Tracer: tracingHooks}, nil)
@ -209,4 +212,5 @@ func testBlockTracesCorrectly(t *testing.T, genesisSpec *core.Genesis, engine co
assertBlockEquals(t, tracer, filepath.Join("testdata", goldenDir, string(model)), len(blocks)) assertBlockEquals(t, tracer, filepath.Join("testdata", goldenDir, string(model)), len(blocks))
}) })
} }
}
} }

6
eth/tracers/internal/tracetest/firehose/helpers_test.go
View file

@ -29,17 +29,17 @@ type firehoseInitLine struct {
type firehoseBlockLines []firehoseBlockLine type firehoseBlockLines []firehoseBlockLine
func newFirehoseTestTracer(t *testing.T, model tracingModel) (*tracers.Firehose, *tracing.Hooks, func()) { func newFirehoseTestTracer(t *testing.T, model tracingModel, concurrentBlockFlushing bool) (*tracers.Firehose, *tracing.Hooks, func()) {
t.Helper() t.Helper()
tracer, err := tracers.NewFirehoseFromRawJSON([]byte(fmt.Sprintf(`{ tracer, err := tracers.NewFirehoseFromRawJSON([]byte(fmt.Sprintf(`{
"concurrentBlockFlushing": true, "concurrentBlockFlushing": %t,
"_private": { "_private": {
"flushToTestBuffer": true, "flushToTestBuffer": true,
"ignoreGenesisBlock": true, "ignoreGenesisBlock": true,
"forcedBackwardCompatibility": %t "forcedBackwardCompatibility": %t
} }
}`, model == tracingModelFirehose2_3))) }`, concurrentBlockFlushing, model == tracingModelFirehose2_3)))
require.NoError(t, err) require.NoError(t, err)
hooks := tracers.NewTracingHooksFromFirehose(tracer) hooks := tracers.NewTracingHooksFromFirehose(tracer)