06a827f · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Move code and formatting/doc changes

06a827fHarendra Kumar8 years ago

src/Streamly/SVar.hs+64−36

compare with master

unifiedsplit

@@ -808,27 +808,8 @@ withDiagMVar sv label action =

808 ]

809 else action

810

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−

830------------------------------------------------------------------------------

811------------------------------------------------------------------------------

831−-- Spawning threads and collecting result in streamed fashion

812+-- Spawning threads

832------------------------------------------------------------------------------

813------------------------------------------------------------------------------

833

814

834-- | A monad that can perform concurrent or parallel IO operations. Streams

815-- | A monad that can perform concurrent or parallel IO operations. Streams

@@ -873,6 +854,10 @@ doFork action (RunInIO mrun) exHandler =

873 exHandler

854 exHandler

874 run (return tid)

855 run (return tid)

875

856

857+------------------------------------------------------------------------------

858+-- Collecting results from child workers in a streamed fashion

859+------------------------------------------------------------------------------

860+

876-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv

861-- XXX Can we make access to remainingWork and yieldRateInfo fields in sv

877-- faster, along with the fields in sv required by send?

862-- faster, along with the fields in sv required by send?

878-- XXX make it noinline

863-- XXX make it noinline

@@ -1067,6 +1052,24 @@ sendStop sv mwinfo = do

1067 return ()

1052 return ()

1068 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)

1053 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)

1069

1054

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+

1070-------------------------------------------------------------------------------

1073-------------------------------------------------------------------------------

1071-- Async

1074-- Async

1072-------------------------------------------------------------------------------

1075-------------------------------------------------------------------------------

@@ -1295,6 +1298,10 @@ updateHeapSeq hpVar seqNo =

1295-- remaining computation at the back of the queue instead of the heap, and

1298-- remaining computation at the back of the queue instead of the heap, and

1296-- increment the sequence number.

1299-- increment the sequence number.

1297

1300

1301+-------------------------------------------------------------------------------

1302+-- Dispatching workers and tracking them

1303+-------------------------------------------------------------------------------

1304+

1298-- Thread tracking is needed for two reasons:

1305-- Thread tracking is needed for two reasons:

1299--

1306--

1300-- 1) Killing threads on exceptions. Threads may not be left to go away by

1307-- 1) Killing threads on exceptions. Threads may not be left to go away by

@@ -1454,17 +1461,17 @@ dispatchWorker yieldCount sv = do

1454 Unlimited -> Limited (fromIntegral n)

1461 Unlimited -> Limited (fromIntegral n)

1455 Limited lim -> Limited $ min lim (fromIntegral n)

1462 Limited lim -> Limited $ min lim (fromIntegral n)

1456

1463

1457− -- XXX for ahead streams shall we take the heap yields into account for

1464+ -- XXX for ahead streams shall we take the heap yields into account

1458− -- controlling the dispatch? We should not dispatch if the heap has

1465+ -- for controlling the dispatch? We should not dispatch if the heap

1459− -- already got the limit covered.

1466+ -- has already got the limit covered.

1460 let dispatch = pushWorker yieldCount sv >> return True

1467 let dispatch = pushWorker yieldCount sv >> return True

1461 in case limit of

1468 in case limit of

1462 Unlimited -> dispatch

1469 Unlimited -> dispatch

1463 -- Note that the use of remainingWork and workerCount is not

1470 -- Note that the use of remainingWork and workerCount is not

1464− -- atomic and the counts may even have changed between reading and

1471+ -- atomic and the counts may even have changed between reading

1465− -- using them here, so this is just approximate logic and we cannot

1472+ -- and using them here, so this is just approximate logic and

1466− -- rely on it for correctness. We may actually dispatch more

1473+ -- we cannot rely on it for correctness. We may actually

1467− -- workers than required.

1474+ -- dispatch more workers than required.

1468 Limited lim | lim > 0 -> dispatch

1475 Limited lim | lim > 0 -> dispatch

1469 _ -> return False

1476 _ -> return False

1470 else do

1477 else do

@@ -1472,6 +1479,10 @@ dispatchWorker yieldCount sv = do

1472 return False

1479 return False

1473 else return False

1480 else return False

1474

1481

1482+-------------------------------------------------------------------------------

1483+-- Dispatch workers with rate control

1484+-------------------------------------------------------------------------------

1485+

1475-- | This is a magic number and it is overloaded, and used at several places to

1486-- | This is a magic number and it is overloaded, and used at several places to

1476-- achieve batching:

1487-- achieve batching:

1477--

1488--

@@ -1711,8 +1722,9 @@ dispatchWorkerPaced sv = do

1711 batch = max 1 $ fromIntegral $

1722 batch = max 1 $ fromIntegral $

1712 minThreadDelay `div` targetLat

1723 minThreadDelay `div` targetLat

1713 -- XXX stagger the workers over a period?

1724 -- XXX stagger the workers over a period?

