git.iko.soy/streamly
on cee4d0f help sign in
streamly/src/Streamly/Internal/Data/Stream/ZipAsync.hs
131 lines 4.6 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
3-- |
4-- Module : Streamly.Internal.Data.Stream.ZipAsync
5-- Copyright : (c) 2017 Composewell Technologies
6-- License : BSD-3-Clause
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
9-- Portability : GHC
10--
11-- To run examples in this module:
12--
13-- >>> import qualified Streamly.Prelude as Stream
14--
15module Streamly.Internal.Data.Stream.ZipAsync
16 ( ZipAsyncM(..)
17 , ZipAsync
18 , consMZipAsync
19 , zipAsyncWithK
20 , zipAsyncWithMK
21 )
22where
24import Streamly.Internal.Control.Concurrent (MonadAsync)
25import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
27import qualified Streamly.Internal.Data.Stream.StreamK as K
28 (mkStream, foldStream, zipWithM, consM)
29import qualified Streamly.Internal.Data.Stream.StreamD as D (fromStreamK)
30import qualified Streamly.Internal.Data.Stream.Serial as Serial
31import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar
32import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar
33import qualified Streamly.Internal.Data.Stream as Stream
34 (fromStreamK, toStreamK)
35import Streamly.Internal.Data.SVar
37import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)
39#include "Instances.hs"
41-- $setup
42-- >>> import qualified Streamly.Prelude as Stream
43-- >>> import Control.Concurrent (threadDelay)
44-- >>> :{
45-- delay n = do
46-- threadDelay (n * 1000000) -- sleep for n seconds
47-- putStrLn (show n ++ " sec") -- print "n sec"
48-- return n -- IO Int
49-- :}
51------------------------------------------------------------------------------
52-- Parallel Zipping
53------------------------------------------------------------------------------
55-- | Like 'zipAsyncWith' but with a monadic zipping function.
56--
57-- @since 0.4.0
58{-# INLINE zipAsyncWithMK #-}
59zipAsyncWithMK :: MonadAsync m
60 => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c
61zipAsyncWithMK f m1 m2 = K.mkStream $ \st yld sng stp -> do
62 sv <- newParallelVar StopNone (adaptState st)
63 SVar.toSVarParallel (adaptState st) sv $ D.fromStreamK m2
64 K.foldStream st yld sng stp
65 $ K.zipWithM f m1 (Stream.toStreamK (SVar.fromSVar sv))
67-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
68-- along with the change of behvaior to end the stream concurrently.
69--
70-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
71-- are evaluated concurrently using the 'ParallelT' concurrent evaluation
72-- style. The maximum number of elements of each stream evaluated in advance
73-- can be controlled by 'maxBuffer'.
74--
75-- The stream ends if stream @a@ or stream @b@ ends. However, if stream @b@
76-- ends while we are still evaluating stream @a@ and waiting for a result then
77-- stream will not end until after the evaluation of stream @a@ finishes. This
78-- behavior can potentially be changed in future to end the stream immediately
79-- as soon as any of the stream end is detected.
80--
81-- @since 0.1.0
82{-# INLINE zipAsyncWithK #-}
83zipAsyncWithK :: MonadAsync m
84 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
85zipAsyncWithK f = zipAsyncWithMK (\a b -> return (f a b))
87------------------------------------------------------------------------------
88-- Parallely Zipping Streams
89------------------------------------------------------------------------------
90--
91-- | For 'ZipAsyncM' streams:
92--
93-- @
94-- (<>) = 'Streamly.Prelude.serial'
95-- (<*>) = 'Streamly.Prelude.serial.zipAsyncWith' id
96-- @
97--
98-- Applicative evaluates the streams being zipped concurrently, the following
99-- would take half the time that it would take in serial zipping:
100--
101-- >>> s = Stream.fromFoldableM $ Prelude.map delay [1, 1, 1]
102-- >>> Stream.toList $ Stream.fromZipAsync $ (,) <$> s <*> s
103-- ...
104-- [(1,1),(1,1),(1,1)]
105--
106-- /Since: 0.2.0 ("Streamly")/
107--
108-- @since 0.8.0
109newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
110 deriving (Semigroup, Monoid)
112-- | An IO stream whose applicative instance zips streams wAsyncly.
113--
114-- /Since: 0.2.0 ("Streamly")/
115--
116-- @since 0.8.0
117type ZipAsync = ZipAsyncM IO
119consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
120consMZipAsync m (ZipAsyncM r) = ZipAsyncM $ K.consM m r
122instance Monad m => Functor (ZipAsyncM m) where
123 {-# INLINE fmap #-}
124 fmap f (ZipAsyncM m) =
125 ZipAsyncM $ Stream.toStreamK $ fmap f (Stream.fromStreamK m)
127instance MonadAsync m => Applicative (ZipAsyncM m) where
128 pure = ZipAsyncM . Stream.toStreamK . Serial.repeat
130 {-# INLINE (<*>) #-}
131 ZipAsyncM m1 <*> ZipAsyncM m2 = ZipAsyncM $ zipAsyncWithK id m1 m2