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
48 -- * Concurrent Stream Vars (SVars)
62import Control.Concurrent (ThreadId, forkIO,
63 myThreadId, threadDelay)
64import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
66import Control.Exception (SomeException (..))
67import qualified Control.Exception.Lifted as EL
68import Control.Monad (when)
69import Control.Monad.Catch (MonadThrow, throwM)
70import Control.Monad.Error.Class (MonadError(..))
71import Control.Monad.IO.Class (MonadIO(..))
72import Control.Monad.Reader.Class (MonadReader(..))
73import Control.Monad.Trans.Class (MonadTrans (lift))
74import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
75import Data.Atomics (atomicModifyIORefCAS,
76 atomicModifyIORefCAS_)
77import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
79import Data.Functor (void)
80import Data.IORef (IORef, modifyIORef, newIORef,
82import Data.Maybe (isNothing)
83import Data.Semigroup (Semigroup(..))
85import qualified Data.Set as S
86import Prelude hiding (repeat)
88------------------------------------------------------------------------------
89-- Parent child thread communication type
90------------------------------------------------------------------------------
92-- | Events that a child thread may send to a parent thread.
95 | ChildStop ThreadId (Maybe SomeException)
97------------------------------------------------------------------------------
98-- State threaded around the monad for thread management
99------------------------------------------------------------------------------
101-- | Conjunction is used for monadic/product style composition. Disjunction is
102-- used for fold/sum style composition. We need to distinguish the two types of
103-- SVars so that the scheduling of the two is independent.
104data SVarTag = Conjunction | Disjunction deriving Eq
106-- | For fairly interleaved parallel composition the sched policy is FIFO
107-- whereas for left biased parallel composition it is LIFO.
108data SVarSched = LIFO | FIFO deriving Eq
110-- | Identify the type of the SVar. Two computations using the same style can
111-- be scheduled on the same SVar.
112data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
114-- | An SVar or a Stream Var is a conduit to the output from multiple streams
115-- running concurrently and asynchronously. An SVar can be thought of as an
116-- asynchronous IO handle. We can write any number of streams to an SVar in a
117-- non-blocking manner and then read them back at any time at any pace. The
118-- SVar would run the streams asynchronously and accumulate results. An SVar
119-- may not really execute the stream completely and accumulate all the results.
120-- However, it ensures that the reader can read the results at whatever paces
121-- it wants to read. The SVar monitors and adapts to the consumer's pace.
123-- An SVar is a mini scheduler, it has an associated runqueue that holds the
124-- stream tasks to be picked and run by a pool of worker threads. It has an
125-- associated output queue where the output stream elements are placed by the
126-- worker threads. A doorBell is used by the worker threads to intimate the
127-- consumer thread about availability of new results in the output queue. More
128-- workers are added to the SVar by 'fromStreamVar' on demand if the output
129-- produced is not keeping pace with the consumer. On bounded SVars, workers
130-- block on the output queue to provide throttling of the producer when the
131-- consumer is not pulling fast enough. The number of workers may even get
132-- reduced depending on the consuming pace.
134-- New work is enqueued either at the time of creation of the SVar or as a
135-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
136-- already enqueued computations get evaluated. See 'joinStreamVar2'.
139 SVar { outputQueue :: IORef [ChildEvent a]
140 , doorBell :: MVar Bool -- wakeup mechanism for outQ
141 , enqueue :: Stream m a -> IO ()
143 , runningThreads :: IORef (Set ThreadId)
144 , queueEmpty :: m Bool
145 , svarStyle :: SVarStyle
148------------------------------------------------------------------------------
150------------------------------------------------------------------------------
152-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
153-- constructed using actions in monad 'm'. It uses stop, singleton and yield
154-- continuations equivalent to the following direct style type:
156-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
158-- To facilitate parallel composition we maintain a local state in an SVar that
159-- is shared across and is used for synchronization of the streams being
162-- The singleton case can be expressed in terms of stop and yield but we have
163-- it as a separate case to optimize composition operations for streams with
164-- single element. We build singleton streams in the implementation of 'pure'
165-- for Applicative and Monad, and in 'lift' for MonadTrans.
169 runStream :: forall r.
170 Maybe (SVar m a) -- local state
172 -> (a -> m r) -- singleton
173 -> (a -> Stream m a -> m r) -- yield
178nil = Stream $ \_ stp _ _ -> stp
180once :: Monad m => m a -> Stream m a
181once m = Stream $ \_ _ single _ -> m >>= single
183{-# INLINE singleton #-}
184-- | Same as @once . return@
185singleton :: a -> Stream m a
186singleton a = Stream $ \_ _ single _ -> single a
188consM :: Monad m => m a -> Stream m a -> Stream m a
189consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
191-- | Same as @consM . return@
192cons :: a -> Stream m a -> Stream m a
193cons a r = Stream $ \_ _ _ yld -> yld a r
195repeat :: a -> Stream m a
196repeat a = let x = cons a x in x
198------------------------------------------------------------------------------
200------------------------------------------------------------------------------
202-- | Concatenates two streams sequentially i.e. the first stream is
203-- exhausted completely before yielding any element from the second stream.
204splice :: Stream m a -> Stream m a -> Stream m a
207 go (Stream m) = Stream $ \_ stp sng yld ->
208 let stop = (runStream m2) Nothing stp sng yld
210 yield a r = yld a (go r)
211 in m Nothing stop single yield
213instance Semigroup (Stream m a) where
216------------------------------------------------------------------------------
218------------------------------------------------------------------------------
220instance Monoid (Stream m a) where
224------------------------------------------------------------------------------
226------------------------------------------------------------------------------
228cosplice :: Stream m a -> Stream m a -> Stream m a
229cosplice m1 m2 = Stream $ \_ stp sng yld -> do
230 let stop = (runStream m2) Nothing stp sng yld
232 yield a r = yld a (cosplice m2 r)
233 (runStream m1) Nothing stop single yield
235------------------------------------------------------------------------------
236-- Spawning threads and collecting result in streamed fashion
237------------------------------------------------------------------------------
239-- | A monad that can perform concurrent or parallel IO operations. Streams
240-- that can be composed concurrently require the underlying monad to be
242type MonadParallel m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
244{-# DEPRECATED MonadAsync "Please use MonadParallel instead." #-}
245type MonadAsync m = MonadParallel m
248doFork :: MonadBaseControl IO m
250 -> (SomeException -> m ())
252doFork action exHandler =
253 EL.mask $ \restore ->
254 liftBaseWith $ \runInIO -> forkIO $ do
255 -- XXX test the exception handling
256 _ <- runInIO $ EL.catch (restore action) exHandler
257 -- XXX restore state here?
260-- XXX exception safety of all atomic/MVar operations
263send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
264send sv msg = liftIO $ do
265 atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
266 -- XXX need a memory barrier? The wake up must happen only after the
267 -- store has finished otherwise we can have lost wakeup problems.
268 void $ tryPutMVar (doorBell sv) True
270{-# INLINE sendStop #-}
271sendStop :: MonadIO m => SVar m a -> m ()
272sendStop sv = liftIO myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
274-- Note: Left associated compositions can grow this queue to a large size
275{-# INLINE enqueueLIFO #-}
276enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
277enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
279runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
280runqueueLIFO sv q = run
287 Nothing -> sendStop sv
288 Just m -> (runStream m) (Just sv) run single yield
290 sendit a = send sv (ChildYield a)
291 single a = sendit a >> run
292 yield a r = sendit a >> (runStream r) (Just sv) run single yield
294 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
296 x : xs -> (xs, Just x)
298{-# INLINE enqueueFIFO #-}
299enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
302runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
303runqueueFIFO sv q = run
310 Nothing -> sendStop sv
311 Just m -> (runStream m) (Just sv) run single yield
313 dequeue = liftIO $ tryPopR q
314 sendit a = send sv (ChildYield a)
315 single a = sendit a >> run
316 yield a r = sendit a >> liftIO (enqueueFIFO q r) >> run
318-- Thread tracking is needed for two reasons:
320-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
321-- themselves because they may run for significant times before going away or
322-- worse they may be stuck in IO and never go away.
324-- 2) To know when all threads are done.
326{-# NOINLINE addThread #-}
327addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
329 liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
331{-# INLINE delThread #-}
332delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
334 liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
336{-# INLINE allThreadsDone #-}
337allThreadsDone :: MonadIO m => SVar m a -> m Bool
338allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
340{-# NOINLINE handleChildException #-}
341handleChildException :: MonadIO m => SVar m a -> SomeException -> m ()
342handleChildException sv e = do
343 tid <- liftIO myThreadId
344 send sv (ChildStop tid (Just e))
346{-# NOINLINE pushWorker #-}
347pushWorker :: MonadParallel m => SVar m a -> m ()
349 doFork (runqueue sv) (handleChildException sv) >>= addThread sv
351-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
352-- Also, if a worker blocks on the output queue we can decide if we want to
353-- block or make it go away entirely, depending on the number of workers and
354-- the type of the queue.
355{-# INLINE sendWorkerWait #-}
356sendWorkerWait :: MonadParallel m => SVar m a -> m ()
357sendWorkerWait sv = do
359 SVarStyle _ LIFO -> liftIO $ threadDelay 200
360 SVarStyle _ FIFO -> liftIO $ threadDelay 0
362 output <- liftIO $ readIORef (outputQueue sv)
363 when (null output) $ do
364 done <- queueEmpty sv
366 then pushWorker sv >> sendWorkerWait sv
367 else void (liftIO $ takeMVar (doorBell sv))
369-- | Pull a stream from an SVar.
370{-# NOINLINE fromStreamVar #-}
371fromStreamVar :: MonadParallel m => SVar m a -> Stream m a
372fromStreamVar sv = Stream $ \_ stp sng yld -> do
373 -- XXX if reading the IORef is costly we can use a flag in the SVar to
374 -- indicate we are done.
375 done <- allThreadsDone sv
379 res <- liftIO $ tryTakeMVar (doorBell sv)
380 when (isNothing res) $ sendWorkerWait sv
381 list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
382 -- To avoid lock overhead we read all events at once instead of reading
383 -- one at a time. We just reverse the list to process the events in the
384 -- order they arrived. Maybe we can use a queue instead?
385 (runStream $ processEvents (reverse list)) Nothing stp sng yld
389 handleException e tid = do
391 -- XXX implement kill async exception handling
392 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
395 {-# INLINE processEvents #-}
396 processEvents [] = Stream $ \_ stp sng yld -> do
397 done <- allThreadsDone sv
399 then (runStream (fromStreamVar sv)) Nothing stp sng yld
402 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
403 let continue = (runStream (processEvents es)) Nothing stp sng yld
404 yield a = yld a (processEvents es)
407 ChildYield a -> yield a
410 Nothing -> delThread sv tid >> continue
411 Just ex -> handleException ex tid
413getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
414getFifoSVar ctype = do
416 outQMv <- newEmptyMVar
417 running <- newIORef S.empty
420 SVar { outputQueue = outQ
422 , runningThreads = running
423 , runqueue = runqueueFIFO sv q
425 , queueEmpty = liftIO $ nullQ q
430getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
431getLifoSVar ctype = do
433 outQMv <- newEmptyMVar
434 running <- newIORef S.empty
436 let checkEmpty = null <$> liftIO (readIORef q)
438 SVar { outputQueue = outQ
440 , runningThreads = running
441 , runqueue = runqueueLIFO sv q
442 , enqueue = enqueueLIFO q
443 , queueEmpty = checkEmpty
448-- | Create a new empty SVar.
449newEmptySVar :: MonadParallel m => SVarStyle -> m (SVar m a)
450newEmptySVar style = do
453 SVarStyle _ FIFO -> getFifoSVar style
454 SVarStyle _ LIFO -> getLifoSVar style
456-- | Create a new SVar and enqueue one stream computation on it.
457newStreamVar1 :: MonadParallel m => SVarStyle -> Stream m a -> m (SVar m a)
458newStreamVar1 style m = do
459 sv <- newEmptySVar style
460 -- Note: We must have all the work on the queue before sending the
461 -- pushworker, otherwise the pushworker may exit before we even get a
463 liftIO $ (enqueue sv) m
467-- | Create a new SVar and enqueue two stream computations on it.
468newStreamVar2 :: MonadParallel m
469 => SVarStyle -> Stream m a -> Stream m a -> m (SVar m a)
470newStreamVar2 style m1 m2 = do
471 -- Note: We must have all the work on the queue before sending the
472 -- pushworker, otherwise the pushworker may exit before we even get a
476 SVarStyle _ FIFO -> do
477 c <- getFifoSVar style
478 (enqueue c) m1 >> (enqueue c) m2
480 SVarStyle _ LIFO -> do
481 c <- getLifoSVar style
482 (enqueue c) m2 >> (enqueue c) m1
487-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
488-- be read back from the SVar using 'fromSVar'.
489toStreamVar :: MonadParallel m => SVar m a -> Stream m a -> m ()
491 liftIO $ (enqueue sv) m
492 done <- allThreadsDone sv
493 -- XXX there may be a race here unless we are running in the consumer
494 -- thread. This is safe only when called from the consumer thread or when
495 -- no consumer is present.
496 when done $ pushWorker sv
498------------------------------------------------------------------------------
499-- Running streams concurrently
500------------------------------------------------------------------------------
502-- Concurrency rate control. Our objective is to create more threads on demand
503-- if the consumer is running faster than us. As soon as we encounter an
504-- Alternative composition we create a push pull pair of threads. We use a
505-- channel for communication between the consumer pulling from the channel and
506-- the producer who is pushing to the channel. The producer creates more threads
507-- if no output is seen on the channel, that is the consumer is running faster.
508-- However this mechanism can be problematic if the initial production latency
509-- is high, we may end up creating too many threads. So we need some way to
510-- monitor and use the latency as well.
512-- TBD We may run computations at the lower level of the composition tree
513-- serially even if they are composed using a parallel combinator. We can use
514-- <> in place of <| and <=> in place of <|>. If we find that a parallel
515-- channel immediately above a computation becomes empty we can switch to
516-- parallelizing the computation. For that we can use a state flag to fork the
517-- rest of the computation at any point of time inside the Monad bind operation
518-- if the consumer is running at a faster speed.
520-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
521-- If we have to dispatch in arbitrary chunksizes we will need to compose the
522-- parallel actions using a data constructor (Free Alternative) instead so that
523-- we can divide it in chunks of arbitrary size before dispatching. If the stream
524-- is composed of hierarchically composed grains of different sizes then we can
525-- always switch to a desired granularity depending on the consumer speed.
527-- TBD for pure work (when we are not in the IO monad) we can divide it into
528-- just the number of CPUs.
530{-# NOINLINE withNewSVar2 #-}
531withNewSVar2 :: MonadParallel m
532 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
533withNewSVar2 style m1 m2 = Stream $ \_ stp sng yld -> do
534 sv <- newStreamVar2 style m1 m2
535 (runStream (fromStreamVar sv)) Nothing stp sng yld
537-- | Join two computations on the currently running 'SVar' queue for concurrent
538-- execution. The 'SVarStyle' required by the current composition context is
539-- passed as one of the parameters. If the style does not match with the style
540-- of the current 'SVar' we create a new 'SVar' and schedule the computations
541-- on that. The newly created SVar joins as one of the computations on the
542-- current SVar queue.
544-- When we are using parallel composition, an SVar is passed around as a state
545-- variable. We try to schedule a new parallel computation on the SVar passed
546-- to us. The first time, when no SVar exists, a new SVar is created.
547-- Subsequently, 'joinStreamVar2' may get called when a computation already
548-- scheduled on the SVar is further evaluated. For example, when (a \<|> b) is
549-- evaluated it calls a 'joinStreamVar2' to put 'a' and 'b' on the current scheduler
550-- queue. However, if the scheduling and composition style of the new
551-- computation being scheduled is different than the style of the current SVar,
552-- then we create a new SVar and schedule it on that.
556-- * (x \<|> y) \<|> (t \<|> u) -- all of them get scheduled on the same SVar
557-- * (x \<|> y) \<|> (t \<| u) -- @t@ and @u@ get scheduled on a new child SVar
558-- because of the scheduling policy change.
559-- * if we 'adapt' a stream of type 'AsyncT' to a stream of type
560-- 'ParallelT', we create a new SVar at the transitioning bind.
561-- * When the stream is switching from disjunctive composition to conjunctive
562-- composition and vice-versa we create a new SVar to isolate the scheduling
565{-# INLINE joinStreamVar2 #-}
566joinStreamVar2 :: MonadParallel m
567 => SVarStyle -> Stream m a -> Stream m a -> Stream m a
568joinStreamVar2 style m1 m2 = Stream $ \svr stp sng yld ->
570 Just sv | svarStyle sv == style ->
571 liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
572 _ -> (runStream (withNewSVar2 style m1 m2)) Nothing stp sng yld
574------------------------------------------------------------------------------
575-- Semigroup and Monoid style compositions for parallel actions
576------------------------------------------------------------------------------
578{-# INLINE coparallel #-}
579coparallel :: MonadParallel m => Stream m a -> Stream m a -> Stream m a
580coparallel = joinStreamVar2 (SVarStyle Disjunction LIFO)
582{-# INLINE parallel #-}
583parallel :: MonadParallel m => Stream m a -> Stream m a -> Stream m a
584parallel = joinStreamVar2 (SVarStyle Disjunction FIFO)
586-------------------------------------------------------------------------------
587-- Functor instace is the same for all types
588-------------------------------------------------------------------------------
590instance Monad m => Functor (Stream m) where
591 fmap f m = Stream $ \_ stp sng yld ->
593 yield a r = yld (f a) (fmap f r)
594 in (runStream m) Nothing stp single yield
596------------------------------------------------------------------------------
597-- Alternative & MonadPlus
598------------------------------------------------------------------------------
600alt :: Stream m a -> Stream m a -> Stream m a
601alt m1 m2 = Stream $ \_ stp sng yld ->
602 let stop = runStream m2 Nothing stp sng yld
603 in runStream m1 Nothing stop sng yld
605-------------------------------------------------------------------------------
607-------------------------------------------------------------------------------
609instance MonadTrans Stream where
612withLocal :: MonadReader r m => (r -> r) -> Stream m a -> Stream m a
614 Stream $ \svr stp sng yld ->
615 let single = local f . sng
616 yield a r = local f $ yld a (withLocal f r)
617 in (runStream m) svr (local f stp) single yield
619-- XXX handle and test cross thread state transfer
622 => Stream m a -> (e -> Stream m a) -> Stream m a
624 Stream $ \svr stp sng yld ->
625 let run x = runStream x svr stp sng yield
626 handle r = r `catchError` \e -> run $ h e
627 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