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