753d091 · streamly · git.iko.soy

git.iko.soy/streamly

Refactor currentTime, add comments, safety notes

commit753d091b75bd0d493b130a7eb66b283a949994e3 authorHarendra Kumar <harendra.kumar@gmail.com> dateMon 27 Jan 2020, 02:25 (7 years ago) parent5054813

5 files changed+90−47

compare with master

unifiedsplit

benchmark/Linear.hs+2−7 benchmark/Streamly/Benchmark/Prelude.hs+5−5 src/Streamly/Internal/Data/Stream/StreamD.hs+36−17 src/Streamly/Internal/Mutable/Prim/Var.hs+9 src/Streamly/Internal/Prelude.hs+38−18

benchmark/Linear.hs+2−7

@@ -161,13 +161,8 @@ 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−

164+ , benchIOSrc serially "currentTime/0.00001s"

165+ $ Ops.currentTime value 0.00001

171166 ]

172167 , bgroup "elimination"

173168 [ bgroup "reduce"

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

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

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

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

179179

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+

180185-------------------------------------------------------------------------------

181186-- Elimination

182187-------------------------------------------------------------------------------

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

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

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

384389

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−

390390{-# INLINE mapM #-}

391391{-# INLINE map' #-}

392392{-# INLINE fmap' #-}

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

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

9393 , enumerateFromToFractional

9494 , enumerateFromThenToFractional

9595

96+ -- ** Time

97+ , currentTime

98+

9699 -- ** Conversions

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

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

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

245248 , filterM

246249 , uniq

247250 , take

251+ , takeByTime

248252 , takeWhile

249253 , takeWhileM

250254 , drop

255+ , dropByTime

251256 , dropWhile

252257 , dropWhileM

253258

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

307312 , mkParallelD

308313

309314 , lastN

310−

311− -- * Time related

312− , takeByTime

313− , dropByTime

314− , currentTime

315315 )

316316where

317317

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

319319import Control.Exception

320320 (Exception, SomeException, AsyncException, fromException)

321−import Control.Monad (void, when)

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

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

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

324324import Control.Monad.Reader (ReaderT)

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

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

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

329329import Data.Functor.Identity (Identity(..))

330+import Data.Int (Int64)

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

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

332333import Data.Word (Word32)

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

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

347348import qualified Prelude

348349

349−import Data.Int (Int64)

350350import Streamly.Internal.Mutable.Prim.Var

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

352352import Streamly.Internal.Data.Time.Units

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

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

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

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

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

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

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

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

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

363364

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

34493450

34503451 {-# INLINE_LATE step' #-}

34513452 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.

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

34533457 tid <- forkManaged

34543458 $ void $ runFold fld

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

42114215 Skip s -> Skip (DropByTimeYield s)

42124216 Stop -> Stop

42134217

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.

42144220{-# INLINE_NORMAL currentTime #-}

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

42164222currentTime g = Stream step Nothing

4217− where

4218− next timeVar = do

4223+

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+

4235+ updateTimeVar timeVar = do

42194236 threadDelay $ delayTime

42204237 MicroSecond64 t <- fromAbsTime <$> getTime Monotonic

42214238 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'

4239+

42284240 {-# INLINE_LATE step #-}

42294241 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.

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

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

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

42324247 return $ Skip $ Just (timeVar, tid)

4248+

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

42344250 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

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

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

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

2020--

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

2222-- 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.

2332

2433module Streamly.Internal.Mutable.Prim.Var

2534 (

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

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

5959 , fromFoldableM

6060 , fromPrimVar

6161

62+ -- ** Time related

63+ , currentTime

64+

6265 -- * Elimination

6366

6467 -- ** Deconstruction

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

421424 -- * Diagnostics

422425 , inspectMode

423426

424− -- * Time related

425− , currentTime

426−

427427 -- * Deprecated

428428 , K.once

429429 , each

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

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

899899fromPrimVar = fromStreamD . D.fromPrimVar

900900

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+

901929------------------------------------------------------------------------------

902930-- Elimination by Folding

903931------------------------------------------------------------------------------

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

37563784-- $ S.enumerateFrom 0

37573785-- @

37583786--

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

3788+--

3789+-- /Internal

3790+--

37593791{-# INLINE pollCounts #-}

37603792pollCounts ::

37613793 (IsStream t, MonadAsync m)

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

37823814-- 1 elements processed

37833815-- @

37843816--

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

3818+--

3819+-- /Internal

37853820{-# INLINE tapRate #-}

37863821tapRate ::

37873822 (IsStream t, MonadAsync m, MonadCatch m)

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

43004335{-# INLINE runStateT #-}

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

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