diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index dab9f61cd6..863fb89276 100644 --- a/eth/tracers/firehose.go +++ b/eth/tracers/firehose.go @@ -192,9 +192,10 @@ type Firehose struct { testingIgnoreGenesisBlock bool // Worker queue for block printing - blockPrintQueue chan *blockPrintJob - flushDone sync.WaitGroup - closeOnce sync.Once + blockPrintQueue chan *blockPrintJob + blockOutputQueue chan *blockOutput + flushDone sync.WaitGroup + closeOnce sync.Once } const FirehoseProtocolVersion = "3.0" @@ -261,11 +262,32 @@ func NewFirehose(config *FirehoseConfig) *Firehose { 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 + go func() { + expected := uint64(1000) // TODO: We will need to determine at which block it starts + 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++ // TODO: Assuming block numbers are linear + } + } + }() } return firehose @@ -1859,13 +1881,13 @@ 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)) } - var buf bytes.Buffer - previousHash := block.PreviousID() previousNum := 0 if block.Number > 0 { @@ -1897,7 +1919,14 @@ func (f *Firehose) printBlockToFirehose(block *pbeth.Block, finalityStatus *Fina buf.WriteString("\n") - f.flushToFirehose(buf.Bytes()) + if f.concurrentBlockFlushing { + f.blockOutputQueue <- &blockOutput{ + blockNum: block.Number, + data: buf.Bytes(), + } + } else { + f.flushToFirehose(buf.Bytes()) + } } // printToFirehose is an easy way to print to Firehose format, it essentially @@ -2852,3 +2881,8 @@ func (f *Firehose) CloseBlockPrintQueue() { }) } } + +type blockOutput struct { + blockNum uint64 + data []byte +} diff --git a/eth/tracers/firehose_concurrency_test.go b/eth/tracers/firehose_concurrency_test.go index 902a62b04d..abcaa79e17 100644 --- a/eth/tracers/firehose_concurrency_test.go +++ b/eth/tracers/firehose_concurrency_test.go @@ -10,6 +10,7 @@ import ( "strconv" "strings" "testing" + "time" ) func TestFirehose_BlockPrintsToFirehose_SingleBlock(t *testing.T) { @@ -92,6 +93,8 @@ func TestFirehose_BlocksPrintToFirehose_MultipleBlocksInOrder(t *testing.T) { f.OnBlockEnd(nil) } + time.Sleep(5 * time.Second) + f.OnClose() output := f.InternalTestingBuffer().String()