1714− -- XXX cannot sleep, as that would mean we cannot process the

1725+ -- XXX cannot sleep, as that would mean we cannot process

1715− -- outputs. need to try a different mechanism to stagger.

1726+ -- the outputs. need to try a different mechanism to

1727+ -- stagger.

1716 -- when (total > batch) $

1728 -- when (total > batch) $

1717 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay

1729 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay

1718 dispatchN (min total batch)

1730 dispatchN (min total batch)

@@ -1738,6 +1750,10 @@ dispatchWorkerPaced sv = do

1738 then dispatchN (n - 1)

1750 then dispatchN (n - 1)

1739 else return False

1751 else return False

1740

1752

1753+-------------------------------------------------------------------------------

1754+-- Worker dispatch and wait loop

1755+-------------------------------------------------------------------------------

1756+

1741sendWorkerDelayPaced :: SVar t m a -> IO ()

1757sendWorkerDelayPaced :: SVar t m a -> IO ()

1742sendWorkerDelayPaced _ = return ()

1758sendWorkerDelayPaced _ = return ()

1743

1759

@@ -1808,18 +1824,18 @@ sendWorkerWait delay dispatch sv = do

1808 -- queued items in the heap even though the outputQueue is empty, and

1824 -- queued items in the heap even though the outputQueue is empty, and

1809 -- we may have active workers which are deadlocked on those items to be

1825 -- we may have active workers which are deadlocked on those items to be

1810 -- processed by the consumer. We should either guarantee that any

1826 -- processed by the consumer. We should either guarantee that any

1811− -- worker, before returning, clears the heap or we send a worker to clear

1827+ -- worker, before returning, clears the heap or we send a worker to

1812− -- it. Normally we always send a worker if no output is seen, but if

1828+ -- clear it. Normally we always send a worker if no output is seen, but

1813− -- the thread limit is reached or we are using pacing then we may not

1829+ -- if the thread limit is reached or we are using pacing then we may

1814− -- send a worker. See the concurrentApplication test in the tests, that

1830+ -- not send a worker. See the concurrentApplication test in the tests,

1815− -- test case requires at least one yield from the producer to not

1831+ -- that test case requires at least one yield from the producer to not

1816 -- deadlock, if the last workers output is stuck in the heap then this

1832 -- deadlock, if the last workers output is stuck in the heap then this

1817 -- test fails. This problem can be extended to n threads when the

1833 -- test fails. This problem can be extended to n threads when the

1818 -- consumer may depend on the evaluation of next n items in the

1834 -- consumer may depend on the evaluation of next n items in the

1819 -- producer stream.

1835 -- producer stream.

1820

1836

1821− -- register for the outputDoorBell before we check the queue so that if we

1837+ -- register for the outputDoorBell before we check the queue so that if

1822− -- sleep because the queue was empty we are guaranteed to get a

1838+ -- we sleep because the queue was empty we are guaranteed to get a

1823 -- doorbell on the next enqueue.

1839 -- doorbell on the next enqueue.

1824

1840

1825 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True

1841 liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True

@@ -1841,6 +1857,10 @@ sendWorkerWait delay dispatch sv = do

1841 (_, len) <- liftIO $ readIORef (outputQueue sv)

1857 (_, len) <- liftIO $ readIORef (outputQueue sv)

1842 when (len <= 0) $ sendWorkerWait delay dispatch sv

1858 when (len <= 0) $ sendWorkerWait delay dispatch sv

1843

1859

1860+-------------------------------------------------------------------------------

1861+-- Reading from the workers' output queue/buffer

1862+-------------------------------------------------------------------------------

1863+

1844{-# INLINE readOutputQRaw #-}

1864{-# INLINE readOutputQRaw #-}

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

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

1846readOutputQRaw sv = do

1866readOutputQRaw sv = do

@@ -1939,6 +1959,10 @@ postProcessPaced sv = do

1939 return r

1959 return r

1940 else return False

1960 else return False

1941

1961

1962+-------------------------------------------------------------------------------

1963+-- Creating an SVar

1964+-------------------------------------------------------------------------------

1965+

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

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

1943getYieldRateInfo st = do

1967getYieldRateInfo st = do

1944 -- convert rate in Hertz to latency in Nanoseconds

1968 -- convert rate in Hertz to latency in Nanoseconds

@@ -2184,6 +2208,10 @@ newParallelVar st = do

2184 mrun <- captureMonadState

2208 mrun <- captureMonadState

2185 liftIO $ getParallelSVar st mrun

2209 liftIO $ getParallelSVar st mrun

2186

2210

2211+-------------------------------------------------------------------------------

2212+-- Write a stream to an SVar

2213+-------------------------------------------------------------------------------

2214+

2187-- XXX this errors out for Parallel/Ahead SVars

2215-- XXX this errors out for Parallel/Ahead SVars

2188-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

2216-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then

2189-- be read back from the SVar using 'fromSVar'.

2217-- be read back from the SVar using 'fromSVar'.