git.iko.soy/streamly
on d134ea3 help sign in
streamly/src/Streamly/Internal/Data/Fold/Concurrent.hs
69 lines 2.0 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Fold.Concurrent
3-- Copyright : (c) 2022 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9-- Eval a fold asynchronously in a separate thread.
10--
11-- Use multiple worker folds to fold a stream and collect the results using
12-- another fold, combine using a monoid. The results can be collected
13-- out-of-order or in-order.
14--
15-- Concurrent append: if one fold's buffer becomes full then use the next one
16-- Concurrent interleave/partition: Round robin to n folds.
17-- Concurrent distribute to multiple folds.
19module Streamly.Internal.Data.Fold.Concurrent
20 (
21 -- * Imports
22 -- $setup
24 -- * Configuration
25 Config
26 , maxBuffer
27 , inspect
29 -- * Combinators
30 -- | Stream combinators using Async channel
32 , eval
33 , parEval
34 )
35where
37import Streamly.Internal.Control.Concurrent (MonadAsync)
38import Streamly.Internal.Data.Fold (Fold)
40import Streamly.Internal.Data.Fold.Concurrent.Channel
42-- $setup
43--
44-- Imports for example snippets in this module.
45--
46-- >>> :m
47-- >>> :set -XFlexibleContexts
48-- >>> import Control.Concurrent (threadDelay)
49-- >>> import qualified Streamly.Data.Array as Array
50-- >>> import qualified Streamly.Data.Fold as Fold
51-- >>> import qualified Streamly.Data.Parser as Parser
52-- >>> import qualified Streamly.Data.Stream as Stream
53-- >>> import qualified Streamly.Internal.Data.Fold.Concurrent as Fold
54-- >>> import Prelude hiding (concatMap, concat)
55-- >>> :{
56-- delay n = do
57-- threadDelay (n * 1000000) -- sleep for n seconds
58-- putStrLn (show n ++ " sec") -- print "n sec"
59-- return n -- IO Int
60-- :}
62-- | Evaluate a stream asynchronously using a channel and serve the consumer
63-- from the evaluation buffer.
64--
65-- >>> eval = Fold.parEval id
66--
67{-# INLINE eval #-}
68eval :: MonadAsync m => Fold m a b -> Fold m a b
69eval = parEval id