git.iko.soy/streamly
on a79caed help sign in
streamly/src/Streamly/Internal/Data/Stream/ZipAsync.hs
134 lines 4.7 kB history raw
1{-# OPTIONS_GHC -Wno-deprecations #-}
2{-# LANGUAGE UndecidableInstances #-}
4-- |
5-- Module : Streamly.Internal.Data.Stream.ZipAsync
6-- Copyright : (c) 2017 Composewell Technologies
7-- License : BSD-3-Clause
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
10-- Portability : GHC
11--
12-- To run examples in this module:
13--
14-- >>> import qualified Streamly.Prelude as Stream
15--
16module Streamly.Internal.Data.Stream.ZipAsync
17 {-# DEPRECATED "Use \"Streamly.Data.Stream.MkType\" instead." #-}
18 ( ZipAsyncM(..)
19 , ZipAsync
20 , consMZipAsync
21 , zipAsyncWithK
22 , zipAsyncWithMK
23 )
24where
26import Streamly.Internal.Control.Concurrent (MonadAsync)
27import Streamly.Internal.Data.StreamK (Stream)
29import qualified Streamly.Internal.Data.StreamK as K
30 (mkStream, foldStream, zipWithM, consM)
31import qualified Streamly.Internal.Data.Stream as D (fromStreamK)
32import qualified Streamly.Internal.Data.Stream.Serial as Serial
33import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar
34import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar
35import qualified Streamly.Internal.Data.Stream.Serial as Stream
36 (fromStreamK, toStreamK)
37import Streamly.Internal.Data.SVar
39import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)
41#include "Instances.hs"
43-- $setup
44-- >>> :set -fno-warn-deprecations
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