Streamly.Internal.Data.Stream.Channel.Dispatcher

Latency collection

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.

collectLatency :: Bool -> SVarStats -> YieldRateInfo -> Bool -> IO (Count, AbsTime, NanoSecond64)

Thread accounting

addThread :: MonadIO m => IORef (Set ThreadId) -> ThreadId -> m ()

delThread :: MonadIO m => IORef (Set ThreadId) -> ThreadId -> m ()

modifyThread :: MonadIO m => IORef (Set ThreadId) -> MVar () -> ThreadId -> m ()

allThreadsDone :: MonadIO m => IORef (Set ThreadId) -> m Bool

This is safe even if we are adding more threads concurrently because if a child thread is adding another thread then anyway workerThreads will not be empty.

recordMaxWorkers :: MonadIO m => IORef Int -> SVarStats -> m ()

Diagnostics

dumpSVarStats :: Bool -> Maybe YieldRateInfo -> SVarStats -> IO String