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