git.iko.soy/streamly
on bfeee8a help sign in
streamly/src/Streamly/Internal/Data/Stream/ZipAsync.hs
133 lines 4.8 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 {-# DEPRECATED "Please use \"Streamly.Internal.Data.Stream.Zip.Concurrent\" instead." #-}
17 ( ZipAsyncM(..)
18 , ZipAsync
19 , consMZipAsync
20 , zipAsyncWithK
21 , zipAsyncWithMK
22 )
23where
25import Streamly.Internal.Control.Concurrent (MonadAsync)
26import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
28import qualified Streamly.Internal.Data.Stream.StreamK as K
29 (mkStream, foldStream, zipWithM, consM)
30import qualified Streamly.Internal.Data.Stream.StreamD as D (fromStreamK)
31import qualified Streamly.Internal.Data.Stream.Serial as Serial
32import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar
33import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar
34import qualified Streamly.Internal.Data.Stream.Serial as Stream
35 (fromStreamK, toStreamK)
36import Streamly.Internal.Data.SVar
38import Prelude hiding (map, repeat, zipWith, errorWithoutStackTrace)
40#include "Instances.hs"
42-- $setup
43-- >>> :set -fno-warn-deprecations
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
67 $ K.zipWithM f m1 (Stream.toStreamK (SVar.fromSVar sv))
69-- XXX Should we rename this to zipParWith or zipParallelWith? This can happen
70-- along with the change of behvaior to end the stream concurrently.
71--
72-- | Like 'zipWith' but zips concurrently i.e. both the streams being zipped
73-- are evaluated concurrently using the 'ParallelT' concurrent evaluation
74-- style. The maximum number of elements of each stream evaluated in advance
75-- can be controlled by 'maxBuffer'.
76--
77-- The stream ends if stream @a@ or stream @b@ ends. However, if stream @b@
78-- ends while we are still evaluating stream @a@ and waiting for a result then
79-- stream will not end until after the evaluation of stream @a@ finishes. This
80-- behavior can potentially be changed in future to end the stream immediately
81-- as soon as any of the stream end is detected.
82--
83-- @since 0.1.0
84{-# INLINE zipAsyncWithK #-}
85zipAsyncWithK :: MonadAsync m
86 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
87zipAsyncWithK f = zipAsyncWithMK (\a b -> return (f a b))
89------------------------------------------------------------------------------
90-- Parallely Zipping Streams
91------------------------------------------------------------------------------
92--
93-- | For 'ZipAsyncM' streams:
94--
95-- @
96-- (<>) = 'Streamly.Prelude.serial'
97-- (<*>) = 'Streamly.Prelude.serial.zipAsyncWith' id
98-- @
99--
100-- Applicative evaluates the streams being zipped concurrently, the following
101-- would take half the time that it would take in serial zipping:
102--
103-- >>> s = Stream.fromFoldableM $ Prelude.map delay [1, 1, 1]
104-- >>> Stream.toList $ Stream.fromZipAsync $ (,) <$> s <*> s
105-- ...
106-- [(1,1),(1,1),(1,1)]
107--
108-- /Since: 0.2.0 ("Streamly")/
109--
110-- @since 0.8.0
111newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: Stream m a}
112 deriving (Semigroup, Monoid)
114-- | An IO stream whose applicative instance zips streams wAsyncly.
115--
116-- /Since: 0.2.0 ("Streamly")/
117--
118-- @since 0.8.0
119type ZipAsync = ZipAsyncM IO
121consMZipAsync :: Monad m => m a -> ZipAsyncM m a -> ZipAsyncM m a
122consMZipAsync m (ZipAsyncM r) = ZipAsyncM $ K.consM m r
124instance Monad m => Functor (ZipAsyncM m) where
125 {-# INLINE fmap #-}
126 fmap f (ZipAsyncM m) =
127 ZipAsyncM $ Stream.toStreamK $ fmap f (Stream.fromStreamK m)
129instance MonadAsync m => Applicative (ZipAsyncM m) where
130 pure = ZipAsyncM . Stream.toStreamK . Serial.repeat
132 {-# INLINE (<*>) #-}
133 ZipAsyncM m1 <*> ZipAsyncM m2 = ZipAsyncM $ zipAsyncWithK id m1 m2