git.iko.soy/streamly
on 8f89d92 help sign in
streamly/src/Asyncly/AsyncT.hs
673 lines 26 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.AsyncT
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.AsyncT
19 ( Stream (..)
20 , MonadAsync
21 , Streaming (..)
23 , fromCallback
25 , runStreaming
26 , toList
27 , async
29 , (<=>)
30 , (<|)
32 , yielding
33 , interleave
34 , parallel
35 , parLeft
36 , CtxType (..)
37 , SchedPolicy (..)
38 , Dimension (..)
40 , foldWith
41 , foldMapWith
42 , forEachWith
43 )
44where
46import Control.Applicative (Alternative (..))
47import Control.Concurrent (ThreadId, forkIO,
48 myThreadId, threadDelay)
49import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
50 tryPutMVar, takeMVar)
51import Control.Exception (SomeException (..), Exception)
52import qualified Control.Exception.Lifted as EL
53import Control.Monad (ap, liftM, MonadPlus(..), mzero,
54 when)
55import Control.Monad.Base (MonadBase (..), liftBaseDefault)
56import Control.Monad.Catch (MonadThrow, throwM)
57import Control.Monad.Error.Class (MonadError(..))
58import Control.Monad.IO.Class (MonadIO(..))
59import Control.Monad.Reader.Class (MonadReader(..))
60import Control.Monad.State.Class (MonadState(..))
61import Control.Monad.Trans.Class (MonadTrans (lift))
62import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
63import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
64 tryPopR, nullQ)
65import Data.Functor (void)
66import Data.IORef (IORef, modifyIORef, newIORef,
67 readIORef)
68import Data.Atomics (atomicModifyIORefCAS,
69 atomicModifyIORefCAS_)
70import Data.Maybe (isNothing)
71import Data.Semigroup (Semigroup(..))
72import Data.Set (Set)
73import qualified Data.Set as S
75------------------------------------------------------------------------------
76-- Concurrency Semantics
77------------------------------------------------------------------------------
78--
79-- Asyncly is essentially a concurrent list transformer. To fold the lists
80-- (AsyncT m a) it provides many different ways of composing them. The Monoid
81-- instance folds lists in a sequential fashion. The Alternative instance folds
82-- a list in a concurrent manner running each action in parallel and therefore
83-- the order of results in the fold is not deterministic, but the results are
84-- fairly interleaved. There are other ways to bind and compose. The
85-- combinations of different types of binds and monoidal compositions can be
86-- used to achieve the desired results.
88------------------------------------------------------------------------------
89-- Parent child thread communication types
90------------------------------------------------------------------------------
92data ChildEvent a =
93 ChildYield a
94 | ChildStop ThreadId (Maybe SomeException)
96------------------------------------------------------------------------------
97-- State threaded around the monad for thread management
98------------------------------------------------------------------------------
100data Dimension = Conjunction | Disjunction deriving Eq
101data SchedPolicy = LIFO | FIFO deriving Eq
102data CtxType = CtxType Dimension SchedPolicy deriving Eq
104data Context m a =
105 Context { outputQueue :: IORef [ChildEvent a]
106 , doorBell :: MVar Bool -- wakeup mechanism for outQ
107 , enqueue :: Stream m a -> IO ()
108 , runqueue :: m ()
109 , runningThreads :: IORef (Set ThreadId)
110 , queueEmpty :: m Bool
111 , ctxType :: CtxType
112 }
114-- | Represents a monadic stream of values of type 'a' resulting from actions
115-- in monad 'm'. 'AsyncT' streams can be composed sequentially or in parallel
116-- in various ways using monadic bind or variants of it and using monoidal
117-- compositions like 'Monoid', 'Alternative' or variants of these.
118newtype Stream m a =
119 Stream {
120 runStream :: forall r.
121 Maybe (Context m a) -- local state
122 -> m r -- stop
123 -> (a -> Maybe (Stream m a) -> m r) -- yield
124 -> m r
125 }
127-- | A monad that can do asynchronous (or parallel) IO operations.
128type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
130yielding :: a -> Stream m a
131yielding a = Stream $ \_ _ yld -> yld a Nothing
133------------------------------------------------------------------------------
134-- Monoid
135------------------------------------------------------------------------------
137-- | '<>' concatenates two 'AsyncT' streams sequentially i.e. the first stream
138-- is exhausted completely before yielding any element from the second stream.
139instance Semigroup (Stream m a) where
140 m1 <> m2 = go m1
141 where
142 go (Stream m) = Stream $ \_ stp yld ->
143 let stop = (runStream m2) Nothing stp yld
144 yield a Nothing = yld a (Just m2)
145 yield a (Just r) = yld a (Just (go r))
146 in m Nothing stop yield
148instance Monoid (Stream m a) where
149 mempty = Stream $ \_ stp _ -> stp
150 mappend = (<>)
152interleave :: Stream m a -> Stream m a -> Stream m a
153interleave m1 m2 = Stream $ \_ stp yld -> do
154 let stop = (runStream m2) Nothing stp yld
155 yield a Nothing = yld a (Just m2)
156 yield a (Just r) = yld a (Just (interleave m2 r))
157 (runStream m1) Nothing stop yield
159------------------------------------------------------------------------------
160-- Functor
161------------------------------------------------------------------------------
163instance Monad m => Functor (Stream m) where
164 fmap = liftM
166------------------------------------------------------------------------------
167-- Applicative
168------------------------------------------------------------------------------
170instance Monad m => Applicative (Stream m) where
171 pure = yielding
172 (<*>) = ap
174------------------------------------------------------------------------------
175-- Monad
176------------------------------------------------------------------------------
178-- | Execute a monadic action sequentially for each element in the 'AsyncT'
179-- stream, i.e. an iteration finishes completely before the next one starts.
180instance Monad m => Monad (Stream m) where
181 return = pure
183 Stream m >>= f = Stream $ \_ stp yld ->
184 let run x = (runStream x) Nothing stp yld
185 yield a Nothing = run $ f a
186 yield a (Just r) = run $ f a <> (r >>= f)
187 in m Nothing stp yield
189------------------------------------------------------------------------------
190-- Alternative
191------------------------------------------------------------------------------
193{-# INLINE doFork #-}
194doFork :: MonadBaseControl IO m
195 => m ()
196 -> (SomeException -> m ())
197 -> m ThreadId
198doFork action exHandler =
199 EL.mask $ \restore ->
200 liftBaseWith $ \runInIO -> forkIO $ do
201 -- XXX test the exception handling
202 _ <- runInIO $ EL.catch (restore action) exHandler
203 -- XXX restore state here?
204 return ()
206-- XXX exception safety of all atomic/MVar operations
208{-# INLINE send #-}
209send :: MonadIO m => Context m a -> ChildEvent a -> m ()
210send ctx msg = liftIO $ do
211 atomicModifyIORefCAS_ (outputQueue ctx) $ \es -> msg : es
212 -- XXX need a memory barrier? The wake up must happen only after the
213 -- store has finished otherwise we can have lost wakeup problems.
214 void $ tryPutMVar (doorBell ctx) True
216{-# INLINE sendStop #-}
217sendStop :: MonadIO m => Context m a -> m ()
218sendStop ctx = liftIO myThreadId >>= \tid -> send ctx (ChildStop tid Nothing)
220-- Note: Left associated operations can grow this queue to a large size
221{-# INLINE enqueueLIFO #-}
222enqueueLIFO :: IORef [Stream m a] -> Stream m a -> IO ()
223enqueueLIFO q m = atomicModifyIORefCAS_ q $ \ ms -> m : ms
225runqueueLIFO :: MonadIO m => Context m a -> IORef [Stream m a] -> m ()
226runqueueLIFO ctx q = run
228 where
230 run = do
231 work <- dequeue
232 case work of
233 Nothing -> sendStop ctx
234 Just m -> (runStream m) (Just ctx) run yield
236 sendit a = send ctx (ChildYield a)
237 yield a Nothing = sendit a >> run
238 yield a (Just r) = sendit a >> (runStream r) (Just ctx) run yield
240 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->
241 case ms of
242 [] -> ([], Nothing)
243 x : xs -> (xs, Just x)
245{-# INLINE enqueueFIFO #-}
246enqueueFIFO :: LinkedQueue (Stream m a) -> Stream m a -> IO ()
247enqueueFIFO = pushL
249runqueueFIFO :: MonadIO m => Context m a -> LinkedQueue (Stream m a) -> m ()
250runqueueFIFO ctx q = run
252 where
254 run = do
255 work <- dequeue
256 case work of
257 Nothing -> sendStop ctx
258 Just m -> (runStream m) (Just ctx) run yield
260 dequeue = liftIO $ tryPopR q
261 sendit a = send ctx (ChildYield a)
262 yield a Nothing = sendit a >> run
263 yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run
265-- Thread tracking is needed for two reasons:
266--
267-- 1) Killing threads on exceptions. Threads may not be allowed to go away by
268-- themselves because they may run for significant times before going away or
269-- worse they may be stuck in IO and never go away.
270--
271-- 2) To know when all threads are done.
273{-# NOINLINE addThread #-}
274addThread :: MonadIO m => Context m a -> ThreadId -> m ()
275addThread ctx tid =
276 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
278{-# INLINE delThread #-}
279delThread :: MonadIO m => Context m a -> ThreadId -> m ()
280delThread ctx tid =
281 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.delete tid s)
283{-# INLINE allThreadsDone #-}
284allThreadsDone :: MonadIO m => Context m a -> m Bool
285allThreadsDone ctx = liftIO $ do
286 readIORef (runningThreads ctx) >>= return . S.null
288{-# NOINLINE handleChildException #-}
289handleChildException :: MonadIO m => Context m a -> SomeException -> m ()
290handleChildException ctx e = do
291 tid <- liftIO myThreadId
292 send ctx (ChildStop tid (Just e))
294{-# NOINLINE pushWorker #-}
295pushWorker :: MonadAsync m => Context m a -> m ()
296pushWorker ctx =
297 doFork (runqueue ctx) (handleChildException ctx) >>= addThread ctx
299-- XXX When the queue is LIFO we can put a limit on the number of dispatches.
300-- Also, if a worker blocks on the output queue we can decide if we want to
301-- block or make it go away entirely, depending on the number of workers and
302-- the type of the queue.
303{-# INLINE sendWorkerWait #-}
304sendWorkerWait :: MonadAsync m => Context m a -> m ()
305sendWorkerWait ctx = do
306 case ctxType ctx of
307 CtxType _ LIFO -> liftIO $ threadDelay 200
308 CtxType _ FIFO -> liftIO $ threadDelay 0
310 output <- liftIO $ readIORef (outputQueue ctx)
311 when (null output) $ do
312 done <- queueEmpty ctx
313 if (not done)
314 then (pushWorker ctx) >> sendWorkerWait ctx
315 else void (liftIO $ takeMVar (doorBell ctx))
317data ContextUsedAfterEOF = ContextUsedAfterEOF deriving Show
318instance Exception ContextUsedAfterEOF
320-- | Pull an AsyncT stream from a context
321{-# NOINLINE pullFromCtx #-}
322pullFromCtx :: MonadAsync m => Context m a -> Stream m a
323pullFromCtx ctx = Stream $ \_ stp yld -> do
324 -- When using an async handle to the context, one may keep using a stale
325 -- context even after it has been fully drained. To detect it gracefully we
326 -- raise an explicit exception.
327 -- XXX if reading the IORef is costly we can use a flag in the context to
328 -- indicate we are done.
329 done <- allThreadsDone ctx
330 when done $ throwM ContextUsedAfterEOF
332 res <- liftIO $ tryTakeMVar (doorBell ctx)
333 when (isNothing res) $ sendWorkerWait ctx
334 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
335 -- To avoid lock overhead we read all events at once instead of reading one
336 -- at a time. We just reverse the list to process the events in the order
337 -- they arrived. Maybe we can use a queue instead?
338 (runStream $ processEvents (reverse list)) Nothing stp yld
340 where
342 handleException e tid = do
343 delThread ctx tid
344 -- XXX implement kill async exception handling
345 -- liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
346 throwM e
348 {-# INLINE processEvents #-}
349 processEvents [] = Stream $ \_ stp yld -> do
350 done <- allThreadsDone ctx
351 if not done
352 then (runStream (pullFromCtx ctx)) Nothing stp yld
353 else stp
355 processEvents (ev : es) = Stream $ \_ stp yld -> do
356 let continue = (runStream (processEvents es)) Nothing stp yld
357 yield a = yld a (Just (processEvents es))
359 case ev of
360 ChildYield a -> yield a
361 ChildStop tid e ->
362 case e of
363 Nothing -> delThread ctx tid >> continue
364 Just ex -> handleException ex tid
366getFifoCtx :: MonadIO m => CtxType -> IO (Context m a)
367getFifoCtx ctype = do
368 outQ <- newIORef []
369 outQMv <- newEmptyMVar
370 running <- newIORef S.empty
371 q <- newQ
372 let ctx =
373 Context { outputQueue = outQ
374 , doorBell = outQMv
375 , runningThreads = running
376 , runqueue = runqueueFIFO ctx q
377 , enqueue = pushL q
378 , queueEmpty = liftIO $ nullQ q
379 , ctxType = ctype
380 }
381 in return ctx
383getLifoCtx :: MonadIO m => CtxType -> IO (Context m a)
384getLifoCtx ctype = do
385 outQ <- newIORef []
386 outQMv <- newEmptyMVar
387 running <- newIORef S.empty
388 q <- newIORef []
389 let checkEmpty = liftIO (readIORef q) >>= return . null
390 let ctx =
391 Context { outputQueue = outQ
392 , doorBell = outQMv
393 , runningThreads = running
394 , runqueue = runqueueLIFO ctx q
395 , enqueue = enqueueLIFO q
396 , queueEmpty = checkEmpty
397 , ctxType = ctype
398 }
399 in return ctx
401-- | Split the original computation in a pull-push pair. The original
402-- computation pulls from a Channel while m1 and m2 push to the channel.
403pushPairToCtx :: MonadAsync m
404 => CtxType -> Stream m a -> Stream m a -> m (Context m a)
405pushPairToCtx ctype m1 m2 = do
406 -- Note: We must have all the work on the queue before sending the
407 -- pushworker, otherwise the pushworker may exit before we even get a
408 -- chance to push.
409 ctx <- liftIO $
410 case ctype of
411 CtxType _ FIFO -> do
412 c <- getFifoCtx ctype
413 (enqueue c) m1 >> (enqueue c) m2
414 return c
415 CtxType _ LIFO -> do
416 c <- getLifoCtx ctype
417 (enqueue c) m2 >> (enqueue c) m1
418 return c
419 pushWorker ctx
420 return ctx
422{-# NOINLINE pushPullFork #-}
423pushPullFork :: MonadAsync m
424 => CtxType -> Stream m a -> Stream m a -> Stream m a
425pushPullFork ct m1 m2 = Stream $ \_ stp yld -> do
426 ctx <- pushPairToCtx ct m1 m2
427 (runStream (pullFromCtx ctx)) Nothing stp yld
429-- Concurrency rate control. Our objective is to create more threads on
430-- demand if the consumer is running faster than us. As soon as we
431-- encounter an Alternative composition we create a push pull pair of
432-- threads. We use a channel for communication between the consumer that
433-- pulls from the channel and the producer that pushes to the channel. The
434-- producer creates more threads if the channel becomes empty at times,
435-- that is the consumer is running faster. However this mechanism can be
436-- problematic if the initial production latency is high, we may end up
437-- creating too many threads. So we need some way to monitor and use the
438-- latency as well.
439--
440-- TBD For quick response we may have to increase the rate in the middle of
441-- a serially running computation. For that we can use a state flag to fork
442-- the rest of the computation at any point of time inside the Monad bind
443-- operation if the consumer is running at a faster speed.
444--
445-- TBD the alternative composition allows us to dispatch a chunkSize of only 1.
446-- If we have to dispatch in arbitrary chunksizes we will need to compose the
447-- parallel actions using a data constructor instead so that we can divide it
448-- in chunks of arbitrary size before dispatch. When batching we can convert
449-- the structure into Alternative batches of Monoid composition. That will also
450-- allow us to dispatch more work to existing threads rather than creating new
451-- threads always.
452--
453-- TBD for pure work (when we are not in the IO monad) we can divide it into
454-- just the number of CPUs.
455--
456-- XXX to rate control left folded structrues we will have to return the
457-- residual work back to the dispatcher. It will also consume a lot of
458-- memory due to queueing of all the work before execution starts.
460-- Note: This is designed to scale for right associated compositions,
461-- therefore always use a right fold for folding bigger structures.
462{-# INLINE parallel #-}
463parallel :: MonadAsync m => CtxType -> Stream m a -> Stream m a -> Stream m a
464parallel ct m1 m2 = Stream $ \ctx stp yld -> do
465 case ctx of
466 Nothing -> (runStream (pushPullFork ct m1 m2)) Nothing stp yld
467 Just c | ctxType c /= ct ->
468 (runStream (pushPullFork ct m1 m2)) Nothing stp yld
469 Just c -> liftIO ((enqueue c) m2) >> (runStream m1) ctx stp yld
471-- | `empty` represents an action that takes non-zero time to complete. Since
472-- all actions take non-zero time, an `Alternative` composition ('<|>') is a
473-- monoidal composition executing all actions in parallel, it is similar to
474-- '<>' except that it runs all the actions in parallel and interleaves their
475-- results fairly.
476instance MonadAsync m => Alternative (Stream m) where
477 empty = mempty
479 {-# INLINE (<|>) #-}
480 (<|>) = parallel (CtxType Disjunction FIFO)
482------------------------------------------------------------------------------
483-- Other monoidal compositions for parallel actions
484------------------------------------------------------------------------------
486{-
487-- | Same as '<>|'.
488parAhead :: Stream m a -> Stream m a -> Stream m a
489parAhead = undefined
491-- | Sequential composition similar to '<>' except that it can execute the
492-- action on the right in parallel ahead of time. Returns the results in
493-- sequential order like '<>' from left to right.
494(<>|) :: Stream m a -> Stream m a -> Stream m a
495(<>|) = parAhead
496-}
498-- | Same as '<|'.
499{-# INLINE parLeft #-}
500parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
501parLeft = parallel (CtxType Disjunction LIFO)
503instance MonadAsync m => MonadPlus (Stream m) where
504 mzero = empty
505 mplus = (<|>)
507-- | The computation specified in the argument is pushed to a new thread and
508-- the context is computation pulls output from the thread.
509pushOneToCtx :: MonadAsync m => CtxType -> Stream m a -> m (Context m a)
510pushOneToCtx ctype m = do
511 ctx <- liftIO $
512 case ctype of
513 CtxType _ FIFO -> do
514 c <- getFifoCtx ctype
515 return c
516 CtxType _ LIFO -> do
517 c <- getLifoCtx ctype
518 return c
519 -- Note: We must have all the work on the queue before sending the
520 -- pushworker, otherwise the pushworker may exit before we even get a
521 -- chance to push.
522 liftIO $ (enqueue ctx) m
523 pushWorker ctx
524 return ctx
526-------------------------------------------------------------------------------
527-- AsyncT transformer
528-------------------------------------------------------------------------------
530instance MonadTrans Stream where
531 lift mx = Stream $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))
533instance (MonadBase b m, Monad m) => MonadBase b (Stream m) where
534 liftBase = liftBaseDefault
536------------------------------------------------------------------------------
537-- Standard transformer instances
538------------------------------------------------------------------------------
540instance MonadIO m => MonadIO (Stream m) where
541 liftIO = lift . liftIO
543instance MonadThrow m => MonadThrow (Stream m) where
544 throwM = lift . throwM
546-- XXX handle and test cross thread state transfer
547instance MonadError e m => MonadError e (Stream m) where
548 throwError = lift . throwError
549 catchError m h = Stream $ \ctx stp yld ->
550 let handle r = r `catchError` \e -> (runStream (h e)) ctx stp yld
551 yield a Nothing = yld a Nothing
552 yield a (Just r) = yld a (Just (catchError r h))
553 in handle $ (runStream m) ctx stp yield
555instance MonadReader r m => MonadReader r (Stream m) where
556 ask = lift ask
557 local f m = Stream $ \ctx stp yld ->
558 let yield a Nothing = local f $ yld a Nothing
559 yield a (Just r) = local f $ yld a (Just (local f r))
560 in (runStream m) ctx (local f stp) yield
562instance MonadState s m => MonadState s (Stream m) where
563 get = lift get
564 put x = lift (put x)
565 state k = lift (state k)
567------------------------------------------------------------------------------
568-- Types that can behave as a Stream
569------------------------------------------------------------------------------
571class Streaming t where
572 toStream :: t m a -> Stream m a
573 fromStream :: Stream m a -> t m a
575-- XXX The async API is useful for exploring each stream arbitrarily when
576-- zipping or merging two streams. We can use a newtype wrapper with a monad
577-- instance that composes like regular streaming libraries to facilitate linear
578-- composition. We will also need a yield API for that.
580-- | Run a computation asynchronously, triggers the computation and returns
581-- another computation (i.e. a promise) that when executed produces the output
582-- from the original computation. Note that the returned action must be
583-- executed exactly once and drained completely. If not executed or not drained
584-- fully we will may have a thread blocked forever and if executed more than
585-- once a ContextUsedAfterEOF exception will be raised.
587async :: (Streaming t, MonadAsync m) => t m a -> m (t m a)
588async m = do
589 ctx <- pushOneToCtx (CtxType Disjunction LIFO) (toStream m)
590 return $ fromStream $ pullFromCtx ctx
592infixr 5 <=>
594-- | Sequential interleaved composition, similar to '<>' except that it fairly
595-- interleaves the two 'AsyncT' streams, yielding an element from each stream
596-- alternately. Should be used only on finite streams.
597{-# INLINE (<=>) #-}
598(<=>) :: Streaming t => t m a -> t m a -> t m a
599m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)
601-- | Parallel composition similar to '<|>' except that it is left-biased
602-- instead of being fair. Action on the left is likely to be given a chance to
603-- execute before the action on the right. If the left action keeps yielding
604-- results without blocking it continues running until it finishes and only
605-- then the right action runs. Unlike '<|>' it can be used on infinite streams.
606{-# INLINE (<|) #-}
607(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a
608m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)
610------------------------------------------------------------------------------
611-- Running the monad
612------------------------------------------------------------------------------
614-- | Run an 'AsyncT' computation, wait for it to finish and discard the
615-- results.
616runStreaming :: (Monad m, Streaming t) => t m a -> m ()
617runStreaming m = go (toStream m)
619 where
621 go m1 = (runStream m1) Nothing stop yield
623 stop = return ()
625 {-# INLINE yield #-}
626 yield _ Nothing = stop
627 yield _ (Just x) = go x
629-- | Collect the results of an 'AsyncT' stream into a list.
630{-# INLINABLE toList #-}
631toList :: (Monad m, Streaming t) => t m a -> m [a]
632toList m = go (toStream m)
634 where
636 go m1 = (runStream m1) Nothing stop yield
638 stop = return []
640 {-# INLINE yield #-}
641 yield a Nothing = return [a]
642 yield a (Just x) = liftM (a :) (go x)
644------------------------------------------------------------------------------
645-- Utilities
646------------------------------------------------------------------------------
648-- | Fold a 'Foldable' container using the given function.
649{-# INLINABLE foldWith #-}
650foldWith :: (Monoid b, Foldable t) => (a -> b -> b) -> t a -> b
651foldWith f = foldr f mempty
653-- | Fold a 'Foldable' container using a function that is a composition of the
654-- two arguments.
655{-# INLINABLE foldMapWith #-}
656foldMapWith :: (Monoid b, Foldable t) =>
657 (b1 -> b -> b) -> (a -> b1) -> t a -> b
658foldMapWith f g = foldr (f . g) mempty
660-- | Fold a 'Foldable' container using a function that is a composition of the
661-- first and the third argument.
662{-# INLINABLE forEachWith #-}
663forEachWith :: (Monoid b, Foldable t) =>
664 (b1 -> b -> b) -> t a -> (a -> b1) -> b
665forEachWith f xs g = foldr (f . g) mempty xs
667------------------------------------------------------------------------------
668-- Convert a callback into an 'AsyncT' computation
669------------------------------------------------------------------------------
671-- | Convert a callback into an 'AsyncT' stream.
672fromCallback :: (forall r. (a -> m r) -> m r) -> Stream m a
673fromCallback k = Stream $ \_ _ yld -> k (\a -> yld a Nothing)