From 3e10bd67ba92d9c4540bc9ca8294b71295de43c9 Mon Sep 17 00:00:00 2001 From: David Zhou Date: Thu, 15 May 2025 09:50:26 -0400 Subject: [PATCH] Buffer size as variable --- eth/tracers/firehose.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/eth/tracers/firehose.go b/eth/tracers/firehose.go index 2bf8867065..5a517770c8 100644 --- a/eth/tracers/firehose.go +++ b/eth/tracers/firehose.go @@ -200,6 +200,7 @@ type Firehose struct { flushStarted bool flushStartBlockNum uint64 flushStartCond *sync.Cond + flushBufferSize int } const FirehoseProtocolVersion = "3.0" @@ -254,7 +255,8 @@ func NewFirehose(config *FirehoseConfig) *Firehose { deferredCallState: NewDeferredCallState(), latestCallEnterSuicided: false, - flushStartCond: sync.NewCond(&sync.Mutex{}), + flushStartCond: sync.NewCond(&sync.Mutex{}), + flushBufferSize: 100, // TODO: Optimal buffer size tbd } if config.private != nil { @@ -268,8 +270,8 @@ func NewFirehose(config *FirehoseConfig) *Firehose { log.Info("Firehose concurrent block flushing enabled, starting", config.ConcurrentBlockFlushing, "block print worker goroutine") - firehose.flushJobQueue = make(chan *blockPrintJob, 100) // TODO: Optimal buffer size tbd - firehose.flushOrderedOutputQueue = make(chan *blockOutput, 100) + firehose.flushJobQueue = make(chan *blockPrintJob, firehose.flushBufferSize) + firehose.flushOrderedOutputQueue = make(chan *blockOutput, firehose.flushBufferSize) for i := 0; i < config.ConcurrentBlockFlushing; i++ { firehose.flushJobWG.Add(1) go firehose.blockPrintWorker()