Copyright	(c) 2017 Composewell Technologies
License	BSD-3-Clause
Maintainer	[email protected]
Stability	experimental
Portability	GHC
Safe Haskell	None
Language	Haskell2010

Streamly.Internal.Data.SVar

Contents

New SVar
- Parallel
- Ahead

Description

Deprecated: SVar is replaced by Channel.

Synopsis

module Streamly.Internal.Data.SVar.Type
send :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> ChildEvent a -> IO Int
ringDoorBell :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO ()
decrementYieldLimit :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO Bool
incrementYieldLimit :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO ()
minThreadDelay :: NanoSecond64
data Work
- = BlockWait NanoSecond64
- | PartialWorker Count
- | ManyWorkers Int Count
estimateWorkers :: Limit -> Count -> Count -> NanoSecond64 -> NanoSecond64 -> NanoSecond64 -> LatencyRange -> Work
isBeyondMaxRate :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> YieldRateInfo -> IO Bool
workerRateControl :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
sendYield :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
sendStop :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> Maybe WorkerInfo -> IO ()
handleChildException :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> SomeException -> IO ()
updateYieldCount :: WorkerInfo -> IO Count
sendStopToProducer :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadIO m => SVar t m a -> m ()
decrementBufferLimit :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO ()
incrementBufferLimit :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO ()
resetBufferLimit :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO ()
workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
sendToProducer :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> ChildEvent a -> IO Int
handleFoldException :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> SomeException -> IO ()
withDiagMVar :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> String -> IO () -> IO ()
printSVar :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> String -> IO ()
collectLatency :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> YieldRateInfo -> Bool -> IO (Count, AbsTime, NanoSecond64)
delThread :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadIO m => SVar t m a -> ThreadId -> m ()
modifyThread :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadIO m => SVar t m a -> ThreadId -> m ()
allThreadsDone :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadIO m => SVar t m a -> m Bool
recordMaxWorkers :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadIO m => SVar t m a -> m ()
dumpSVar :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO String
pushWorker :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => Count -> SVar t m a -> m ()
dispatchWorker :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => Count -> SVar t m a -> m Bool
dispatchWorkerPaced :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVar t m a -> m Bool
sendWorkerWait :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => (SVar t m a -> IO ()) -> (SVar t m a -> m Bool) -> SVar t m a -> m ()
sendWorkerDelay :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO ()
sendWorkerDelayPaced :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO ()
pushWorkerPar :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
readOutputQRaw :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO ([ChildEvent a], Int)
readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)
cleanupSVar :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO ()
readOutputQPaced :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVar t m a -> m [ChildEvent a]
readOutputQBounded :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVar t m a -> m [ChildEvent a]
postProcessPaced :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVar t m a -> m Bool
postProcessBounded :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVar t m a -> m Bool
cleanupSVarFromWorker :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO ()
getYieldRateInfo :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. State t m a -> IO (Maybe YieldRateInfo)
newSVarStats :: IO SVarStats
newParallelVar :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVarStopStyle -> State t m a -> m (SVar t m a)
enqueueAhead :: forall t (m :: Type -> Type) a. SVar t m a -> IORef ([t m a], Int) -> (RunInIO m, t m a) -> IO ()
reEnqueueAhead :: forall t (m :: Type -> Type) a. 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 :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a
- = Clearing
- | Waiting Int
- | Ready (Entry Int (AheadHeapEntry t m a))
dequeueFromHeap :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int) -> IO (HeapDequeueResult t m a)
dequeueFromHeapSeq :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int) -> Int -> IO (HeapDequeueResult t m a)
requeueOnHeapTop :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int) -> Entry Int (AheadHeapEntry t m a) -> Int -> IO ()
updateHeapSeq :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. 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)

Documentation

module Streamly.Internal.Data.SVar.Type

send :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> ChildEvent a -> IO Int Source #

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 :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO () Source #

decrementYieldLimit :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO Bool Source #

incrementYieldLimit :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO () Source #

minThreadDelay :: NanoSecond64 Source #

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

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.
Collected latencies are computed and transferred to measured latency after a minimum of this period.

data Work Source #

Constructors

BlockWait NanoSecond64
PartialWorker Count
ManyWorkers Int Count

Instances

Instances details

Show Work Source #
Instance details Defined in Streamly.Internal.Data.SVar.Worker Methods showsPrec :: Int -> Work -> ShowS # show :: Work -> String # showList :: [Work] -> ShowS #

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

isBeyondMaxRate :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> YieldRateInfo -> IO Bool Source #

workerRateControl :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool Source #

sendYield :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool Source #

sendStop :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> Maybe WorkerInfo -> IO () Source #

handleChildException :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> SomeException -> IO () Source #

updateYieldCount :: WorkerInfo -> IO Count Source #

sendStopToProducer :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadIO m => SVar t m a -> m () Source #

decrementBufferLimit :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO () Source #

incrementBufferLimit :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO () Source #

resetBufferLimit :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO () Source #

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

sendToProducer :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> ChildEvent a -> IO Int Source #

handleFoldException :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> SomeException -> IO () Source #

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

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

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

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

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

allThreadsDone :: forall m (t :: (Type -> Type) -> Type -> Type) a. 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.

recordMaxWorkers :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadIO m => SVar t m a -> m () Source #

dumpSVar :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO String Source #

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

dispatchWorker :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => Count -> SVar t m a -> m Bool Source #

dispatchWorkerPaced :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVar t m a -> m Bool Source #

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

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

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

pushWorkerPar :: forall m (t :: (Type -> Type) -> Type -> Type) a. 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.

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

readOutputQRaw :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO ([ChildEvent a], Int) Source #

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

cleanupSVar :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO () Source #

readOutputQPaced :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVar t m a -> m [ChildEvent a] Source #

readOutputQBounded :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVar t m a -> m [ChildEvent a] Source #

postProcessPaced :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVar t m a -> m Bool Source #

postProcessBounded :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVar t m a -> m Bool Source #

cleanupSVarFromWorker :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. SVar t m a -> IO () Source #

New SVar

getYieldRateInfo :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. State t m a -> IO (Maybe YieldRateInfo) Source #

newSVarStats :: IO SVarStats Source #

Parallel

newParallelVar :: forall m (t :: (Type -> Type) -> Type -> Type) a. MonadAsync m => SVarStopStyle -> State t m a -> m (SVar t m a) Source #

Ahead

enqueueAhead :: forall t (m :: Type -> Type) a. SVar t m a -> IORef ([t m a], Int) -> (RunInIO m, t m a) -> IO () Source #

reEnqueueAhead :: forall t (m :: Type -> Type) a. SVar t m a -> IORef ([t m a], Int) -> t m a -> IO () Source #

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

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

data HeapDequeueResult (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a Source #

Constructors

Clearing
Waiting Int
Ready (Entry Int (AheadHeapEntry t m a))

dequeueFromHeap :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int) -> IO (HeapDequeueResult t m a) Source #

dequeueFromHeapSeq :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int) -> Int -> IO (HeapDequeueResult t m a) Source #

requeueOnHeapTop :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int) -> Entry Int (AheadHeapEntry t m a) -> Int -> IO () Source #

updateHeapSeq :: forall (t :: (Type -> Type) -> Type -> Type) (m :: Type -> Type) a. IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int) -> Int -> IO () Source #

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

heapIsSane :: Maybe Int -> Int -> Bool Source #

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