@@ -48,6 +48,10 @@ module Streamly.Core
4848 , wAsync
4949 , parallel
5050
51+ -- * applications
52+ , applyWith
53+ , runWith
54+
5155 -- * zip
5256 , zipWith
5357 , zipAsyncWith
@@ -66,18 +70,10 @@ module Streamly.Core
6670 )
6771where
6872
69−-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark,
70−-- but keep it on for now to debug problems quickly if and when they happen.
71−#define DIAGNOSTICS
72−
7373import Control.Concurrent (ThreadId, myThreadId, threadDelay)
7474import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
7575 tryPutMVar, takeMVar, putMVar)
7676import Control.Exception (SomeException (..))
77−#ifdef DIAGNOSTICS
78−import Control.Exception (catch,
79− BlockedIndefinitelyOnMVar(..))
80−#endif
8177import qualified Control.Exception.Lifted as EL
8278import Control.Monad (when)
8379import Control.Monad.Catch (MonadThrow, throwM)
@@ -105,6 +101,17 @@ import GHC.Exts
105101import GHC.Conc (ThreadId(..))
106102import GHC.IO (IO(..))
107103
104+-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark,
105+-- but keep it on for now to debug problems quickly if and when they happen.
106+#define DIAGNOSTICS
107+
108+#ifdef DIAGNOSTICS
109+import Control.Exception (catches, throwIO, Handler(..),
110+ BlockedIndefinitelyOnMVar(..),
111+ BlockedIndefinitelyOnSTM(..))
112+import System.IO (hPutStrLn, stderr)
113+#endif
114+
108115------------------------------------------------------------------------------
109116-- Parent child thread communication type
110117------------------------------------------------------------------------------
@@ -193,41 +200,63 @@ data SVar m a =
193200{-# NOINLINE dumpSVar #-}
194201dumpSVar :: SVar m a -> IO String
195202dumpSVar sv = do
203+ tid <- myThreadId
196204 (oqList, oqLen) <- readIORef $ outputQueue sv
197205 db <- tryTakeMVar $ doorBell sv
198206 srn <- tryTakeMVar $ siren sv
199− (oheap, oheapSize) <- readIORef $ outputHeap sv
200− hsrn <- tryTakeMVar $ heapSiren sv
201− (wq, wqSize) <- readIORef $ workQueue sv
207+ aheadDump <-
208+ if svarStyle sv == AheadVar
209+ then do
210+ (oheap, oheapSize) <- readIORef $ outputHeap sv
211+ hsrn <- tryTakeMVar $ heapSiren sv
212+ (wq, wqSize) <- readIORef $ workQueue sv
213+ return $ unlines
214+ [ "heap length computed = " ++ show (H.size oheap)
215+ , "heap length maintained = " ++ show oheapSize
216+ , "heap siren = " ++ show hsrn
217+ , "work queue length computed = " ++ show (length wq)
218+ , "work queue length maintained = " ++ show wqSize
219+ ]
220+ else return []
221+
202222 waiting <- readIORef $ waitingForWork sv
203223 rthread <- readIORef $ runningThreads sv
204224 workers <- readIORef $ activeWorkers sv
205225 -- XXX queueEmpty should be made IO return type
206226
207227 return $ unlines
208− [ "style = " ++ show (svarStyle sv)
228+ [ "tid = " ++ show tid
229+ , "style = " ++ show (svarStyle sv)
209230 , "outputQueue length computed = " ++ show (length oqList)
210231 , "outputQueue length maintained = " ++ show oqLen
211232 , "output doorBell = " ++ show db
212233 , "worker siren = " ++ show srn
213− , "heap length computed = " ++ show (H.size oheap)
214− , "heap length maintained = " ++ show oheapSize
215− , "heap siren = " ++ show hsrn
216− , "work queue length computed = " ++ show (length wq)
217− , "work queue length maintained = " ++ show wqSize
218− , "waitingForWork = " ++ show waiting
234+ ]
235+ ++ aheadDump ++ unlines
236+ [ "waitingForWork = " ++ show waiting
219237 , "running threads = " ++ show rthread
220238 , "running thread count = " ++ show workers
221239 ]
222240
223241{-# NOINLINE mvarExcHandler #-}
224242mvarExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
225−mvarExcHandler sv label BlockedIndefinitelyOnMVar = do
243+mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
226244 svInfo <- dumpSVar sv
227− error $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo
245+ hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo
246+ throwIO e
247+
248+{-# NOINLINE stmExcHandler #-}
249+stmExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
250+stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
251+ svInfo <- dumpSVar sv
252+ hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo
253+ throwIO e
228254
229255withDBGMVar :: SVar m a -> String -> IO () -> IO ()
230−withDBGMVar sv label action = action `catch` mvarExcHandler sv label
256+withDBGMVar sv label action =
257+ action `catches` [ Handler (mvarExcHandler sv label)
258+ , Handler (stmExcHandler sv label)
259+ ]
231260#else
232261withDBGMVar :: SVar m a -> String -> IO () -> IO ()
233262withDBGMVar _ _ action = action
@@ -759,7 +788,7 @@ delThread sv tid =
759788-- If present then delete else add. This takes care of out of order add and
760789-- delete i.e. a delete arriving before we even added a thread.
761790-- This occurs when the forked thread is done even before the 'addThread' right
762−-- after the fork is executed.
791+-- after the fork gets a chance to run.
763792{-# INLINE modifyThread #-}
764793modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
765794modifyThread sv tid = do
@@ -797,7 +826,10 @@ pushWorker sv = do
797826-- using a CAS based modification.
798827{-# NOINLINE pushWorkerPar #-}
799828pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
800−pushWorkerPar sv m =
829+pushWorkerPar sv m = do
830+ -- We do not use activeWorkers in case of ParallelVar but still there is no
831+ -- harm in maintaining it correctly.
832+ liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
801833 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
802834
803835-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
@@ -1050,8 +1082,11 @@ newStreamVar1 style m = do
10501082 -- Note: We must have all the work on the queue before sending the
10511083 -- pushworker, otherwise the pushworker may exit before we even get a
10521084 -- chance to push.
1053− liftIO $ (enqueue sv) m
1054− pushWorker sv
1085+ if style == ParallelVar
1086+ then pushWorkerPar sv m
1087+ else do
1088+ liftIO $ (enqueue sv) m
1089+ pushWorker sv
10551090 return sv
10561091
10571092-- | Create a new SVar and enqueue one stream computation on it.
@@ -1245,6 +1280,26 @@ _alt m1 m2 = Stream $ \_ stp sng yld ->
12451280 let stop = runStream m2 Nothing stp sng yld
12461281 in runStream m1 Nothing stop sng yld
12471282
1283+------------------------------------------------------------------------------
1284+-- Stream to stream function application
1285+------------------------------------------------------------------------------
1286+
1287+applyWith :: MonadAsync m
1288+ => SVarStyle -> (Stream m a -> Stream m b) -> Stream m a -> Stream m b
1289+applyWith style f m = Stream $ \svr stp sng yld -> do
1290+ sv <- newStreamVar1 style m
1291+ runStream (f $ fromStreamVar sv) svr stp sng yld
1292+
1293+------------------------------------------------------------------------------
1294+-- Stream runner function application
1295+------------------------------------------------------------------------------
1296+
1297+runWith :: MonadAsync m
1298+ => SVarStyle -> (Stream m a -> m b) -> Stream m a -> m b
1299+runWith style f m = do
1300+ sv <- newStreamVar1 style m
1301+ f $ fromStreamVar sv
1302+
12481303------------------------------------------------------------------------------
12491304-- Zipping
12501305------------------------------------------------------------------------------
@@ -1261,10 +1316,6 @@ zipWith f m1 m2 = go m1 m2
12611316 yield1 a ra = merge a ra
12621317 (runStream mx) Nothing stp single1 yield1
12631318
1264−mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)
1265−mkAsync m = newStreamVar1 AsyncVar m
1266− >>= return . fromStreamVar
1267−
12681319zipAsyncWith :: MonadAsync m
12691320 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
12701321zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
@@ -1272,6 +1323,12 @@ zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
12721323 mb <- mkAsync m2
12731324 (runStream (zipWith f ma mb)) Nothing stp sng yld
12741325
1326+ where
1327+
1328+ mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)
1329+ mkAsync m = newStreamVar1 AsyncVar m
1330+ >>= return . fromStreamVar
1331+
12751332-------------------------------------------------------------------------------
12761333-- Transformers
12771334-------------------------------------------------------------------------------