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
246type MonadParallel m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
248{-# DEPRECATED MonadAsync "Please use MonadParallel instead." #-}
249type MonadAsync m = MonadParallel m
252doFork :: MonadBaseControl IO m
254 -> (SomeException -> m ())
256doFork action exHandler =
257 EL.mask $ \restore ->
258 liftBaseWith $ \runInIO -> forkIO $ do
259 -- XXX test the exception handling
260 _ <- runInIO $ EL.catch (restore action) exHandler
261 -- XXX restore state here?
264-- XXX exception safety of all atomic/MVar operations
267send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
268send sv msg = liftIO $ do
269 atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
270 -- XXX need a memory barrier? The wake up must happen only after the
271 -- store has finished otherwise we can have lost wakeup problems.
272 void $ tryPutMVar (doorBell sv) True
274{-# INLINE sendStop #-}
275sendStop :: MonadIO m => SVar m a -> m ()
276sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
278-- Note: Left associated compositions can grow this queue to a large size
279{-# INLINE enqueueLIFO #-}
280enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
281enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
283runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
284runqueueLIFO sv q = run
291 Nothing -> sendStop sv
292 Just m -> (runStream m) (Just sv) run single yield
294 sendit a = send sv (ChildYield a)
295 single a = sendit a >> run
296 yield a r = sendit a >> (runStream r) (Just sv) run single yield
298 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
300 x : xs -> (xs, Just x)
302{-# INLINE enqueueFIFO #-}
303enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
306runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
307runqueueFIFO sv q = run
314 Nothing -> sendStop sv
315 Just m -> (runStream m) (Just sv) run single yield
317 dequeue = liftIO $ tryPopR q
318 sendit a = send sv (ChildYield a)
319 single a = sendit a >> run
320 yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run
322-- Thread tracking is needed for two reasons:
324-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
325-- themselves because they may run for significant times before going away or
326-- worse they may be stuck in IO and never go away.
328-- 2) To know when all threads are done.
330{-# NOINLINE addThread #-}
331addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
333 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
335{-# INLINE delThread #-}
336delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
338 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
340{-# INLINE allThreadsDone #-}
341allThreadsDone :: MonadIO m => SVar m a -> m Bool
342allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
344{-# NOINLINE handleChildException #-}
345handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
346handleChildException sv e = do
347 tid <- liftIO myThreadId
348 send sv (ChildStop tid (Just e))
350{-# NOINLINE pushWorker #-}
351pushWorker :: MonadParallel m => SVar m a -> m ()
353 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
355-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
356-- Also, if a worker blocks on the output queue we can decide if we want to
357-- block or make it go away entirely, depending on the number of workers and
358-- the type of the queue.
359{-# INLINE sendWorkerWait #-}
360sendWorkerWait :: MonadParallel m => SVar m a -> m ()
361sendWorkerWait sv = do
363 SVarStyle _ LIFO -> liftIO $ threadDelay 200
364 SVarStyle _ FIFO -> liftIO $ threadDelay 0
366 output <- liftIO $ readIORef (outputQueue sv)
367 when (null output) $ do
368 done <- queueEmpty sv
370 then pushWorker sv >> sendWorkerWait sv
371 else void (liftIO $ takeMVar (doorBell sv))
373-- | Pull a stream from an SVar.
374{-# NOINLINE fromStreamVar #-}
375fromStreamVar :: MonadParallel m => SVar m a -> Stream m a
376fromStreamVar sv = Stream $ \_ stp sng yld -> do
377 -- XXX if reading the IORef is costly we can use a flag in the SVar to
378 -- indicate we are done.
379 done <- allThreadsDone sv
383 res <- liftIO $ tryTakeMVar (doorBell sv)
384 when (isNothing res) $ sendWorkerWait sv
385 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
386 -- To avoid lock overhead we read all events at once instead of reading
387 -- one at a time. We just reverse the list to process the events in the
388 -- order they arrived. Maybe we can use a queue instead?
389 (runStream $ processEvents (reverse list)) Nothing stp sng yld
393 handleException e tid = do
395 -- XXX implement kill async exception handling
396 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
399 {-# INLINE processEvents #-}
400 processEvents [] = Stream $ \_ stp sng yld -> do
401 done <- allThreadsDone sv
403 then (runStream (fromStreamVar sv)) Nothing stp sng yld
406 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
407 let continue = (runStream (processEvents es)) Nothing stp sng yld
408 yield a = yld a (processEvents es)
411 ChildYield a -> yield a
414 Nothing -> delThread sv tid >> continue
415 Just ex -> handleException ex tid
417getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
418getFifoSVar ctype = do
420 outQMv <- newEmptyMVar
421 running <- newIORef S.empty
424 SVar { outputQueue = outQ
426 , runningThreads = running
427 , runqueue = runqueueFIFO sv q
429 , queueEmpty = liftIO $ nullQ q
434getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
435getLifoSVar ctype = do
437 outQMv <- newEmptyMVar
438 running <- newIORef S.empty
440 let checkEmpty = null <$> liftIO (readIORef q)
442 SVar { outputQueue = outQ
444 , runningThreads = running
445 , runqueue = runqueueLIFO sv q
446 , enqueue = enqueueLIFO q
447 , queueEmpty = checkEmpty
452-- | Create a new empty SVar.
453newEmptySVar :: MonadParallel m => SVarStyle -> m (SVar m a)
454newEmptySVar style = do
457 SVarStyle _ FIFO -> getFifoSVar style
458 SVarStyle _ LIFO -> getLifoSVar style
460-- | Create a new SVar and enqueue one stream computation on it.
461newStreamVar1 :: MonadParallel m => SVarStyle -> Stream m a -> m (SVar m a)
462newStreamVar1 style m = do
463 sv <- newEmptySVar style
464 -- Note: We must have all the work on the queue before sending the
465 -- pushworker, otherwise the pushworker may exit before we even get a
467 liftIO $ (enqueue sv) m
471-- | Create a new SVar and enqueue two stream computations on it.
472newStreamVar2 :: MonadParallel m
473 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
474newStreamVar2 style m1 m2 = do
475 -- Note: We must have all the work on the queue before sending the
476 -- pushworker, otherwise the pushworker may exit before we even get a
480 SVarStyle _ FIFO -> do
481 c <- getFifoSVar style
482 (enqueue c) m1 >> (enqueue c) m2
484 SVarStyle _ LIFO -> do
485 c <- getLifoSVar style
486 (enqueue c) m2 >> (enqueue c) m1
491-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
492-- be read back from the SVar using 'fromSVar'.
493toStreamVar :: MonadParallel m => SVar m a -> Stream m a -> m ()
495 liftIO $ (enqueue sv) m
496 done <- allThreadsDone sv
497 -- XXX there may be a race here unless we are running in the consumer
498 -- thread. This is safe only when called from the consumer thread or when
499 -- no consumer is present.
500 when done $ pushWorker sv
502------------------------------------------------------------------------------
503-- Running streams concurrently
504------------------------------------------------------------------------------
506-- Concurrency rate control. Our objective is to create more threads on demand
507-- if the consumer is running faster than us. As soon as we encounter an
508-- Alternative composition we create a push pull pair of threads. We use a
509-- channel for communication between the consumer pulling from the channel and
510-- the producer who is pushing to the channel. The producer creates more threads
511-- if no output is seen on the channel, that is the consumer is running faster.
512-- However this mechanism can be problematic if the initial production latency
513-- is high, we may end up creating too many threads. So we need some way to
514-- monitor and use the latency as well.
516-- TBD We may run computations at the lower level of the composition tree
517-- serially even if they are composed using a parallel combinator. We can use
518-- <> in place of <| and <=> in place of <|>. If we find that a parallel
519-- channel immediately above a computation becomes empty we can switch to
520-- parallelizing the computation. For that we can use a state flag to fork the
521-- rest of the computation at any point of time inside the Monad bind operation
522-- if the consumer is running at a faster speed.
524-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
525-- If we have to dispatch in arbitrary chunksizes we will need to compose the
526-- parallel actions using a data constructor (Free Alternative) instead so that
527-- we can divide it in chunks of arbitrary size before dispatching. If the stream
528-- is composed of hierarchically composed grains of different sizes then we can
529-- always switch to a desired granularity depending on the consumer speed.
531-- TBD for pure work (when we are not in the IO monad) we can divide it into
532-- just the number of CPUs.
534{-# NOINLINE withNewSVar2 #-}
535withNewSVar2 :: MonadParallel m
536 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
537withNewSVar2 style m1 m2 = Stream $ \_ stp sng yld -> do
538 sv <- newStreamVar2 style m1 m2
539 (runStream (fromStreamVar sv)) Nothing stp sng yld
541-- | Join two computations on the currently running 'SVar' queue for concurrent
542-- execution. The 'SVarStyle' required by the current composition context is
543-- passed as one of the parameters. If the style does not match with the style
544-- of the current 'SVar' we create a new 'SVar' and schedule the computations
545-- on that. The newly created SVar joins as one of the computations on the
546-- current SVar queue.
548-- When we are using parallel composition, an SVar is passed around as a state
549-- variable. We try to schedule a new parallel computation on the SVar passed
550-- to us. The first time, when no SVar exists, a new SVar is created.
551-- Subsequently, 'joinStreamVar2' may get called when a computation already
552-- scheduled on the SVar is further evaluated. For example, when (a \<|> b) is
553-- evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on the current scheduler
554-- queue. However, if the scheduling and composition style of the new
555-- computation being scheduled is different than the style of the current SVar,
556-- then we create a new SVar and schedule it on that.
560-- * (x \<|> y) \<|> (t \<|> u) -- all of them get scheduled on the same SVar
561-- * (x \<|> y) \<|> (t \<| u) -- @t@ and @u@ get scheduled on a new child SVar
562-- because of the scheduling policy change.
563-- * if we 'adapt' a stream of type 'AsyncT' to a stream of type
564-- 'ParallelT', we create a new SVar at the transitioning bind.
565-- * When the stream is switching from disjunctive composition to conjunctive
566-- composition and vice-versa we create a new SVar to isolate the scheduling
569{-# INLINE joinStreamVar2 #-}
570joinStreamVar2 :: MonadParallel m
571 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
572joinStreamVar2 style m1 m2 = Stream $ \svr stp sng yld ->
574 Just sv | svarStyle sv == style ->
575 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
576 _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp sng yld
578------------------------------------------------------------------------------
579-- Semigroup and Monoid style compositions for parallel actions
580------------------------------------------------------------------------------
582{-# INLINE coparallel #-}
583coparallel :: MonadParallel m => Stream m a -> Stream m a -> Stream m a
584coparallel = joinStreamVar2 (SVarStyle Disjunction LIFO)
586{-# INLINE parallel #-}
587parallel :: MonadParallel m => Stream m a -> Stream m a -> Stream m a
588parallel = joinStreamVar2 (SVarStyle Disjunction FIFO)
590-------------------------------------------------------------------------------
591-- Functor instace is the same for all types
592-------------------------------------------------------------------------------
594instance Monad m => Functor (Stream m) where
595 fmap f m = Stream $ \_ stp sng yld ->
597 yield a r = yld (f a) (fmap f r)
598 in (runStream m) Nothing stp single yield
600------------------------------------------------------------------------------
601-- Alternative & MonadPlus
602------------------------------------------------------------------------------
604alt :: Stream m a -> Stream m a -> Stream m a
605alt m1 m2 = Stream $ \_ stp sng yld ->
606 let stop = runStream m2 Nothing stp sng yld
607 in runStream m1 Nothing stop sng yld
609------------------------------------------------------------------------------
611------------------------------------------------------------------------------
613zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
614zipWith f m1 m2 = go m1 m2
616 go mx my = Stream $ \_ stp sng yld -> do
618 let single2 b = sng (f a b)
619 yield2 b rb = yld (f a b) (go ra rb)
620 in (runStream my) Nothing stp single2 yield2
621 let single1 a = merge a nil
622 yield1 a ra = merge a ra
623 (runStream mx) Nothing stp single1 yield1
625async :: MonadParallel m => Stream m a -> m (Stream m a)
626async m = newStreamVar1 (SVarStyle Disjunction LIFO) m
627 >>= return . fromStreamVar
629zipParallelWith :: MonadParallel m
630 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
631zipParallelWith f m1 m2 = Stream $ \_ stp sng yld -> do
634 (runStream (zipWith f ma mb)) Nothing stp sng yld
636-------------------------------------------------------------------------------
638-------------------------------------------------------------------------------
640instance MonadTrans Stream where
643withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
645 Stream $ \svr stp sng yld ->
646 let single = local f . sng
647 yield a r = local f $ yld a (withLocal f r)
648 in (runStream m) svr (local f stp) single yield
650-- XXX handle and test cross thread state transfer
653 => Stream m a -> (e -> Stream m a) -> Stream m a
655 Stream $ \svr stp sng yld ->
656 let run x = runStream x svr stp sng yield
657 handle r = r `catchError` \e -> run $ h e
658 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