1979370 · streamly · git.iko.soy

benchmark/Linear.hs+6

compare with master

unifiedsplit

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

26import qualified Streamly.Prelude as S

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

28

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

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

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

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

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

31import qualified Streamly.Internal.Prelude as IP

32import qualified Streamly.Internal.Prelude as IP

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

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

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

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

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

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

359 , 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+

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

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

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

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

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

367 , 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

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

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

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

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

56

57

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

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

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

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

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

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

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

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

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

384

385

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+

385{-# INLINE mapM #-}

394{-# INLINE mapM #-}

386{-# INLINE map' #-}

395{-# INLINE map' #-}

387{-# INLINE fmap' #-}

396{-# INLINE fmap' #-}

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

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

307 , foldParallel

308

309 , lastN

310+

311+ -- * Time related

312+ , takeByTime

313+ , dropByTime

310 )

314 )

311where

315where

312

316

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

342

346

343import Streamly.Internal.Mutable.Prim.Var

347import Streamly.Internal.Mutable.Prim.Var

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

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

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

350+

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

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

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

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

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

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

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

4034 foldFunc i

4040 foldFunc i

4035 | i < n = RB.unsafeFoldRingM

4041 | i < n = RB.unsafeFoldRingM

4036 | otherwise = RB.unsafeFoldRingFullM

4042 | 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

413 -- * Diagnostics

414 , inspectMode

415

416+ -- * Time related

417+ , takeByTime

418+ , dropByTime

419+

416 -- * Deprecated

420 -- * Deprecated

417 , K.once

421 , K.once

418 , each

422 , each

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

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

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

471import Streamly.Internal.Data.Time.Units

475import Streamly.Internal.Data.Time.Units

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

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

473− toAbsTime)

477+ toAbsTime, TimeUnit64)

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

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

475

479

476import Streamly.Internal.Data.Strict

480import Streamly.Internal.Data.Strict

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

4191{-# INLINE runStateT #-}

4195{-# INLINE runStateT #-}

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

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

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

4197runStateT 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

412 , exceptions >= 0.8 && < 0.11

413 default-language: Haskell2010

414

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+

415test-suite pure-streams-base

428test-suite pure-streams-base

416 import: test-options

429 import: test-options

417 type: exitcode-stdio-1.0

430 type: exitcode-stdio-1.0

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

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

1+{-# LANGUAGE ScopedTypeVariables #-}

2+

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+