Streamly.Internal.Data.SVar.Worker

Adjusting Limits

decrementYieldLimit :: SVar t m a -> IO Bool

incrementYieldLimit :: SVar t m a -> IO ()

decrementBufferLimit :: SVar t m a -> IO ()

incrementBufferLimit :: SVar t m a -> IO ()

resetBufferLimit :: SVar t m a -> IO ()

Rate Control

data Work

Constructors

BlockWait NanoSecond64

PartialWorker Count

ManyWorkers Int Count

Instances

Show Work

Defined in Streamly.Internal.Data.SVar.Worker

Methods

showsPrec :: Int -> Work -> ShowS

show :: Work -> String

showList :: [Work] -> ShowS

isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool

estimateWorkers :: Limit -> Count -> Count -> NanoSecond64 -> NanoSecond64 -> NanoSecond64 -> LatencyRange -> Work

updateYieldCount :: WorkerInfo -> IO Count

minThreadDelay :: NanoSecond64

This is a magic number and it is overloaded, and used at several places to achieve batching:

1. If we have to sleep to slowdown this is the minimum period that we accumulate before we sleep. Also, workers do not stop until this much sleep time is accumulated.
2. Collected latencies are computed and transferred to measured latency after a minimum of this period.

workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool

workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()

Send Events

send :: SVar t m a -> ChildEvent a -> IO Int

This function is used by the producer threads to queue output for the consumer thread to consume. Returns whether the queue has more space.

ringDoorBell :: SVar t m a -> IO ()

Yield

sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool

sendToProducer :: SVar t m a -> ChildEvent a -> IO Int

Stop

sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()

sendStopToProducer :: MonadIO m => SVar t m a -> m ()

Exception

handleChildException :: SVar t m a -> SomeException -> IO ()

handleFoldException :: SVar t m a -> SomeException -> IO ()