5054813 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add currentTime and its benchmarks

5054813adithyaov7 years ago

benchmark/Linear.hs+7

compare with master

unifiedsplit

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

161 -- These are essentially cons and consM

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

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

164 ]

171 ]

165 , bgroup "elimination"

172 , bgroup "elimination"

166 [ bgroup "reduce"

173 [ bgroup "reduce"

benchmark/Streamly/Benchmark/Prelude.hs+5

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

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

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

384

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+

385{-# INLINE mapM #-}

390{-# INLINE mapM #-}

386{-# INLINE map' #-}

391{-# INLINE map' #-}

387{-# INLINE fmap' #-}

392{-# INLINE fmap' #-}

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

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

311 -- * Time related

312 , takeByTime

313 , dropByTime

314+ , currentTime

314 )

315 )

315where

316where

316

317

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

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

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

346import qualified Prelude

347import qualified Prelude

347

348

349+import Data.Int (Int64)

348import Streamly.Internal.Mutable.Prim.Var

350import Streamly.Internal.Mutable.Prim.Var

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

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

350import Streamly.Internal.Data.Time.Units

352import Streamly.Internal.Data.Time.Units

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

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

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

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

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

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

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

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

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

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

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

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

360

363

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

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

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

4208 Skip s -> Skip (DropByTimeYield s)

4211 Skip s -> Skip (DropByTimeYield s)

4209 Stop -> Stop

4212 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

421 -- * Diagnostics

422 , inspectMode

423

424+ -- * Time related

425+ , currentTime

426+

424 -- * Deprecated

427 -- * Deprecated

425 , K.once

428 , K.once

426 , each

429 , each

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

4297{-# INLINE runStateT #-}

4300{-# INLINE runStateT #-}

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

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

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

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