5054813 · streamly · git.iko.soy

git.iko.soy/streamly

Add currentTime and its benchmarks

commit50548138d6b0f41a17008ab60a695338d7cac0a7 authoradithyaov <adi.obilisetty@gmail.com> dateWed 1 Jan 2020, 15:01 (7 years ago) parentb19d96f

4 files changed+56

compare with master

unifiedsplit

benchmark/Linear.hs+7 benchmark/Streamly/Benchmark/Prelude.hs+5 src/Streamly/Internal/Data/Stream/StreamD.hs+26 src/Streamly/Internal/Prelude.hs+18

benchmark/Linear.hs+7

@@ -161,6 +161,13 @@ main = do

161161 -- These are essentially cons and consM

162162 , benchIOSrc serially "fromFoldable" (Ops.sourceFromFoldable value)

163163 , benchIOSrc serially "fromFoldableM" (Ops.sourceFromFoldableM value)

164+

165+ -- Compare this with takeAll

166+ , bench "currentTime: 0.00001 sec granularity" $ nfIO $ Ops.currentTime value 0.00001

167+ , bench "currentTime: 0.0001 sec granularity" $ nfIO $ Ops.currentTime value 0.0001

168+ , bench "currentTime: 0.001 sec granularity" $ nfIO $ Ops.currentTime value 0.001

169+ , bench "currentTime: 0.01 sec granularity" $ nfIO $ Ops.currentTime value 0.01

170+

164171 ]

165172 , bgroup "elimination"

166173 [ bgroup "reduce"

benchmark/Streamly/Benchmark/Prelude.hs+5

@@ -382,6 +382,11 @@ filterAllOut,

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

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

384384

385+-- XXX Change granularity and compare with takeAll

386+{-# INLINE currentTime #-}

387+currentTime :: S.MonadAsync m => Int -> Double -> m ()

388+currentTime value g = S.drain $ S.take value $ Internal.currentTime g

389+

385390{-# INLINE mapM #-}

386391{-# INLINE map' #-}

387392{-# INLINE fmap' #-}

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

@@ -311,6 +311,7 @@ module Streamly.Internal.Data.Stream.StreamD

311311 -- * Time related

312312 , takeByTime

313313 , dropByTime

314+ , currentTime

314315 )

315316where

316317

@@ -345,6 +346,7 @@ import qualified Control.Monad.Reader as Reader

345346import qualified Control.Monad.State.Strict as State

346347import qualified Prelude

347348

349+import Data.Int (Int64)

348350import Streamly.Internal.Mutable.Prim.Var

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

350352import Streamly.Internal.Data.Time.Units

@@ -355,6 +357,7 @@ import Streamly.Internal.Memory.Array.Types (Array(..))

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

356358import Streamly.Internal.Data.Pipe.Types (Pipe(..), PipeState(..))

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

360+import Streamly.Internal.Data.Time.Units (MicroSecond64(..), fromAbsTime, toAbsTime, AbsTime)

358361import Streamly.Internal.Data.Unfold.Types (Unfold(..))

359362import Streamly.Internal.Data.Strict (Tuple3'(..))

360363

@@ -4207,3 +4210,26 @@ dropByTime duration (Stream step1 state1) = Stream step (DropByTimeInit state1)

42074210 Yield x s -> Yield x (DropByTimeYield s)

42084211 Skip s -> Skip (DropByTimeYield s)

42094212 Stop -> Stop

4213+

4214+{-# INLINE_NORMAL currentTime #-}

4215+currentTime :: MonadAsync m => Double -> Stream m AbsTime

4216+currentTime g = Stream step Nothing

4217+ where

4218+ next timeVar = do

4219+ threadDelay $ delayTime

4220+ MicroSecond64 t <- fromAbsTime <$> getTime Monotonic

4221+ modifyVar' timeVar (const t)

4222+ next timeVar

4223+ g' = g * 10 ^ (6 :: Int)

4224+ {-# INLINE delayTime #-}

4225+ delayTime = if g' >= fromIntegral (maxBound :: Int)

4226+ then maxBound

4227+ else round g'

4228+ {-# INLINE_LATE step #-}

4229+ step _ Nothing = do

4230+ timeVar <- liftIO $ newVar (0 :: Int64)

4231+ tid <- forkManaged $ liftIO $ void $ next timeVar

4232+ return $ Skip $ Just (timeVar, tid)

4233+ step _ s@(Just (timeVar, _)) = do

4234+ a <- liftIO $ readVar timeVar

4235+ return $ Yield (toAbsTime (MicroSecond64 a)) s

src/Streamly/Internal/Prelude.hs+18

@@ -421,6 +421,9 @@ module Streamly.Internal.Prelude

421421 -- * Diagnostics

422422 , inspectMode

423423

424+ -- * Time related

425+ , currentTime

426+

424427 -- * Deprecated

425428 , K.once

426429 , each

@@ -4297,3 +4300,18 @@ usingStateT s f xs = evalStateT s $ f $ liftInner xs

42974300{-# INLINE runStateT #-}

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

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

4303+

4304+------------------------------------------------------------------------------

4305+-- Time related

4306+------------------------------------------------------------------------------

4307+

4308+-- | /currentTime g/ returns a stream of 'AbsTime'. The time is updated every

4309+-- /g/ seconds. Between any 2 updates, the stream will contain the same

4310+-- element. Getting the absolute time is a costly operation and hence the time

4311+-- taken to generate this stream depends on the granularity /g/. If /g/ is very

4312+-- low, the time taken to generate this the elements of this stream will be

4313+-- very high. Conversely, if the granularity is high, the time taken to

4314+-- generate the elements of this stream will be low.

4315+{-# INLINE currentTime #-}

4316+currentTime :: (IsStream t, MonadAsync m) => Double -> t m AbsTime

4317+currentTime g = fromStreamD $ D.currentTime g