@@ -1,69 +1,31 @@
1−{-# LANGUAGE UnboxedTuples #-}
2−
31-- |
42-- Module : Streamly.Internal.Data.SVar
53-- Copyright : (c) 2017 Composewell Technologies
6−--
7−-- License : BSD3
4+-- License : BSD-3-Clause
85-- Maintainer : streamly@composewell.com
96-- Stability : experimental
107-- Portability : GHC
11−
8+--
9+--
1210module Streamly.Internal.Data.SVar
1311 (
14− MonadAsync
15− , SVarStyle (..)
16− , SVarStopStyle (..)
17− , SVar (..)
18−
19− -- State threaded around the stream
20− , Limit (..)
21− , State (streamVar)
22− , defState
23− , adaptState
24− , getMaxThreads
25− , setMaxThreads
26− , getMaxBuffer
27− , setMaxBuffer
28− , getStreamRate
29− , setStreamRate
30− , setStreamLatency
31− , getYieldLimit
32− , setYieldLimit
33− , getInspectMode
34− , setInspectMode
35− , recordMaxWorkers
12+ module Streamly.Internal.Data.SVar.Type
13+ , module Streamly.Internal.Data.SVar.Worker
14+ , module Streamly.Internal.Data.SVar.Dispatch
15+ , module Streamly.Internal.Data.SVar.Pull
3616
37− , cleanupSVar
38− , cleanupSVarFromWorker
17+ -- * New SVar
18+ , getYieldRateInfo
19+ , newSVarStats
3920
40− -- SVar related
41− , newAheadVar
21+ -- ** Parallel
4222 , newParallelVar
43− , captureMonadState
44− , RunInIO (..)
4523
46− , WorkerInfo (..)
47− , YieldRateInfo (..)
48− , ThreadAbort (..)
49− , ChildEvent (..)
50− , AheadHeapEntry (..)
51− , send
52− , sendToProducer
53− , sendYield
54− , sendStop
55− , sendStopToProducer
56− , enqueueLIFO
57− , enqueueFIFO
24+ -- ** Ahead
5825 , enqueueAhead
5926 , reEnqueueAhead
60− , pushWorkerPar
61− , handleChildException
62− , handleFoldException
63−
6427 , queueEmptyAhead
6528 , dequeueAhead
66−
6729 , HeapDequeueResult(..)
6830 , dequeueFromHeap
6931 , dequeueFromHeapSeq
@@ -71,1166 +33,98 @@ module Streamly.Internal.Data.SVar
7133 , updateHeapSeq
7234 , withIORef
7335 , heapIsSane
74−
75− , Rate (..)
76− , getYieldRateInfo
77− , newSVarStats
78− , collectLatency
79− , workerUpdateLatency
80− , isBeyondMaxRate
81− , workerRateControl
82− , updateYieldCount
83− , decrementYieldLimit
84− , incrementYieldLimit
85− , decrementBufferLimit
86− , incrementBufferLimit
87− , postProcessBounded
88− , postProcessPaced
89− , readOutputQBounded
90− , readOutputQPaced
91− , readOutputQBasic
92− , dispatchWorkerPaced
93− , sendFirstWorker
94− , delThread
95− , modifyThread
96− , doFork
97− , fork
98− , forkManaged
99−
100− , toStreamVar
101− , SVarStats (..)
102− , dumpSVar
103− , printSVar
104− , withDiagMVar
36+ , newAheadVar
10537 )
10638where
10739
108−import Control.Concurrent (ThreadId, myThreadId, threadDelay, throwTo)
109−import Control.Concurrent.MVar
110− (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar,
111− tryReadMVar)
112−import Control.Exception
113− (SomeException(..), assert, Exception, catches,
114− throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
115− BlockedIndefinitelyOnSTM(..))
116−import Control.Monad (when)
117−import Control.Monad.IO.Class (MonadIO(..))
118−import Control.Monad.Trans.Control (MonadBaseControl, control)
119−import Streamly.Internal.Data.Atomics
120− (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
121− storeLoadBarrier)
122−import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
123−import Data.Functor (void)
124−import Data.Heap (Heap, Entry(..))
125−import Data.Int (Int64)
126−import Data.IORef
127− (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
128−import Data.Kind (Type)
129−import Data.Maybe (fromJust, fromMaybe)
130−#if __GLASGOW_HASKELL__ < 808
131−import Data.Semigroup ((<>))
132−#endif
133−import Data.Set (Set)
134−import System.IO (hPutStrLn, stderr)
40+#include "inline.hs"
13541
42+import Control.Concurrent (myThreadId, takeMVar)
43+import Control.Concurrent.MVar (newEmptyMVar, tryPutMVar, tryTakeMVar, newMVar)
44+import Control.Exception (assert)
45+import Control.Monad (void)
46+import Control.Monad.IO.Class (MonadIO(liftIO))
47+import Data.Heap (Heap, Entry(..))
48+import Data.IORef (newIORef, readIORef)
49+import Data.IORef (IORef, atomicModifyIORef)
13650import Streamly.Internal.Control.Concurrent
137− (MonadAsync, RunInIO(..), doFork, fork, forkManaged)
138−import Streamly.Internal.Data.Time.Clock.Type (Clock(..), getTime)
139−import Streamly.Internal.Data.Time.Units
140− (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
141− fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
51+ (MonadAsync, captureMonadState, RunInIO(..))
52+import Streamly.Internal.Data.Atomics
53+ (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier)
54+import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
55+import Streamly.Internal.Data.Time.Units (NanoSecond64(..))
14256
14357import qualified Data.Heap as H
144−import qualified Data.Set as S
145−
146−newtype Count = Count Int64
147− deriving ( Eq
148− , Read
149− , Show
150− , Enum
151− , Bounded
152− , Num
153− , Real
154− , Integral
155− , Ord
156− )
58+import qualified Data.Set as S
15759
158−------------------------------------------------------------------------------
159−-- Parent child thread communication type
160−------------------------------------------------------------------------------
60+import Streamly.Internal.Data.SVar.Dispatch
61+import Streamly.Internal.Data.SVar.Pull
62+import Streamly.Internal.Data.SVar.Type
63+import Streamly.Internal.Data.SVar.Worker
16164
162−data ThreadAbort = ThreadAbort deriving Show
163−
164−instance Exception ThreadAbort
165−
166−-- | Events that a child thread may send to a parent thread.
167−data ChildEvent a =
168− ChildYield a
169− | ChildStop ThreadId (Maybe SomeException)
170−
171−-- | Sorting out-of-turn outputs in a heap for Ahead style streams
172−data AheadHeapEntry (t :: (Type -> Type) -> Type -> Type) m a =
173− AheadEntryNull
174− | AheadEntryPure a
175− | AheadEntryStream (RunInIO m, t m a)
176−#undef Type
177−
178−------------------------------------------------------------------------------
179−-- State threaded around the monad for thread management
180−------------------------------------------------------------------------------
181−
182−-- | Identify the type of the SVar. Two computations using the same style can
183−-- be scheduled on the same SVar.
184−data SVarStyle =
185− AsyncVar -- depth first concurrent
186− | WAsyncVar -- breadth first concurrent
187− | ParallelVar -- all parallel
188− | AheadVar -- Concurrent look ahead
189− deriving (Eq, Show)
190−
191−-- | An SVar or a Stream Var is a conduit to the output from multiple streams
192−-- running concurrently and asynchronously. An SVar can be thought of as an
193−-- asynchronous IO handle. We can write any number of streams to an SVar in a
194−-- non-blocking manner and then read them back at any time at any pace. The
195−-- SVar would run the streams asynchronously and accumulate results. An SVar
196−-- may not really execute the stream completely and accumulate all the results.
197−-- However, it ensures that the reader can read the results at whatever paces
198−-- it wants to read. The SVar monitors and adapts to the consumer's pace.
199−--
200−-- An SVar is a mini scheduler, it has an associated workLoop that holds the
201−-- stream tasks to be picked and run by a pool of worker threads. It has an
202−-- associated output queue where the output stream elements are placed by the
203−-- worker threads. A outputDoorBell is used by the worker threads to intimate the
204−-- consumer thread about availability of new results in the output queue. More
205−-- workers are added to the SVar by 'fromStreamVar' on demand if the output
206−-- produced is not keeping pace with the consumer. On bounded SVars, workers
207−-- block on the output queue to provide throttling of the producer when the
208−-- consumer is not pulling fast enough. The number of workers may even get
209−-- reduced depending on the consuming pace.
210−--
211−-- New work is enqueued either at the time of creation of the SVar or as a
212−-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
213−-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
214−
215−-- We measure the individual worker latencies to estimate the number of workers
216−-- needed or the amount of time we have to sleep between dispatches to achieve
217−-- a particular rate when controlled pace mode it used.
218−data WorkerInfo = WorkerInfo
219− { workerYieldMax :: Count -- 0 means unlimited
220− -- total number of yields by the worker till now
221− , workerYieldCount :: IORef Count
222− -- yieldCount at start, timestamp
223− , workerLatencyStart :: IORef (Count, AbsTime)
224− }
225−
226−-- | Specifies the stream yield rate in yields per second (@Hertz@).
227−-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
228−-- allow only as many yields as we have accumulated as per 'rateGoal' since the
229−-- start of time. If the consumer or the producer is slower or faster, the
230−-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
231−-- between the two by increasing or decreasing the pull rate from the producer.
232−-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
233−-- much as 'rateBuffer'.
234−--
235−-- 'rateLow' puts a bound on how low the instantaneous rate can go when
236−-- recovering the rate gap. In other words, it determines the maximum yield
237−-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
238−-- rate can go when recovering the rate gap. In other words, it determines the
239−-- minimum yield latency. We reduce the latency by increasing concurrency,
240−-- therefore we can say that it puts an upper bound on concurrency.
241−--
242−-- If the 'rateGoal' is 0 or negative the stream never yields a value.
243−-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
244−--
245−-- /Since: 0.5.0 ("Streamly")/
246−--
247−-- @since 0.8.0
248−data Rate = Rate
249− { rateLow :: Double -- ^ The lower rate limit
250− , rateGoal :: Double -- ^ The target rate we want to achieve
251− , rateHigh :: Double -- ^ The upper rate limit
252− , rateBuffer :: Int -- ^ Maximum slack from the goal
253− }
254−
255−data LatencyRange = LatencyRange
256− { minLatency :: NanoSecond64
257− , maxLatency :: NanoSecond64
258− } deriving Show
259−
260−-- Rate control.
261−data YieldRateInfo = YieldRateInfo
262− { svarLatencyTarget :: NanoSecond64
263− , svarLatencyRange :: LatencyRange
264− , svarRateBuffer :: Int
265−
266− -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
267− -- read by the worker threads
268− , svarGainedLostYields :: IORef Count
269−
270− -- Actual latency/througput as seen from the consumer side, we count the
271− -- yields and the time it took to generates those yields. This is used to
272− -- increase or decrease the number of workers needed to achieve the desired
273− -- rate. The idle time of workers is adjusted in this, so that we only
274− -- account for the rate when the consumer actually demands data.
275− -- XXX interval latency is enough, we can move this under diagnostics build
276− -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
277− -- read by the worker threads
278− , svarAllTimeLatency :: IORef (Count, AbsTime)
279−
280− -- XXX Worker latency specified by the user to be used before the first
281− -- actual measurement arrives. Not yet implemented
282− , workerBootstrapLatency :: Maybe NanoSecond64
283−
284− -- After how many yields the worker should update the latency information.
285− -- If the latency is high, this count is kept lower and vice-versa. XXX If
286− -- the latency suddenly becomes too high this count may remain too high for
287− -- long time, in such cases the consumer can change it.
288− -- 0 means no latency computation
289− -- XXX this is derivable from workerMeasuredLatency, can be removed.
290− -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
291− -- read by the worker threads
292− , workerPollingInterval :: IORef Count
293−
294− -- This is in progress latency stats maintained by the workers which we
295− -- empty into workerCollectedLatency stats at certain intervals - whenever
296− -- we process the stream elements yielded in this period. The first count
297− -- is all yields, the second count is only those yields for which the
298− -- latency was measured to be non-zero (note that if the timer resolution
299− -- is low the measured latency may be zero e.g. on JS platform).
300− -- [LOCKING] Locked access. Modified by the consumer thread as well as
301− -- worker threads. Workers modify it periodically based on
302− -- workerPollingInterval and not on every yield to reduce the locking
303− -- overhead.
304− -- (allYieldCount, yieldCount, timeTaken)
305− , workerPendingLatency :: IORef (Count, Count, NanoSecond64)
306−
307− -- This is the second level stat which is an accmulation from
308− -- workerPendingLatency stats. We keep accumulating latencies in this
309− -- bucket until we have stats for a sufficient period and then we reset it
310− -- to start collecting for the next period and retain the computed average
311− -- latency for the last period in workerMeasuredLatency.
312− -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
313− -- read by the worker threads
314− -- (allYieldCount, yieldCount, timeTaken)
315− , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)
316−
317− -- Latency as measured by workers, aggregated for the last period.
318− -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
319− -- read by the worker threads
320− , workerMeasuredLatency :: IORef NanoSecond64
321− }
322−
323−data SVarStats = SVarStats {
324− totalDispatches :: IORef Int
325− , maxWorkers :: IORef Int
326− , maxOutQSize :: IORef Int
327− , maxHeapSize :: IORef Int
328− , maxWorkQSize :: IORef Int
329− , avgWorkerLatency :: IORef (Count, NanoSecond64)
330− , minWorkerLatency :: IORef NanoSecond64
331− , maxWorkerLatency :: IORef NanoSecond64
332− , svarStopTime :: IORef (Maybe AbsTime)
333−}
334−
335−-- This is essentially a 'Maybe Word' type
336−data Limit = Unlimited | Limited Word deriving Show
337−
338−instance Eq Limit where
339− Unlimited == Unlimited = True
340− Unlimited == Limited _ = False
341− Limited _ == Unlimited = False
342− Limited x == Limited y = x == y
343−
344−instance Ord Limit where
345− Unlimited <= Unlimited = True
346− Unlimited <= Limited _ = False
347− Limited _ <= Unlimited = True
348− Limited x <= Limited y = x <= y
349−
350−-- When to stop the composed stream.
351−data SVarStopStyle =
352− StopNone -- stops only when all streams are finished
353− | StopAny -- stop when any stream finishes
354− | StopBy -- stop when a specific stream finishes
355− deriving (Eq, Show)
356−
357−-- | Buffering policy for persistent push workers (in ParallelT). In a pull
358−-- style SVar (in AsyncT, AheadT etc.), the consumer side dispatches workers on
359−-- demand, workers terminate if the buffer is full or if the consumer is not
360−-- cosuming fast enough. In a push style SVar, a worker is dispatched only
361−-- once, workers are persistent and keep pushing work to the consumer via a
362−-- bounded buffer. If the buffer becomes full the worker either blocks, or it
363−-- can drop an item from the buffer to make space.
364−--
365−-- Pull style SVars are useful in lazy stream evaluation whereas push style
366−-- SVars are useful in strict left Folds.
367−--
368−-- XXX Maybe we can separate the implementation in two different types instead
369−-- of using a common SVar type.
370−--
371−data PushBufferPolicy =
372− PushBufferDropNew -- drop the latest element and continue
373− | PushBufferDropOld -- drop the oldest element and continue
374− | PushBufferBlock -- block the thread until space
375− -- becomes available
376−
377−-- IMPORTANT NOTE: we cannot update the SVar after generating it as we have
378−-- references to the original SVar stored in several functions which will keep
379−-- pointing to the original data and the new updates won't reflect there.
380−-- Any updateable parts must be kept in mutable references (IORef).
381−--
382−data SVar t m a = SVar
383− {
384− -- Read only state
385− svarStyle :: SVarStyle
386− , svarMrun :: RunInIO m
387− , svarStopStyle :: SVarStopStyle
388− , svarStopBy :: IORef ThreadId
389−
390− -- Shared output queue (events, length)
391− -- XXX For better efficiency we can try a preallocated array type (perhaps
392− -- something like a vector) that allows an O(1) append. That way we will
393− -- avoid constructing and reversing the list. Possibly we can also avoid
394− -- the GC copying overhead. When the size increases we should be able to
395− -- allocate the array in chunks.
396− --
397− -- [LOCKING] Frequent locked access. This is updated by workers on each
398− -- yield and once in a while read by the consumer thread. This could have
399− -- big locking overhead if the number of workers is high.
400− --
401− -- XXX We can use a per-CPU data structure to reduce the locking overhead.
402− -- However, a per-cpu structure cannot guarantee the exact sequence in
403− -- which the elements were added, though that may not be important.
404− , outputQueue :: IORef ([ChildEvent a], Int)
405−
406− -- [LOCKING] Infrequent MVar. Used when the outputQ transitions from empty
407− -- to non-empty, or a work item is queued by a worker to the work queue and
408− -- needDoorBell is set by the consumer.
409− , outputDoorBell :: MVar () -- signal the consumer about output
410− , readOutputQ :: m [ChildEvent a]
411− , postProcess :: m Bool
412−
413− -- channel to send events from the consumer to the worker. Used to send
414− -- exceptions from a fold driver to the fold computation running as a
415− -- consumer thread in the concurrent fold cases. Currently only one event
416− -- is sent by the fold so we do not really need a queue for it.
417− , outputQueueFromConsumer :: IORef ([ChildEvent a], Int)
418− , outputDoorBellFromConsumer :: MVar ()
419−
420− -- Combined/aggregate parameters
421− -- This is truncated to maxBufferLimit if set to more than that. Otherwise
422− -- potentially each worker may yield one value to the buffer in the worst
423− -- case exceeding the requested buffer size.
424− , maxWorkerLimit :: Limit
425− , maxBufferLimit :: Limit
426− -- These two are valid and used only when maxBufferLimit is Limited.
427− , pushBufferSpace :: IORef Count
428− , pushBufferPolicy :: PushBufferPolicy
429− -- [LOCKING] The consumer puts this MVar after emptying the buffer, workers
430− -- block on it when the buffer becomes full. No overhead unless the buffer
431− -- becomes full.
432− , pushBufferMVar :: MVar ()
433−
434− -- [LOCKING] Read only access by consumer when dispatching a worker.
435− -- Decremented by workers when picking work and undo decrement if the
436− -- worker does not yield a value.
437− , remainingWork :: Maybe (IORef Count)
438− , yieldRateInfo :: Maybe YieldRateInfo
439−
440− -- Used only by bounded SVar types
441− , enqueue :: (RunInIO m, t m a) -> IO ()
442− , isWorkDone :: IO Bool
443− , isQueueDone :: IO Bool
444− , needDoorBell :: IORef Bool
445− , workLoop :: Maybe WorkerInfo -> m ()
446−
447− -- Shared, thread tracking
448− -- [LOCKING] Updated unlocked only by consumer thread in case of
449− -- Async/Ahead style SVars. Updated locked by worker threads in case of
450− -- Parallel style.
451− , workerThreads :: IORef (Set ThreadId)
452− -- [LOCKING] Updated locked by consumer thread when dispatching a worker
453− -- and by the worker threads when the thread stops. This is read unsafely
454− -- at several places where we want to rely on an approximate value.
455− , workerCount :: IORef Int
456− , accountThread :: ThreadId -> m ()
457− , workerStopMVar :: MVar ()
458−
459− , svarStats :: SVarStats
460− -- to track garbage collection of SVar
461− , svarRef :: Maybe (IORef ())
462−
463− -- Only for diagnostics
464− , svarInspectMode :: Bool
465− , svarCreator :: ThreadId
466− , outputHeap :: IORef ( Heap (Entry Int (AheadHeapEntry t m a))
467− , Maybe Int)
468− -- Shared work queue (stream, seqNo)
469− , aheadWorkQueue :: IORef ([t m a], Int)
470− }
471−
472−-------------------------------------------------------------------------------
473−-- State for concurrency control
474−-------------------------------------------------------------------------------
475−
476−-- XXX we can put the resettable fields in a oneShotConfig field and others in
477−-- a persistentConfig field. That way reset would be fast and scalable
478−-- irrespective of the number of fields.
479−--
480−-- XXX make all these Limited types and use phantom types to distinguish them
481−data State t m a = State
482− { -- one shot configuration, automatically reset for each API call
483− streamVar :: Maybe (SVar t m a)
484− , _yieldLimit :: Maybe Count
485−
486− -- persistent configuration, state that remains valid until changed by
487− -- an explicit setting via a combinator.
488− , _threadsHigh :: Limit
489− , _bufferHigh :: Limit
490− -- XXX these two can be collapsed into a single type
491− , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency
492− , _maxStreamRate :: Maybe Rate
493− , _inspectMode :: Bool
494− }
495−
496−-------------------------------------------------------------------------------
497−-- State defaults and reset
498−-------------------------------------------------------------------------------
499−
500−-- A magical value for the buffer size arrived at by running the smallest
501−-- possible task and measuring the optimal value of the buffer for that. This
502−-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
503−-- core-i7 processor.
504−magicMaxBuffer :: Word
505−magicMaxBuffer = 1500
506−
507−defaultMaxThreads, defaultMaxBuffer :: Limit
508−defaultMaxThreads = Limited magicMaxBuffer
509−defaultMaxBuffer = Limited magicMaxBuffer
510−
511−-- The fields prefixed by an _ are not to be accessed or updated directly but
512−-- via smart accessor APIs.
513−defState :: State t m a
514−defState = State
515− { streamVar = Nothing
516− , _yieldLimit = Nothing
517− , _threadsHigh = defaultMaxThreads
518− , _bufferHigh = defaultMaxBuffer
519− , _maxStreamRate = Nothing
520− , _streamLatency = Nothing
521− , _inspectMode = False
522− }
523−
524−-- XXX if perf gets affected we can have all the Nothing params in a single
525−-- structure so that we reset is fast. We can also use rewrite rules such that
526−-- reset occurs only in concurrent streams to reduce the impact on serial
527−-- streams.
528−-- We can optimize this so that we clear it only if it is a Just value, it
529−-- results in slightly better perf for zip/zipM but the performance of scan
530−-- worsens a lot, it does not fuse.
531−--
532−-- XXX This has a side effect of clearing the SVar and yieldLimit, therefore it
533−-- should not be used to convert from the same type to the same type, unless
534−-- you want to clear the SVar. For clearing the SVar you should be using the
535−-- appropriate unStream functions instead.
536−--
537−-- | Adapt the stream state from one type to another.
538−adaptState :: State t m a -> State t n b
539−adaptState st = st
540− { streamVar = Nothing
541− , _yieldLimit = Nothing
542− }
543−
544−-------------------------------------------------------------------------------
545−-- Smart get/set routines for State
546−-------------------------------------------------------------------------------
547−
548−-- Use get/set routines instead of directly accessing the State fields
549−setYieldLimit :: Maybe Int64 -> State t m a -> State t m a
550−setYieldLimit lim st =
551− st { _yieldLimit =
552− case lim of
553− Nothing -> Nothing
554− Just n ->
555− if n <= 0
556− then Just 0
557− else Just (fromIntegral n)
558− }
559−
560−getYieldLimit :: State t m a -> Maybe Count
561−getYieldLimit = _yieldLimit
562−
563−setMaxThreads :: Int -> State t m a -> State t m a
564−setMaxThreads n st =
565− st { _threadsHigh =
566− if n < 0
567− then Unlimited
568− else if n == 0
569− then defaultMaxThreads
570− else Limited (fromIntegral n)
571− }
572−
573−getMaxThreads :: State t m a -> Limit
574−getMaxThreads = _threadsHigh
575−
576−setMaxBuffer :: Int -> State t m a -> State t m a
577−setMaxBuffer n st =
578− st { _bufferHigh =
579− if n < 0
580− then Unlimited
581− else if n == 0
582− then defaultMaxBuffer
583− else Limited (fromIntegral n)
584− }
585−
586−getMaxBuffer :: State t m a -> Limit
587−getMaxBuffer = _bufferHigh
588−
589−setStreamRate :: Maybe Rate -> State t m a -> State t m a
590−setStreamRate r st = st { _maxStreamRate = r }
591−
592−getStreamRate :: State t m a -> Maybe Rate
593−getStreamRate = _maxStreamRate
594−
595−setStreamLatency :: Int -> State t m a -> State t m a
596−setStreamLatency n st =
597− st { _streamLatency =
598− if n <= 0
599− then Nothing
600− else Just (fromIntegral n)
601− }
602−
603−getStreamLatency :: State t m a -> Maybe NanoSecond64
604−getStreamLatency = _streamLatency
605−
606−setInspectMode :: State t m a -> State t m a
607−setInspectMode st = st { _inspectMode = True }
608−
609−getInspectMode :: State t m a -> Bool
610−getInspectMode = _inspectMode
611−
612−-------------------------------------------------------------------------------
613−-- Cleanup
61465-------------------------------------------------------------------------------
615−
616−cleanupSVar :: SVar t m a -> IO ()
617−cleanupSVar sv = do
618− workers <- readIORef (workerThreads sv)
619− Prelude.mapM_ (`throwTo` ThreadAbort)
620− workers
621−
622−cleanupSVarFromWorker :: SVar t m a -> IO ()
623−cleanupSVarFromWorker sv = do
624− workers <- readIORef (workerThreads sv)
625− self <- myThreadId
626− Prelude.mapM_ (`throwTo` ThreadAbort)
627− (Prelude.filter (/= self) $ S.toList workers)
628−
629−-------------------------------------------------------------------------------
630−-- Worker latency data collection
631−-------------------------------------------------------------------------------
632−
633−-- Every once in a while workers update the latencies and check the yield rate.
634−-- They return if we are above the expected yield rate. If we check too often
635−-- it may impact performance, if we check less often we may have a stale
636−-- picture. We update every minThreadDelay but we translate that into a yield
637−-- count based on latency so that the checking overhead is little.
638−--
639−-- XXX use a generation count to indicate that the value is updated. If the
640−-- value is updated an existing worker must check it again on the next yield.
641−-- Otherwise it is possible that we may keep updating it and because of the mod
642−-- worker keeps skipping it.
643−updateWorkerPollingInterval :: YieldRateInfo -> NanoSecond64 -> IO ()
644−updateWorkerPollingInterval yinfo latency = do
645− let periodRef = workerPollingInterval yinfo
646− cnt = max 1 $ minThreadDelay `div` latency
647− period = min cnt (fromIntegral magicMaxBuffer)
648−
649− writeIORef periodRef (fromIntegral period)
650−
651−{-# INLINE recordMinMaxLatency #-}
652−recordMinMaxLatency :: SVar t m a -> NanoSecond64 -> IO ()
653−recordMinMaxLatency sv new = do
654− let ss = svarStats sv
655− minLat <- readIORef (minWorkerLatency ss)
656− when (new < minLat || minLat == 0) $
657− writeIORef (minWorkerLatency ss) new
658−
659− maxLat <- readIORef (maxWorkerLatency ss)
660− when (new > maxLat) $ writeIORef (maxWorkerLatency ss) new
661−
662−recordAvgLatency :: SVar t m a -> (Count, NanoSecond64) -> IO ()
663−recordAvgLatency sv (count, time) = do
664− let ss = svarStats sv
665− modifyIORef (avgWorkerLatency ss) $
666− \(cnt, t) -> (cnt + count, t + time)
667−
668−-- Pour the pending latency stats into a collection bucket
669−{-# INLINE collectWorkerPendingLatency #-}
670−collectWorkerPendingLatency
671− :: IORef (Count, Count, NanoSecond64)
672− -> IORef (Count, Count, NanoSecond64)
673− -> IO (Count, Maybe (Count, NanoSecond64))
674−collectWorkerPendingLatency cur col = do
675− (fcount, count, time) <- atomicModifyIORefCAS cur $ \v -> ((0,0,0), v)
676−
677− (fcnt, cnt, t) <- readIORef col
678− let totalCount = fcnt + fcount
679− latCount = cnt + count
680− latTime = t + time
681− writeIORef col (totalCount, latCount, latTime)
682−
683− assert (latCount == 0 || latTime /= 0) (return ())
684− let latPair =
685− if latCount > 0 && latTime > 0
686− then Just (latCount, latTime)
687− else Nothing
688− return (totalCount, latPair)
689−
690−{-# INLINE shouldUseCollectedBatch #-}
691−shouldUseCollectedBatch
692− :: Count
693− -> NanoSecond64
694− -> NanoSecond64
695− -> NanoSecond64
696− -> Bool
697−shouldUseCollectedBatch collectedYields collectedTime newLat prevLat =
698− let r = fromIntegral newLat / fromIntegral prevLat :: Double
699− in (collectedYields > fromIntegral magicMaxBuffer)
700− || (collectedTime > minThreadDelay)
701− || (prevLat > 0 && (r > 2 || r < 0.5))
702− || (prevLat == 0)
703−
704−-- Returns a triple, (1) yield count since last collection, (2) the base time
705−-- when we started counting, (3) average latency in the last measurement
706−-- period. The former two are used for accurate measurement of the going rate
707−-- whereas the average is used for future estimates e.g. how many workers
708−-- should be maintained to maintain the rate.
709−-- CAUTION! keep it in sync with getWorkerLatency
710−collectLatency :: SVar t m a
711− -> YieldRateInfo
712− -> Bool
713− -> IO (Count, AbsTime, NanoSecond64)
714−collectLatency sv yinfo drain = do
715− let cur = workerPendingLatency yinfo
716− col = workerCollectedLatency yinfo
717− longTerm = svarAllTimeLatency yinfo
718− measured = workerMeasuredLatency yinfo
719−
720− (newCount, newLatPair) <- collectWorkerPendingLatency cur col
721− (lcount, ltime) <- readIORef longTerm
722− prevLat <- readIORef measured
723−
724− let newLcount = lcount + newCount
725− retWith lat = return (newLcount, ltime, lat)
726−
727− case newLatPair of
728− Nothing -> retWith prevLat
729− Just (count, time) -> do
730− let newLat = time `div` fromIntegral count
731− when (svarInspectMode sv) $ recordMinMaxLatency sv newLat
732− -- When we have collected a significant sized batch we compute the
733− -- new latency using that batch and return the new latency,
734− -- otherwise we return the previous latency derived from the
735− -- previous batch.
736− if shouldUseCollectedBatch newCount time newLat prevLat || drain
737− then do
738− -- XXX make this NOINLINE?
739− updateWorkerPollingInterval yinfo (max newLat prevLat)
740− when (svarInspectMode sv) $ recordAvgLatency sv (count, time)
741− writeIORef col (0, 0, 0)
742− writeIORef measured ((prevLat + newLat) `div` 2)
743− modifyIORef longTerm $ \(_, t) -> (newLcount, t)
744− retWith newLat
745− else retWith prevLat
746−
747−-------------------------------------------------------------------------------
748−-- Dumping the SVar for debug/diag
66+-- Creating an SVar
74967-------------------------------------------------------------------------------
75068
751−dumpSVarStats :: SVar t m a -> SVarStats -> SVarStyle -> IO String
752−dumpSVarStats sv ss style = do
753− case yieldRateInfo sv of
754− Nothing -> return ()
755− Just yinfo -> do
756− _ <- liftIO $ collectLatency sv yinfo True
757− return ()
758−
759− dispatches <- readIORef $ totalDispatches ss
760− maxWrk <- readIORef $ maxWorkers ss
761− maxOq <- readIORef $ maxOutQSize ss
762− maxHp <- readIORef $ maxHeapSize ss
763− minLat <- readIORef $ minWorkerLatency ss
764− maxLat <- readIORef $ maxWorkerLatency ss
765− (avgCnt, avgTime) <- readIORef $ avgWorkerLatency ss
766− (svarCnt, svarGainLossCnt, svarLat) <- case yieldRateInfo sv of
767− Nothing -> return (0, 0, 0)
768− Just yinfo -> do
769− (cnt, startTime) <- readIORef $ svarAllTimeLatency yinfo
770− if cnt > 0
771− then do
772− t <- readIORef (svarStopTime ss)
773− gl <- readIORef (svarGainedLostYields yinfo)
774− case t of
775− Nothing -> do
776− now <- getTime Monotonic
777− let interval = diffAbsTime64 now startTime
778− return (cnt, gl, interval `div` fromIntegral cnt)
779− Just stopTime -> do
780− let interval = diffAbsTime64 stopTime startTime
781− return (cnt, gl, interval `div` fromIntegral cnt)
782− else return (0, 0, 0)
783−
784− return $ unlines
785− [ "total dispatches = " <> show dispatches
786− , "max workers = " <> show maxWrk
787− , "max outQSize = " <> show maxOq
788− <> (if style == AheadVar
789− then "\nheap max size = " <> show maxHp
790− else "")
791− <> (if minLat > 0
792− then "\nmin worker latency = " <> showNanoSecond64 minLat
793− else "")
794− <> (if maxLat > 0
795− then "\nmax worker latency = " <> showNanoSecond64 maxLat
796− else "")
797− <> (if avgCnt > 0
798− then let lat = avgTime `div` fromIntegral avgCnt
799− in "\navg worker latency = " <> showNanoSecond64 lat
800− else "")
801− <> (if svarLat > 0
802− then "\nSVar latency = " <> showRelTime64 svarLat
803− else "")
804− <> (if svarCnt > 0
805− then "\nSVar yield count = " <> show svarCnt
806− else "")
807− <> (if svarGainLossCnt > 0
808− then "\nSVar gain/loss yield count = " <> show svarGainLossCnt
809− else "")
810− ]
811−
812−{-# NOINLINE dumpSVar #-}
813−dumpSVar :: SVar t m a -> IO String
814−dumpSVar sv = do
815− (oqList, oqLen) <- readIORef $ outputQueue sv
816− db <- tryReadMVar $ outputDoorBell sv
817− aheadDump <-
818− if svarStyle sv == AheadVar
819− then do
820− (oheap, oheapSeq) <- readIORef $ outputHeap sv
821− (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
822− return $ unlines
823− [ "heap length = " <> show (H.size oheap)
824− , "heap seqeunce = " <> show oheapSeq
825− , "work queue length = " <> show (length wq)
826− , "work queue sequence = " <> show wqSeq
827− ]
828− else return []
829−
830− let style = svarStyle sv
831− waiting <-
832− if style /= ParallelVar
833− then readIORef $ needDoorBell sv
834− else return False
835− rthread <- readIORef $ workerThreads sv
836− workers <- readIORef $ workerCount sv
837− stats <- dumpSVarStats sv (svarStats sv) (svarStyle sv)
838−
839− return $ unlines
840− [
841− "Creator tid = " <> show (svarCreator sv),
842− "style = " <> show (svarStyle sv)
843− , "---------CURRENT STATE-----------"
844− , "outputQueue length computed = " <> show (length oqList)
845− , "outputQueue length maintained = " <> show oqLen
846− -- XXX print the types of events in the outputQueue, first 5
847− , "outputDoorBell = " <> show db
848− ]
849− <> aheadDump
850− <> unlines
851− [ "needDoorBell = " <> show waiting
852− , "running threads = " <> show rthread
853− -- XXX print the status of first 5 threads
854− , "running thread count = " <> show workers
855− ]
856− <> "---------STATS-----------\n"
857− <> stats
858−
859−printSVar :: SVar t m a -> String -> IO ()
860−printSVar sv how = do
861− svInfo <- dumpSVar sv
862− hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo
863−
864−-- MVar diagnostics has some overhead - around 5% on AsyncT null benchmark, we
865−-- can keep it on in production to debug problems quickly if and when they
866−-- happen, but it may result in unexpected output when threads are left hanging
867−-- until they are GCed because the consumer went away.
868−
869−{-# NOINLINE mvarExcHandler #-}
870−mvarExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
871−mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
872− svInfo <- dumpSVar sv
873− hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
874− throwIO e
875−
876−{-# NOINLINE stmExcHandler #-}
877−stmExcHandler :: SVar t m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
878−stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
879− svInfo <- dumpSVar sv
880− hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
881− throwIO e
882−
883−withDiagMVar :: SVar t m a -> String -> IO () -> IO ()
884−withDiagMVar sv label action =
885− if svarInspectMode sv
886− then
887− action `catches` [ Handler (mvarExcHandler sv label)
888− , Handler (stmExcHandler sv label)
889− ]
890− else action
891−
892−------------------------------------------------------------------------------
893−-- Spawning threads
894−------------------------------------------------------------------------------
895−
896−-- | When we run computations concurrently, we completely isolate the state of
897−-- the concurrent computations from the parent computation. The invariant is
898−-- that we should never be running two concurrent computations in the same
899−-- thread without using the runInIO function. Also, we should never be running
900−-- a concurrent computation in the parent thread, otherwise it may affect the
901−-- state of the parent which is against the defined semantics of concurrent
902−-- execution.
903−captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
904−captureMonadState = control $ \run -> run (return $ RunInIO run)
905−
906−------------------------------------------------------------------------------
907−-- Collecting results from child workers in a streamed fashion
908−------------------------------------------------------------------------------
909−
910−-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
911−-- faster, along with the fields in sv required by send?
912−-- XXX make it noinline
913−--
914−-- XXX we may want to employ an increment and decrement in batches when the
915−-- througput is high or when the cost of synchronization is high. For example
916−-- if the application is distributed then inc/dec of a shared variable may be
917−-- very costly.
918−--
919−-- A worker decrements the yield limit before it executes an action. However,
920−-- the action may not result in an element being yielded, in that case we have
921−-- to increment the yield limit.
922−--
923−-- Note that we need it to be an Int type so that we have the ability to undo a
924−-- decrement that takes it below zero.
925−{-# INLINE decrementYieldLimit #-}
926−decrementYieldLimit :: SVar t m a -> IO Bool
927−decrementYieldLimit sv =
928− case remainingWork sv of
929− Nothing -> return True
930− Just ref -> do
931− r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
932− return $ r >= 1
933−
934−{-# INLINE incrementYieldLimit #-}
935−incrementYieldLimit :: SVar t m a -> IO ()
936−incrementYieldLimit sv =
937− case remainingWork sv of
938− Nothing -> return ()
939− Just ref -> atomicModifyIORefCAS_ ref (+ 1)
940−
941−-- XXX exception safety of all atomic/MVar operations
942−
943−-- TBD Each worker can have their own queue and the consumer can empty one
944−-- queue at a time, that way contention can be reduced.
945−
946−-- XXX Only yields should be counted in the buffer limit and not the Stop
947−-- events.
948−
949−{-# INLINE decrementBufferLimit #-}
950−decrementBufferLimit :: SVar t m a -> IO ()
951−decrementBufferLimit sv =
952− case maxBufferLimit sv of
953− Unlimited -> return ()
954− Limited _ -> do
955− let ref = pushBufferSpace sv
956− old <- atomicModifyIORefCAS ref $ \x ->
957− (if x >= 1 then x - 1 else x, x)
958− when (old <= 0) $
959− case pushBufferPolicy sv of
960− PushBufferBlock -> blockAndRetry
961− PushBufferDropNew -> do
962− -- We just drop one item and proceed. It is possible
963− -- that by the time we drop the item the consumer
964− -- thread might have run and created space in the
965− -- buffer, but we do not care about that condition.
966− -- This is not pedantically correct but it should be
967− -- fine in practice.
968− -- XXX we may want to drop only if n == maxBuf
969− -- otherwise we must have space in the buffer and a
970− -- decrement should be possible.
971− block <- atomicModifyIORefCAS (outputQueue sv) $
972− \(es, n) ->
973− case es of
974− [] -> (([],n), True)
975− _ : xs -> ((xs, n - 1), False)
976− when block blockAndRetry
977− -- XXX need a dequeue or ring buffer for this
978− PushBufferDropOld -> undefined
69+getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
70+getYieldRateInfo st = do
71+ -- convert rate in Hertz to latency in Nanoseconds
72+ let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
73+ case getStreamRate st of
74+ Just (Rate low goal high buf) ->
75+ let l = rateToLatency goal
76+ minl = rateToLatency high
77+ maxl = rateToLatency low
78+ in mkYieldRateInfo l (LatencyRange minl maxl) buf
79+ Nothing -> return Nothing
97980
98081 where
98182
982− blockAndRetry = do
983− let ref = pushBufferSpace sv
984− liftIO $ takeMVar (pushBufferMVar sv)
985− old <- atomicModifyIORefCAS ref $ \x ->
986− (if x >= 1 then x - 1 else x, x)
987− -- When multiple threads sleep on takeMVar, the first thread would
988− -- wakeup due to a putMVar by the consumer, but the rest of the threads
989− -- would have to put back the MVar after taking it and decrementing the
990− -- buffer count, otherwise all other threads will remain asleep.
991− if old >= 1
992− then void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
993− -- We do not put the MVar back in this case, instead we
994− -- wait for the consumer to put it.
995− else blockAndRetry
996−
997−{-# INLINE incrementBufferLimit #-}
998−incrementBufferLimit :: SVar t m a -> IO ()
999−incrementBufferLimit sv =
1000− case maxBufferLimit sv of
1001− Unlimited -> return ()
1002− Limited _ -> do
1003− atomicModifyIORefCAS_ (pushBufferSpace sv) (+ 1)
1004− writeBarrier
1005− void $ liftIO $ tryPutMVar (pushBufferMVar sv) ()
1006−
1007−{-# INLINE resetBufferLimit #-}
1008−resetBufferLimit :: SVar t m a -> IO ()
1009−resetBufferLimit sv =
1010− case maxBufferLimit sv of
1011− Unlimited -> return ()
1012− Limited n -> atomicModifyIORefCAS_ (pushBufferSpace sv)
1013− (const (fromIntegral n))
1014−
1015−{-# INLINE sendWithDoorBell #-}
1016−sendWithDoorBell ::
1017− IORef ([ChildEvent a], Int) -> MVar () -> ChildEvent a -> IO Int
1018−sendWithDoorBell q bell msg = do
1019− -- XXX can the access to outputQueue be made faster somehow?
1020− oldlen <- atomicModifyIORefCAS q $ \(es, n) ->
1021− ((msg : es, n + 1), n)
1022− when (oldlen <= 0) $ do
1023− -- The wake up must happen only after the store has finished otherwise
1024− -- we can have lost wakeup problems.
1025− writeBarrier
1026− -- Since multiple workers can try this at the same time, it is possible
1027− -- that we may put a spurious MVar after the consumer has already seen
1028− -- the output. But that's harmless, at worst it may cause the consumer
1029− -- to read the queue again and find it empty.
1030− -- The important point is that the consumer is guaranteed to receive a
1031− -- doorbell if something was added to the queue after it empties it.
1032− void $ tryPutMVar bell ()
1033− return oldlen
1034−
1035−-- | This function is used by the producer threads to queue output for the
1036−-- consumer thread to consume. Returns whether the queue has more space.
1037−send :: SVar t m a -> ChildEvent a -> IO Int
1038−send sv = sendWithDoorBell (outputQueue sv) (outputDoorBell sv)
1039−
1040−-- There is no bound implemented on the buffer, this is assumed to be low
1041−-- traffic.
1042−sendToProducer :: SVar t m a -> ChildEvent a -> IO Int
1043−sendToProducer sv msg = do
1044− -- In case the producer stream is blocked on pushing to the fold buffer
1045− -- then wake it up so that it can check for the stop event or exception
1046− -- being sent to it otherwise we will be deadlocked.
1047− void $ tryPutMVar (pushBufferMVar sv) ()
1048− sendWithDoorBell (outputQueueFromConsumer sv)
1049− (outputDoorBellFromConsumer sv) msg
1050−
1051−-- {-# NOINLINE sendStopToProducer #-}
1052−sendStopToProducer :: MonadIO m => SVar t m a -> m ()
1053−sendStopToProducer sv = liftIO $ do
1054− tid <- myThreadId
1055− void $ sendToProducer sv (ChildStop tid Nothing)
1056−
1057−workerCollectLatency :: WorkerInfo -> IO (Maybe (Count, NanoSecond64))
1058−workerCollectLatency winfo = do
1059− (cnt0, t0) <- readIORef (workerLatencyStart winfo)
1060− cnt1 <- readIORef (workerYieldCount winfo)
1061− let cnt = cnt1 - cnt0
1062−
1063− if cnt > 0
1064− then do
1065− t1 <- getTime Monotonic
1066− let period = fromRelTime64 $ diffAbsTime64 t1 t0
1067− writeIORef (workerLatencyStart winfo) (cnt1, t1)
1068− return $ Just (cnt, period)
1069− else return Nothing
1070−
1071−-- XXX There are a number of gotchas in measuring latencies.
1072−-- 1) We measure latencies only when a worker yields a value
1073−-- 2) It is possible that a stream calls the stop continuation, in which case
1074−-- the worker would not yield a value and we would not account that worker in
1075−-- latencies. Even though this case should ideally be accounted we do not
1076−-- account it because we cannot or do not distinguish it from the case
1077−-- described next.
1078−-- 3) It is possible that a worker returns without yielding anything because it
1079−-- never got a chance to pick up work.
1080−-- 4) If the system timer resolution is lower than the latency, the latency
1081−-- computation turns out to be zero.
1082−--
1083−-- We can fix this if we measure the latencies by counting the work items
1084−-- picked rather than based on the outputs yielded.
1085−workerUpdateLatency :: YieldRateInfo -> WorkerInfo -> IO ()
1086−workerUpdateLatency yinfo winfo = do
1087− r <- workerCollectLatency winfo
1088− case r of
1089− Just (cnt, period) -> do
1090− -- NOTE: On JS platform the timer resolution could be pretty low. When
1091− -- the timer resolution is low, measurement of latencies could be
1092− -- tricky. All the worker latencies will turn out to be zero if they
1093− -- are lower than the resolution. We only take into account those
1094− -- measurements which are more than the timer resolution.
1095−
1096− let ref = workerPendingLatency yinfo
1097− (cnt1, t1) = if period > 0 then (cnt, period) else (0, 0)
1098− atomicModifyIORefCAS_ ref $
1099− \(fc, n, t) -> (fc + cnt, n + cnt1, t + t1)
1100− Nothing -> return ()
1101−
1102−updateYieldCount :: WorkerInfo -> IO Count
1103−updateYieldCount winfo = do
1104− cnt <- readIORef (workerYieldCount winfo)
1105− let cnt1 = cnt + 1
1106− writeIORef (workerYieldCount winfo) cnt1
1107− return cnt1
1108−
1109−isBeyondMaxYield :: Count -> WorkerInfo -> Bool
1110−isBeyondMaxYield cnt winfo =
1111− let ymax = workerYieldMax winfo
1112− in ymax /= 0 && cnt >= ymax
1113−
1114−-- XXX we should do rate control periodically based on the total yields rather
1115−-- than based on the worker local yields as other workers may have yielded more
1116−-- and we should stop based on the aggregate yields. However, latency update
1117−-- period can be based on individual worker yields.
1118−{-# NOINLINE checkRatePeriodic #-}
1119−checkRatePeriodic :: SVar t m a
1120− -> YieldRateInfo
1121− -> WorkerInfo
1122− -> Count
1123− -> IO Bool
1124−checkRatePeriodic sv yinfo winfo ycnt = do
1125− i <- readIORef (workerPollingInterval yinfo)
1126− -- XXX use generation count to check if the interval has been updated
1127− if i /= 0 && (ycnt `mod` i) == 0
1128− then do
1129− workerUpdateLatency yinfo winfo
1130− -- XXX not required for parallel streams
1131− isBeyondMaxRate sv yinfo
1132− else return False
1133−
1134−-- CAUTION! this also updates the yield count and therefore should be called
1135−-- only when we are actually yielding an element.
1136−{-# NOINLINE workerRateControl #-}
1137−workerRateControl :: SVar t m a -> YieldRateInfo -> WorkerInfo -> IO Bool
1138−workerRateControl sv yinfo winfo = do
1139− cnt <- updateYieldCount winfo
1140− beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt
1141− return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)
1142−
1143−-- XXX we should do rate control here but not latency update in case of ahead
1144−-- streams. latency update must be done when we yield directly to outputQueue
1145−-- or when we yield to heap.
1146−--
1147−-- returns whether the worker should continue (True) or stop (False).
1148−{-# INLINE sendYield #-}
1149−sendYield :: SVar t m a -> Maybe WorkerInfo -> ChildEvent a -> IO Bool
1150−sendYield sv mwinfo msg = do
1151− oldlen <- send sv msg
1152− let limit = maxBufferLimit sv
1153− bufferSpaceOk <- case limit of
1154− Unlimited -> return True
1155− Limited lim -> do
1156− active <- readIORef (workerCount sv)
1157− return $ (oldlen + 1) < (fromIntegral lim - active)
1158− rateLimitOk <-
1159− case mwinfo of
1160− Just winfo ->
1161− case yieldRateInfo sv of
1162− Nothing -> return True
1163− Just yinfo -> workerRateControl sv yinfo winfo
1164− Nothing -> return True
1165− return $ bufferSpaceOk && rateLimitOk
1166−
1167−{-# INLINE workerStopUpdate #-}
1168−workerStopUpdate :: WorkerInfo -> YieldRateInfo -> IO ()
1169−workerStopUpdate winfo info = do
1170− i <- readIORef (workerPollingInterval info)
1171− when (i /= 0) $ workerUpdateLatency info winfo
1172−
1173−{-# INLINABLE sendStop #-}
1174−sendStop :: SVar t m a -> Maybe WorkerInfo -> IO ()
1175−sendStop sv mwinfo = do
1176− atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
1177− case (mwinfo, yieldRateInfo sv) of
1178− (Just winfo, Just info) ->
1179− workerStopUpdate winfo info
1180− _ ->
1181− return ()
1182− myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
1183−
1184−-------------------------------------------------------------------------------
1185−-- Doorbell
1186−-------------------------------------------------------------------------------
1187−
1188−{-# INLINE ringDoorBell #-}
1189−ringDoorBell :: SVar t m a -> IO ()
1190−ringDoorBell sv = do
1191− storeLoadBarrier
1192− w <- readIORef $ needDoorBell sv
1193− when w $ do
1194− -- Note: the sequence of operations is important for correctness here.
1195− -- We need to set the flag to false strictly before sending the
1196− -- outputDoorBell, otherwise the outputDoorBell may get processed too
1197− -- early and then we may set the flag to False to later making the
1198− -- consumer lose the flag, even without receiving a outputDoorBell.
1199− atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1200− void $ tryPutMVar (outputDoorBell sv) ()
1201−
1202−-------------------------------------------------------------------------------
1203−-- Async
1204−-------------------------------------------------------------------------------
1205−
1206−-- Note: For purely right associated expressions this queue should have at most
1207−-- one element. It grows to more than one when we have left associcated
1208−-- expressions. Large left associated compositions can grow this to a
1209−-- large size
1210−{-# INLINE enqueueLIFO #-}
1211−enqueueLIFO ::
1212− SVar t m a -> IORef [(RunInIO m, t m a)] -> (RunInIO m, t m a) -> IO ()
1213−enqueueLIFO sv q m = do
1214− atomicModifyIORefCAS_ q $ \ms -> m : ms
1215− ringDoorBell sv
83+ mkYieldRateInfo latency latRange buf = do
84+ measured <- newIORef 0
85+ wcur <- newIORef (0,0,0)
86+ wcol <- newIORef (0,0,0)
87+ now <- getTime Monotonic
88+ wlong <- newIORef (0,now)
89+ period <- newIORef 1
90+ gainLoss <- newIORef (Count 0)
121691
1217−-------------------------------------------------------------------------------
1218−-- WAsync
1219−-------------------------------------------------------------------------------
92+ return $ Just YieldRateInfo
93+ { svarLatencyTarget = latency
94+ , svarLatencyRange = latRange
95+ , svarRateBuffer = buf
96+ , svarGainedLostYields = gainLoss
97+ , workerBootstrapLatency = getStreamLatency st
98+ , workerPollingInterval = period
99+ , workerMeasuredLatency = measured
100+ , workerPendingLatency = wcur
101+ , workerCollectedLatency = wcol
102+ , svarAllTimeLatency = wlong
103+ }
1220104
1221−-- XXX we can use the Ahead style sequence/heap mechanism to make the best
1222−-- effort to always try to finish the streams on the left side of an expression
1223−-- first as long as possible.
105+newSVarStats :: IO SVarStats
106+newSVarStats = do
107+ disp <- newIORef 0
108+ maxWrk <- newIORef 0
109+ maxOq <- newIORef 0
110+ maxHs <- newIORef 0
111+ maxWq <- newIORef 0
112+ avgLat <- newIORef (0, NanoSecond64 0)
113+ maxLat <- newIORef (NanoSecond64 0)
114+ minLat <- newIORef (NanoSecond64 0)
115+ stpTime <- newIORef Nothing
1224116
1225−{-# INLINE enqueueFIFO #-}
1226−enqueueFIFO ::
1227− SVar t m a
1228− -> LinkedQueue (RunInIO m, t m a)
1229− -> (RunInIO m, t m a)
1230− -> IO ()
1231−enqueueFIFO sv q m = do
1232− pushL q m
1233− ringDoorBell sv
117+ return SVarStats
118+ { totalDispatches = disp
119+ , maxWorkers = maxWrk
120+ , maxOutQSize = maxOq
121+ , maxHeapSize = maxHs
122+ , maxWorkQSize = maxWq
123+ , avgWorkerLatency = avgLat
124+ , minWorkerLatency = minLat
125+ , maxWorkerLatency = maxLat
126+ , svarStopTime = stpTime
127+ }
1234128
1235129-------------------------------------------------------------------------------
1236130-- Ahead
@@ -1424,748 +318,6 @@ updateHeapSeq hpVar seqNo =
1424318 atomicModifyIORef_ hpVar $ \(hp, snum) ->
1425319 assert (heapIsSane snum seqNo) (hp, Just seqNo)
1426320
1427−-------------------------------------------------------------------------------
1428−-- WAhead
1429−-------------------------------------------------------------------------------
1430−
1431−-- XXX To be implemented. Use a linked queue like WAsync and put back the
1432−-- remaining computation at the back of the queue instead of the heap, and
1433−-- increment the sequence number.
1434−
1435−-------------------------------------------------------------------------------
1436−-- Dispatching workers and tracking them
1437−-------------------------------------------------------------------------------
1438−
1439−-- Thread tracking is needed for two reasons:
1440−--
1441−-- 1) Killing threads on exceptions. Threads may not be left to go away by
1442−-- themselves because they may run for significant times before going away or
1443−-- worse they may be stuck in IO and never go away.
1444−--
1445−-- 2) To know when all threads are done and the stream has ended.
1446−
1447−{-# NOINLINE addThread #-}
1448−addThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1449−addThread sv tid =
1450− liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
1451−
1452−-- This is cheaper than modifyThread because we do not have to send a
1453−-- outputDoorBell This can make a difference when more workers are being
1454−-- dispatched.
1455−{-# INLINE delThread #-}
1456−delThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1457−delThread sv tid =
1458− liftIO $ modifyIORef (workerThreads sv) (S.delete tid)
1459−
1460−-- If present then delete else add. This takes care of out of order add and
1461−-- delete i.e. a delete arriving before we even added a thread.
1462−-- This occurs when the forked thread is done even before the 'addThread' right
1463−-- after the fork gets a chance to run.
1464−{-# INLINE modifyThread #-}
1465−modifyThread :: MonadIO m => SVar t m a -> ThreadId -> m ()
1466−modifyThread sv tid = do
1467− changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
1468− if S.member tid old
1469− then let new = S.delete tid old in (new, new)
1470− else let new = S.insert tid old in (new, old)
1471− when (null changed) $
1472− liftIO $ do
1473− writeBarrier
1474− void $ tryPutMVar (outputDoorBell sv) ()
1475−
1476−-- | This is safe even if we are adding more threads concurrently because if
1477−-- a child thread is adding another thread then anyway 'workerThreads' will
1478−-- not be empty.
1479−{-# INLINE allThreadsDone #-}
1480−allThreadsDone :: MonadIO m => SVar t m a -> m Bool
1481−allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
1482−
1483−{-# NOINLINE handleChildException #-}
1484−handleChildException :: SVar t m a -> SomeException -> IO ()
1485−handleChildException sv e = do
1486− tid <- myThreadId
1487− void $ send sv (ChildStop tid (Just e))
1488−
1489−{-# NOINLINE handleFoldException #-}
1490−handleFoldException :: SVar t m a -> SomeException -> IO ()
1491−handleFoldException sv e = do
1492− tid <- myThreadId
1493− void $ sendToProducer sv (ChildStop tid (Just e))
1494−
1495−{-# NOINLINE recordMaxWorkers #-}
1496−recordMaxWorkers :: MonadIO m => SVar t m a -> m ()
1497−recordMaxWorkers sv = liftIO $ do
1498− active <- readIORef (workerCount sv)
1499− maxWrk <- readIORef (maxWorkers $ svarStats sv)
1500− when (active > maxWrk) $ writeIORef (maxWorkers $ svarStats sv) active
1501− modifyIORef (totalDispatches $ svarStats sv) (+1)
1502−
1503−{-# NOINLINE pushWorker #-}
1504−pushWorker :: MonadAsync m => Count -> SVar t m a -> m ()
1505−pushWorker yieldMax sv = do
1506− liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1507− when (svarInspectMode sv) $ recordMaxWorkers sv
1508− -- This allocation matters when significant number of workers are being
1509− -- sent. We allocate it only when needed.
1510− winfo <-
1511− case yieldRateInfo sv of
1512− Nothing -> return Nothing
1513− Just _ -> liftIO $ do
1514− cntRef <- newIORef 0
1515− t <- getTime Monotonic
1516− lat <- newIORef (0, t)
1517− return $ Just WorkerInfo
1518− { workerYieldMax = yieldMax
1519− , workerYieldCount = cntRef
1520− , workerLatencyStart = lat
1521− }
1522− doFork (workLoop sv winfo) (svarMrun sv) (handleChildException sv)
1523− >>= addThread sv
1524−
1525−-- XXX we can push the workerCount modification in accountThread and use the
1526−-- same pushWorker for Parallel case as well.
1527−--
1528−-- | In contrast to pushWorker which always happens only from the consumer
1529−-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
1530−-- producer side. So we need to use a thread safe modification of
1531−-- workerThreads. Alternatively, we can use a CreateThread event to avoid
1532−-- using a CAS based modification.
1533−{-# INLINE pushWorkerPar #-}
1534−pushWorkerPar
1535− :: MonadAsync m
1536− => SVar t m a -> (Maybe WorkerInfo -> m ()) -> m ()
1537−pushWorkerPar sv wloop =
1538− if svarInspectMode sv
1539− then forkWithDiag
1540− else doFork (wloop Nothing) (svarMrun sv) (handleChildException sv)
1541− >>= modifyThread sv
1542−
1543− where
1544−
1545− {-# NOINLINE forkWithDiag #-}
1546− forkWithDiag = do
1547− -- We do not use workerCount in case of ParallelVar but still there is
1548− -- no harm in maintaining it correctly.
1549− liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
1550− recordMaxWorkers sv
1551− -- This allocation matters when significant number of workers are being
1552− -- sent. We allocate it only when needed. The overhead increases by 4x.
1553− winfo <-
1554− case yieldRateInfo sv of
1555− Nothing -> return Nothing
1556− Just _ -> liftIO $ do
1557− cntRef <- newIORef 0
1558− t <- getTime Monotonic
1559− lat <- newIORef (0, t)
1560− return $ Just WorkerInfo
1561− { workerYieldMax = 0
1562− , workerYieldCount = cntRef
1563− , workerLatencyStart = lat
1564− }
1565−
1566− doFork (wloop winfo) (svarMrun sv) (handleChildException sv)
1567− >>= modifyThread sv
1568−
1569−-- Returns:
1570−-- True: can dispatch more
1571−-- False: cannot dispatch any more
1572−dispatchWorker :: MonadAsync m => Count -> SVar t m a -> m Bool
1573−dispatchWorker yieldCount sv = do
1574− let workerLimit = maxWorkerLimit sv
1575− -- XXX in case of Ahead streams we should not send more than one worker
1576− -- when the work queue is done but heap is not done.
1577− done <- liftIO $ isWorkDone sv
1578− -- Note, "done" may not mean that the work is actually finished if there
1579− -- are workers active, because there may be a worker which has not yet
1580− -- queued the leftover work.
1581− if not done
1582− then do
1583− qDone <- liftIO $ isQueueDone sv
1584− -- This count may not be accurate as it is decremented by the workers
1585− -- and we have no synchronization with that decrement.
1586− active <- liftIO $ readIORef $ workerCount sv
1587− if not qDone
1588− then do
1589− -- Note that we may deadlock if the previous workers (tasks in the
1590− -- stream) wait/depend on the future workers (tasks in the stream)
1591− -- executing. In that case we should either configure the maxWorker
1592− -- count to higher or use parallel style instead of ahead or async
1593− -- style.
1594− limit <- case remainingWork sv of
1595− Nothing -> return workerLimit
1596− Just ref -> do
1597− n <- liftIO $ readIORef ref
1598− case yieldRateInfo sv of
1599− Just _ -> return workerLimit
1600− Nothing ->
1601− return $
1602− case workerLimit of
1603− Unlimited -> Limited (fromIntegral n)
1604− Limited lim -> Limited $ min lim (fromIntegral n)
1605−
1606− -- XXX for ahead streams shall we take the heap yields into account
1607− -- for controlling the dispatch? We should not dispatch if the heap
1608− -- has already got the limit covered.
1609− let dispatch = pushWorker yieldCount sv >> return True
1610− in case limit of
1611− Unlimited -> dispatch
1612− -- Note that the use of remainingWork and workerCount is not
1613− -- atomic and the counts may even have changed between reading
1614− -- and using them here, so this is just approximate logic and
1615− -- we cannot rely on it for correctness. We may actually
1616− -- dispatch more workers than required.
1617− Limited lim | lim > fromIntegral active -> dispatch
1618− _ -> return False
1619− else do
1620− when (active <= 0) $ pushWorker 0 sv
1621− return False
1622− else return False
1623−
1624−-------------------------------------------------------------------------------
1625−-- Dispatch workers with rate control
1626−-------------------------------------------------------------------------------
1627−
1628−-- | This is a magic number and it is overloaded, and used at several places to
1629−-- achieve batching:
1630−--
1631−-- 1. If we have to sleep to slowdown this is the minimum period that we
1632−-- accumulate before we sleep. Also, workers do not stop until this much
1633−-- sleep time is accumulated.
1634−-- 3. Collected latencies are computed and transferred to measured latency
1635−-- after a minimum of this period.
1636−minThreadDelay :: NanoSecond64
1637−minThreadDelay = 1000000
1638−
1639−-- | Another magic number! When we have to start more workers to cover up a
1640−-- number of yields that we are lagging by then we cannot start one worker for
1641−-- each yield because that may be a very big number and if the latency of the
1642−-- workers is low these number of yields could be very high. We assume that we
1643−-- run each extra worker for at least this much time.
1644−rateRecoveryTime :: NanoSecond64
1645−rateRecoveryTime = 1000000
1646−
1647−-- We either block, or send one worker with limited yield count or one or more
1648−-- workers with unlimited yield count.
1649−data Work
1650− = BlockWait NanoSecond64
1651− | PartialWorker Count
1652− | ManyWorkers Int Count
1653− deriving Show
1654−
1655−-- XXX we can use phantom types to distinguish the duration/latency/expectedLat
1656−estimateWorkers
1657− :: Limit
1658− -> Count
1659− -> Count
1660− -> NanoSecond64
1661− -> NanoSecond64
1662− -> NanoSecond64
1663− -> LatencyRange
1664− -> Work
1665−estimateWorkers workerLimit svarYields gainLossYields
1666− svarElapsed wLatency targetLat range =
1667− -- XXX we can have a maxEfficiency combinator as well which runs the
1668− -- producer at the maximal efficiency i.e. the number of workers are chosen
1669− -- such that the latency is minimum or within a range. Or we can call it
1670− -- maxWorkerLatency.
1671− --
1672− let
1673− -- How many workers do we need to achieve the required rate?
1674− --
1675− -- When the workers are IO bound we can increase the throughput by
1676− -- increasing the number of workers as long as the IO device has enough
1677− -- capacity to process all the requests concurrently. If the IO
1678− -- bandwidth is saturated increasing the workers won't help. Also, if
1679− -- the CPU utilization in processing all these requests exceeds the CPU
1680− -- bandwidth, then increasing the number of workers won't help.
1681− --
1682− -- When the workers are purely CPU bound, increasing the workers beyond
1683− -- the number of CPUs won't help.
1684− --
1685− -- TODO - measure the CPU and IO requirements of the workers. Have a
1686− -- way to specify the max bandwidth of the underlying IO mechanism and
1687− -- use that to determine the max rate of workers, and also take the CPU
1688− -- bandwidth into account. We can also discover the IO bandwidth if we
1689− -- know that we are not CPU bound, then how much steady state rate are
1690− -- we able to achieve. Design tests for CPU bound and IO bound cases.
1691−
1692− -- Calculate how many yields are we ahead or behind to match the exact
1693− -- required rate. Based on that we increase or decrease the effective
1694− -- workers.
1695− --
1696− -- When the worker latency is lower than required latency we begin with
1697− -- a yield and then wait rather than first waiting and then yielding.
1698− targetYields = (svarElapsed + wLatency + targetLat - 1) `div` targetLat
1699− effectiveYields = svarYields + gainLossYields
1700− deltaYields = fromIntegral targetYields - effectiveYields
1701−
1702− -- We recover the deficit by running at a higher/lower rate for a
1703− -- certain amount of time. To keep the effective rate in reasonable
1704− -- limits we use rateRecoveryTime, minLatency and maxLatency.
1705− in if deltaYields > 0
1706− then
1707− let deltaYieldsFreq :: Double
1708− deltaYieldsFreq =
1709− fromIntegral deltaYields /
1710− fromIntegral rateRecoveryTime
1711− yieldsFreq = 1.0 / fromIntegral targetLat
1712− totalYieldsFreq = yieldsFreq + deltaYieldsFreq
1713− requiredLat = NanoSecond64 $ round $ 1.0 / totalYieldsFreq
1714− adjustedLat = min (max requiredLat (minLatency range))
1715− (maxLatency range)
1716− in assert (adjustedLat > 0) $
1717− if wLatency <= adjustedLat
1718− then PartialWorker deltaYields
1719− else let workers = withLimit $ wLatency `div` adjustedLat
1720− limited = min workers (fromIntegral deltaYields)
1721− in ManyWorkers (fromIntegral limited) deltaYields
1722− else
1723− let expectedDuration = fromIntegral effectiveYields * targetLat
1724− sleepTime = expectedDuration - svarElapsed
1725− maxSleepTime = maxLatency range - wLatency
1726− s = min sleepTime maxSleepTime
1727− in assert (sleepTime >= 0) $
1728− -- if s is less than 0 it means our maxSleepTime is less
1729− -- than the worker latency.
1730− if s > 0 then BlockWait s else ManyWorkers 1 (Count 0)
1731− where
1732− withLimit n =
1733− case workerLimit of
1734− Unlimited -> n
1735− Limited x -> min n (fromIntegral x)
1736−
1737−-- | Get the worker latency without resetting workerPendingLatency
1738−-- Returns (total yield count, base time, measured latency)
1739−-- CAUTION! keep it in sync with collectLatency
1740−getWorkerLatency :: YieldRateInfo -> IO (Count, AbsTime, NanoSecond64)
1741−getWorkerLatency yinfo = do
1742− let cur = workerPendingLatency yinfo
1743− col = workerCollectedLatency yinfo
1744− longTerm = svarAllTimeLatency yinfo
1745− measured = workerMeasuredLatency yinfo
1746−
1747− (curTotalCount, curCount, curTime) <- readIORef cur
1748− (colTotalCount, colCount, colTime) <- readIORef col
1749− (lcount, ltime) <- readIORef longTerm
1750− prevLat <- readIORef measured
1751−
1752− let latCount = colCount + curCount
1753− latTime = colTime + curTime
1754− totalCount = colTotalCount + curTotalCount
1755− newLat =
1756− if latCount > 0 && latTime > 0
1757− then let lat = latTime `div` fromIntegral latCount
1758− -- XXX Give more weight to new?
1759− in (lat + prevLat) `div` 2
1760− else prevLat
1761− return (lcount + totalCount, ltime, newLat)
1762−
1763−isBeyondMaxRate :: SVar t m a -> YieldRateInfo -> IO Bool
1764−isBeyondMaxRate sv yinfo = do
1765− (count, tstamp, wLatency) <- getWorkerLatency yinfo
1766− now <- getTime Monotonic
1767− let duration = fromRelTime64 $ diffAbsTime64 now tstamp
1768− let targetLat = svarLatencyTarget yinfo
1769− gainLoss <- readIORef (svarGainedLostYields yinfo)
1770− let work = estimateWorkers (maxWorkerLimit sv) count gainLoss duration
1771− wLatency targetLat (svarLatencyRange yinfo)
1772− cnt <- readIORef $ workerCount sv
1773− return $ case work of
1774− -- XXX set the worker's maxYields or polling interval based on yields
1775− PartialWorker _yields -> cnt > 1
1776− ManyWorkers n _ -> cnt > n
1777− BlockWait _ -> True
1778−
1779−-- XXX in case of ahead style stream we need to take the heap size into account
1780−-- because we return the workers on the basis of that which causes a condition
1781−-- where we keep dispatching and they keep returning. So we must have exactly
1782−-- the same logic for not dispatching and for returning.
1783−--
1784−-- Returns:
1785−-- True: can dispatch more
1786−-- False: full, no more dispatches
1787−dispatchWorkerPaced :: MonadAsync m => SVar t m a -> m Bool
1788−dispatchWorkerPaced sv = do
1789− let yinfo = fromJust $ yieldRateInfo sv
1790− (svarYields, svarElapsed, wLatency) <- do
1791− now <- liftIO $ getTime Monotonic
1792− (yieldCount, baseTime, lat) <-
1793− liftIO $ collectLatency sv yinfo False
1794− let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
1795− let latency =
1796− if lat == 0
1797− then fromMaybe lat (workerBootstrapLatency yinfo)
1798− else lat
1799−
1800− return (yieldCount, elapsed, latency)
1801−
1802− if wLatency == 0
1803− -- Need to measure the latency with a single worker before we can perform
1804− -- any computation.
1805− then return False
1806− else do
1807− let workerLimit = maxWorkerLimit sv
1808− let targetLat = svarLatencyTarget yinfo
1809− let range = svarLatencyRange yinfo
1810− gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
1811− let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
1812− wLatency targetLat range
1813−
1814− -- XXX we need to take yieldLimit into account here. If we are at the
1815− -- end of the limit as well as the time, we should not be sleeping.
1816− -- If we are not actually planning to dispatch any more workers we need
1817− -- to take that in account.
1818− case work of
1819− BlockWait s -> do
1820− assert (s >= 0) (return ())
1821− -- XXX note that when we return from here we will block waiting
1822− -- for the result from the existing worker. If that takes too
1823− -- long we won't be able to send another worker until the
1824− -- result arrives.
1825− --
1826− -- Sleep only if there are no active workers, otherwise we will
1827− -- defer the output of those. Note we cannot use workerCount
1828− -- here as it is not a reliable way to ensure there are
1829− -- definitely no active workers. When workerCount is 0 we may
1830− -- still have a Stop event waiting in the outputQueue.
1831− done <- allThreadsDone sv
1832− when done $ void $ do
1833− let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
1834− liftIO $ threadDelay (fromIntegral us)
1835− dispatchWorker 1 sv
1836− return False
1837− PartialWorker yields -> do
1838− assert (yields > 0) (return ())
1839− updateGainedLostYields yinfo yields
1840−
1841− done <- allThreadsDone sv
1842− when done $ void $ dispatchWorker yields sv
1843− return False
1844− ManyWorkers netWorkers yields -> do
1845− assert (netWorkers >= 1) (return ())
1846− assert (yields >= 0) (return ())
1847− updateGainedLostYields yinfo yields
1848−
1849− let periodRef = workerPollingInterval yinfo
1850− ycnt = max 1 $ yields `div` fromIntegral netWorkers
1851− period = min ycnt (fromIntegral magicMaxBuffer)
1852−
1853− old <- liftIO $ readIORef periodRef
1854− when (period < old) $
1855− liftIO $ writeIORef periodRef period
1856−
1857− cnt <- liftIO $ readIORef $ workerCount sv
1858− if cnt < netWorkers
1859− then do
1860− let total = netWorkers - cnt
1861− batch = max 1 $ fromIntegral $
1862− minThreadDelay `div` targetLat
1863− -- XXX stagger the workers over a period?
1864− -- XXX cannot sleep, as that would mean we cannot process
1865− -- the outputs. need to try a different mechanism to
1866− -- stagger.
1867− -- when (total > batch) $
1868− -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
1869− dispatchN (min total batch)
1870− else return False
1871−
1872− where
1873−
1874− updateGainedLostYields yinfo yields = do
1875− let buf = fromIntegral $ svarRateBuffer yinfo
1876− when (yields /= 0 && abs yields > buf) $ do
1877− let delta =
1878− if yields > 0
1879− then yields - buf
1880− else yields + buf
1881− liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
1882−
1883− dispatchN n =
1884− if n == 0
1885− then return True
1886− else do
1887− r <- dispatchWorker 0 sv
1888− if r
1889− then dispatchN (n - 1)
1890− else return False
1891−
1892−-------------------------------------------------------------------------------
1893−-- Worker dispatch and wait loop
1894−-------------------------------------------------------------------------------
1895−
1896−sendWorkerDelayPaced :: SVar t m a -> IO ()
1897−sendWorkerDelayPaced _ = return ()
1898−
1899−sendWorkerDelay :: SVar t m a -> IO ()
1900−sendWorkerDelay _sv =
1901− -- XXX we need a better way to handle this than hardcoded delays. The
1902− -- delays may be different for different systems.
1903− -- If there is a usecase where this is required we can create a combinator
1904− -- to set it as a config in the state.
1905− {-
1906− do
1907− ncpu <- getNumCapabilities
1908− if ncpu <= 1
1909− then
1910− if (svarStyle sv == AheadVar)
1911− then threadDelay 100
1912− else threadDelay 25
1913− else
1914− if (svarStyle sv == AheadVar)
1915− then threadDelay 100
1916− else threadDelay 10
1917− -}
1918− return ()
1919−
1920−{-# NOINLINE sendWorkerWait #-}
1921−sendWorkerWait
1922− :: MonadAsync m
1923− => (SVar t m a -> IO ())
1924− -> (SVar t m a -> m Bool)
1925− -> SVar t m a
1926− -> m ()
1927−sendWorkerWait delay dispatch sv = do
1928− -- Note that we are guaranteed to have at least one outstanding worker when
1929− -- we enter this function. So if we sleep we are guaranteed to be woken up
1930− -- by an outputDoorBell, when the worker exits.
1931−
1932− liftIO $ delay sv
1933− (_, n) <- liftIO $ readIORef (outputQueue sv)
1934− when (n <= 0) $ do
1935− -- The queue may be empty temporarily if the worker has dequeued the
1936− -- work item but has not enqueued the remaining part yet. For the same
1937− -- reason, a worker may come back if it tries to dequeue and finds the
1938− -- queue empty, even though the whole work has not finished yet.
1939−
1940− -- If we find that the queue is empty, but it may be empty
1941− -- temporarily, when we checked it. If that's the case we might
1942− -- sleep indefinitely unless the active workers produce some
1943− -- output. We may deadlock specially if the otuput from the active
1944− -- workers depends on the future workers that we may never send.
1945− -- So in case the queue was temporarily empty set a flag to inform
1946− -- the enqueue to send us a doorbell.
1947−
1948− -- Note that this is just a best effort mechanism to avoid a
1949− -- deadlock. Deadlocks may still happen if for some weird reason
1950− -- the consuming computation shares an MVar or some other resource
1951− -- with the producing computation and gets blocked on that resource
1952− -- and therefore cannot do any pushworker to add more threads to
1953− -- the producer. In such cases the programmer should use a parallel
1954− -- style so that all the producers are scheduled immediately and
1955− -- unconditionally. We can also use a separate monitor thread to
1956− -- push workers instead of pushing them from the consumer, but then
1957− -- we are no longer using pull based concurrency rate adaptation.
1958− --
1959− -- XXX update this in the tutorial.
1960− --
1961− -- Having pending active workers does not mean that we are guaranteed
1962− -- to be woken up if we sleep. In case of Ahead streams, there may be
1963− -- queued items in the heap even though the outputQueue is empty, and
1964− -- we may have active workers which are deadlocked on those items to be
1965− -- processed by the consumer. We should either guarantee that any
1966− -- worker, before returning, clears the heap or we send a worker to
1967− -- clear it. Normally we always send a worker if no output is seen, but
1968− -- if the thread limit is reached or we are using pacing then we may
1969− -- not send a worker. See the concurrentApplication test in the tests,
1970− -- that test case requires at least one yield from the producer to not
1971− -- deadlock, if the last workers output is stuck in the heap then this
1972− -- test fails. This problem can be extended to n threads when the
1973− -- consumer may depend on the evaluation of next n items in the
1974− -- producer stream.
1975−
1976− -- register for the outputDoorBell before we check the queue so that if
1977− -- we sleep because the queue was empty we are guaranteed to get a
1978− -- doorbell on the next enqueue.
1979−
1980− liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
1981− liftIO storeLoadBarrier
1982− canDoMore <- dispatch sv
1983−
1984− -- XXX test for the case when we miss sending a worker when the worker
1985− -- count is more than 1500.
1986− --
1987− -- XXX Assert here that if the heap is not empty then there is at
1988− -- least one outstanding worker. Otherwise we could be sleeping
1989− -- forever.
1990−
1991− if canDoMore
1992− then sendWorkerWait delay dispatch sv
1993− else do
1994− liftIO $ withDiagMVar sv "sendWorkerWait: nothing to do"
1995− $ takeMVar (outputDoorBell sv)
1996− (_, len) <- liftIO $ readIORef (outputQueue sv)
1997− when (len <= 0) $ sendWorkerWait delay dispatch sv
1998−
1999−-------------------------------------------------------------------------------
2000−-- Reading from the workers' output queue/buffer
2001−-------------------------------------------------------------------------------
2002−
2003−{-# INLINE readOutputQBasic #-}
2004−readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)
2005−readOutputQBasic q = atomicModifyIORefCAS q $ \x -> (([],0), x)
2006−
2007−{-# INLINE readOutputQRaw #-}
2008−readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
2009−readOutputQRaw sv = do
2010− (list, len) <- readOutputQBasic (outputQueue sv)
2011− when (svarInspectMode sv) $ do
2012− let ref = maxOutQSize $ svarStats sv
2013− oqLen <- readIORef ref
2014− when (len > oqLen) $ writeIORef ref len
2015− return (list, len)
2016−
2017−readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2018−readOutputQBounded sv = do
2019− (list, len) <- liftIO $ readOutputQRaw sv
2020− -- When there is no output seen we dispatch more workers to help
2021− -- out if there is work pending in the work queue.
2022− if len <= 0
2023− then blockingRead
2024− else do
2025− -- send a worker proactively, if needed, even before we start
2026− -- processing the output. This may degrade single processor
2027− -- perf but improves multi-processor, because of more
2028− -- parallelism
2029− sendOneWorker
2030− return list
2031−
2032− where
2033−
2034− sendOneWorker = do
2035− cnt <- liftIO $ readIORef $ workerCount sv
2036− when (cnt <= 0) $ do
2037− done <- liftIO $ isWorkDone sv
2038− when (not done) (pushWorker 0 sv)
2039−
2040− {-# INLINE blockingRead #-}
2041− blockingRead = do
2042− sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
2043− liftIO (fst `fmap` readOutputQRaw sv)
2044−
2045−readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
2046−readOutputQPaced sv = do
2047− (list, len) <- liftIO $ readOutputQRaw sv
2048− if len <= 0
2049− then blockingRead
2050− else do
2051− -- XXX send a worker proactively, if needed, even before we start
2052− -- processing the output.
2053− void $ dispatchWorkerPaced sv
2054− return list
2055−
2056− where
2057−
2058− {-# INLINE blockingRead #-}
2059− blockingRead = do
2060− sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
2061− liftIO (fst `fmap` readOutputQRaw sv)
2062−
2063−postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
2064−postProcessBounded sv = do
2065− workersDone <- allThreadsDone sv
2066− -- There may still be work pending even if there are no workers pending
2067− -- because all the workers may return if the outputQueue becomes full. In
2068− -- that case send off a worker to kickstart the work again.
2069− --
2070− -- Note that isWorkDone can only be safely checked if all workers are done.
2071− -- When some workers are in progress they may have decremented the yield
2072− -- Limit and later ending up incrementing it again. If we look at the yield
2073− -- limit in that window we may falsely say that it is 0 and therefore we
2074− -- are done.
2075− if workersDone
2076− then do
2077− r <- liftIO $ isWorkDone sv
2078− -- Note that we need to guarantee a worker, therefore we cannot just
2079− -- use dispatchWorker which may or may not send a worker.
2080− when (not r) (pushWorker 0 sv)
2081− -- XXX do we need to dispatch many here?
2082− -- void $ dispatchWorker sv
2083− return r
2084− else return False
2085−
2086−postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
2087−postProcessPaced sv = do
2088− workersDone <- allThreadsDone sv
2089− -- XXX If during consumption we figure out we are getting delayed then we
2090− -- should trigger dispatch there as well. We should try to check on the
2091− -- workers after consuming every n item from the buffer?
2092− if workersDone
2093− then do
2094− r <- liftIO $ isWorkDone sv
2095− when (not r) $ do
2096− void $ dispatchWorkerPaced sv
2097− -- Note that we need to guarantee a worker since the work is not
2098− -- finished, therefore we cannot just rely on dispatchWorkerPaced
2099− -- which may or may not send a worker.
2100− noWorker <- allThreadsDone sv
2101− when noWorker $ pushWorker 0 sv
2102− return r
2103− else return False
2104−
2105−-------------------------------------------------------------------------------
2106−-- Creating an SVar
2107−-------------------------------------------------------------------------------
2108−
2109−getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
2110−getYieldRateInfo st = do
2111− -- convert rate in Hertz to latency in Nanoseconds
2112− let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
2113− case getStreamRate st of
2114− Just (Rate low goal high buf) ->
2115− let l = rateToLatency goal
2116− minl = rateToLatency high
2117− maxl = rateToLatency low
2118− in mkYieldRateInfo l (LatencyRange minl maxl) buf
2119− Nothing -> return Nothing
2120−
2121− where
2122−
2123− mkYieldRateInfo latency latRange buf = do
2124− measured <- newIORef 0
2125− wcur <- newIORef (0,0,0)
2126− wcol <- newIORef (0,0,0)
2127− now <- getTime Monotonic
2128− wlong <- newIORef (0,now)
2129− period <- newIORef 1
2130− gainLoss <- newIORef (Count 0)
2131−
2132− return $ Just YieldRateInfo
2133− { svarLatencyTarget = latency
2134− , svarLatencyRange = latRange
2135− , svarRateBuffer = buf
2136− , svarGainedLostYields = gainLoss
2137− , workerBootstrapLatency = getStreamLatency st
2138− , workerPollingInterval = period
2139− , workerMeasuredLatency = measured
2140− , workerPendingLatency = wcur
2141− , workerCollectedLatency = wcol
2142− , svarAllTimeLatency = wlong
2143− }
2144−
2145−newSVarStats :: IO SVarStats
2146−newSVarStats = do
2147− disp <- newIORef 0
2148− maxWrk <- newIORef 0
2149− maxOq <- newIORef 0
2150− maxHs <- newIORef 0
2151− maxWq <- newIORef 0
2152− avgLat <- newIORef (0, NanoSecond64 0)
2153− maxLat <- newIORef (NanoSecond64 0)
2154− minLat <- newIORef (NanoSecond64 0)
2155− stpTime <- newIORef Nothing
2156−
2157− return SVarStats
2158− { totalDispatches = disp
2159− , maxWorkers = maxWrk
2160− , maxOutQSize = maxOq
2161− , maxHeapSize = maxHs
2162− , maxWorkQSize = maxWq
2163− , avgWorkerLatency = avgLat
2164− , minWorkerLatency = minLat
2165− , maxWorkerLatency = maxLat
2166− , svarStopTime = stpTime
2167− }
2168−
2169321-- XXX remove polymorphism in t, inline f
2170322getAheadSVar :: MonadAsync m
2171323 => State t m a
@@ -2268,6 +420,34 @@ getAheadSVar st f mrun = do
2268420 (xs, _) <- readIORef q
2269421 return $ null xs
2270422
423+{-# INLINABLE newAheadVar #-}
424+newAheadVar :: MonadAsync m
425+ => State t m a
426+ -> t m a
427+ -> ( IORef ([t m a], Int)
428+ -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
429+ -> State t m a
430+ -> SVar t m a
431+ -> Maybe WorkerInfo
432+ -> m ())
433+ -> m (SVar t m a)
434+newAheadVar st m wloop = do
435+ mrun <- captureMonadState
436+ sv <- liftIO $ getAheadSVar st wloop mrun
437+ sendFirstWorker sv m
438+
439+-------------------------------------------------------------------------------
440+-- WAhead
441+-------------------------------------------------------------------------------
442+
443+-- XXX To be implemented. Use a linked queue like WAsync and put back the
444+-- remaining computation at the back of the queue instead of the heap, and
445+-- increment the sequence number.
446+
447+-------------------------------------------------------------------------------
448+-- Parallel
449+-------------------------------------------------------------------------------
450+
2271451getParallelSVar :: MonadIO m
2272452 => SVarStopStyle -> State t m a -> RunInIO m -> IO (SVar t m a)
2273453getParallelSVar ss st mrun = do
@@ -2349,62 +529,9 @@ getParallelSVar ss st mrun = do
2349529 void $ tryPutMVar (pushBufferMVar sv) ()
2350530 return r
2351531
2352−sendFirstWorker :: MonadAsync m => SVar t m a -> t m a -> m (SVar t m a)
2353−sendFirstWorker sv m = do
2354− -- Note: We must have all the work on the queue before sending the
2355− -- pushworker, otherwise the pushworker may exit before we even get a
2356− -- chance to push.
2357− runIn <- captureMonadState
2358− liftIO $ enqueue sv (runIn, m)
2359− case yieldRateInfo sv of
2360− Nothing -> pushWorker 0 sv
2361− Just yinfo ->
2362− if svarLatencyTarget yinfo == maxBound
2363− then liftIO $ threadDelay maxBound
2364− else pushWorker 1 sv
2365− return sv
2366−
2367−{-# INLINABLE newAheadVar #-}
2368−newAheadVar :: MonadAsync m
2369− => State t m a
2370− -> t m a
2371− -> ( IORef ([t m a], Int)
2372− -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
2373− -> State t m a
2374− -> SVar t m a
2375− -> Maybe WorkerInfo
2376− -> m ())
2377− -> m (SVar t m a)
2378−newAheadVar st m wloop = do
2379− mrun <- captureMonadState
2380− sv <- liftIO $ getAheadSVar st wloop mrun
2381− sendFirstWorker sv m
2382−
2383532{-# INLINABLE newParallelVar #-}
2384533newParallelVar :: MonadAsync m
2385534 => SVarStopStyle -> State t m a -> m (SVar t m a)
2386535newParallelVar ss st = do
2387536 mrun <- captureMonadState
2388537 liftIO $ getParallelSVar ss st mrun
2389−
2390−-------------------------------------------------------------------------------
2391−-- Write a stream to an SVar
2392−-------------------------------------------------------------------------------
2393−
2394−-- XXX this errors out for Parallel/Ahead SVars
2395−-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
2396−-- be read back from the SVar using 'fromSVar'.
2397−toStreamVar :: MonadAsync m => SVar t m a -> t m a -> m ()
2398−toStreamVar sv m = do
2399− runIn <- captureMonadState
2400− liftIO $ enqueue sv (runIn, m)
2401− done <- allThreadsDone sv
2402− -- XXX This is safe only when called from the consumer thread or when no
2403− -- consumer is present. There may be a race if we are not running in the
2404− -- consumer thread.
2405− -- XXX do this only if the work queue is not empty. The work may have been
2406− -- carried out by existing workers.
2407− when done $
2408− case yieldRateInfo sv of
2409− Nothing -> pushWorker 0 sv
2410− Just _ -> pushWorker 1 sv