git.iko.soy/streamly
on 7654d48 help sign in
streamly/src/Asyncly/Core.hs
702 lines 28 kB history raw
1{-# LANGUAGE ConstraintKinds #-}
2{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE FlexibleInstances #-}
4{-# LANGUAGE MultiParamTypeClasses #-}
5{-# LANGUAGE RankNTypes #-}
6{-# LANGUAGE UndecidableInstances #-} -- XXX
8-- |
9-- Module : Asyncly.Core
10-- Copyright : (c) 2017 Harendra Kumar
11--
12-- License : BSD3
13-- Maintainer : harendra.kumar@gmail.com
14-- Stability : experimental
15-- Portability : GHC
16--
17--
18module Asyncly.Core
19 ( Stream (..)
20 , MonadAsync
21 , Streaming (..)
23 -- * Construction
24 , streamBuild
25 , yielding
27 -- * Elimination
28 , streamFold
29 , runStreaming
31 -- * Composition
32 , interleave
33 , (<=>)
35 -- * Asynchronous Operations
36 , CtxType (..)
37 , SchedPolicy (..)
38 , Dimension (..)
39 , EndOfStream (..)
40 , parallel
41 , parAlt
42 , parLeft
43 , (<|)
44 , async
46 -- * Fold Utilities
47 , foldWith
48 , foldMapWith
49 , forEachWith
50 )
51where
53import Control.Applicative (Alternative (..))
54import Control.Concurrent (ThreadId, forkIO,
55 myThreadId, threadDelay)
56import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
57 tryPutMVar, takeMVar)
58import Control.Exception (SomeException (..), Exception)
59import qualified Control.Exception.Lifted as EL
60import Control.Monad (MonadPlus(..), mzero, when)
61import Control.Monad.Base (MonadBase (..), liftBaseDefault)
62import Control.Monad.Catch (MonadThrow, throwM)
63import Control.Monad.Error.Class (MonadError(..))
64import Control.Monad.IO.Class (MonadIO(..))
65import Control.Monad.Reader.Class (MonadReader(..))
66import Control.Monad.State.Class (MonadState(..))
67import Control.Monad.Trans.Class (MonadTrans (lift))
68import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
69import Data.Atomics (atomicModifyIORefCAS,
70 atomicModifyIORefCAS_)
71import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
72 tryPopR, nullQ)
73import Data.Functor (void)
74import Data.IORef (IORef, modifyIORef, newIORef,
75 readIORef)
76import Data.Maybe (isNothing)
77import Data.Semigroup (Semigroup(..))
78import Data.Set (Set)
79import qualified Data.Set as S
81------------------------------------------------------------------------------
82-- Parent child thread communication type
83------------------------------------------------------------------------------
85-- | Events that a child thread may send to a parent thread.
86data ChildEvent a =
87 ChildYield a
88 | ChildStop ThreadId (Maybe SomeException)
90------------------------------------------------------------------------------
91-- State threaded around the monad for thread management
92------------------------------------------------------------------------------
94-- | Conjunction is used for monadic/product style composition. Disjunction is
95-- used for fold/sum style composition. We need to distiguish the two types of
96-- contexts so that the scheduing of the two is independent.
97data Dimension = Conjunction | Disjunction deriving Eq
99-- | For fairly interleaved parallel composition the sched policy is FIFO
100-- whereas for left biased parallel composition it is LIFO.
101data SchedPolicy = LIFO | FIFO deriving Eq
103-- | Identify the type of the context used for parallel composition.
104data CtxType = CtxType Dimension SchedPolicy deriving Eq
106-- | A Context represents a parallel composition. It has a runqueue which holds
107-- the tasks that are to be picked by a pool of worker threads. It has an
108-- output queue where the results are placed by the worker threads. A
109-- doorBell is used by the worker threads to intimate the consumer thread
110-- about availability of new results. New work is usually enqueued as a result
111-- of executing the parallel combinators i.e. '<|' and '<|>'.
112data Context m a =
113 Context { outputQueue :: IORef [ChildEvent a]
114 , doorBell :: MVar Bool -- wakeup mechanism for outQ
115 , enqueue :: Stream m a -> IO ()
116 , runqueue :: m ()
117 , runningThreads :: IORef (Set ThreadId)
118 , queueEmpty :: m Bool
119 , ctxType :: CtxType
120 }
122------------------------------------------------------------------------------
123-- The stream type
124------------------------------------------------------------------------------
126-- TBD use a functor instead of the bare type a?
128-- | Represents a monadic stream of values of type 'a' constructed using
129-- actions in monad 'm'. Streams can be composed sequentially or in parallel in
130-- product style compositions (monadic bind multiplies streams in a ListT
131-- fashion) and using sum style compositions like 'Semigroup', 'Monoid',
132-- 'Alternative' or variants of these.
133newtype Stream m a =
134 Stream {
135 runStream :: forall r.
136 Maybe (Context m a) -- local state
137 -> m r -- stop
138 -> (a -> Maybe (Stream m a) -> m r) -- yield
139 -> m r
140 }
142-- | A monad that can perform asynchronous/concurrent IO operations.
143type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
145-- | Yield a singleton value in a stream.
146yielding :: a -> Stream m a
147yielding a = Stream $ \_ _ yld -> yld a Nothing
149------------------------------------------------------------------------------
150-- Semigroup
151------------------------------------------------------------------------------
153-- | '<>' concatenates two streams sequentially i.e. the first stream is
154-- exhausted completely before yielding any element from the second stream.
155instance Semigroup (Stream m a) where
156 m1 <> m2 = go m1
157 where
158 go (Stream m) = Stream $ \_ stp yld ->
159 let stop = (runStream m2) Nothing stp yld
160 yield a Nothing = yld a (Just m2)
161 yield a (Just r) = yld a (Just (go r))
162 in m Nothing stop yield
164------------------------------------------------------------------------------
165-- Monoid
166------------------------------------------------------------------------------
168instance Monoid (Stream m a) where
169 mempty = Stream $ \_ stp _ -> stp
170 mappend = (<>)
172------------------------------------------------------------------------------
173-- Interleave
174------------------------------------------------------------------------------
176-- | Same as '<=>'.
177interleave :: Stream m a -> Stream m a -> Stream m a
178interleave m1 m2 = Stream $ \_ stp yld -> do
179 let stop = (runStream m2) Nothing stp yld
180 yield a Nothing = yld a (Just m2)
181 yield a (Just r) = yld a (Just (interleave m2 r))
182 (runStream m1) Nothing stop yield
184------------------------------------------------------------------------------
185-- Spawning threads and collecting result in streamed fashion
186------------------------------------------------------------------------------
188{-# INLINE doFork #-}
189doFork :: MonadBaseControl IO m
190 => m ()
191 -> (SomeException -> m ())
192 -> m ThreadId
193doFork action exHandler =
194 EL.mask $ \restore ->
195 liftBaseWith $ \runInIO -> forkIO $ do
196 -- XXX test the exception handling
197 _ <- runInIO $ EL.catch (restore action) exHandler
198 -- XXX restore state here?
199 return ()
201-- XXX exception safety of all atomic/MVar operations
203{-# INLINE send #-}
204send :: MonadIO m => Context m a -> ChildEvent a -> m ()
205send ctx msg = liftIO $ do
206 atomicModifyIORefCAS_ (outputQueue ctx) $ \es -> msg : es
207 -- XXX need a memory barrier? The wake up must happen only after the
208 -- store has finished otherwise we can have lost wakeup problems.
209 void $ tryPutMVar (doorBell ctx) True
211{-# INLINE sendStop #-}
212sendStop :: MonadIO m => Context m a -> m ()
213sendStop ctx = liftIO myThreadId >>= \tid -> send ctx (ChildStop tid Nothing)
215-- Note: Left associated operations can grow this queue to a large size
216{-# INLINE enqueueLIFO #-}
217enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
218enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
220runqueueLIFO :: MonadIO m => Context m a -> IORef [Stream m a] -> m ()
221runqueueLIFO ctx q = run
223 where
225 run = do
226 work <- dequeue
227 case work of
228 Nothing -> sendStop ctx
229 Just m -> (runStream m) (Just ctx) run yield
231 sendit a = send ctx (ChildYield a)
232 yield a Nothing = sendit a >> run
233 yield a (Just r) = sendit a >> (runStream r) (Just ctx) run yield
235 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->
236 case ms of
237 [] -> ([], Nothing)
238 x : xs -> (xs, Just x)
240{-# INLINE enqueueFIFO #-}
241enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
242enqueueFIFO = pushL
244runqueueFIFO :: MonadIO m => Context m a -> LinkedQueue (Stream m a) -> m ()
245runqueueFIFO ctx q = run
247 where
249 run = do
250 work <- dequeue
251 case work of
252 Nothing -> sendStop ctx
253 Just m -> (runStream m) (Just ctx) run yield
255 dequeue = liftIO $ tryPopR q
256 sendit a = send ctx (ChildYield a)
257 yield a Nothing = sendit a >> run
258 yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
260-- Thread tracking is needed for two reasons:
261--
262-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
263-- themselves because they may run for significant times before going away or
264-- worse they may be stuck in IO and never go away.
265--
266-- 2) To know when all threads are done.
268{-# NOINLINE addThread #-}
269addThread :: MonadIO m => Context m a -> ThreadId -> m ()
270addThread ctx tid =
271 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
273{-# INLINE delThread #-}
274delThread :: MonadIO m => Context m a -> ThreadId -> m ()
275delThread ctx tid =
276 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.delete tid s)
278{-# INLINE allThreadsDone #-}
279allThreadsDone :: MonadIO m => Context m a -> m Bool
280allThreadsDone ctx = liftIO $ do
281 readIORef (runningThreads ctx) >>= return . S.null
283{-# NOINLINE handleChildException #-}
284handleChildException :: MonadIO m => Context m a -> SomeException -> m ()
285handleChildException ctx e = do
286 tid <- liftIO myThreadId
287 send ctx (ChildStop tid (Just e))
289{-# NOINLINE pushWorker #-}
290pushWorker :: MonadAsync m => Context m a -> m ()
291pushWorker ctx =
292 doFork (runqueue ctx) (handleChildException ctx) >>= addThread ctx
294-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
295-- Also, if a worker blocks on the output queue we can decide if we want to
296-- block or make it go away entirely, depending on the number of workers and
297-- the type of the queue.
298{-# INLINE sendWorkerWait #-}
299sendWorkerWait :: MonadAsync m => Context m a -> m ()
300sendWorkerWait ctx = do
301 case ctxType ctx of
302 CtxType _ LIFO -> liftIO $ threadDelay 200
303 CtxType _ FIFO -> liftIO $ threadDelay 0
305 output <- liftIO $ readIORef (outputQueue ctx)
306 when (null output) $ do
307 done <- queueEmpty ctx
308 if (not done)
309 then (pushWorker ctx) >> sendWorkerWait ctx
310 else void (liftIO $ takeMVar (doorBell ctx))
312-- | An 'async' stream has finished but is still being used.
313data EndOfStream = EndOfStream deriving Show
314instance Exception EndOfStream
316-- | Pull a stream from a context
317{-# NOINLINE pullFromCtx #-}
318pullFromCtx :: MonadAsync m => Context m a -> Stream m a
319pullFromCtx ctx = Stream $ \_ stp yld -> do
320 -- When using an async handle to the context, one may keep using a stale
321 -- context even after it has been fully drained. To detect it gracefully we
322 -- raise an explicit exception.
323 -- XXX if reading the IORef is costly we can use a flag in the context to
324 -- indicate we are done.
325 done <- allThreadsDone ctx
326 when done $ throwM EndOfStream
328 res <- liftIO $ tryTakeMVar (doorBell ctx)
329 when (isNothing res) $ sendWorkerWait ctx
330 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
331 -- To avoid lock overhead we read all events at once instead of reading one
332 -- at a time. We just reverse the list to process the events in the order
333 -- they arrived. Maybe we can use a queue instead?
334 (runStream $ processEvents (reverse list)) Nothing stp yld
336 where
338 handleException e tid = do
339 delThread ctx tid
340 -- XXX implement kill async exception handling
341 -- liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
342 throwM e
344 {-# INLINE processEvents #-}
345 processEvents [] = Stream $ \_ stp yld -> do
346 done <- allThreadsDone ctx
347 if not done
348 then (runStream (pullFromCtx ctx)) Nothing stp yld
349 else stp
351 processEvents (ev : es) = Stream $ \_ stp yld -> do
352 let continue = (runStream (processEvents es)) Nothing stp yld
353 yield a = yld a (Just (processEvents es))
355 case ev of
356 ChildYield a -> yield a
357 ChildStop tid e ->
358 case e of
359 Nothing -> delThread ctx tid >> continue
360 Just ex -> handleException ex tid
362getFifoCtx :: MonadIO m => CtxType -> IO (Context m a)
363getFifoCtx ctype = do
364 outQ <- newIORef []
365 outQMv <- newEmptyMVar
366 running <- newIORef S.empty
367 q <- newQ
368 let ctx =
369 Context { outputQueue = outQ
370 , doorBell = outQMv
371 , runningThreads = running
372 , runqueue = runqueueFIFO ctx q
373 , enqueue = pushL q
374 , queueEmpty = liftIO $ nullQ q
375 , ctxType = ctype
376 }
377 in return ctx
379getLifoCtx :: MonadIO m => CtxType -> IO (Context m a)
380getLifoCtx ctype = do
381 outQ <- newIORef []
382 outQMv <- newEmptyMVar
383 running <- newIORef S.empty
384 q <- newIORef []
385 let checkEmpty = liftIO (readIORef q) >>= return . null
386 let ctx =
387 Context { outputQueue = outQ
388 , doorBell = outQMv
389 , runningThreads = running
390 , runqueue = runqueueLIFO ctx q
391 , enqueue = enqueueLIFO q
392 , queueEmpty = checkEmpty
393 , ctxType = ctype
394 }
395 in return ctx
397-- | Split the original computation in a pull-push pair. The original
398-- computation pulls from a Channel while m1 and m2 push to the channel.
399pushPairToCtx :: MonadAsync m
400 => CtxType -> Stream m a -> Stream m a -> m (Context m a)
401pushPairToCtx ctype m1 m2 = do
402 -- Note: We must have all the work on the queue before sending the
403 -- pushworker, otherwise the pushworker may exit before we even get a
404 -- chance to push.
405 ctx <- liftIO $
406 case ctype of
407 CtxType _ FIFO -> do
408 c <- getFifoCtx ctype
409 (enqueue c) m1 >> (enqueue c) m2
410 return c
411 CtxType _ LIFO -> do
412 c <- getLifoCtx ctype
413 (enqueue c) m2 >> (enqueue c) m1
414 return c
415 pushWorker ctx
416 return ctx
418-- | The computation specified in the argument is pushed to a new thread and
419-- the context is returned. The returned context can be used to pull the output
420-- of the computation.
421pushOneToCtx :: MonadAsync m => CtxType -> Stream m a -> m (Context m a)
422pushOneToCtx ctype m = do
423 ctx <- liftIO $
424 case ctype of
425 CtxType _ FIFO -> do
426 c <- getFifoCtx ctype
427 return c
428 CtxType _ LIFO -> do
429 c <- getLifoCtx ctype
430 return c
431 -- Note: We must have all the work on the queue before sending the
432 -- pushworker, otherwise the pushworker may exit before we even get a
433 -- chance to push.
434 liftIO $ (enqueue ctx) m
435 pushWorker ctx
436 return ctx
438{-# NOINLINE pushPullFork #-}
439pushPullFork :: MonadAsync m
440 => CtxType -> Stream m a -> Stream m a -> Stream m a
441pushPullFork ct m1 m2 = Stream $ \_ stp yld -> do
442 ctx <- pushPairToCtx ct m1 m2
443 (runStream (pullFromCtx ctx)) Nothing stp yld
445-- Concurrency rate control. Our objective is to create more threads on demand
446-- if the consumer is running faster than us. As soon as we encounter an
447-- Alternative composition we create a push pull pair of threads. We use a
448-- channel for communication between the consumer pulling from the channel and
449-- the producer who pushing to the channel. The producer creates more threads
450-- if no output is seen on the channel, that is the consumer is running faster.
451-- However this mechanism can be problematic if the initial production latency
452-- is high, we may end up creating too many threads. So we need some way to
453-- monitor and use the latency as well.
454--
455-- TBD We may run computations at the lower level of the composition tree
456-- serially even if they are composed using a parallel combinator. We can use
457-- <> in place of <| and <=> in place of <|>. If we find that a parallel
458-- channel immediately above a computation becomes empty we can switch to
459-- parallelizing the computation. For that we can use a state flag to fork the
460-- rest of the computation at any point of time inside the Monad bind operation
461-- if the consumer is running at a faster speed.
462--
463-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
464-- If we have to dispatch in arbitrary chunksizes we will need to compose the
465-- parallel actions using a data constructor (Free Alternative) instead so that
466-- we can divide it in chunks of arbitrary size before dispatch. If the stream
467-- is composed of hierarchically composed grains of different sizes then we can
468-- always switch to a desired granularity depending on the consumer speed.
469--
470-- TBD for pure work (when we are not in the IO monad) we can divide it into
471-- just the number of CPUs.
472--
473-- XXX to rate control left folded structrues we will have to return the
474-- residual work back to the dispatcher. It will also consume a lot of
475-- memory due to queueing of all the work before execution starts.
477-- | Compose two streams in parallel using a scheduling policy specified by
478-- 'CtxType'. Note: This is designed to scale for right associated
479-- compositions, therefore always use a right fold for folding bigger
480-- structures.
481{-# INLINE parallel #-}
482parallel :: MonadAsync m => CtxType -> Stream m a -> Stream m a -> Stream m a
483parallel ct m1 m2 = Stream $ \ctx stp yld -> do
484 case ctx of
485 Nothing -> (runStream (pushPullFork ct m1 m2)) Nothing stp yld
486 Just c | ctxType c /= ct ->
487 (runStream (pushPullFork ct m1 m2)) Nothing stp yld
488 Just c -> liftIO ((enqueue c) m2) >> (runStream m1) ctx stp yld
490------------------------------------------------------------------------------
491-- Semigroup and Monoid style compositions for parallel actions
492------------------------------------------------------------------------------
494{-
495-- | Same as '<>|'.
496parAhead :: Stream m a -> Stream m a -> Stream m a
497parAhead = undefined
499-- | Sequential composition similar to '<>' except that it can execute the
500-- action on the right in parallel ahead of time. Returns the results in
501-- sequential order like '<>' from left to right.
502(<>|) :: Stream m a -> Stream m a -> Stream m a
503(<>|) = parAhead
504-}
506-- | Same as '<|>'.
507{-# INLINE parAlt #-}
508parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
509parAlt = parallel (CtxType Disjunction FIFO)
511-- | Same as '<|'.
512{-# INLINE parLeft #-}
513parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
514parLeft = parallel (CtxType Disjunction LIFO)
516-------------------------------------------------------------------------------
517-- Instances (only used for deriving newtype instances)
518-------------------------------------------------------------------------------
520-- Stream type is not exposed, these instances are only for deriving instances
521-- for the newtype wrappers based on Stream.
523-- Dummy Instances, defined to enable the definition of other instances that
524-- require a Monad constraint. Must be defined by the newtypes.
526instance Monad m => Functor (Stream m) where
527 fmap = undefined
529instance Monad m => Applicative (Stream m) where
530 pure = undefined
531 (<*>) = undefined
533instance Monad m => Monad (Stream m) where
534 return = pure
535 (>>=) = undefined
537------------------------------------------------------------------------------
538-- Alternative & MonadPlus
539------------------------------------------------------------------------------
541-- | `empty` represents an action that takes non-zero time to complete. Since
542-- all actions take non-zero time, an `Alternative` composition ('<|>') is a
543-- monoidal composition executing all actions in parallel, it is similar to
544-- '<>' except that it runs all the actions in parallel and interleaves their
545-- results fairly.
546instance MonadAsync m => Alternative (Stream m) where
547 empty = mempty
548 (<|>) = parAlt
550instance MonadAsync m => MonadPlus (Stream m) where
551 mzero = empty
552 mplus = (<|>)
554-------------------------------------------------------------------------------
555-- Transformer
556-------------------------------------------------------------------------------
558instance MonadTrans Stream where
559 lift mx = Stream $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))
561instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where
562 liftBase = liftBaseDefault
564------------------------------------------------------------------------------
565-- Standard transformer instances
566------------------------------------------------------------------------------
568instance MonadIO m => MonadIO (Stream m) where
569 liftIO = lift . liftIO
571instance MonadThrow m => MonadThrow (Stream m) where
572 throwM = lift . throwM
574-- XXX handle and test cross thread state transfer
575instance MonadError e m => MonadError e (Stream m) where
576 throwError = lift . throwError
577 catchError m h = Stream $ \ctx stp yld ->
578 let handle r = r `catchError` \e -> (runStream (h e)) ctx stp yld
579 yield a Nothing = yld a Nothing
580 yield a (Just r) = yld a (Just (catchError r h))
581 in handle $ (runStream m) ctx stp yield
583instance MonadReader r m => MonadReader r (Stream m) where
584 ask = lift ask
585 local f m = Stream $ \ctx stp yld ->
586 let yield a Nothing = local f $ yld a Nothing
587 yield a (Just r) = local f $ yld a (Just (local f r))
588 in (runStream m) ctx (local f stp) yield
590instance MonadState s m => MonadState s (Stream m) where
591 get = lift get
592 put x = lift (put x)
593 state k = lift (state k)
595------------------------------------------------------------------------------
596-- Types that can behave as a Stream
597------------------------------------------------------------------------------
599-- | Types that can represent a stream of elements.
600class Streaming t where
601 toStream :: t m a -> Stream m a
602 fromStream :: Stream m a -> t m a
604------------------------------------------------------------------------------
605-- Some basic type agnostic (working on the Streaming type class) APIs
606------------------------------------------------------------------------------
608-- XXX The async API is useful for exploring each stream arbitrarily when
609-- zipping or merging two streams. We can use a newtype wrapper with a monad
610-- instance that composes like regular streaming libraries to facilitate linear
611-- composition. We will also need a yield API for that.
613-- | Run a computation asynchronously, triggers the computation and returns
614-- another computation that when executed produces the output generated by the
615-- original computation. Note that the returned action must be executed exactly
616-- once and drained completely. If not executed or not drained fully we may
617-- have a thread blocked forever and if executed more than once a
618-- 'EndOfStream' exception will be raised.
620async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
621async m = do
622 ctx <- pushOneToCtx (CtxType Disjunction LIFO) (toStream m)
623 return $ fromStream $ pullFromCtx ctx
625infixr 5 <=>
627-- | Sequential interleaved composition, in contrast to '<>' this operator
628-- fairly interleaves the two streams instead of appending them; yielding one
629-- element from each stream alternately. Unlike '<>' it cannot be used to fold
630-- an infinite container of streams.
631{-# INLINE (<=>) #-}
632(<=>) :: Streaming t => t m a -> t m a -> t m a
633m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
635-- | Parallel interleaved composition, in contrast to '<|>' this operator
636-- "merges" streams in a left biased manner rather than fairly interleaving
637-- them. It keeps yielding from the stream on the left as long as it can. If
638-- the left stream blocks or cannot keep up with the pace of the consumer it
639-- can yield from the stream on the right in parallel. Unlike '<|>' it can be
640-- used to fold infinite containers of streams.
641{-# INLINE (<|) #-}
642(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
643m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
645------------------------------------------------------------------------------
646-- Constructing a stream
647------------------------------------------------------------------------------
649-- XXX Need to accept a context as well
650-- | Build a stream from its church encoding.
651streamBuild :: Streaming t
652 => (forall r. (a -> Maybe (t m a) -> m r) -> m r -> m r) -> t m a
653streamBuild f = fromStream $ Stream $ \_ stp yld ->
654 let yield a Nothing = yld a Nothing
655 yield a (Just r) = yld a (Just (toStream r))
656 in f yield stp
658------------------------------------------------------------------------------
659-- Destroying a stream
660------------------------------------------------------------------------------
662-- | Fold a stream using its church encoding. The first argument is the
663-- "step" function consuming one element and the rest of the stream. The second
664-- argument is the "done" function that is called when the stream is over.
665streamFold :: Streaming t => (a -> Maybe (t m a) -> m r) -> m r -> t m a -> m r
666streamFold step done m =
667 let yield a Nothing = step a Nothing
668 yield a (Just x) = step a (Just (fromStream x))
669 in (runStream (toStream m)) Nothing done yield
671-- | Run a streaming composition, discard the results.
672runStreaming :: (Monad m, Streaming t) => t m a -> m ()
673runStreaming m = go (toStream m)
674 where
675 go m1 =
676 let stop = return ()
677 yield _ Nothing = stop
678 yield _ (Just x) = go x
679 in (runStream m1) Nothing stop yield
681------------------------------------------------------------------------------
682-- Utilities
683------------------------------------------------------------------------------
685-- | Fold a 'Foldable' container using the given function.
686{-# INLINABLE foldWith #-}
687foldWith :: (Monoid b, Foldable t) => (a -> b -> b) -> t a -> b
688foldWith f = foldr f mempty
690-- | Fold a 'Foldable' container using a function that is a composition of the
691-- two arguments.
692{-# INLINABLE foldMapWith #-}
693foldMapWith :: (Monoid b, Foldable t) =>
694 (b1 -> b -> b) -> (a -> b1) -> t a -> b
695foldMapWith f g = foldr (f . g) mempty
697-- | Fold a 'Foldable' container using a function that is a composition of the
698-- first and the third argument.
699{-# INLINABLE forEachWith #-}
700forEachWith :: (Monoid b, Foldable t) =>
701 (b1 -> b -> b) -> t a -> (a -> b1) -> b
702forEachWith f xs g = foldr (f . g) mempty xs