753d091 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Refactor currentTime, add comments, safety notes

753d091Harendra Kumar7 years ago

benchmark/Linear.hs+2−7

compare with master

unifiedsplit

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

161 -- These are essentially cons and consM

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

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

164−

164+ , benchIOSrc serially "currentTime/0.00001s"

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

165+ $ Ops.currentTime value 0.00001

170−

171 ]

166 ]

172 , bgroup "elimination"

167 , bgroup "elimination"

173 [ bgroup "reduce"

168 [ bgroup "reduce"

benchmark/Streamly/Benchmark/Prelude.hs+5−5

@@ -177,6 +177,11 @@ sourceFromFoldable value n = S.fromFoldable [n..n+value]

177sourceFromFoldableM :: (S.IsStream t, S.MonadAsync m) => Int -> Int -> t m Int

178sourceFromFoldableM value n = S.fromFoldableM (Prelude.fmap return [n..n+value])

179

180+{-# INLINE currentTime #-}

181+currentTime :: (S.IsStream t, S.MonadAsync m)

182+ => Int -> Double -> Int -> t m AbsTime

183+currentTime value g _ = S.take value $ Internal.currentTime g

184+

180-------------------------------------------------------------------------------

185-------------------------------------------------------------------------------

181-- Elimination

186-- Elimination

182-------------------------------------------------------------------------------

187-------------------------------------------------------------------------------

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

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

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

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

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

384

389

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−

390{-# INLINE mapM #-}

391{-# INLINE map' #-}

392{-# INLINE fmap' #-}

src/Streamly/Internal/Data/Stream/StreamD.hs+36−17

@@ -93,6 +93,9 @@ module Streamly.Internal.Data.Stream.StreamD

93 , enumerateFromToFractional

94 , enumerateFromThenToFractional

96+ -- ** Time

97+ , currentTime

98+

96 -- ** Conversions

99 -- ** Conversions

97 -- | Transform an input structure into a stream.

100 -- | Transform an input structure into a stream.

98 -- | Direct style stream does not support @fromFoldable@.

101 -- | Direct style stream does not support @fromFoldable@.

@@ -245,9 +248,11 @@ module Streamly.Internal.Data.Stream.StreamD

245 , filterM

248 , filterM

246 , uniq

249 , uniq

247 , take

250 , take

251+ , takeByTime

248 , takeWhile

252 , takeWhile

249 , takeWhileM

253 , takeWhileM

250 , drop

254 , drop

255+ , dropByTime

251 , dropWhile

256 , dropWhile

252 , dropWhileM

257 , dropWhileM

253

258

@@ -307,18 +312,13 @@ module Streamly.Internal.Data.Stream.StreamD

307 , mkParallelD

312 , mkParallelD

308

313

309 , lastN

314 , lastN

310−

311− -- * Time related

312− , takeByTime

313− , dropByTime

314− , currentTime

315 )

316where

317

318import Control.Concurrent (killThread, myThreadId, takeMVar, threadDelay)

319import Control.Exception

320 (Exception, SomeException, AsyncException, fromException)

321−import Control.Monad (void, when)

321+import Control.Monad (void, when, forever)

322import Control.Monad.Catch (MonadCatch, throwM)

323import Control.Monad.IO.Class (MonadIO(..))

324import Control.Monad.Reader (ReaderT)

@@ -327,6 +327,7 @@ import Control.Monad.Trans (MonadTrans(lift))

327import Control.Monad.Trans.Control (MonadBaseControl)

328import Data.Bits (shiftR, shiftL, (.|.), (.&.))

329import Data.Functor.Identity (Identity(..))

330+import Data.Int (Int64)

330import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef, IORef)

331import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef, IORef)

331import Data.Maybe (fromJust, isJust, isNothing)

332import Data.Maybe (fromJust, isJust, isNothing)

332import Data.Word (Word32)

333import Data.Word (Word32)

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

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

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

347import qualified Prelude

348import qualified Prelude

348

349

349−import Data.Int (Int64)

350import Streamly.Internal.Mutable.Prim.Var

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

352import Streamly.Internal.Data.Time.Units

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

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

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

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

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

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

361+ (MicroSecond64(..), fromAbsTime, toAbsTime, AbsTime)

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

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

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

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

363

364

@@ -3449,6 +3450,9 @@ pollCounts transf fld (Stream step state) = Stream step' Nothing

3449

3450

3450 {-# INLINE_LATE step' #-}

3451 {-# INLINE_LATE step' #-}

3451 step' _ Nothing = do

3452 step' _ Nothing = do

3453+ -- As long as we are using an "Int" for counts lockfree reads from

3454+ -- Var should work correctly on both 32-bit and 64-bit machines.

3455+ -- However, an Int on a 32-bit machine may overflow quickly.

3452 countVar <- liftIO $ newVar (0 :: Int)

3456 countVar <- liftIO $ newVar (0 :: Int)

3453 tid <- forkManaged

3457 tid <- forkManaged

3454 $ void $ runFold fld

3458 $ void $ runFold fld

@@ -4211,25 +4215,40 @@ dropByTime duration (Stream step1 state1) = Stream step (DropByTimeInit state1)

4211 Skip s -> Skip (DropByTimeYield s)

4215 Skip s -> Skip (DropByTimeYield s)

4212 Stop -> Stop

4216 Stop -> Stop

4213

4217

4218+-- XXX we should move this to stream generation section of this file. Also, the

4219+-- take/drop combinators above should be moved to filtering section.

4214{-# INLINE_NORMAL currentTime #-}

4220{-# INLINE_NORMAL currentTime #-}

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

4221currentTime :: MonadAsync m => Double -> Stream m AbsTime

4216currentTime g = Stream step Nothing

4222currentTime g = Stream step Nothing

4223+

4217− where

4224+ where

4225+

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

4227+

4228+ -- XXX should have a minimum granularity to avoid high CPU usage?

4229+ {-# INLINE delayTime #-}

4230+ delayTime =

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

4232+ then maxBound

4233+ else round g'

4234+

4218− next timeVar = do

4235+ updateTimeVar timeVar = do

4219 threadDelay $ delayTime

4236 threadDelay $ delayTime

4220 MicroSecond64 t <- fromAbsTime <$> getTime Monotonic

4237 MicroSecond64 t <- fromAbsTime <$> getTime Monotonic

4221 modifyVar' timeVar (const t)

4238 modifyVar' timeVar (const t)

4222− next timeVar

4239+

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 #-}

4240 {-# INLINE_LATE step #-}

4229 step _ Nothing = do

4241 step _ Nothing = do

4242+ -- XXX note that this is safe only on a 64-bit machine. On a 32-bit

4243+ -- machine a 64-bit 'Var' cannot be read consistently without a lock

4244+ -- while another thread is writing to it.

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

4245 timeVar <- liftIO $ newVar (0 :: Int64)

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

4246+ tid <- forkManaged $ liftIO $ forever (updateTimeVar timeVar)

4232 return $ Skip $ Just (timeVar, tid)

4247 return $ Skip $ Just (timeVar, tid)

4248+

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

4249 step _ s@(Just (timeVar, _)) = do

4234 a <- liftIO $ readVar timeVar

4250 a <- liftIO $ readVar timeVar

4251+ -- XXX we can perhaps use an AbsTime64 using a 64 bit Int for

4252+ -- efficiency. or maybe we can use a representation using Double for

4253+ -- floating precision time

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

4254 return $ Yield (toAbsTime (MicroSecond64 a)) s

src/Streamly/Internal/Mutable/Prim/Var.hs+9

@@ -20,6 +20,15 @@

20--

21-- A mutable variable in a mutation capable monad (IO/ST) holding a 'Prim'

22-- value. This allows fast modification because of unboxed storage.

23+--

24+-- = Multithread Consistency Notes

25+--

26+-- In general, any value that straddles a machine word cannot be guaranteed to

27+-- be consistently read from another thread without a lock. GHC heap objects

28+-- are always machine word aligned, therefore, a 'Var' is also word aligned. On

29+-- a 64-bit platform, writing a 64-bit aligned type from one thread and reading

30+-- it from another thread should give consistent old or new value. The same

31+-- holds true for 32-bit values on a 32-bit platform.

24module Streamly.Internal.Mutable.Prim.Var

33module Streamly.Internal.Mutable.Prim.Var

25 (

34 (

src/Streamly/Internal/Prelude.hs+38−18

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

59 , fromFoldableM

60 , fromPrimVar

62+ -- ** Time related

63+ , currentTime

64+

62 -- * Elimination

65 -- * Elimination

64 -- ** Deconstruction

67 -- ** Deconstruction

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

421 -- * Diagnostics

424 -- * Diagnostics

422 , inspectMode

425 , inspectMode

423

426

424− -- * Time related

425− , currentTime

426−

427 -- * Deprecated

428 , K.once

429 , each

@@ -898,6 +898,34 @@ fromHandle h = go

898fromPrimVar :: (IsStream t, MonadIO m, Prim a) => Var IO a -> t m a

899fromPrimVar = fromStreamD . D.fromPrimVar

900

901+------------------------------------------------------------------------------

902+-- Time related

903+------------------------------------------------------------------------------

904+

905+-- XXX Some related/interesting combinators:

906+--

907+-- 1) emit the relative time elapsed since last evaluation. That would just be

908+-- a rollingMap on the currentTime stream.

909+--

910+-- 2) Generate ticks at specified interval. Drop ticks when blocked.

911+-- ticks :: Double -> t m ()

912+--

913+-- 3) Emit relative time at specified tick interval. If a tick is dropped

914+-- combine the interval with the next tick.

915+-- ticks :: Double -> t m RelTime

916+--

917+-- | @currentTime g@ returns a stream of absolute timestamps using a clock of

918+-- granularity @g@ specified in seconds. A low granularity clock is more

919+-- expensive in terms of CPU usage.

920+--

921+-- Note: This API is not safe on 32-bit machines.

922+--

923+-- /Internal/

924+--

925+{-# INLINE currentTime #-}

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

927+currentTime g = fromStreamD $ D.currentTime g

928+

901------------------------------------------------------------------------------

929------------------------------------------------------------------------------

902-- Elimination by Folding

930-- Elimination by Folding

903------------------------------------------------------------------------------

931------------------------------------------------------------------------------

@@ -3756,6 +3784,10 @@ tapAsync f xs = D.fromStreamD $ D.tapAsync f (D.toStreamD xs)

3756-- $ S.enumerateFrom 0

3784-- $ S.enumerateFrom 0

3757-- @

3785-- @

3758--

3786--

3787+-- Note: This may not work correctly on 32-bit machines.

3788+--

3789+-- /Internal

3790+--

3759{-# INLINE pollCounts #-}

3791{-# INLINE pollCounts #-}

3760pollCounts ::

3792pollCounts ::

3761 (IsStream t, MonadAsync m)

3793 (IsStream t, MonadAsync m)

@@ -3782,6 +3814,9 @@ pollCounts transf f xs =

3782-- 1 elements processed

3814-- 1 elements processed

3783-- @

3815-- @

3784--

3816--

3817+-- Note: This may not work correctly on 32-bit machines.

3818+--

3819+-- /Internal

3785{-# INLINE tapRate #-}

3820{-# INLINE tapRate #-}

3786tapRate ::

3821tapRate ::

3787 (IsStream t, MonadAsync m, MonadCatch m)

3822 (IsStream t, MonadAsync m, MonadCatch m)

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

4300{-# INLINE runStateT #-}

4335{-# INLINE runStateT #-}

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

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

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

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