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