1979370 · streamly · git.iko.soy

git.iko.soy/streamly

Add takeByTime and dropByTime combinators

commit197937037382165f4df7e5e13814a68c6fda6364 authoradithyaov <adi.obilisetty@gmail.com> dateWed 11 Dec 2019, 17:31 (7 years ago) parent750c7fe

6 files changed+173−1

compare with master

unifiedsplit

benchmark/Linear.hs+6 benchmark/Streamly/Benchmark/Prelude.hs+9 src/Streamly/Internal/Data/Stream/StreamD.hs+59 src/Streamly/Internal/Prelude.hs+27−1 streamly.cabal+13 test/Streamly/Test/Internal/Prelude.hs+59

benchmark/Linear.hs+6

@@ -26,6 +26,7 @@ import qualified Streamly.Memory.Array as A

2626import qualified Streamly.Prelude as S

2727import qualified Streamly.Internal.Data.Sink as Sink

2828

29+import Streamly.Internal.Data.Time.Units

2930import qualified Streamly.Internal.Memory.Array as IA

3031import qualified Streamly.Internal.Data.Fold as IFL

3132import qualified Streamly.Internal.Prelude as IP

@@ -356,6 +357,11 @@ main = do

356357 , benchIOSink value "filter-all-out" (Ops.filterAllOut value 1)

357358 , benchIOSink value "filter-all-in" (Ops.filterAllIn value 1)

358359 , benchIOSink value "take-all" (Ops.takeAll value 1)

360+

361+ -- XXX Compare this to takeAll

362+ , benchIOSink value "takeByTime: All" (Ops.takeByTime (NanoSecond64 maxBound) 1)

363+ , benchIOSink value "dropByTime: All" (Ops.dropByTime (NanoSecond64 maxBound) 1)

364+

359365 , benchIOSink value "takeWhile-true" (Ops.takeWhileTrue value 1)

360366 --, benchIOSink value "takeWhileM-true" (Ops.takeWhileMTrue 1)

361367 , benchIOSink value "drop-one" (Ops.dropOne 1)

benchmark/Streamly/Benchmark/Prelude.hs+9

@@ -53,6 +53,7 @@ import qualified Streamly.Internal.Data.Fold as FL

5353import qualified Streamly.Internal.Data.Unfold as UF

5454import qualified Streamly.Internal.Data.Pipe as Pipe

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

56+import Streamly.Internal.Data.Time.Units

5657

5758-- To detect memory leak issues use larger streams.

5859-- See Note in .cabal file on how to generate streams of specific size.

@@ -382,6 +383,14 @@ filterAllOut,

382383mapMaybeM :: S.MonadAsync m => Int -> Stream m Int -> m ()

383384intersperse :: S.MonadAsync m => Int -> Int -> Stream m Int -> m ()

384385

