@@ -0,0 +1,710 @@
1+-- |
2+-- Module : Streamly.Internal.Data.Stream.Channel.Types
3+-- Copyright : (c) 2017 Composewell Technologies
4+-- License : BSD-3-Clause
5+-- Maintainer : streamly@composewell.com
6+-- Stability : experimental
7+-- Portability : GHC
8+--
9+-- A Channel is a place where streams join and new streams start. This module
10+-- defines low level data structures and functions to build channels. For
11+-- concrete Channels see the Channel modules of specific stream types.
12+--
13+-- A Channel is a conduit to the output from multiple streams running
14+-- concurrently and asynchronously. A channel can be thought of as an
15+-- asynchronous IO handle. We can write any number of streams to a channel in a
16+-- non-blocking manner and then read them back at any time at any pace. The
17+-- channel would run the streams asynchronously and accumulate results. A
18+-- channel may not really execute the stream completely and accumulate all the
19+-- results. However, it ensures that the reader can read the results at
20+-- whatever pace it wants to read. The channel monitors and adapts to the
21+-- consumer's pace.
22+--
23+-- A channel is a mini scheduler, it has an associated workLoop that holds the
24+-- stream tasks to be picked and run by a pool of worker threads. It has an
25+-- associated output queue where the output stream elements are placed by the
26+-- worker threads. An outputDoorBell is used by the worker threads to intimate the
27+-- consumer thread about availability of new results in the output queue. More
28+-- workers are added to the channel by 'fromChannel' on demand if the output
29+-- produced is not keeping pace with the consumer. On bounded channels, workers
30+-- block on the output queue to provide throttling of the producer when the
31+-- consumer is not pulling fast enough. The number of workers may even get
32+-- reduced depending on the consuming pace.
33+--
34+module Streamly.Internal.Data.Stream.Channel.Types
35+ (
36+ -- * Types
37+ Count (..)
38+ , Limit (..)
39+ , ThreadAbort (..)
40+ , ChildEvent (..)
41+
42+ -- * Stats
43+ , SVarStats (..)
44+ , newSVarStats
45+
46+ -- * Rate Control
47+ , WorkerInfo (..)
48+ , LatencyRange (..)
49+ , YieldRateInfo (..)
50+ , newRateInfo
51+
52+ -- * Output queue
53+ , readOutputQRaw
54+ , readOutputQBasic
55+ , ringDoorBell
56+
57+ -- * Yield Limit
58+ , decrementYieldLimit
59+ , incrementYieldLimit
60+
61+ -- * Configuration
62+ , Rate (..)
63+ , Config
64+
65+ -- ** Default config
66+ , magicMaxBuffer
67+ , defaultConfig
68+
69+ -- ** Set config
70+ , maxThreads
71+ , maxBuffer
72+ , maxYields
73+ , inspectMode
74+
75+ , rate
76+ , avgRate
77+ , minRate
78+ , maxRate
79+ , constRate
80+
81+ -- ** Get config
82+ , getMaxThreads
83+ , getMaxBuffer
84+ , getStreamRate
85+ , getStreamLatency
86+ , setStreamLatency
87+ , getYieldLimit
88+ , getInspectMode
89+
90+ -- * Cleanup
91+ , cleanupSVar
92+
93+ -- * Diagnostics
94+ , dumpCreator
95+ , dumpOutputQ
96+ , dumpDoorBell
97+ , dumpNeedDoorBell
98+ , dumpRunningThreads
99+ , dumpWorkerCount
100+
101+ , withDiagMVar
102+ , printSVar
103+ )
104+where
105+
106+import Control.Concurrent (ThreadId, myThreadId, throwTo, MVar, tryReadMVar)
107+import Control.Concurrent.MVar (tryPutMVar)
108+import Control.Exception
109+ ( SomeException(..), Exception, catches, throwIO, Handler(..)
110+ , BlockedIndefinitelyOnMVar(..), BlockedIndefinitelyOnSTM(..))
111+import Control.Monad (void, when)
112+import Data.Int (Int64)
113+import Data.IORef (IORef, newIORef, readIORef, writeIORef)
114+import Data.Set (Set)
115+import Streamly.Internal.Data.Atomics
116+ (atomicModifyIORefCAS, atomicModifyIORefCAS_, storeLoadBarrier)
117+import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
118+import Streamly.Internal.Data.Time.Units (AbsTime, NanoSecond64(..))
119+import System.IO (hPutStrLn, stderr)
120+
121+import qualified Data.Set as Set
122+
123+------------------------------------------------------------------------------
124+-- Common types
125+------------------------------------------------------------------------------
126+
127+newtype Count = Count Int64
128+ deriving ( Eq
129+ , Read
130+ , Show
131+ , Enum
132+ , Bounded
133+ , Num
134+ , Real
135+ , Integral
136+ , Ord
137+ )
138+
139+-- XXX We can use maxBound for unlimited?
140+
141+-- This is essentially a 'Maybe Word' type
142+data Limit = Unlimited | Limited Word deriving Show
143+
144+instance Eq Limit where
145+ Unlimited == Unlimited = True
146+ Unlimited == Limited _ = False
147+ Limited _ == Unlimited = False
148+ Limited x == Limited y = x == y
149+
150+instance Ord Limit where
151+ Unlimited <= Unlimited = True
152+ Unlimited <= Limited _ = False
153+ Limited _ <= Unlimited = True
154+ Limited x <= Limited y = x <= y
155+
156+------------------------------------------------------------------------------
157+-- Parent child thread communication type
158+------------------------------------------------------------------------------
159+
160+data ThreadAbort = ThreadAbort deriving Show
161+
162+instance Exception ThreadAbort
163+
164+-- | Events that a child thread may send to a parent thread.
165+data ChildEvent a =
166+ ChildYield a
167+ | ChildStop ThreadId (Maybe SomeException)
168+
169+-- | We measure the individual worker latencies to estimate the number of workers
170+-- needed or the amount of time we have to sleep between dispatches to achieve
171+-- a particular rate when controlled pace mode it used.
172+data WorkerInfo = WorkerInfo
173+ {
174+ -- | 0 means unlimited
175+ workerYieldMax :: Count
176+ -- | total number of yields by the worker till now
177+ , workerYieldCount :: IORef Count
178+ -- | yieldCount at start, timestamp
179+ , workerLatencyStart :: IORef (Count, AbsTime)
180+ }
181+
182+data LatencyRange = LatencyRange
183+ { minLatency :: NanoSecond64
184+ , maxLatency :: NanoSecond64
185+ } deriving Show
186+
187+-- | Rate control.
188+data YieldRateInfo = YieldRateInfo
189+ { svarLatencyTarget :: NanoSecond64
190+ , svarLatencyRange :: LatencyRange
191+ , svarRateBuffer :: Int
192+
193+ -- | [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
194+ -- read by the worker threads
195+ , svarGainedLostYields :: IORef Count
196+
197+ -- | Actual latency/througput as seen from the consumer side, we count the
198+ -- yields and the time it took to generates those yields. This is used to
199+ -- increase or decrease the number of workers needed to achieve the desired
200+ -- rate. The idle time of workers is adjusted in this, so that we only
201+ -- account for the rate when the consumer actually demands data.
202+ -- XXX interval latency is enough, we can move this under diagnostics build
203+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
204+ -- read by the worker threads
205+ , svarAllTimeLatency :: IORef (Count, AbsTime)
206+
207+ -- XXX Worker latency specified by the user to be used before the first
208+ -- actual measurement arrives. Not yet implemented
209+ , workerBootstrapLatency :: Maybe NanoSecond64
210+
211+ -- | After how many yields the worker should update the latency information.
212+ -- If the latency is high, this count is kept lower and vice-versa. XXX If
213+ -- the latency suddenly becomes too high this count may remain too high for
214+ -- long time, in such cases the consumer can change it.
215+ -- 0 means no latency computation
216+ -- XXX this is derivable from workerMeasuredLatency, can be removed.
217+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
218+ -- read by the worker threads
219+ , workerPollingInterval :: IORef Count
220+
221+ -- | This is in progress latency stats maintained by the workers which we
222+ -- empty into workerCollectedLatency stats at certain intervals - whenever
223+ -- we process the stream elements yielded in this period. The first count
224+ -- is all yields, the second count is only those yields for which the
225+ -- latency was measured to be non-zero (note that if the timer resolution
226+ -- is low the measured latency may be zero e.g. on JS platform).
227+ -- [LOCKING] Locked access. Modified by the consumer thread as well as
228+ -- worker threads. Workers modify it periodically based on
229+ -- workerPollingInterval and not on every yield to reduce the locking
230+ -- overhead.
231+ -- (allYieldCount, yieldCount, timeTaken)
232+ , workerPendingLatency :: IORef (Count, Count, NanoSecond64)
233+
234+ -- | This is the second level stat which is an accmulation from
235+ -- workerPendingLatency stats. We keep accumulating latencies in this
236+ -- bucket until we have stats for a sufficient period and then we reset it
237+ -- to start collecting for the next period and retain the computed average
238+ -- latency for the last period in workerMeasuredLatency.
239+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
240+ -- read by the worker threads
241+ -- (allYieldCount, yieldCount, timeTaken)
242+ , workerCollectedLatency :: IORef (Count, Count, NanoSecond64)
243+
244+ -- | Latency as measured by workers, aggregated for the last period.
245+ -- [LOCKING] Unlocked access. Modified by the consumer thread and unsafely
246+ -- read by the worker threads
247+ , workerMeasuredLatency :: IORef NanoSecond64
248+ }
249+
250+data SVarStats = SVarStats {
251+ totalDispatches :: IORef Int
252+ , maxWorkers :: IORef Int
253+ , maxOutQSize :: IORef Int
254+ , maxHeapSize :: IORef Int
255+ , maxWorkQSize :: IORef Int
256+ , avgWorkerLatency :: IORef (Count, NanoSecond64)
257+ , minWorkerLatency :: IORef NanoSecond64
258+ , maxWorkerLatency :: IORef NanoSecond64
259+ , svarStopTime :: IORef (Maybe AbsTime)
260+}
261+
262+-------------------------------------------------------------------------------
263+-- Config
264+-------------------------------------------------------------------------------
265+
266+-- | Specifies the stream yield rate in yields per second (@Hertz@).
267+-- We keep accumulating yield credits at 'rateGoal'. At any point of time we
268+-- allow only as many yields as we have accumulated as per 'rateGoal' since the
269+-- start of time. If the consumer or the producer is slower or faster, the
270+-- actual rate may fall behind or exceed 'rateGoal'. We try to recover the gap
271+-- between the two by increasing or decreasing the pull rate from the producer.
272+-- However, if the gap becomes more than 'rateBuffer' we try to recover only as
273+-- much as 'rateBuffer'.
274+--
275+-- 'rateLow' puts a bound on how low the instantaneous rate can go when
276+-- recovering the rate gap. In other words, it determines the maximum yield
277+-- latency. Similarly, 'rateHigh' puts a bound on how high the instantaneous
278+-- rate can go when recovering the rate gap. In other words, it determines the
279+-- minimum yield latency. We reduce the latency by increasing concurrency,
280+-- therefore we can say that it puts an upper bound on concurrency.
281+--
282+-- If the 'rateGoal' is 0 or negative the stream never yields a value.
283+-- If the 'rateBuffer' is 0 or negative we do not attempt to recover.
284+--
285+-- /Since: 0.5.0 ("Streamly")/
286+--
287+-- @since 0.8.0
288+data Rate = Rate
289+ { rateLow :: Double -- ^ The lower rate limit
290+ , rateGoal :: Double -- ^ The target rate we want to achieve
291+ , rateHigh :: Double -- ^ The upper rate limit
292+ , rateBuffer :: Int -- ^ Maximum slack from the goal
293+ }
294+
295+-- XXX we can put the resettable fields in a oneShotConfig field and others in
296+-- a persistentConfig field. That way reset would be fast and scalable
297+-- irrespective of the number of fields.
298+--
299+-- XXX make all these Limited types and use phantom types to distinguish them
300+
301+-- | An abstract type for specifying the configuration parameters of a
302+-- 'Channel'. Use @Config -> Config@ modifier functions to modify the
303+-- default configuration. See the modifiers for default values.
304+--
305+data Config = Config
306+ { -- one shot configuration, automatically reset for each API call
307+ -- streamVar :: Maybe (SVar t m a)
308+ _yieldLimit :: Maybe Count
309+
310+ -- persistent configuration, state that remains valid until changed by
311+ -- an explicit setting via a combinator.
312+ , _threadsHigh :: Limit
313+ , _bufferHigh :: Limit
314+
315+ -- XXX these two can be collapsed into a single type
316+ , _streamLatency :: Maybe NanoSecond64 -- bootstrap latency
317+ , _maxStreamRate :: Maybe Rate
318+ , _inspectMode :: Bool
319+ }
320+
321+-------------------------------------------------------------------------------
322+-- State defaults and reset
323+-------------------------------------------------------------------------------
324+
325+-- | A magical value for the buffer size arrived at by running the smallest
326+-- possible task and measuring the optimal value of the buffer for that. This
327+-- is obviously dependent on hardware, this figure is based on a 2.2GHz intel
328+-- core-i7 processor.
329+magicMaxBuffer :: Word
330+magicMaxBuffer = 1500
331+
332+defaultMaxThreads, defaultMaxBuffer :: Limit
333+defaultMaxThreads = Limited magicMaxBuffer
334+defaultMaxBuffer = Limited magicMaxBuffer
335+
336+-- | The fields prefixed by an _ are not to be accessed or updated directly but
337+-- via smart accessor APIs. Use get/set routines instead of directly accessing
338+-- the Config fields
339+defaultConfig :: Config
340+defaultConfig = Config
341+ { -- streamVar = Nothing
342+ _yieldLimit = Nothing
343+ , _threadsHigh = defaultMaxThreads
344+ , _bufferHigh = defaultMaxBuffer
345+ , _maxStreamRate = Nothing
346+ , _streamLatency = Nothing
347+ , _inspectMode = False
348+ }
349+
350+-------------------------------------------------------------------------------
351+-- Smart get/set routines for State
352+-------------------------------------------------------------------------------
353+
354+maxYields :: Maybe Int64 -> Config -> Config
355+maxYields lim st =
356+ st { _yieldLimit =
357+ case lim of
358+ Nothing -> Nothing
359+ Just n ->
360+ if n <= 0
361+ then Just 0
362+ else Just (fromIntegral n)
363+ }
364+
365+getYieldLimit :: Config -> Maybe Count
366+getYieldLimit = _yieldLimit
367+
368+-- | Specify the maximum number of threads that can be spawned by the channel.
369+-- A value of 0 resets the thread limit to default, a negative value means
370+-- there is no limit. The default value is 1500.
371+--
372+-- When the actions in a stream are IO bound, having blocking IO calls, this
373+-- option can be used to control the maximum number of in-flight IO requests.
374+-- When the actions are CPU bound this option can be used to control the amount
375+-- of CPU used by the stream.
376+--
377+maxThreads :: Int -> Config -> Config
378+maxThreads n st =
379+ st { _threadsHigh =
380+ if n < 0
381+ then Unlimited
382+ else if n == 0
383+ then defaultMaxThreads
384+ else Limited (fromIntegral n)
385+ }
386+
387+getMaxThreads :: Config -> Limit
388+getMaxThreads = _threadsHigh
389+
390+-- | Specify the maximum size of the buffer for storing the results from
391+-- concurrent computations. If the buffer becomes full we stop spawning more
392+-- concurrent tasks until there is space in the buffer.
393+-- A value of 0 resets the buffer size to default, a negative value means
394+-- there is no limit. The default value is 1500.
395+--
396+-- CAUTION! using an unbounded 'maxBuffer' value (i.e. a negative value)
397+-- coupled with an unbounded 'maxThreads' value is a recipe for disaster in
398+-- presence of infinite streams, or very large streams. Especially, it must
399+-- not be used when 'pure' is used in 'ZipAsyncM' streams as 'pure' in
400+-- applicative zip streams generates an infinite stream causing unbounded
401+-- concurrent generation with no limit on the buffer or threads.
402+--
403+maxBuffer :: Int -> Config -> Config
404+maxBuffer n st =
405+ st { _bufferHigh =
406+ if n < 0
407+ then Unlimited
408+ else if n == 0
409+ then defaultMaxBuffer
410+ else Limited (fromIntegral n)
411+ }
412+
413+getMaxBuffer :: Config -> Limit
414+getMaxBuffer = _bufferHigh
415+
416+-- | Specify the pull rate of a channel.
417+-- A 'Nothing' value resets the rate to default which is unlimited. When the
418+-- rate is specified, concurrent production may be ramped up or down
419+-- automatically to achieve the specified yield rate. The specific behavior for
420+-- different styles of 'Rate' specifications is documented under 'Rate'. The
421+-- effective maximum production rate achieved by a channel is governed by:
422+--
423+-- * The 'maxThreads' limit
424+-- * The 'maxBuffer' limit
425+-- * The maximum rate that the stream producer can achieve
426+-- * The maximum rate that the stream consumer can achieve
427+--
428+rate :: Maybe Rate -> Config -> Config
429+rate r st = st { _maxStreamRate = r }
430+
431+getStreamRate :: Config -> Maybe Rate
432+getStreamRate = _maxStreamRate
433+
434+setStreamLatency :: Int -> Config -> Config
435+setStreamLatency n st =
436+ st { _streamLatency =
437+ if n <= 0
438+ then Nothing
439+ else Just (fromIntegral n)
440+ }
441+
442+getStreamLatency :: Config -> Maybe NanoSecond64
443+getStreamLatency = _streamLatency
444+
445+inspectMode :: Config -> Config
446+inspectMode st = st { _inspectMode = True }
447+
448+getInspectMode :: Config -> Bool
449+getInspectMode = _inspectMode
450+
451+-------------------------------------------------------------------------------
452+-- Initialization
453+-------------------------------------------------------------------------------
454+
455+newRateInfo :: Config -> IO (Maybe YieldRateInfo)
456+newRateInfo st = do
457+ -- convert rate in Hertz to latency in Nanoseconds
458+ let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
459+ case getStreamRate st of
460+ Just (Rate low goal high buf) ->
461+ let l = rateToLatency goal
462+ minl = rateToLatency high
463+ maxl = rateToLatency low
464+ in mkYieldRateInfo l (LatencyRange minl maxl) buf
465+ Nothing -> return Nothing
466+
467+ where
468+
469+ mkYieldRateInfo latency latRange buf = do
470+ measured <- newIORef 0
471+ wcur <- newIORef (0,0,0)
472+ wcol <- newIORef (0,0,0)
473+ now <- getTime Monotonic
474+ wlong <- newIORef (0,now)
475+ period <- newIORef 1
476+ gainLoss <- newIORef (Count 0)
477+
478+ return $ Just YieldRateInfo
479+ { svarLatencyTarget = latency
480+ , svarLatencyRange = latRange
481+ , svarRateBuffer = buf
482+ , svarGainedLostYields = gainLoss
483+ , workerBootstrapLatency = getStreamLatency st
484+ , workerPollingInterval = period
485+ , workerMeasuredLatency = measured
486+ , workerPendingLatency = wcur
487+ , workerCollectedLatency = wcol
488+ , svarAllTimeLatency = wlong
489+ }
490+
491+newSVarStats :: IO SVarStats
492+newSVarStats = do
493+ disp <- newIORef 0
494+ maxWrk <- newIORef 0
495+ maxOq <- newIORef 0
496+ maxHs <- newIORef 0
497+ maxWq <- newIORef 0
498+ avgLat <- newIORef (0, NanoSecond64 0)
499+ maxLat <- newIORef (NanoSecond64 0)
500+ minLat <- newIORef (NanoSecond64 0)
501+ stpTime <- newIORef Nothing
502+
503+ return SVarStats
504+ { totalDispatches = disp
505+ , maxWorkers = maxWrk
506+ , maxOutQSize = maxOq
507+ , maxHeapSize = maxHs
508+ , maxWorkQSize = maxWq
509+ , avgWorkerLatency = avgLat
510+ , minWorkerLatency = minLat
511+ , maxWorkerLatency = maxLat
512+ , svarStopTime = stpTime
513+ }
514+
515+-------------------------------------------------------------------------------
516+-- Rate
517+-------------------------------------------------------------------------------
518+
519+-- | Same as @rate (Just $ Rate (r/2) r (2*r) maxBound)@
520+--
521+-- Specifies the average production rate of a stream in number of yields
522+-- per second (i.e. @Hertz@). Concurrent production is ramped up or down
523+-- automatically to achieve the specified average yield rate. The rate can
524+-- go down to half of the specified rate on the lower side and double of
525+-- the specified rate on the higher side.
526+--
527+avgRate :: Double -> Config -> Config
528+avgRate r = rate (Just $ Rate (r/2) r (2*r) maxBound)
529+
530+-- | Same as @rate (Just $ Rate r r (2*r) maxBound)@
531+--
532+-- Specifies the minimum rate at which the stream should yield values. As
533+-- far as possible the yield rate would never be allowed to go below the
534+-- specified rate, even though it may possibly go above it at times, the
535+-- upper limit is double of the specified rate.
536+--
537+minRate :: Double -> Config -> Config
538+minRate r = rate (Just $ Rate r r (2*r) maxBound)
539+
540+-- | Same as @rate (Just $ Rate (r/2) r r maxBound)@
541+--
542+-- Specifies the maximum rate at which the stream should yield values. As
543+-- far as possible the yield rate would never be allowed to go above the
544+-- specified rate, even though it may possibly go below it at times, the
545+-- lower limit is half of the specified rate. This can be useful in
546+-- applications where certain resource usage must not be allowed to go
547+-- beyond certain limits.
548+--
549+maxRate :: Double -> Config -> Config
550+maxRate r = rate (Just $ Rate (r/2) r r maxBound)
551+
552+-- | Same as @rate (Just $ Rate r r r 0)@
553+--
554+-- Specifies a constant yield rate. If for some reason the actual rate
555+-- goes above or below the specified rate we do not try to recover it by
556+-- increasing or decreasing the rate in future. This can be useful in
557+-- applications like graphics frame refresh where we need to maintain a
558+-- constant refresh rate.
559+--
560+constRate :: Double -> Config -> Config
561+constRate r = rate (Just $ Rate r r r 0)
562+
563+-------------------------------------------------------------------------------
564+-- Channel yield count
565+-------------------------------------------------------------------------------
566+
567+-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
568+-- faster, along with the fields in sv required by send?
569+-- XXX make it noinline
570+--
571+-- XXX we may want to employ an increment and decrement in batches when the
572+-- througput is high or when the cost of synchronization is high. For example
573+-- if the application is distributed then inc/dec of a shared variable may be
574+-- very costly.
575+
576+-- | A worker decrements the yield limit before it executes an action. However,
577+-- the action may not result in an element being yielded, in that case we have
578+-- to increment the yield limit.
579+--
580+-- Note that we need it to be an Int type so that we have the ability to undo a
581+-- decrement that takes it below zero.
582+{-# INLINE decrementYieldLimit #-}
583+decrementYieldLimit :: Maybe (IORef Count) -> IO Bool
584+decrementYieldLimit remaining =
585+ case remaining of
586+ Nothing -> return True
587+ Just ref -> do
588+ r <- atomicModifyIORefCAS ref $ \x -> (x - 1, x)
589+ return $ r >= 1
590+
591+{-# INLINE incrementYieldLimit #-}
592+incrementYieldLimit :: Maybe (IORef Count) -> IO ()
593+incrementYieldLimit remaining =
594+ case remaining of
595+ Nothing -> return ()
596+ Just ref -> atomicModifyIORefCAS_ ref (+ 1)
597+
598+-------------------------------------------------------------------------------
599+-- Output queue
600+-------------------------------------------------------------------------------
601+
602+{-# INLINE readOutputQBasic #-}
603+readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)
604+readOutputQBasic q = atomicModifyIORefCAS q $ \x -> (([],0), x)
605+
606+{-# INLINE readOutputQRaw #-}
607+readOutputQRaw ::
608+ IORef ([ChildEvent a], Int) -> Maybe SVarStats -> IO ([ChildEvent a], Int)
609+readOutputQRaw q stats = do
610+ (list, len) <- readOutputQBasic q
611+ case stats of
612+ Just ss -> do
613+ let ref = maxOutQSize ss
614+ oqLen <- readIORef ref
615+ when (len > oqLen) $ writeIORef ref len
616+ Nothing -> return ()
617+ return (list, len)
618+
619+{-# INLINE ringDoorBell #-}
620+ringDoorBell :: IORef Bool -> MVar () -> IO ()
621+ringDoorBell needBell bell = do
622+ storeLoadBarrier
623+ w <- readIORef needBell
624+ when w $ do
625+ -- Note: the sequence of operations is important for correctness here.
626+ -- We need to set the flag to false strictly before sending the
627+ -- outputDoorBell, otherwise the outputDoorBell may get processed too
628+ -- early and then we may set the flag to False to later making the
629+ -- consumer lose the flag, even without receiving a outputDoorBell.
630+ atomicModifyIORefCAS_ needBell (const False)
631+ void $ tryPutMVar bell ()
632+
633+-------------------------------------------------------------------------------
634+-- Diagnostics
635+-------------------------------------------------------------------------------
636+
637+dumpCreator :: Show a => a -> String
638+dumpCreator tid = "Creator tid = " <> show tid
639+
640+dumpOutputQ :: (Foldable t, Show a1) => IORef (t a2, a1) -> IO String
641+dumpOutputQ q = do
642+ (oqList, oqLen) <- readIORef q
643+ return $ unlines
644+ [ "outputQueue length computed = " <> show (length oqList)
645+ , "outputQueue length maintained = " <> show oqLen
646+ ]
647+
648+dumpDoorBell :: Show a => MVar a -> IO String
649+dumpDoorBell mvar = do
650+ db <- tryReadMVar mvar
651+ return $ "outputDoorBell = " <> show db
652+
653+dumpNeedDoorBell :: Show a => IORef a -> IO String
654+dumpNeedDoorBell ref = do
655+ waiting <- readIORef ref
656+ return $ "needDoorBell = " <> show waiting
657+
658+dumpRunningThreads :: Show a => IORef a -> IO String
659+dumpRunningThreads ref = do
660+ rthread <- readIORef ref
661+ return $ "running threads = " <> show rthread
662+
663+dumpWorkerCount :: Show a => IORef a -> IO String
664+dumpWorkerCount ref = do
665+ workers <- readIORef ref
666+ return $ "running thread count = " <> show workers
667+
668+{-# NOINLINE mvarExcHandler #-}
669+mvarExcHandler :: IO String -> String -> BlockedIndefinitelyOnMVar -> IO ()
670+mvarExcHandler dump label e@BlockedIndefinitelyOnMVar = do
671+ svInfo <- dump
672+ hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnMVar\n" <> svInfo
673+ throwIO e
674+
675+{-# NOINLINE stmExcHandler #-}
676+stmExcHandler :: IO String -> String -> BlockedIndefinitelyOnSTM -> IO ()
677+stmExcHandler dump label e@BlockedIndefinitelyOnSTM = do
678+ svInfo <- dump
679+ hPutStrLn stderr $ label <> " " <> "BlockedIndefinitelyOnSTM\n" <> svInfo
680+ throwIO e
681+
682+-- | MVar diagnostics has some overhead - around 5% on AsyncT null benchmark, we
683+-- can keep it on in production to debug problems quickly if and when they
684+-- happen, but it may result in unexpected output when threads are left hanging
685+-- until they are GCed because the consumer went away.
686+withDiagMVar :: Bool -> IO String -> String -> IO () -> IO ()
687+withDiagMVar inspect dump label action =
688+ if inspect
689+ then
690+ action `catches` [ Handler (mvarExcHandler dump label)
691+ , Handler (stmExcHandler dump label)
692+ ]
693+ else action
694+
695+printSVar :: IO String -> String -> IO ()
696+printSVar dump how = do
697+ svInfo <- dump
698+ hPutStrLn stderr $ "\n" <> how <> "\n" <> svInfo
699+
700+-------------------------------------------------------------------------------
701+-- Cleanup
702+-------------------------------------------------------------------------------
703+
704+-- Can be called from a worker as well, avoids throwing to itself
705+cleanupSVar :: IORef (Set ThreadId) -> IO ()
706+cleanupSVar workerSet = do
707+ workers <- readIORef workerSet
708+ self <- myThreadId
709+ Prelude.mapM_ (`throwTo` ThreadAbort)
710+ (Prelude.filter (/= self) $ Set.toList workers)