@@ -808,27 +808,8 @@ withDiagMVar sv label action =
808808 ]
809809 else action
810810
811−-------------------------------------------------------------------------------
812−-- CAS
813−-------------------------------------------------------------------------------
814−
815−{-# INLINE ringDoorBell #-}
816−ringDoorBell :: SVar t m a -> IO ()
817−ringDoorBell sv = do
818− storeLoadBarrier
819− w <- readIORef $ needDoorBell sv
820− when w $ do
821− -- Note: the sequence of operations is important for correctness here.
822− -- We need to set the flag to false strictly before sending the
823− -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
824− -- then we may set the flag to False to later making the consumer lose
825− -- the flag, even without receiving a outputDoorBell.
826− atomicModifyIORefCAS_ (needDoorBell sv) (const False)
827− void $ tryPutMVar (outputDoorBell sv) ()
828−
829−
830811------------------------------------------------------------------------------
831−-- Spawning threads and collecting result in streamed fashion
812+-- Spawning threads
832813------------------------------------------------------------------------------
833814
834815-- | A monad that can perform concurrent or parallel IO operations. Streams
@@ -873,6 +854,10 @@ doFork action (RunInIO mrun) exHandler =
873854 exHandler
874855 run (return tid)
875856
857+------------------------------------------------------------------------------
858+-- Collecting results from child workers in a streamed fashion
859+------------------------------------------------------------------------------
860+
876861-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv
877862-- faster, along with the fields in sv required by send?
878863-- XXX make it noinline
@@ -1067,6 +1052,24 @@ sendStop sv mwinfo = do
10671052 return ()
10681053 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
10691054
1055+-------------------------------------------------------------------------------
1056+-- Doorbell
1057+-------------------------------------------------------------------------------
1058+
1059+{-# INLINE ringDoorBell #-}
1060+ringDoorBell :: SVar t m a -> IO ()
1061+ringDoorBell sv = do
1062+ storeLoadBarrier
1063+ w <- readIORef $ needDoorBell sv
1064+ when w $ do
1065+ -- Note: the sequence of operations is important for correctness here.
1066+ -- We need to set the flag to false strictly before sending the
1067+ -- outputDoorBell, otherwise the outputDoorBell may get processed too
1068+ -- early and then we may set the flag to False to later making the
1069+ -- consumer lose the flag, even without receiving a outputDoorBell.
1070+ atomicModifyIORefCAS_ (needDoorBell sv) (const False)
1071+ void $ tryPutMVar (outputDoorBell sv) ()
1072+
10701073-------------------------------------------------------------------------------
10711074-- Async
10721075-------------------------------------------------------------------------------
@@ -1295,6 +1298,10 @@ updateHeapSeq hpVar seqNo =
12951298-- remaining computation at the back of the queue instead of the heap, and
12961299-- increment the sequence number.
12971300
1301+-------------------------------------------------------------------------------
1302+-- Dispatching workers and tracking them
1303+-------------------------------------------------------------------------------
1304+
12981305-- Thread tracking is needed for two reasons:
12991306--
13001307-- 1) Killing threads on exceptions. Threads may not be left to go away by
@@ -1454,17 +1461,17 @@ dispatchWorker yieldCount sv = do
14541461 Unlimited -> Limited (fromIntegral n)
14551462 Limited lim -> Limited $ min lim (fromIntegral n)
14561463
1457− -- XXX for ahead streams shall we take the heap yields into account for
1458− -- controlling the dispatch? We should not dispatch if the heap has
1459− -- already got the limit covered.
1464+ -- XXX for ahead streams shall we take the heap yields into account
1465+ -- for controlling the dispatch? We should not dispatch if the heap
1466+ -- has already got the limit covered.
14601467 let dispatch = pushWorker yieldCount sv >> return True
14611468 in case limit of
14621469 Unlimited -> dispatch
14631470 -- Note that the use of remainingWork and workerCount is not
1464− -- atomic and the counts may even have changed between reading and
1465− -- using them here, so this is just approximate logic and we cannot
1466− -- rely on it for correctness. We may actually dispatch more
1467− -- workers than required.
1471+ -- atomic and the counts may even have changed between reading
1472+ -- and using them here, so this is just approximate logic and
1473+ -- we cannot rely on it for correctness. We may actually
1474+ -- dispatch more workers than required.
14681475 Limited lim | lim > 0 -> dispatch
14691476 _ -> return False
14701477 else do
@@ -1472,6 +1479,10 @@ dispatchWorker yieldCount sv = do
14721479 return False
14731480 else return False
14741481
1482+-------------------------------------------------------------------------------
1483+-- Dispatch workers with rate control
1484+-------------------------------------------------------------------------------
1485+
14751486-- | This is a magic number and it is overloaded, and used at several places to
14761487-- achieve batching:
14771488--
@@ -1711,8 +1722,9 @@ dispatchWorkerPaced sv = do
17111722 batch = max 1 $ fromIntegral $
17121723 minThreadDelay `div` targetLat
17131724 -- XXX stagger the workers over a period?
1714− -- XXX cannot sleep, as that would mean we cannot process the
1715− -- outputs. need to try a different mechanism to stagger.
1725+ -- XXX cannot sleep, as that would mean we cannot process
1726+ -- the outputs. need to try a different mechanism to
1727+ -- stagger.
17161728 -- when (total > batch) $
17171729 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
17181730 dispatchN (min total batch)
@@ -1738,6 +1750,10 @@ dispatchWorkerPaced sv = do
17381750 then dispatchN (n - 1)
17391751 else return False
17401752
1753+-------------------------------------------------------------------------------
1754+-- Worker dispatch and wait loop
1755+-------------------------------------------------------------------------------
1756+
17411757sendWorkerDelayPaced :: SVar t m a -> IO ()
17421758sendWorkerDelayPaced _ = return ()
17431759
@@ -1808,18 +1824,18 @@ sendWorkerWait delay dispatch sv = do
18081824 -- queued items in the heap even though the outputQueue is empty, and
18091825 -- we may have active workers which are deadlocked on those items to be
18101826 -- processed by the consumer. We should either guarantee that any
1811− -- worker, before returning, clears the heap or we send a worker to clear
1812− -- it. Normally we always send a worker if no output is seen, but if
1813− -- the thread limit is reached or we are using pacing then we may not
1814− -- send a worker. See the concurrentApplication test in the tests, that
1815− -- test case requires at least one yield from the producer to not
1827+ -- worker, before returning, clears the heap or we send a worker to
1828+ -- clear it. Normally we always send a worker if no output is seen, but
1829+ -- if the thread limit is reached or we are using pacing then we may
1830+ -- not send a worker. See the concurrentApplication test in the tests,
1831+ -- that test case requires at least one yield from the producer to not
18161832 -- deadlock, if the last workers output is stuck in the heap then this
18171833 -- test fails. This problem can be extended to n threads when the
18181834 -- consumer may depend on the evaluation of next n items in the
18191835 -- producer stream.
18201836
1821− -- register for the outputDoorBell before we check the queue so that if we
1822− -- sleep because the queue was empty we are guaranteed to get a
1837+ -- register for the outputDoorBell before we check the queue so that if
1838+ -- we sleep because the queue was empty we are guaranteed to get a
18231839 -- doorbell on the next enqueue.
18241840
18251841 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
@@ -1841,6 +1857,10 @@ sendWorkerWait delay dispatch sv = do
18411857 (_, len) <- liftIO $ readIORef (outputQueue sv)
18421858 when (len <= 0) $ sendWorkerWait delay dispatch sv
18431859
1860+-------------------------------------------------------------------------------
1861+-- Reading from the workers' output queue/buffer
1862+-------------------------------------------------------------------------------
1863+
18441864{-# INLINE readOutputQRaw #-}
18451865readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
18461866readOutputQRaw sv = do
@@ -1939,6 +1959,10 @@ postProcessPaced sv = do
19391959 return r
19401960 else return False
19411961
1962+-------------------------------------------------------------------------------
1963+-- Creating an SVar
1964+-------------------------------------------------------------------------------
1965+
19421966getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
19431967getYieldRateInfo st = do
19441968 -- convert rate in Hertz to latency in Nanoseconds
@@ -2184,6 +2208,10 @@ newParallelVar st = do
21842208 mrun <- captureMonadState
21852209 liftIO $ getParallelSVar st mrun
21862210
2211+-------------------------------------------------------------------------------
2212+-- Write a stream to an SVar
2213+-------------------------------------------------------------------------------
2214+
21872215-- XXX this errors out for Parallel/Ahead SVars
21882216-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
21892217-- be read back from the SVar using 'fromSVar'.