386+{-# INLINE takeByTime #-}

387+takeByTime :: (MonadIO m, TimeUnit64 t) => t -> Int -> Stream m Int -> m ()

388+takeByTime i n = composeN n (Internal.takeByTime i)

389+

390+{-# INLINE dropByTime #-}

391+dropByTime :: (MonadIO m, TimeUnit64 t) => t -> Int -> Stream m Int -> m ()

392+dropByTime i n = composeN n (Internal.dropByTime i)

393+

385394{-# INLINE mapM #-}

386395{-# INLINE map' #-}

387396{-# INLINE fmap' #-}

src/Streamly/Internal/Data/Stream/StreamD.hs+59

@@ -307,6 +307,10 @@ module Streamly.Internal.Data.Stream.StreamD

307307 , foldParallel

308308

309309 , lastN

310+

311+ -- * Time related

312+ , takeByTime

313+ , dropByTime

310314 )

311315where

312316

@@ -342,6 +346,8 @@ import qualified Prelude

342346

343347import Streamly.Internal.Mutable.Prim.Var

344348 (Prim, Var, readVar, newVar, modifyVar')

349+import Streamly.Internal.Data.Time.Units

350+

345351import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)

346352import Streamly.Internal.Memory.Array.Types (Array(..))

347353import Streamly.Internal.Data.Fold.Types (Fold(..))

@@ -4034,3 +4040,56 @@ lastN n = Fold step initial done

40344040 foldFunc i

40354041 | i < n = RB.unsafeFoldRingM

40364042 | otherwise = RB.unsafeFoldRingFullM

4043+

4044+------------------------------------------------------------------------------

4045+-- Time related

4046+------------------------------------------------------------------------------

4047+

4048+-- XXX This does not fuse. Removing the if function results in fusion though.

4049+-- XXX This function will fuse with the use of fusion-plugin.

4050+-- XXX This function also fuses if compiled with -funfolding-use-threshold=500

4051+{-# INLINE_NORMAL takeByTime #-}

4052+takeByTime :: (MonadIO m, TimeUnit64 t) => t -> Stream m a -> Stream m a

4053+takeByTime lim' (Stream step state) = Stream step' (state, Nothing)

4054+ where

4055+ lim = toRelTime64 lim'

4056+ {-# INLINE_LATE step' #-}

4057+ step' _ (s, Nothing) = do

4058+ t <- liftIO $ getTime Monotonic

4059+ return $ Skip (s, Just t)

4060+ step' gst (s, jt@(Just t)) = do

4061+ ns <- step gst s

4062+ t' <- liftIO $ getTime Monotonic

4063+ if diffAbsTime64 t' t > lim

4064+ then return Stop

4065+ else return $ case ns of

4066+ Yield a s' -> Yield a (s', jt)

4067+ Skip s' -> Skip (s', jt)

4068+ Stop -> Stop

4069+

4070+-- XXX This does not fuse. Removing the if function results in fusion though.

4071+-- XXX This function will fuse with the use of fusion-plugin.

4072+-- XXX This function also fuses if compiled with -funfolding-use-threshold=500

4073+{-# INLINE_NORMAL dropByTime #-}

4074+dropByTime :: (MonadIO m, TimeUnit64 t) => t -> Stream m a -> Stream m a

4075+dropByTime lim' (Stream step state) = Stream step' (state, Nothing)

4076+ where

4077+ lim = toRelTime64 lim'

4078+ {-# INLINE_LATE step' #-}

4079+ step' _ (s, Nothing) = do

4080+ t <- liftIO $ getTime Monotonic

4081+ return $ Skip (s, Just t)

4082+ step' gst (s, jt@(Just t)) = do

4083+ t' <- liftIO $ getTime Monotonic

4084+ ns <- step gst s

4085+ if diffAbsTime64 t' t <= lim

4086+ then return $

4087+ case ns of

4088+ Yield _ s' -> Skip (s', jt)

4089+ Skip s' -> Skip (s', jt)

4090+ Stop -> Stop

4091+ else return $

4092+ case ns of

4093+ Yield a s' -> Yield a (s', jt)

4094+ Skip s' -> Skip (s', jt)

4095+ Stop -> Stop

src/Streamly/Internal/Prelude.hs+27−1

@@ -413,6 +413,10 @@ module Streamly.Internal.Prelude

413413 -- * Diagnostics

414414 , inspectMode

415415

416+ -- * Time related

417+ , takeByTime

418+ , dropByTime

419+

416420 -- * Deprecated

417421 , K.once

418422 , each

@@ -470,7 +474,7 @@ import Streamly.Internal.Data.Stream.Zip (ZipSerialM)

470474import Streamly.Internal.Data.Pipe.Types (Pipe (..))

471475import Streamly.Internal.Data.Time.Units

472476 (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,

473− toAbsTime)

477+ toAbsTime, TimeUnit64)

474478import Streamly.Internal.Mutable.Prim.Var (Prim, Var)

475479

476480import Streamly.Internal.Data.Strict

@@ -4191,3 +4195,25 @@ usingStateT s f xs = evalStateT s $ f $ liftInner xs

41914195{-# INLINE runStateT #-}

41924196runStateT :: Monad m => s -> SerialT (StateT s m) a -> SerialT m (s, a)

41934197runStateT s xs = fromStreamD $ D.runStateT s (toStreamD xs)

4198+

4199+------------------------------------------------------------------------------

4200+-- Time related

4201+------------------------------------------------------------------------------

4202+

4203+-- | /takeByTime d/ keeps taking elements from the stream until d duration has

4204+-- passed. We first check the if the duration has expired, if yes, then we

4205+-- 'Stop' else we take the element, executing the corresponding monadic

4206+-- action. The only guarentee is, the element will not be taken (the action

4207+-- will not begin) if the duration has expired. If the duration has expired

4208+-- during the execution of the monadic action, the next element will not be

4209+-- taken but the current action will run to completion.

4210+{-# INLINE takeByTime #-}

4211+takeByTime ::(MonadIO m, IsStream t, TimeUnit64 u) => u -> t m a -> t m a

4212+takeByTime d = fromStreamD . D.takeByTime d . toStreamD

4213+

4214+-- | /dropByTime d/ keeps dropping elements from the stream until d duration has

4215+-- passed. The first element is taken only after the duration has expired.

4216+{-# INLINE dropByTime #-}

4217+dropByTime ::(MonadIO m, IsStream t, TimeUnit64 u) => u -> t m a -> t m a

4218+dropByTime d = fromStreamD . D.dropByTime d . toStreamD

4219+

streamly.cabal+13

@@ -412,6 +412,19 @@ test-suite test

412412 , exceptions >= 0.8 && < 0.11

413413 default-language: Haskell2010

414414

415+test-suite internal-prelude-test

416+ import: test-options

417+ type: exitcode-stdio-1.0

418+ main-is: Streamly/Test/Internal/Prelude.hs

419+ js-sources: jsbits/clock.js

420+ hs-source-dirs: test

421+ build-depends:

422+ streamly

423+ , base >= 4.8 && < 5

424+ , QuickCheck >= 2.10 && < 2.14

425+ , hspec >= 2.0 && < 3

426+ default-language: Haskell2010

427+

415428test-suite pure-streams-base

416429 import: test-options

417430 type: exitcode-stdio-1.0

test/Streamly/Test/Internal/Prelude.hsnew file+59

@@ -0,0 +1,59 @@

1+{-# LANGUAGE ScopedTypeVariables #-}

3+-- |

4+-- Module : Main

6+--

7+-- License : BSD-3-Clause

8+-- Maintainer : streamly@composewell.com

9+-- Stability : experimental

10+-- Portability : GHC

11+--

12+module Main (main) where

13+

14+import Test.Hspec.QuickCheck

15+import Test.QuickCheck (Property, forAll, choose)

16+import Test.QuickCheck.Monadic (monadicIO, assert, run)

17+

18+import Test.Hspec as H

19+

20+import qualified Streamly.Prelude as S

21+import qualified Streamly.Internal.Prelude as SI

22+

23+import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

24+import Streamly.Internal.Data.Time.Units

25+import Data.Int (Int64)

26+

27+tenPow8 :: Int64

28+tenPow8 = 10^(8 :: Int)

29+

30+tenPow7 :: Int64

31+tenPow7 = 10^(7 :: Int)

32+

33+testTakeByTime :: Property

34+testTakeByTime =

35+ forAll (choose (1, 10)) $ \(n :: Int64) -> monadicIO $ do

36+ let lim = NanoSecond64 $ n * tenPow8

37+ let graceTime = NanoSecond64 $ 8 * tenPow7

38+ t <- run $ getTime Monotonic

39+ t'' <- run $ S.last $ SI.takeByTime lim $ S.repeatM (getTime Monotonic)

40+ case t'' of

41+ Nothing -> assert True

42+ Just t' -> assert $ diffAbsTime64 t' t <= toRelTime64 (lim + graceTime)

43+

44+testDropByTime :: Property

45+testDropByTime =

46+ forAll (choose (1, 10)) $ \(n :: Int64) -> monadicIO $ do

47+ let lim = NanoSecond64 $ n * tenPow8

48+ t <- run $ getTime Monotonic

49+ _ <- run $ S.drain $ S.take 1 $ SI.dropByTime lim $ S.repeat ()

50+ t' <- run $ getTime Monotonic

51+ assert $ diffAbsTime64 t' t >= toRelTime64 lim

52+

53+main :: IO ()

54+main =

55+ hspec $

56+ describe "Time combinators" $ do

57+ prop "takeByTime" testTakeByTime

58+ prop "dropByTime" testDropByTime

59+