Streamly.Internal.Data.SVar

Deprecated: SVar is replaced by Channel.

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

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

Read Output

readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)

readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)

readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]

readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]

Postprocess Hook After Reading

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

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

Release Resources

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

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

New SVar

getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)

newSVarStats :: IO SVarStats

Parallel

newParallelVar :: MonadAsync m => SVarStopStyle -> State t m a -> m (SVar t m a)

Ahead

enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> (RunInIO m, t m a) -> IO ()

reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()

queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool

dequeueAhead :: MonadIO m => IORef ([t m a], Int) -> m (Maybe (t m a, Int))

data HeapDequeueResult t m a

Constructors

Clearing

Waiting Int

Ready (Entry Int (AheadHeapEntry t m a))

dequeueFromHeap :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int) -> IO (HeapDequeueResult t m a)

dequeueFromHeapSeq :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int) -> Int -> IO (HeapDequeueResult t m a)

requeueOnHeapTop :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int) -> Entry Int (AheadHeapEntry t m a) -> Int -> IO ()

updateHeapSeq :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int) -> Int -> IO ()

withIORef :: IORef a -> (a -> IO b) -> IO b

heapIsSane :: Maybe Int -> Int -> Bool

newAheadVar :: MonadAsync m => State t m a -> t m a -> (IORef ([t m a], Int) -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int) -> State t m a -> SVar t m a -> Maybe WorkerInfo -> m ()) -> m (SVar t m a)