ba4d23b · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add correct impl of takeInterval & dropInterval to Stream.Parallel

ba4d23bAdithya Kumar4 years ago

benchmark/Streamly/Benchmark/Data/Stream/Transform.hs+12−17

compare with master

unifiedsplit

@@ -35,6 +35,7 @@ import qualified Streamly.Internal.Data.Unfold as Unfold

35import qualified Streamly.Internal.Data.Stream.IsStream as Stream

36import Streamly.Internal.Data.Time.Units

37#else

38+import qualified Streamly.Internal.Data.Stream.Parallel as Stream

38import qualified Streamly.Internal.Data.Stream as Stream

39import qualified Streamly.Internal.Data.Stream as Stream

39#endif

40#endif

@@ -301,17 +302,17 @@ takeWhileTrue value n = composeN n $ Stream.takeWhile (<= (value + 1))

301takeWhileMTrue :: MonadIO m => Int -> Int -> Stream m Int -> m ()

302takeWhileMTrue :: MonadIO m => Int -> Int -> Stream m Int -> m ()

302takeWhileMTrue value n = composeN n $ Stream.takeWhileM (return . (<= (value + 1)))

303takeWhileMTrue value n = composeN n $ Stream.takeWhileM (return . (<= (value + 1)))

303

304

304−#ifdef USE_PRELUDE

