@@ -1,3 +1,4 @@
1+{-# LANGUAGE CPP #-}
12{-# LANGUAGE ConstraintKinds #-}
23{-# LANGUAGE FlexibleContexts #-}
34{-# LANGUAGE FlexibleInstances #-}
@@ -25,13 +26,19 @@ module Streamly.Core
2526 -- * Streams
2627 , Stream (..)
2728
28− -- * Construction
29+ -- * Construction (pure)
30+ , nil
31+ , cons
2932 , singleton
3033 , once
31− , cons
32− , consM
3334 , repeat
34− , nil
35+
36+ -- * Construction (monadic)
37+ , consM
38+ , consMAhead
39+ , consMAsync
40+ , consMWAsync
41+ , consMParallel
3542
3643 -- * Semigroup Style Composition
3744 , serial
@@ -41,13 +48,11 @@ module Streamly.Core
4148 , wAsync
4249 , parallel
4350
44− -- * Alternative
45− , alt
46−
4751 -- * zip
4852 , zipWith
4953 , zipAsyncWith
5054
55+ -- XXX move to Streams.hs
5156 -- * Transformers
5257 , withLocal
5358 , withCatchError
@@ -55,18 +60,24 @@ module Streamly.Core
5560 -- * Concurrent Stream Vars (SVars)
5661 , SVar
5762 , SVarStyle (..)
58− , newEmptySVar
5963 , newStreamVar1
60− , joinStreamVarAsync
6164 , fromStreamVar
6265 , toStreamVar
6366 )
6467where
6568
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+
6673import Control.Concurrent (ThreadId, myThreadId, threadDelay)
6774import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
6875 tryPutMVar, takeMVar, putMVar)
6976import Control.Exception (SomeException (..))
77+#ifdef DIAGNOSTICS
78+import Control.Exception (catch,
79+ BlockedIndefinitelyOnMVar(..))
80+#endif
7081import qualified Control.Exception.Lifted as EL
7182import Control.Monad (when)
7283import Control.Monad.Catch (MonadThrow, throwM)
@@ -120,7 +131,7 @@ data SVarStyle =
120131 | WAsyncVar -- breadth first concurrent
121132 | ParallelVar -- all parallel
122133 | AheadVar -- Concurrent look ahead
123− deriving Eq
134+ deriving (Eq, Show)
124135
125136-- | An SVar or a Stream Var is a conduit to the output from multiple streams
126137-- running concurrently and asynchronously. An SVar can be thought of as an
@@ -170,6 +181,7 @@ data SVar m a =
170181 , workQueue :: IORef ([Stream m a], Int)
171182 , enqueue :: Stream m a -> IO ()
172183 , queueEmpty :: m Bool
184+ , waitingForWork :: IORef Bool
173185 , runqueue :: m ()
174186
175187 -- Shared, thread tracking
@@ -177,6 +189,50 @@ data SVar m a =
177189 , activeWorkers :: IORef Int
178190 }
179191
192+#ifdef DIAGNOSTICS
193+{-# NOINLINE dumpSVar #-}
194+dumpSVar :: SVar m a -> IO String
195+dumpSVar sv = do
196+ (oqList, oqLen) <- readIORef $ outputQueue sv
197+ db <- tryTakeMVar $ doorBell sv
198+ srn <- tryTakeMVar $ siren sv
199+ (oheap, oheapSize) <- readIORef $ outputHeap sv
200+ hsrn <- tryTakeMVar $ heapSiren sv
201+ (wq, wqSize) <- readIORef $ workQueue sv
202+ waiting <- readIORef $ waitingForWork sv
203+ rthread <- readIORef $ runningThreads sv
204+ workers <- readIORef $ activeWorkers sv
205+ -- XXX queueEmpty should be made IO return type
206+
207+ return $ unlines
208+ [ "style = " ++ show (svarStyle sv)
209+ , "outputQueue length computed = " ++ show (length oqList)
210+ , "outputQueue length maintained = " ++ show oqLen
211+ , "output doorBell = " ++ show db
212+ , "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
219+ , "running threads = " ++ show rthread
220+ , "running thread count = " ++ show workers
221+ ]
222+
223+{-# NOINLINE mvarExcHandler #-}
224+mvarExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
225+mvarExcHandler sv label BlockedIndefinitelyOnMVar = do
226+ svInfo <- dumpSVar sv
227+ error $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo
228+
229+withDBGMVar :: SVar m a -> String -> IO () -> IO ()
230+withDBGMVar sv label action = action `catch` mvarExcHandler sv label
231+#else
232+withDBGMVar :: SVar m a -> String -> IO () -> IO ()
233+withDBGMVar _ _ action = action
234+#endif
235+
180236-- Slightly faster version of CAS. Gained some improvement by avoiding the use
181237-- of "evaluate" because we know we do not have exceptions in fn.
182238{-# INLINE atomicModifyIORefCAS #-}
@@ -350,7 +406,7 @@ send sv msg = do
350406 -- we will be woken up by the consumer blowing the siren.
351407 -- Note: current length after adding our output is len + 1
352408 else when (len + 1 >= 1500) $ do
353− takeMVar (siren sv)
409+ withDBGMVar sv "send: wait for siren" $ takeMVar (siren sv)
354410 (_, n) <- readIORef (outputQueue sv)
355411 when (n <= 1500) $ void $ tryPutMVar (siren sv) ()
356412
@@ -367,8 +423,22 @@ sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
367423-- expressions. Large left associated compositions can grow this to a
368424-- large size
369425{-# INLINE enqueueLIFO #-}
370−enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
371−enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
426+enqueueLIFO :: SVar m a -> IORef [Stream m a] -> Stream m a -> IO ()
427+enqueueLIFO sv q m = do
428+ w <- readIORef $ waitingForWork sv
429+ if w
430+ then withDoorBell
431+ else atomicModifyIORefCAS_ q $ \ms -> m : ms
432+
433+ where
434+
435+ withDoorBell = do
436+ v <- atomicModifyIORefCAS q $ \ ms ->
437+ case ms of
438+ [] -> (m : ms, True)
439+ _ -> (m : ms, False)
440+ when v $ void $ tryPutMVar (doorBell sv) ()
441+ atomicModifyIORefCAS_ (waitingForWork sv) (const False)
372442
373443runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
374444runqueueLIFO sv q = run
@@ -394,8 +464,19 @@ runqueueLIFO sv q = run
394464-------------------------------------------------------------------------------
395465
396466{-# INLINE enqueueFIFO #-}
397−enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
398−enqueueFIFO = pushL
467+enqueueFIFO :: SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> IO ()
468+enqueueFIFO sv q m = do
469+ w <- readIORef $ waitingForWork sv
470+ if w
471+ then withDoorBell
472+ else pushL q m
473+
474+ where
475+
476+ withDoorBell = do
477+ emp <- nullQ q
478+ pushL q m
479+ when emp $ void $ tryPutMVar (doorBell sv) ()
399480
400481runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
401482runqueueFIFO sv q = run
@@ -411,7 +492,7 @@ runqueueFIFO sv q = run
411492 dequeue = liftIO $ tryPopR q
412493 sendit a = liftIO $ send sv (ChildYield a)
413494 single a = sendit a >> run
414− yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run
495+ yield a r = sendit a >> liftIO (enqueueFIFO sv q r) >> run
415496
416497-------------------------------------------------------------------------------
417498-- Parallel
@@ -499,7 +580,8 @@ sendToHeap sv msg seqNo = do
499580 if added
500581 then do
501582 when (len + 1 >= maxHeapSize) $ do
502− takeMVar (heapSiren sv)
583+ withDBGMVar sv "sendToHeap: wait for heapSiren"
584+ $ takeMVar (heapSiren sv)
503585 (h, _) <- readIORef (outputHeap sv)
504586 when (H.size h <= maxHeapSize) $ void $ tryPutMVar (siren sv) ()
505587 return False
@@ -508,7 +590,7 @@ sendToHeap sv msg seqNo = do
508590 AheadYieldSingle a -> send sv (ChildYield a)
509591 AheadYieldMany a mv -> do
510592 send sv (ChildYield a)
511− putMVar mv ()
593+ withDBGMVar sv "sendToHeap: wakeup stream" $ putMVar mv ()
512594 return True
513595
514596popFromHeap
@@ -547,7 +629,7 @@ drainHeap sv nextSeqNo = do
547629 drainHeap sv nextSeqNo
548630 AheadYieldMany a mv -> do
549631 send sv (ChildYield a)
550− putMVar mv ()
632+ withDBGMVar sv "drainHeap: wakeup stream" $ putMVar mv ()
551633
552634-- | The thread that has the token has the responsibility to move any pending
553635-- entries in the heap to the outputQueue. When the token is assigned to a
@@ -585,11 +667,15 @@ handleOutput sv haveToken prevSeqNo seqNo a ev = do
585667-- associated expression, we queue the right expression and execute the left.
586668-- Thererefore the queue never has more than on item in it.
587669{-# INLINE enqueueAhead #-}
588−enqueueAhead :: IORef ([Stream m a], Int) -> Stream m a -> IO ()
589−enqueueAhead q m = do
670+enqueueAhead :: SVar m a -> IORef ([Stream m a], Int) -> Stream m a -> IO ()
671+enqueueAhead sv q m = do
672+ w <- readIORef $ waitingForWork sv
590673 atomicModifyIORefCAS_ q $ \ case
591674 ([], n) -> ([m], n + 1) -- increment sequence
592675 _ -> error "not empty"
676+ when w $ do
677+ void $ tryPutMVar (doorBell sv) ()
678+ atomicModifyIORefCAS_ (waitingForWork sv) (const False)
593679
594680runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
595681runqueueAhead sv q = run False (-1)
@@ -637,7 +723,9 @@ runqueueAhead sv q = run False (-1)
637723 gotToken <-
638724 handleOutput sv tStatus prevSeqNo seqNo a
639725 (AheadYieldMany a mv)
640− when (not gotToken) $ void $ takeMVar mv
726+ when (not gotToken) $ void $
727+ withDBGMVar sv "runqueueAhead: stream waiting"
728+ $ takeMVar mv
641729 (runStream r) (Just sv) stopNext singleNext yieldNext
642730
643731 (runStream m) (Just sv) stop single yield
@@ -725,32 +813,76 @@ sendWorkerWait sv = do
725813 -- If there is no output pending to process and there is no worker to be
726814 -- sent then we block, so that we do not keep looping fruitlessly.
727815
816+ -- Note that we are guaranteed to have at least one outstanding worker when
817+ -- we enter this function. So if we sleep we are guaranteed to be woken up.
818+ -- Note that the worker count is only decremented during event processing
819+ -- in fromStreamVar and therefore it is safe to read and use it without a
820+ -- lock.
821+
728822 -- XXX we should wait in such a way so that we can be immediately
729− -- interrupted by a doorBell.
823+ -- interrupted by a doorBell. We can sleep on the doorbell and use a
824+ -- monitor thread to send us a doorbell.
730825 liftIO $ threadDelay 200
731826 (_, n) <- liftIO $ readIORef (outputQueue sv)
732827 when (n <= 0) $ do
733− -- XXX the queue may be empty temporarily if the worker has dequeued
734− -- the work item but has not enqueued the remaining part yet. For the
735− -- same reason, a worker may come back if it tries to dequeue when the
736− -- queue is empty, even though the whole work has not finished yet.
828+ -- The queue may be empty temporarily if the worker has dequeued the
829+ -- work item but has not enqueued the remaining part yet. For the same
830+ -- reason, a worker may come back if it tries to dequeue and finds the
831+ -- queue empty, even though the whole work has not finished yet.
737832 done <- queueEmpty sv
738833 if not done
739834 then do
740835 cnt <- liftIO $ readIORef $ activeWorkers sv
836+ -- Note that we may deadlock if the previous workers (tasks in the
837+ -- stream) wait/depend on the future workers (tasks in the stream)
838+ -- executing. In that case we should either configure the maxWorker
839+ -- count to higher or use parallel style instead of ahead or async
840+ -- style.
741841 if (cnt < 1500)
742842 then do
743843 pushWorker sv
744844 sendWorkerWait sv
745− else liftIO $ takeMVar (doorBell sv)
746− else liftIO $ takeMVar (doorBell sv)
845+ else liftIO $ withDBGMVar sv "sendWorkerWait: wait for workers"
846+ $ takeMVar (doorBell sv)
847+ else do
848+ -- We found that the queue was empty, but it may be empty
849+ -- temporarily, when we checked it. If that's the case we might
850+ -- sleep indefinitely unless the active workers produce some
851+ -- output. We may deadlock specially if the otuput from the other
852+ -- workers depends on the future workers that we may never send.
853+ -- So in case the queue was temporarily empty set a flag to inform
854+ -- the enqueue to send us a doorbell.
855+
856+ liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True
857+
858+ -- check again, this time we have set the waitingForWork flag so we
859+ -- are guaranteed to get a doorbell in case the status changed
860+ -- before we could sleep.
861+ --
862+ -- Note that this is just a best effort mechanism to avoid a
863+ -- deadlock. Deadlocks may still happen if for some weird reason
864+ -- the consuming computation shares an MVar or some other resource
865+ -- with the producing computation and gets blocked on that resource
866+ -- and therefore cannot do any pushworker to add more threads to
867+ -- the producer. In such cases the programmer should use a parallel
868+ -- style so that all the producers are scheduled immediately and
869+ -- unconditionally. We can also use a separate monitor thread to
870+ -- push workers instead of pushing them from the consumer, but then
871+ -- we are no longer using pull based concurrency rate adaptation.
872+ --
873+ -- XXX update this in the tutorial.
874+ emp <- queueEmpty sv
875+ when emp $
876+ liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
877+ $ takeMVar (doorBell sv)
747878
748879-- | Pull a stream from an SVar.
749880{-# NOINLINE fromStreamVar #-}
750881fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
751882fromStreamVar sv = Stream $ \_ stp sng yld -> do
752883 if svarStyle sv == ParallelVar
753− then liftIO $ takeMVar (doorBell sv)
884+ then liftIO $ withDBGMVar sv "fromStreamVar: doorbell"
885+ $ takeMVar (doorBell sv)
754886 else do
755887 res <- liftIO $ tryTakeMVar (doorBell sv)
756888 when (isNothing res) $ sendWorkerWait sv
@@ -798,6 +930,7 @@ getFifoSVar ctype = do
798930 outQMv <- newEmptyMVar
799931 hooter <- newEmptyMVar
800932 active <- newIORef 0
933+ wfw <- newIORef False
801934 running <- newIORef S.empty
802935 q <- newQ
803936 let sv =
@@ -809,8 +942,9 @@ getFifoSVar ctype = do
809942 , runningThreads = running
810943 , workQueue = undefined
811944 , runqueue = runqueueFIFO sv q
812− , enqueue = pushL q
945+ , enqueue = enqueueFIFO sv q
813946 , queueEmpty = liftIO $ nullQ q
947+ , waitingForWork = wfw
814948 , svarStyle = ctype
815949 , activeWorkers = active
816950 }
@@ -822,6 +956,7 @@ getLifoSVar ctype = do
822956 outQMv <- newEmptyMVar
823957 hooter <- newEmptyMVar
824958 active <- newIORef 0
959+ wfw <- newIORef False
825960 running <- newIORef S.empty
826961 q <- newIORef []
827962 let checkEmpty = null <$> liftIO (readIORef q)
@@ -834,8 +969,9 @@ getLifoSVar ctype = do
834969 , runningThreads = running
835970 , workQueue = undefined
836971 , runqueue = runqueueLIFO sv q
837− , enqueue = enqueueLIFO q
972+ , enqueue = enqueueLIFO sv q
838973 , queueEmpty = checkEmpty
974+ , waitingForWork = wfw
839975 , svarStyle = ctype
840976 , activeWorkers = active
841977 }
@@ -847,6 +983,7 @@ getParSVar style = do
847983 outQMv <- newEmptyMVar
848984 hooter <- newEmptyMVar
849985 active <- newIORef 0
986+ wfw <- newIORef False
850987 running <- newIORef S.empty
851988 let sv =
852989 SVar { outputQueue = outQ
@@ -859,6 +996,7 @@ getParSVar style = do
859996 , runqueue = undefined
860997 , enqueue = undefined
861998 , queueEmpty = undefined
999+ , waitingForWork = wfw
8621000 , svarStyle = style
8631001 , activeWorkers = active
8641002 }
@@ -872,6 +1010,7 @@ getAheadSVar style = do
8721010 hooter <- newEmptyMVar
8731011 hsiren <- newEmptyMVar
8741012 active <- newIORef 0
1013+ wfw <- newIORef False
8751014 running <- newIORef S.empty
8761015 q <- newIORef ([], -1)
8771016 let checkEmpty = liftIO $ do
@@ -888,6 +1027,7 @@ getAheadSVar style = do
8881027 , runqueue = runqueueAhead sv q
8891028 , enqueue = undefined
8901029 , queueEmpty = checkEmpty
1030+ , waitingForWork = wfw
8911031 , svarStyle = style
8921032 , activeWorkers = active
8931033 }
@@ -921,7 +1061,7 @@ newStreamVarAhead m = do
9211061 -- Note: We must have all the work on the queue before sending the
9221062 -- pushworker, otherwise the pushworker may exit before we even get a
9231063 -- chance to push.
924− liftIO $ enqueueAhead (workQueue sv) m
1064+ liftIO $ enqueueAhead sv (workQueue sv) m
9251065 pushWorker sv
9261066 return sv
9271067
@@ -1033,7 +1173,7 @@ ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
10331173ahead m1 m2 = Stream $ \svr stp sng yld -> do
10341174 case svr of
10351175 Just sv | svarStyle sv == AheadVar -> do
1036− liftIO $ enqueueAhead (workQueue sv) m2
1176+ liftIO $ enqueueAhead sv (workQueue sv) m2
10371177 -- Always run the left side on a new SVar to avoid complexity in
10381178 -- sequencing results. This means the left side cannot further
10391179 -- split into more ahead computations on the same SVar.
@@ -1043,9 +1183,15 @@ ahead m1 m2 = Stream $ \svr stp sng yld -> do
10431183 (runStream (fromStreamVar sv)) Nothing stp sng yld
10441184 where
10451185 concurrently ma mb = Stream $ \svr stp sng yld -> do
1046− liftIO $ enqueueAhead (workQueue (fromJust svr)) mb
1186+ liftIO $ enqueueAhead (fromJust svr) (workQueue (fromJust svr)) mb
10471187 (runStream ma) Nothing stp sng yld
10481188
1189+-- | XXX we can implement it more efficienty by directly implementing instead
1190+-- of combining streams using ahead.
1191+{-# INLINE consMAhead #-}
1192+consMAhead :: MonadAsync m => m a -> Stream m a -> Stream m a
1193+consMAhead m r = once m `ahead` r
1194+
10491195------------------------------------------------------------------------------
10501196-- Semigroup and Monoid style compositions for parallel actions
10511197------------------------------------------------------------------------------
@@ -1054,14 +1200,32 @@ ahead m1 m2 = Stream $ \svr stp sng yld -> do
10541200async :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
10551201async = joinStreamVarAsync AsyncVar
10561202
1203+-- | XXX we can implement it more efficienty by directly implementing instead
1204+-- of combining streams using async.
1205+{-# INLINE consMAsync #-}
1206+consMAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
1207+consMAsync m r = once m `async` r
1208+
10571209{-# INLINE wAsync #-}
10581210wAsync :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
10591211wAsync = joinStreamVarAsync WAsyncVar
10601212
1213+-- | XXX we can implement it more efficienty by directly implementing instead
1214+-- of combining streams using wAsync.
1215+{-# INLINE consMWAsync #-}
1216+consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
1217+consMWAsync m r = once m `wAsync` r
1218+
10611219{-# INLINE parallel #-}
10621220parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
10631221parallel = joinStreamVarPar ParallelVar
10641222
1223+-- | XXX we can implement it more efficienty by directly implementing instead
1224+-- of combining streams using parallel.
1225+{-# INLINE consMParallel #-}
1226+consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a
1227+consMParallel m r = once m `parallel` r
1228+
10651229-------------------------------------------------------------------------------
10661230-- Functor instace is the same for all types
10671231-------------------------------------------------------------------------------
@@ -1076,8 +1240,8 @@ instance Monad m => Functor (Stream m) where
10761240-- Alternative & MonadPlus
10771241------------------------------------------------------------------------------
10781242
1079−alt :: Stream m a -> Stream m a -> Stream m a
1080−alt m1 m2 = Stream $ \_ stp sng yld ->
1243+_alt :: Stream m a -> Stream m a -> Stream m a
1244+_alt m1 m2 = Stream $ \_ stp sng yld ->
10811245 let stop = runStream m2 Nothing stp sng yld
10821246 in runStream m1 Nothing stop sng yld
10831247