1{-# LANGUAGE ConstraintKinds #-}
2{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE FlexibleInstances #-}
4{-# LANGUAGE LambdaCase #-}
5{-# LANGUAGE MultiParamTypeClasses #-}
6{-# LANGUAGE RankNTypes #-}
7{-# LANGUAGE UndecidableInstances #-} -- XXX
10-- Module : Streamly.Core
11-- Copyright : (c) 2017 Harendra Kumar
14-- Maintainer : harendra.kumar@gmail.com
15-- Stability : experimental
22 , MonadAsync -- deprecated
35 -- * Semigroup Style Composition
52 -- * Concurrent Stream Vars (SVars)
66import Control.Concurrent (ThreadId, forkIO,
67 myThreadId, threadDelay)
68import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
70import Control.Exception (SomeException (..))
71import qualified Control.Exception.Lifted as EL
72import Control.Monad (when)
73import Control.Monad.Catch (MonadThrow, throwM)
74import Control.Monad.Error.Class (MonadError(..))
75import Control.Monad.IO.Class (MonadIO(..))
76import Control.Monad.Reader.Class (MonadReader(..))
77import Control.Monad.Trans.Class (MonadTrans (lift))
78import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
79import Data.Atomics (atomicModifyIORefCAS,
80 atomicModifyIORefCAS_)
81import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
83import Data.Functor (void)
84import Data.IORef (IORef, modifyIORef, newIORef,
86import Data.Maybe (isNothing)
87import Data.Semigroup (Semigroup(..))
89import qualified Data.Set as S
90import Prelude hiding (repeat, zipWith)
92------------------------------------------------------------------------------
93-- Parent child thread communication type
94------------------------------------------------------------------------------
96-- | Events that a child thread may send to a parent thread.
99 | ChildStop ThreadId (Maybe SomeException)
101------------------------------------------------------------------------------
102-- State threaded around the monad for thread management
103------------------------------------------------------------------------------
105-- | Conjunction is used for monadic/product style composition. Disjunction is
106-- used for fold/sum style composition. We need to distinguish the two types of
107-- SVars so that the scheduling of the two is independent.
108data SVarTag = Conjunction | Disjunction deriving Eq
110-- | For fairly interleaved parallel composition the sched policy is FIFO
111-- whereas for left biased parallel composition it is LIFO.
112data SVarSched = LIFO | FIFO deriving Eq
114-- | Identify the type of the SVar. Two computations using the same style can
115-- be scheduled on the same SVar.
116data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
118-- | An SVar or a Stream Var is a conduit to the output from multiple streams
119-- running concurrently and asynchronously. An SVar can be thought of as an
120-- asynchronous IO handle. We can write any number of streams to an SVar in a
121-- non-blocking manner and then read them back at any time at any pace. The
122-- SVar would run the streams asynchronously and accumulate results. An SVar
123-- may not really execute the stream completely and accumulate all the results.
124-- However, it ensures that the reader can read the results at whatever paces
125-- it wants to read. The SVar monitors and adapts to the consumer's pace.
127-- An SVar is a mini scheduler, it has an associated runqueue that holds the
128-- stream tasks to be picked and run by a pool of worker threads. It has an
129-- associated output queue where the output stream elements are placed by the
130-- worker threads. A doorBell is used by the worker threads to intimate the
131-- consumer thread about availability of new results in the output queue. More
132-- workers are added to the SVar by 'fromStreamVar' on demand if the output
133-- produced is not keeping pace with the consumer. On bounded SVars, workers
134-- block on the output queue to provide throttling of the producer when the
135-- consumer is not pulling fast enough. The number of workers may even get
136-- reduced depending on the consuming pace.
138-- New work is enqueued either at the time of creation of the SVar or as a
139-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
140-- already enqueued computations get evaluated. See 'joinStreamVar2'.
143 SVar { outputQueue :: IORef [ChildEvent a]
144 , doorBell :: MVar Bool -- wakeup mechanism for outQ
145 , enqueue :: Stream m a -> IO ()
147 , runningThreads :: IORef (Set ThreadId)
148 , queueEmpty :: m Bool
149 , svarStyle :: SVarStyle
152------------------------------------------------------------------------------
154------------------------------------------------------------------------------
156-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
157-- constructed using actions in monad 'm'. It uses stop, singleton and yield
158-- continuations equivalent to the following direct style type:
160-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
162-- To facilitate parallel composition we maintain a local state in an SVar that
163-- is shared across and is used for synchronization of the streams being
166-- The singleton case can be expressed in terms of stop and yield but we have
167-- it as a separate case to optimize composition operations for streams with
168-- single element. We build singleton streams in the implementation of 'pure'
169-- for Applicative and Monad, and in 'lift' for MonadTrans.
173 runStream :: forall r.
174 Maybe (SVar m a) -- local state
176 -> (a -> m r) -- singleton
177 -> (a -> Stream m a -> m r) -- yield
182nil = Stream $ \_ stp _ _ -> stp
184once :: Monad m => m a -> Stream m a
185once m = Stream $ \_ _ single _ -> m >>= single
187{-# INLINE singleton #-}
188-- | Same as @once . return@
189singleton :: a -> Stream m a
190singleton a = Stream $ \_ _ single _ -> single a
192consM :: Monad m => m a -> Stream m a -> Stream m a
193consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
195-- | Same as @consM . return@
196cons :: a -> Stream m a -> Stream m a
197cons a r = Stream $ \_ _ _ yld -> yld a r
199repeat :: a -> Stream m a
200repeat a = let x = cons a x in x
202------------------------------------------------------------------------------
204------------------------------------------------------------------------------
206-- | Concatenates two streams sequentially i.e. the first stream is
207-- exhausted completely before yielding any element from the second stream.
208splice :: Stream m a -> Stream m a -> Stream m a
211 go (Stream m) = Stream $ \_ stp sng yld ->
212 let stop = (runStream m2) Nothing stp sng yld
214 yield a r = yld a (go r)
215 in m Nothing stop single yield
217instance Semigroup (Stream m a) where
220------------------------------------------------------------------------------
222------------------------------------------------------------------------------
224instance Monoid (Stream m a) where
228------------------------------------------------------------------------------
230------------------------------------------------------------------------------
232cosplice :: Stream m a -> Stream m a -> Stream m a
233cosplice m1 m2 = Stream $ \_ stp sng yld -> do
234 let stop = (runStream m2) Nothing stp sng yld
236 yield a r = yld a (cosplice m2 r)
237 (runStream m1) Nothing stop single yield
239------------------------------------------------------------------------------
240-- Spawning threads and collecting result in streamed fashion
241------------------------------------------------------------------------------
243-- | A monad that can perform concurrent or parallel IO operations. Streams
244-- that can be composed concurrently require the underlying monad to be
248type MonadParallel m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
252{-# DEPRECATED MonadAsync "Please use MonadParallel instead." #-}
253type MonadAsync m = MonadParallel m
256doFork :: MonadBaseControl IO m
258 -> (SomeException -> m ())
260doFork action exHandler =
261 EL.mask $ \restore ->
262 liftBaseWith $ \runInIO -> forkIO $ do
263 _ <- runInIO $ EL.catch (restore action) exHandler
264 -- XXX restore state here?
267-- XXX exception safety of all atomic/MVar operations
270send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
271send sv msg = liftIO $ do
272 atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
273 -- XXX need a memory barrier? The wake up must happen only after the
274 -- store has finished otherwise we can have lost wakeup problems.
275 void $ tryPutMVar (doorBell sv) True
277{-# INLINE sendStop #-}
278sendStop :: MonadIO m => SVar m a -> m ()
279sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
281-- Note: Left associated compositions can grow this queue to a large size
282{-# INLINE enqueueLIFO #-}
283enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
284enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
286runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
287runqueueLIFO sv q = run
294 Nothing -> sendStop sv
295 Just m -> (runStream m) (Just sv) run single yield
297 sendit a = send sv (ChildYield a)
298 single a = sendit a >> run
299 yield a r = sendit a >> (runStream r) (Just sv) run single yield
301 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
303 x : xs -> (xs, Just x)
305{-# INLINE enqueueFIFO #-}
306enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
309runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
310runqueueFIFO sv q = run
317 Nothing -> sendStop sv
318 Just m -> (runStream m) (Just sv) run single yield
320 dequeue = liftIO $ tryPopR q
321 sendit a = send sv (ChildYield a)
322 single a = sendit a >> run
323 yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run
325-- Thread tracking is needed for two reasons:
327-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
328-- themselves because they may run for significant times before going away or
329-- worse they may be stuck in IO and never go away.
331-- 2) To know when all threads are done.
333{-# NOINLINE addThread #-}
334addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
336 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
338{-# INLINE delThread #-}
339delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
341 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
343{-# INLINE allThreadsDone #-}
344allThreadsDone :: MonadIO m => SVar m a -> m Bool
345allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
347{-# NOINLINE handleChildException #-}
348handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
349handleChildException sv e = do
350 tid <- liftIO myThreadId
351 send sv (ChildStop tid (Just e))
353{-# NOINLINE pushWorker #-}
354pushWorker :: MonadParallel m => SVar m a -> m ()
356 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
358-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
359-- Also, if a worker blocks on the output queue we can decide if we want to
360-- block or make it go away entirely, depending on the number of workers and
361-- the type of the queue.
362{-# INLINE sendWorkerWait #-}
363sendWorkerWait :: MonadParallel m => SVar m a -> m ()
364sendWorkerWait sv = do
366 SVarStyle _ LIFO -> liftIO $ threadDelay 200
367 SVarStyle _ FIFO -> liftIO $ threadDelay 0
369 output <- liftIO $ readIORef (outputQueue sv)
370 when (null output) $ do
371 done <- queueEmpty sv
373 then pushWorker sv >> sendWorkerWait sv
374 else void (liftIO $ takeMVar (doorBell sv))
376-- | Pull a stream from an SVar.
377{-# NOINLINE fromStreamVar #-}
378fromStreamVar :: MonadParallel m => SVar m a -> Stream m a
379fromStreamVar sv = Stream $ \_ stp sng yld -> do
380 -- XXX if reading the IORef is costly we can use a flag in the SVar to
381 -- indicate we are done.
382 done <- allThreadsDone sv
386 res <- liftIO $ tryTakeMVar (doorBell sv)
387 when (isNothing res) $ sendWorkerWait sv
388 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
389 -- To avoid lock overhead we read all events at once instead of reading
390 -- one at a time. We just reverse the list to process the events in the
391 -- order they arrived. Maybe we can use a queue instead?
392 (runStream $ processEvents (reverse list)) Nothing stp sng yld
396 handleException e tid = do
398 -- XXX implement kill async exception handling
399 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
402 {-# INLINE processEvents #-}
403 processEvents [] = Stream $ \_ stp sng yld -> do
404 done <- allThreadsDone sv
406 then (runStream (fromStreamVar sv)) Nothing stp sng yld
409 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
410 let continue = (runStream (processEvents es)) Nothing stp sng yld
411 yield a = yld a (processEvents es)
414 ChildYield a -> yield a
417 Nothing -> delThread sv tid >> continue
418 Just ex -> handleException ex tid
420getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
421getFifoSVar ctype = do
423 outQMv <- newEmptyMVar
424 running <- newIORef S.empty
427 SVar { outputQueue = outQ
429 , runningThreads = running
430 , runqueue = runqueueFIFO sv q
432 , queueEmpty = liftIO $ nullQ q
437getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
438getLifoSVar ctype = do
440 outQMv <- newEmptyMVar
441 running <- newIORef S.empty
443 let checkEmpty = null <$> liftIO (readIORef q)
445 SVar { outputQueue = outQ
447 , runningThreads = running
448 , runqueue = runqueueLIFO sv q
449 , enqueue = enqueueLIFO q
450 , queueEmpty = checkEmpty
455-- | Create a new empty SVar.
456newEmptySVar :: MonadParallel m => SVarStyle -> m (SVar m a)
457newEmptySVar style = do
460 SVarStyle _ FIFO -> getFifoSVar style
461 SVarStyle _ LIFO -> getLifoSVar style
463-- | Create a new SVar and enqueue one stream computation on it.
464newStreamVar1 :: MonadParallel m => SVarStyle -> Stream m a -> m (SVar m a)
465newStreamVar1 style m = do
466 sv <- newEmptySVar style
467 -- Note: We must have all the work on the queue before sending the
468 -- pushworker, otherwise the pushworker may exit before we even get a
470 liftIO $ (enqueue sv) m
474-- | Create a new SVar and enqueue two stream computations on it.
475newStreamVar2 :: MonadParallel m
476 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
477newStreamVar2 style m1 m2 = do
478 -- Note: We must have all the work on the queue before sending the
479 -- pushworker, otherwise the pushworker may exit before we even get a
483 SVarStyle _ FIFO -> do
484 c <- getFifoSVar style
485 (enqueue c) m1 >> (enqueue c) m2
487 SVarStyle _ LIFO -> do
488 c <- getLifoSVar style
489 (enqueue c) m2 >> (enqueue c) m1
494-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
495-- be read back from the SVar using 'fromSVar'.
496toStreamVar :: MonadParallel m => SVar m a -> Stream m a -> m ()
498 liftIO $ (enqueue sv) m
499 done <- allThreadsDone sv
500 -- XXX This is safe only when called from the consumer thread or when no
501 -- consumer is present. There may be a race if we are not running in the
503 when done $ pushWorker sv
505------------------------------------------------------------------------------
506-- Running streams concurrently
507------------------------------------------------------------------------------
509-- Concurrency rate control.
511-- Our objective is to create more threads on demand if the consumer is running
512-- faster than us. As soon as we encounter a concurrent composition we create a
513-- push pull pair of threads. We use an SVar for communication between the
514-- consumer, pulling from the SVar and the producer who is pushing to the SVar.
515-- The producer creates more threads if the SVar drains and becomes empty, that
516-- is the consumer is running faster.
518-- XXX Note 1: This mechanism can be problematic if the initial production
519-- latency is high, we may end up creating too many threads. So we need some
520-- way to monitor and use the latency as well. Having a limit on the dispatches
521-- (programmer controlled) may also help.
523-- TBD Note 2: We may want to run computations at the lower level of the
524-- composition tree serially even when they are composed using a parallel
525-- combinator. We can use 'splice' in place of 'coparallel' and 'cosplice' in
526-- place of 'parallel'. If we find that an SVar immediately above a computation
527-- gets drained empty we can switch to parallelizing the computation. For that
528-- we can use a state flag to fork the rest of the computation at any point of
529-- time inside the Monad bind operation if the consumer is running at a faster
532-- TBD Note 3: the binary operation ('parallel') composition allows us to
533-- dispatch a chunkSize of only 1. If we have to dispatch in arbitrary
534-- chunksizes we will need to compose the parallel actions using a data
535-- constructor (A Free container) instead so that we can divide it in chunks of
536-- arbitrary size before dispatching. If the stream is composed of
537-- hierarchically composed grains of different sizes then we can always switch
538-- to a desired granularity depending on the consumer speed.
540-- TBD Note 4: for pure work (when we are not in the IO monad) we can divide it
541-- into just the number of CPUs.
543{-# NOINLINE withNewSVar2 #-}
544withNewSVar2 :: MonadParallel m
545 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
546withNewSVar2 style m1 m2 = Stream $ \_ stp sng yld -> do
547 sv <- newStreamVar2 style m1 m2
548 (runStream (fromStreamVar sv)) Nothing stp sng yld
550-- | Join two computations on the currently running 'SVar' queue for concurrent
551-- execution. When we are using parallel composition, an SVar is passed around
552-- as a state variable. We try to schedule a new parallel computation on the
553-- SVar passed to us. The first time, when no SVar exists, a new SVar is
554-- created. Subsequently, 'joinStreamVar2' may get called when a computation
555-- already scheduled on the SVar is further evaluated. For example, when (a
556-- `parallel` b) is evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on
557-- the current scheduler queue.
559-- The 'SVarStyle' required by the current composition context is passed as one
560-- of the parameters. If the scheduling and composition style of the new
561-- computation being scheduled is different than the style of the current SVar,
562-- then we create a new SVar and schedule it on that. The newly created SVar
563-- joins as one of the computations on the current SVar queue.
565-- Cases when we need to switch to a new SVar:
567-- * (x `parallel` y) `parallel` (t `parallel` u) -- all of them get scheduled on the same SVar
568-- * (x `parallel` y) `parallel` (t `coparallel` u) -- @t@ and @u@ get scheduled on a new child SVar
569-- because of the scheduling policy change.
570-- * if we 'adapt' a stream of type 'Coparallel' to a stream of type
571-- 'Parallel', we create a new SVar at the transitioning bind.
572-- * When the stream is switching from disjunctive composition to conjunctive
573-- composition and vice-versa we create a new SVar to isolate the scheduling
576{-# INLINE joinStreamVar2 #-}
577joinStreamVar2 :: MonadParallel m
578 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
579joinStreamVar2 style m1 m2 = Stream $ \svr stp sng yld ->
581 Just sv | svarStyle sv == style ->
582 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
583 _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp sng yld
585------------------------------------------------------------------------------
586-- Semigroup and Monoid style compositions for parallel actions
587------------------------------------------------------------------------------
589{-# INLINE coparallel #-}
590coparallel :: MonadParallel m => Stream m a -> Stream m a -> Stream m a
591coparallel = joinStreamVar2 (SVarStyle Disjunction LIFO)
593{-# INLINE parallel #-}
594parallel :: MonadParallel m => Stream m a -> Stream m a -> Stream m a
595parallel = joinStreamVar2 (SVarStyle Disjunction FIFO)
597-------------------------------------------------------------------------------
598-- Functor instace is the same for all types
599-------------------------------------------------------------------------------
601instance Monad m => Functor (Stream m) where
602 fmap f m = Stream $ \_ stp sng yld ->
604 yield a r = yld (f a) (fmap f r)
605 in (runStream m) Nothing stp single yield
607------------------------------------------------------------------------------
608-- Alternative & MonadPlus
609------------------------------------------------------------------------------
611alt :: Stream m a -> Stream m a -> Stream m a
612alt m1 m2 = Stream $ \_ stp sng yld ->
613 let stop = runStream m2 Nothing stp sng yld
614 in runStream m1 Nothing stop sng yld
616------------------------------------------------------------------------------
618------------------------------------------------------------------------------
620zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
621zipWith f m1 m2 = go m1 m2
623 go mx my = Stream $ \_ stp sng yld -> do
625 let single2 b = sng (f a b)
626 yield2 b rb = yld (f a b) (go ra rb)
627 in (runStream my) Nothing stp single2 yield2
628 let single1 a = merge a nil
629 yield1 a ra = merge a ra
630 (runStream mx) Nothing stp single1 yield1
632async :: MonadParallel m => Stream m a -> m (Stream m a)
633async m = newStreamVar1 (SVarStyle Disjunction LIFO) m
634 >>= return . fromStreamVar
636zipParallelWith :: MonadParallel m
637 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
638zipParallelWith f m1 m2 = Stream $ \_ stp sng yld -> do
641 (runStream (zipWith f ma mb)) Nothing stp sng yld
643-------------------------------------------------------------------------------
645-------------------------------------------------------------------------------
647instance MonadTrans Stream where
650withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
652 Stream $ \svr stp sng yld ->
653 let single = local f . sng
654 yield a r = local f $ yld a (withLocal f r)
655 in (runStream m) svr (local f stp) single yield
657-- XXX handle and test cross thread state transfer
660 => Stream m a -> (e -> Stream m a) -> Stream m a
662 Stream $ \svr stp sng yld ->
663 let run x = runStream x svr stp sng yield
664 handle r = r `catchError` \e -> run $ h e
665 yield a r = yld a (withCatchError r h)
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list