305{-# INLINE takeInterval #-}

306−takeInterval :: NanoSecond64 -> Int -> Stream IO Int -> IO ()

306+takeInterval :: Double -> Int -> Stream IO Int -> IO ()

307takeInterval i n = composeN n (Stream.takeInterval i)

308

309+-- Inspection testing is disabled for takeInterval

310+-- Enable it when looking at it throughly

309#ifdef INSPECTION

311#ifdef INSPECTION

310-- inspect $ hasNoType 'takeInterval ''SPEC

312-- inspect $ hasNoType 'takeInterval ''SPEC

311−inspect $ hasNoTypeClasses 'takeInterval

313+-- inspect $ hasNoTypeClasses 'takeInterval

312-- inspect $ 'takeInterval `hasNoType` ''D.Step

314-- inspect $ 'takeInterval `hasNoType` ''D.Step

313#endif

315#endif

314−#endif

315

316

316{-# INLINE dropOne #-}

317{-# INLINE dropOne #-}

317dropOne :: MonadIO m => Int -> Stream m Int -> m ()

318dropOne :: MonadIO m => Int -> Stream m Int -> m ()

@@ -338,16 +339,18 @@ dropWhileFalse value n = composeN n $ Stream.dropWhile (> (value + 1))

338{-# INLINE _intervalsOfSum #-}

339{-# INLINE _intervalsOfSum #-}

339_intervalsOfSum :: MonadAsync m => Double -> Int -> Stream m Int -> m ()

340_intervalsOfSum :: MonadAsync m => Double -> Int -> Stream m Int -> m ()

340_intervalsOfSum i n = composeN n (Stream.intervalsOf i FL.sum)

341_intervalsOfSum i n = composeN n (Stream.intervalsOf i FL.sum)

342+#endif

341

343

342{-# INLINE dropInterval #-}

344{-# INLINE dropInterval #-}

343−dropInterval :: NanoSecond64 -> Int -> Stream IO Int -> IO ()

345+dropInterval :: Double -> Int -> Stream IO Int -> IO ()

344dropInterval i n = composeN n (Stream.dropInterval i)

346dropInterval i n = composeN n (Stream.dropInterval i)

345

347

348+-- Inspection testing is disabled for dropInterval

349+-- Enable it when looking at it throughly

346#ifdef INSPECTION

350#ifdef INSPECTION

347−inspect $ hasNoTypeClasses 'dropInterval

351+-- inspect $ hasNoTypeClasses 'dropInterval

348-- inspect $ 'dropInterval `hasNoType` ''D.Step

352-- inspect $ 'dropInterval `hasNoType` ''D.Step

349#endif

353#endif

350−#endif

351

354

352{-# INLINE findIndices #-}

355{-# INLINE findIndices #-}

353findIndices :: MonadIO m => Int -> Int -> Stream m Int -> m ()

356findIndices :: MonadIO m => Int -> Int -> Stream m Int -> m ()

@@ -403,16 +406,8 @@ o_1_space_filtering value =

403 -- , benchIOSink value "takeWhileM-true" (_takeWhileMTrue value 1)

406 -- , benchIOSink value "takeWhileM-true" (_takeWhileMTrue value 1)

404 , benchIOSink value "drop-one" (dropOne 1)

407 , benchIOSink value "drop-one" (dropOne 1)

405 , benchIOSink value "drop-all" (dropAll value 1)

408 , benchIOSink value "drop-all" (dropAll value 1)

406−#ifdef USE_PRELUDE

407− , benchIOSink

408− value

409− "takeInterval-all"

409+ , benchIOSink value "takeInterval-all" (takeInterval 10000 1)

410− (takeInterval (NanoSecond64 maxBound) 1)

411− , benchIOSink

412− value

413− "dropInterval-all"

410+ , benchIOSink value "dropInterval-all" (dropInterval 10000 1)

414− (dropInterval (NanoSecond64 maxBound) 1)

415−#endif

416 , benchIOSink value "dropWhile-true" (dropWhileTrue value 1)

411 , benchIOSink value "dropWhile-true" (dropWhileTrue value 1)

417 -- , benchIOSink value "dropWhileM-true" (_dropWhileMTrue value 1)

412 -- , benchIOSink value "dropWhileM-true" (_dropWhileMTrue value 1)

418 , benchIOSink

413 , benchIOSink

core/src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+1−84

@@ -78,11 +78,9 @@ module Streamly.Internal.Data.Stream.StreamD.Transform

78 -- * Trimming

79 -- | Produce a subset of the stream trimmed at ends.

80 , take

81− , takeByTime

82 , takeWhile

81 , takeWhile

83 , takeWhileM

82 , takeWhileM

84 , drop

83 , drop

85− , dropByTime

86 , dropWhile

84 , dropWhile

87 , dropWhileM

85 , dropWhileM

@@ -133,17 +131,14 @@ import Control.Monad.Trans.Class (MonadTrans(lift))

133import Data.Maybe (fromJust, isJust)

131import Data.Maybe (fromJust, isJust)

134import Fusion.Plugin.Types (Fuse(..))

132import Fusion.Plugin.Types (Fuse(..))

135import GHC.Types (SPEC(..))

133import GHC.Types (SPEC(..))

136−import qualified Data.Set as Set

137

134

138import Streamly.Internal.Control.Concurrent (MonadAsync)

135import Streamly.Internal.Control.Concurrent (MonadAsync)

139import Streamly.Internal.Control.ForkLifted (forkManaged)

136import Streamly.Internal.Control.ForkLifted (forkManaged)

140import Streamly.Internal.Data.Fold.Type (Fold(..))

137import Streamly.Internal.Data.Fold.Type (Fold(..))

141import Streamly.Internal.Data.Pipe.Type (Pipe(..), PipeState(..))

138import Streamly.Internal.Data.Pipe.Type (Pipe(..), PipeState(..))

142import Streamly.Internal.Data.SVar.Type (defState, adaptState)

139import Streamly.Internal.Data.SVar.Type (defState, adaptState)

143−import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

144−import Streamly.Internal.Data.Time.Units

145− (TimeUnit64, toRelTime64, diffAbsTime64)

146

140

141+import qualified Data.Set as Set

147import qualified Streamly.Internal.Data.Fold.Type as FL

142import qualified Streamly.Internal.Data.Fold.Type as FL

148import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed

143import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed

149import qualified Streamly.Internal.Data.Pipe.Type as Pipe

144import qualified Streamly.Internal.Data.Pipe.Type as Pipe

@@ -791,84 +786,6 @@ deleteBy eq x (Stream step state) = Stream step' (state, False)

791-- Trimming

786-- Trimming

792------------------------------------------------------------------------------

787------------------------------------------------------------------------------

793

788

794−-- XXX using getTime in the loop can be pretty expensive especially for

795−-- computations where iterations are lightweight. We have the following

796−-- options:

797−--

798−-- 1) Run a timeout thread updating a flag asynchronously and check that

799−-- flag here, that way we can have a cheap termination check.

800−--

801−-- 2) Use COARSE clock to get time with lower resolution but more efficiently.

802−--

803−-- 3) Use rdtscp/rdtsc to get time directly from the processor, compute the

804−-- termination value of rdtsc in the beginning and then in each iteration just

805−-- get rdtsc and check if we should terminate.

806−--

807−data TakeByTime st s

808− = TakeByTimeInit st

809− | TakeByTimeCheck st s

810− | TakeByTimeYield st s

811−

812−{-# INLINE_NORMAL takeByTime #-}

813−takeByTime :: (MonadIO m, TimeUnit64 t) => t -> Stream m a -> Stream m a

814−takeByTime duration (Stream step1 state1) = Stream step (TakeByTimeInit state1)

815− where

816−

817− lim = toRelTime64 duration

818−

819− {-# INLINE_LATE step #-}

820− step _ (TakeByTimeInit _) | lim == 0 = return Stop

821− step _ (TakeByTimeInit st) = do

822− t0 <- liftIO $ getTime Monotonic

823− return $ Skip (TakeByTimeYield st t0)

824− step _ (TakeByTimeCheck st t0) = do

825− t <- liftIO $ getTime Monotonic

826− return $

827− if diffAbsTime64 t t0 > lim

828− then Stop

829− else Skip (TakeByTimeYield st t0)

830− step gst (TakeByTimeYield st t0) = do

831− r <- step1 gst st

832− return $ case r of

833− Yield x s -> Yield x (TakeByTimeCheck s t0)

834− Skip s -> Skip (TakeByTimeCheck s t0)

835− Stop -> Stop

836−

837−data DropByTime st s x

838− = DropByTimeInit st

839− | DropByTimeGen st s

840− | DropByTimeCheck st s x

841− | DropByTimeYield st

842−

843−{-# INLINE_NORMAL dropByTime #-}

844−dropByTime :: (MonadIO m, TimeUnit64 t) => t -> Stream m a -> Stream m a

845−dropByTime duration (Stream step1 state1) = Stream step (DropByTimeInit state1)

846− where

847−

848− lim = toRelTime64 duration

849−

850− {-# INLINE_LATE step #-}

851− step _ (DropByTimeInit st) = do

852− t0 <- liftIO $ getTime Monotonic

853− return $ Skip (DropByTimeGen st t0)

854− step gst (DropByTimeGen st t0) = do

855− r <- step1 gst st

856− return $ case r of

857− Yield x s -> Skip (DropByTimeCheck s t0 x)

858− Skip s -> Skip (DropByTimeGen s t0)

859− Stop -> Stop

860− step _ (DropByTimeCheck st t0 x) = do

861− t <- liftIO $ getTime Monotonic

862− if diffAbsTime64 t t0 <= lim

863− then return $ Skip $ DropByTimeGen st t0

864− else return $ Yield x $ DropByTimeYield st

865− step gst (DropByTimeYield st) = do

866− r <- step1 gst st

867− return $ case r of

868− Yield x s -> Yield x (DropByTimeYield s)

869− Skip s -> Skip (DropByTimeYield s)

870− Stop -> Stop

871−

872-- Adapted from the vector package

789-- Adapted from the vector package

873{-# INLINE_NORMAL drop #-}

790{-# INLINE_NORMAL drop #-}

874drop :: Monad m => Int -> Stream m a -> Stream m a

791drop :: Monad m => Int -> Stream m a -> Stream m a

src/Streamly/Internal/Data/Stream/IsStream/Transform.hs+2−42

@@ -85,7 +85,6 @@ module Streamly.Internal.Data.Stream.IsStream.Transform

85 -- | Produce a subset of the stream trimmed at ends.

87 , take

88− , takeInterval

89 , takeLast

88 , takeLast

90 , takeLastInterval

89 , takeLastInterval

91 , takeWhile

90 , takeWhile

@@ -93,7 +92,6 @@ module Streamly.Internal.Data.Stream.IsStream.Transform

93 , takeWhileLast

92 , takeWhileLast

94 , takeWhileAround

93 , takeWhileAround

95 , drop

94 , drop

96− , dropInterval

97 , dropLast

95 , dropLast

98 , dropLastInterval

96 , dropLastInterval

99 , dropWhile

97 , dropWhile

@@ -250,7 +248,7 @@ import Streamly.Internal.Data.Stream.IsStream.Type

250 (IsStream(..), fromStreamD, toStreamD, toConsK)

248 (IsStream(..), fromStreamD, toStreamD, toConsK)

251import Streamly.Internal.Data.Stream.Serial (SerialT)

249import Streamly.Internal.Data.Stream.Serial (SerialT)

252import Streamly.Internal.Data.SVar (Rate(..))

250import Streamly.Internal.Data.SVar (Rate(..))

253−import Streamly.Internal.Data.Time.Units (TimeUnit64, AbsTime, RelTime64)

251+import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)

254

252

255import qualified Streamly.Data.Fold as FL

253import qualified Streamly.Data.Fold as FL

256import qualified Streamly.Internal.Data.Stream.Parallel as Par

254import qualified Streamly.Internal.Data.Stream.Parallel as Par

@@ -259,7 +257,7 @@ import qualified Streamly.Internal.Data.Stream.StreamD as D

259 (transform, foldrT, tap, tapOffsetEvery, pollCounts, mapM, scanOnce

257 (transform, foldrT, tap, tapOffsetEvery, pollCounts, mapM, scanOnce

260 , scanMany, postscanOnce, scanlx', scanlM', scanl', postscanl', prescanl'

258 , scanMany, postscanOnce, scanlx', scanlM', scanl', postscanl', prescanl'

261 , prescanlM', scanl1M', scanl1', filter, filterM, uniq, deleteBy, takeWhileM

259 , prescanlM', scanl1M', scanl1', filter, filterM, uniq, deleteBy, takeWhileM

262− , takeByTime, dropWhile, dropWhileM, dropByTime, insertBy, intersperse

260+ , dropWhile, dropWhileM, insertBy, intersperse

263 , intersperseM_, intersperseSuffix, intersperseSuffix_

261 , intersperseM_, intersperseSuffix, intersperseSuffix_

264 , intersperseSuffixBySpan, indexed, indexedR, rollingMap, rollingMapM

262 , intersperseSuffixBySpan, indexed, indexedR, rollingMap, rollingMapM

265 , rollingMap2, mapMaybe, mapMaybeM)

263 , rollingMap2, mapMaybe, mapMaybeM)

@@ -1059,26 +1057,6 @@ takeWhileAround :: -- (IsStream t, Monad m) =>

1059 (a -> Bool) -> t m a -> t m a

1057 (a -> Bool) -> t m a -> t m a

1060takeWhileAround = undefined -- fromStreamD $ D.takeWhileAround n $ toStreamD m

1058takeWhileAround = undefined -- fromStreamD $ D.takeWhileAround n $ toStreamD m

1061

1059

1062−-- | @takeInterval duration@ yields stream elements upto specified time

1063−-- @duration@. The duration starts when the stream is evaluated for the first

1064−-- time, before the first element is yielded. The time duration is checked

1065−-- before generating each element, if the duration has expired the stream

1066−-- stops.

1067−--

1068−-- The total time taken in executing the stream is guaranteed to be /at least/

1069−-- @duration@, however, because the duration is checked before generating an

1070−-- element, the upper bound is indeterminate and depends on the time taken in

1071−-- generating and processing the last element.

1072−--

1073−-- No element is yielded if the duration is zero. At least one element is

1074−-- yielded if the duration is non-zero.

1075−--

1076−-- /Pre-release/

1077−--

1078−{-# INLINE takeInterval #-}

1079−takeInterval ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a

1080−takeInterval d = fromStreamD . D.takeByTime d . toStreamD

1081−

1082-- | Drop elements in the stream as long as the predicate succeeds and then

1060-- | Drop elements in the stream as long as the predicate succeeds and then

1083-- take the rest of the stream.

1061-- take the rest of the stream.

1084--

1062--

@@ -1094,24 +1072,6 @@ dropWhile p m = fromStreamD $ D.dropWhile p $ toStreamD m

1094dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a

1072dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a

1095dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m

1073dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m

1096

1074

1097−-- | @dropInterval duration@ drops stream elements until specified @duration@ has

1098−-- passed. The duration begins when the stream is evaluated for the first

1099−-- time. The time duration is checked /after/ generating a stream element, the

1100−-- element is yielded if the duration has expired otherwise it is dropped.

1101−--

1102−-- The time elapsed before starting to generate the first element is /at most/

1103−-- @duration@, however, because the duration expiry is checked after the

1104−-- element is generated, the lower bound is indeterminate and depends on the

1105−-- time taken in generating an element.

1106−--

1107−-- All elements are yielded if the duration is zero.

1108−--

1109−-- /Pre-release/

1110−--

1111−{-# INLINE dropInterval #-}

1112−dropInterval ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a

1113−dropInterval d = fromStreamD . D.dropByTime d . toStreamD

1114−

1115-- | Drop @n@ elements at the end of the stream.

1075-- | Drop @n@ elements at the end of the stream.

1116--

1076--

1117-- O(n) space, where n is the number elements dropped.

1077-- O(n) space, where n is the number elements dropped.

src/Streamly/Internal/Data/Stream/Parallel.hs+116−22

@@ -11,7 +11,7 @@

11--

12-- To run examples in this module:

13--

14−-- >>> import qualified Streamly.Prelude as Stream

14+-- >>> import qualified Streamly.Data.Stream as Stream

15-- >>> import Control.Concurrent (threadDelay)

16-- >>> :{

17-- delay n = do

@@ -42,10 +42,15 @@ module Streamly.Internal.Data.Stream.Parallel

43 -- * Callbacks

44 , newCallbackStream

45+

46+ -- * Combinators

47+ , interjectSuffix

48+ , takeInterval

49+ , dropInterval

45 )

50 )

46where

51where

48−import Control.Concurrent (myThreadId, takeMVar)

53+import Control.Concurrent (myThreadId, takeMVar, threadDelay)

49import Control.Monad (when)

54import Control.Monad (when)

50import Control.Monad.Base (MonadBase(..), liftBaseDefault)

55import Control.Monad.Base (MonadBase(..), liftBaseDefault)

51import Control.Monad.Catch (MonadThrow, throwM)

56import Control.Monad.Catch (MonadThrow, throwM)

@@ -56,37 +61,36 @@ import Control.Monad.State.Class (MonadState(..))

56import Control.Monad.Trans.Class (MonadTrans(lift))

61import Control.Monad.Trans.Class (MonadTrans(lift))

57import Data.Functor (void)

62import Data.Functor (void)

58import Data.IORef (readIORef, writeIORef)

63import Data.IORef (readIORef, writeIORef)

59−import Data.Maybe (fromJust)

64+import Data.Maybe (fromJust, isNothing)

60#if __GLASGOW_HASKELL__ < 808

65#if __GLASGOW_HASKELL__ < 808

61import Data.Semigroup (Semigroup(..))

66import Data.Semigroup (Semigroup(..))

62#endif

67#endif

63−import Prelude hiding (map)

64−

65−import qualified Data.Set as Set

67import Streamly.Data.Fold (Fold)

69import Streamly.Data.Fold (Fold)

68import Streamly.Internal.Control.Concurrent (MonadAsync)

70import Streamly.Internal.Control.Concurrent (MonadAsync)

69import Streamly.Internal.Data.Stream.StreamD.Type (Step(..))

71import Streamly.Internal.Data.Stream.StreamD.Type (Step(..))

70−import Streamly.Internal.Data.Stream.StreamK.Type (Stream)

72+import Streamly.Internal.Data.Stream.Type (Stream)

74+import qualified Data.Set as Set

72import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

75import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

73− (foldStreamShared, mkStream, foldStream, fromEffect

76+ (Stream, foldStreamShared, mkStream, foldStream, fromEffect

74 , nil, concatMapWith, fromPure, bindWith, withLocal)

77 , nil, concatMapWith, fromPure, bindWith, withLocal)

75import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

78import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

76 (Stream(..), mapM, toStreamK, fromStreamK)

79 (Stream(..), mapM, toStreamK, fromStreamK)

77import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar

80import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar

78import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar

81import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar

79import qualified Streamly.Internal.Data.Stream as Stream

82import qualified Streamly.Internal.Data.Stream as Stream

80− (toStreamK, fromStreamK)

83+ (catMaybes, dropWhile, fromStreamK, repeat, sequence, takeWhile, toStreamK)

82import Streamly.Internal.Data.SVar

85import Streamly.Internal.Data.SVar

86+import Prelude hiding (map)

84#include "inline.hs"

88#include "inline.hs"

85#include "Instances.hs"

89#include "Instances.hs"

87--

91--

88-- $setup

92-- $setup

89−-- >>> import qualified Streamly.Prelude as Stream

93+-- >>> import qualified Streamly.Data.Stream as Stream

90-- >>> import Control.Concurrent (threadDelay)

94-- >>> import Control.Concurrent (threadDelay)

91-- >>> :{

95-- >>> :{

92-- delay n = do

96-- delay n = do

@@ -106,7 +110,7 @@ import Streamly.Internal.Data.SVar

106{-# NOINLINE runOne #-}

110{-# NOINLINE runOne #-}

107runOne

111runOne

108 :: MonadIO m

112 :: MonadIO m

109− => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()

113+ => State K.Stream m a -> K.Stream m a -> Maybe WorkerInfo -> m ()

110runOne st m0 winfo =

114runOne st m0 winfo =

111 case getYieldLimit st of

115 case getYieldLimit st of

112 Nothing -> go m0

116 Nothing -> go m0

@@ -129,7 +133,7 @@ runOne st m0 winfo =

129

133

130runOneLimited

134runOneLimited

131 :: MonadIO m

135 :: MonadIO m

132− => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()

136+ => State K.Stream m a -> K.Stream m a -> Maybe WorkerInfo -> m ()

133runOneLimited st m0 winfo = go m0

137runOneLimited st m0 winfo = go m0

134

138

135 where

139 where

@@ -166,7 +170,7 @@ runOneLimited st m0 winfo = go m0

166

170

167{-# NOINLINE forkSVarPar #-}

171{-# NOINLINE forkSVarPar #-}

168forkSVarPar :: MonadAsync m

172forkSVarPar :: MonadAsync m

169− => SVarStopStyle -> Stream m a -> Stream m a -> Stream m a

173+ => SVarStopStyle -> K.Stream m a -> K.Stream m a -> K.Stream m a

170forkSVarPar ss m r = K.mkStream $ \st yld sng stp -> do

174forkSVarPar ss m r = K.mkStream $ \st yld sng stp -> do

171 sv <- newParallelVar ss st

175 sv <- newParallelVar ss st

172 pushWorkerPar sv (runOne st{streamVar = Just sv} m)

176 pushWorkerPar sv (runOne st{streamVar = Just sv} m)

@@ -179,8 +183,13 @@ forkSVarPar ss m r = K.mkStream $ \st yld sng stp -> do

179 K.foldStream st yld sng stp $ Stream.toStreamK (SVar.fromSVar sv)

183 K.foldStream st yld sng stp $ Stream.toStreamK (SVar.fromSVar sv)

180

184

181{-# INLINE joinStreamVarPar #-}

185{-# INLINE joinStreamVarPar #-}

182−joinStreamVarPar :: MonadAsync m

186+joinStreamVarPar ::

187+ MonadAsync m

188+ => SVarStyle

183− => SVarStyle -> SVarStopStyle -> Stream m a -> Stream m a -> Stream m a

189+ -> SVarStopStyle

190+ -> K.Stream m a

191+ -> K.Stream m a

192+ -> K.Stream m a

184joinStreamVarPar style ss m1 m2 = K.mkStream $ \st yld sng stp ->

193joinStreamVarPar style ss m1 m2 = K.mkStream $ \st yld sng stp ->

185 case streamVar st of

194 case streamVar st of

186 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do

195 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do

@@ -230,7 +239,7 @@ joinStreamVarPar style ss m1 m2 = K.mkStream $ \st yld sng stp ->

230-------------------------------------------------------------------------------

239-------------------------------------------------------------------------------

231

240

232{-# INLINE parallelK #-}

241{-# INLINE parallelK #-}

233−parallelK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

242+parallelK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a

234parallelK = joinStreamVarPar ParallelVar StopNone

243parallelK = joinStreamVarPar ParallelVar StopNone

235

244

236-- | XXX we can implement it more efficienty by directly implementing instead

245-- | XXX we can implement it more efficienty by directly implementing instead

@@ -248,7 +257,7 @@ consM m (ParallelT r) = ParallelT $ parallelK (K.fromEffect m) r

248--

257--

249-- /Pre-release/

258-- /Pre-release/

250{-# INLINE parallelFstK #-}

259{-# INLINE parallelFstK #-}

251−parallelFstK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

260+parallelFstK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a

252parallelFstK = joinStreamVarPar ParallelVar StopBy

261parallelFstK = joinStreamVarPar ParallelVar StopBy

253

262

254-- This is a race like combinator for streams.

263-- This is a race like combinator for streams.

@@ -258,7 +267,7 @@ parallelFstK = joinStreamVarPar ParallelVar StopBy

258--

267--

259-- /Pre-release/

268-- /Pre-release/

260{-# INLINE parallelMinK #-}

269{-# INLINE parallelMinK #-}

261−parallelMinK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

270+parallelMinK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a

262parallelMinK = joinStreamVarPar ParallelVar StopAny

271parallelMinK = joinStreamVarPar ParallelVar StopAny

263

272

264------------------------------------------------------------------------------

273------------------------------------------------------------------------------

@@ -269,7 +278,7 @@ parallelMinK = joinStreamVarPar ParallelVar StopAny

269--

278--

270-- /Pre-release/

279-- /Pre-release/

271--

280--

272−mkParallelK :: MonadAsync m => Stream m a -> Stream m a

281+mkParallelK :: MonadAsync m => K.Stream m a -> K.Stream m a

273mkParallelK m = K.mkStream $ \st yld sng stp -> do

282mkParallelK m = K.mkStream $ \st yld sng stp -> do

274 sv <- newParallelVar StopNone (adaptState st)

283 sv <- newParallelVar StopNone (adaptState st)

275 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)

284 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)

@@ -342,7 +351,8 @@ mkParallelD m = D.Stream step Nothing

342--

351--

343-- /Pre-release/

352-- /Pre-release/

344{-# INLINE tapAsyncK #-}

353{-# INLINE tapAsyncK #-}

354+tapAsyncK ::

345−tapAsyncK :: MonadAsync m => (Stream m a -> m b) -> Stream m a -> Stream m a

355+ MonadAsync m => (K.Stream m a -> m b) -> K.Stream m a -> K.Stream m a

346tapAsyncK f m = K.mkStream $ \st yld sng stp -> do

356tapAsyncK f m = K.mkStream $ \st yld sng stp -> do

347 sv <- SVar.newFoldSVar st (f . Stream.toStreamK)

357 sv <- SVar.newFoldSVar st (f . Stream.toStreamK)

348 K.foldStreamShared st yld sng stp

358 K.foldStreamShared st yld sng stp

@@ -420,7 +430,7 @@ tapAsyncF f (D.Stream step1 state1) = D.Stream step TapInit

420-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/

430-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/

421--

431--

422-- @since 0.8.0

432-- @since 0.8.0

423−newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}

433+newtype ParallelT m a = ParallelT {getParallelT :: K.Stream m a}

424 deriving (MonadTrans)

434 deriving (MonadTrans)

425

435

426-- | A parallely composing IO stream of elements of type @a@.

436-- | A parallely composing IO stream of elements of type @a@.

@@ -507,7 +517,7 @@ MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)

507-- /Pre-release/

517-- /Pre-release/

508--

518--

509{-# INLINE_NORMAL newCallbackStream #-}

519{-# INLINE_NORMAL newCallbackStream #-}

510−newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)

520+newCallbackStream :: MonadAsync m => m (a -> m (), K.Stream m a)

511newCallbackStream = do

521newCallbackStream = do

512 sv <- newParallelVar StopNone defState

522 sv <- newParallelVar StopNone defState

513

523

@@ -520,3 +530,87 @@ newCallbackStream = do

520 -- XXX we can return an SVar and then the consumer can unfold from the

530 -- XXX we can return an SVar and then the consumer can unfold from the

521 -- SVar?

531 -- SVar?

522 return (callback, D.toStreamK (SVar.fromSVarD sv))

532 return (callback, D.toStreamK (SVar.fromSVarD sv))

533+

534+------------------------------------------------------------------------------

535+-- Combinators

536+------------------------------------------------------------------------------

537+

538+{-# INLINE parallelFst #-}

539+parallelFst :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

540+parallelFst m1 m2 =

541+ Stream.fromStreamK

542+ $ parallelFstK (Stream.toStreamK m1) (Stream.toStreamK m2)

543+

544+-- | Intersperse a monadic action into the input stream after every @n@

545+-- seconds.

546+--

547+-- >>> import qualified Streamly.Data.Fold as Fold

548+-- >>> import qualified Streamly.Internal.Data.Stream.Parallel as Parallel

549+-- >>> Stream.fold Fold.drain $ Parallel.interjectSuffix 1.05 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"

550+-- h,e,l,l,o

551+--

552+-- /Pre-release/

553+{-# INLINE interjectSuffix #-}

554+interjectSuffix :: MonadAsync m => Double -> m a -> Stream m a -> Stream m a

555+interjectSuffix n f xs = xs `parallelFst` repeatM timed

556+ where timed = liftIO (threadDelay (round $ n * 1000000)) >> f

557+ repeatM = Stream.sequence . Stream.repeat

558+

559+-- XXX Notes from D.takeByTime (which was removed)

560+-- XXX using getTime in the loop can be pretty expensive especially for

561+-- computations where iterations are lightweight. We have the following

562+-- options:

563+--

564+-- 1) Run a timeout thread updating a flag asynchronously and check that

565+-- flag here, that way we can have a cheap termination check.

566+--

567+-- 2) Use COARSE clock to get time with lower resolution but more efficiently.

568+--

569+-- 3) Use rdtscp/rdtsc to get time directly from the processor, compute the

570+-- termination value of rdtsc in the beginning and then in each iteration just

571+-- get rdtsc and check if we should terminate.

572+

573+

574+-- | @takeInterval duration@ yields stream elements upto specified time

575+-- @duration@ in seconds. The duration starts when the stream is evaluated for

576+-- the first time, before the first element is yielded. The time duration is

577+-- checked before generating each element, if the duration has expired the

578+-- stream stops.

579+--

580+-- The total time taken in executing the stream is guaranteed to be /at least/

581+-- @duration@, however, because the duration is checked before generating an

582+-- element, the upper bound is indeterminate and depends on the time taken in

583+-- generating and processing the last element.

584+--

585+-- No element is yielded if the duration is zero. At least one element is

586+-- yielded if the duration is non-zero.

587+--

588+-- /Pre-release/

589+--

590+{-# INLINE takeInterval #-}

591+takeInterval :: MonadAsync m => Double -> Stream m a -> Stream m a

592+takeInterval d =

593+ Stream.catMaybes

594+ . Stream.takeWhile isNothing

595+ . interjectSuffix d (return Nothing) . fmap Just

596+

597+-- | @dropInterval duration@ drops stream elements until specified @duration@ in

598+-- seconds has passed. The duration begins when the stream is evaluated for the

599+-- first time. The time duration is checked /after/ generating a stream element,

600+-- the element is yielded if the duration has expired otherwise it is dropped.

601+--

602+-- The time elapsed before starting to generate the first element is /at most/

603+-- @duration@, however, because the duration expiry is checked after the

604+-- element is generated, the lower bound is indeterminate and depends on the

605+-- time taken in generating an element.

606+--

607+-- All elements are yielded if the duration is zero.

608+--

609+-- /Pre-release/

610+--

611+{-# INLINE dropInterval #-}

612+dropInterval :: MonadAsync m => Double -> Stream m a -> Stream m a

613+dropInterval d =

614+ Stream.catMaybes

615+ . Stream.dropWhile isNothing

616+ . interjectSuffix d (return Nothing) . fmap Just