Streamly.Internal.Data.SVar.Dispatch

Latency collection

collectLatency :: SVar t m a -> YieldRateInfo -> Bool -> IO (Count, AbsTime, NanoSecond64)

Diagnostics

withDiagMVar :: SVar t m a -> String -> IO () -> IO ()

dumpSVar :: SVar t m a -> IO String

printSVar :: SVar t m a -> String -> IO ()

Thread accounting

delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()

modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()

allThreadsDone :: MonadIO m => SVar t m a -> 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.

Dispatching

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

pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()

pushWorkerPar :: MonadAsync m => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()

In contrast to pushWorker which always happens only from the consumer thread, a pushWorkerPar can happen concurrently from multiple threads on the producer side. So we need to use a thread safe modification of workerThreads. Alternatively, we can use a CreateThread event to avoid using a CAS based modification.

dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool

dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool

sendWorkerWait :: MonadAsync m => (SVar t m a -> IO ()) -> (SVar t m a -> m Bool) -> SVar t m a -> m ()

sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)

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

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