Copyright	(c) 2017 Composewell Technologies
License	BSD-3-Clause
Maintainer	streamly@composewell.com
Stability	experimental
Portability	GHC
Safe Haskell	Safe-Inferred
Language	Haskell2010

Streamly.Internal.Data.SVar.Dispatch

Contents

Latency collection
Diagnostics
Thread accounting
Dispatching

Description

Synopsis

collectLatency :: SVar t m a -> YieldRateInfo -> Bool -> IO (Count, AbsTime, NanoSecond64)
withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
dumpSVar :: SVar t m a -> IO String
printSVar :: SVar t m a -> String -> IO ()
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
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 ()
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 ()

Latency collection

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

Diagnostics

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

dumpSVar :: SVar t m a -> IO String Source #

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

Thread accounting

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

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

allThreadsDone :: MonadIO m => SVar t m a -> m Bool Source #

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 () Source #

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

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

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 Source #

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

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

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

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

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