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

DEV: Multiple goroutines implementation

Browse source
This commit is contained in:
David Zhou 2025-05-13 09:48:22 -04:00
parent 7df9b21013
commit 6a3c000a4d
2 changed files with 12 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
eth/tracers/firehose.go
View file

@ -259,9 +259,13 @@ func NewFirehose(config *FirehoseConfig) *Firehose {
if config.ConcurrentBlockFlushing { if config.ConcurrentBlockFlushing {
log.Info("Concurrent block flushing enabled: starting goroutine...") log.Info("Concurrent block flushing enabled: starting goroutine...")
firehose.blockPrintQueue = make(chan *blockPrintJob, 100) const numWorkers = 10 // TODO: This becomes a parameter
firehose.flushDone.Add(1) firehose.blockPrintQueue = make(chan *blockPrintJob, 100) // TODO: Optimal buffer size tbd
go firehose.blockPrintWorker()
for i := 0; i < numWorkers; i++ {
firehose.flushDone.Add(1)
go firehose.blockPrintWorker()
}
} }
return firehose return firehose
@ -1860,8 +1864,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 var buf bytes.Buffer
f.outputBuffer.Reset()
previousHash := block.PreviousID() previousHash := block.PreviousID()
previousNum := 0 previousNum := 0
@ -1881,9 +1884,9 @@ 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!
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 { 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))
} }
@ -1892,9 +1895,9 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina
panic(fmt.Errorf("closing encoder should have been infaillible: %w", err)) panic(fmt.Errorf("closing encoder should have been infaillible: %w", err))
} }
f.outputBuffer.WriteString("\n") buf.WriteString("\n")
f.flushToFirehose(f.outputBuffer.Bytes()) f.flushToFirehose(buf.Bytes())
} }
// printToFirehose is an easy way to print to Firehose format, it essentially // printToFirehose is an easy way to print to Firehose format, it essentially

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

@ -38,7 +38,6 @@ 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 _, concurrent := range []bool{true, false} {
for _, folder := range testFolders { for _, folder := range testFolders {
name := filepath.Base(folder) name := filepath.Base(folder)