git.iko.soy/streamly
on master help sign in
Add Timed API for stream generations
commitb528bae8ce08ee42368b326521b0549712b1e8d2 authorRanjeet Kumar Ranjan <ranjeetr@composewell.com> dateMon 27 Jun 2022, 17:18 (4 years ago) parent4bb8b7c
2 files changed+134−2
compare with master
unifiedsplit
core/src/Streamly/Internal/Data/Stream/Bottom.hs+67−1 core/src/Streamly/Internal/Data/Stream/Generate.hs+67−1
core/src/Streamly/Internal/Data/Stream/Bottom.hs+67−1
@@ -14,6 +14,9 @@ module Streamly.Internal.Data.Stream.Bottom
1414 -- * Generation
1515 fromPure
1616 , fromEffect
17+ , timesWith
18+ , absTimesWith
19+ , relTimesWith
1720
1821 -- * Folds
1922 , fold
@@ -55,12 +58,12 @@ where
5558
5659import Control.Monad.IO.Class (MonadIO(..))
5760import Streamly.Internal.Data.Fold.Type (Fold (..))
61+import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64, addToAbsTime64)
5862import Streamly.Internal.Data.Unboxed (Unboxed)
5963import Streamly.Internal.System.IO (defaultChunkSize)
6064
6165import qualified Streamly.Internal.Data.Array.Unboxed.Type as A
6266import qualified Streamly.Internal.Data.Fold as Fold
63−import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
6467import qualified Streamly.Internal.Data.Stream.StreamK as K
6568import qualified Streamly.Internal.Data.Stream.StreamD as D
6669
@@ -84,6 +87,69 @@ import Streamly.Internal.Data.Stream.Type
8487-- >>> import qualified Streamly.Internal.Data.Parser as Parser
8588-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
8689
90+------------------------------------------------------------------------------
91+-- Generation - Time related
92+------------------------------------------------------------------------------
93+
94+-- | @timesWith g@ returns a stream of time value tuples. The first component
95+-- of the tuple is an absolute time reference (epoch) denoting the start of the
96+-- stream and the second component is a time relative to the reference.
97+--
98+-- The argument @g@ specifies the granularity of the relative time in seconds.
99+-- A lower granularity clock gives higher precision but is more expensive in
100+-- terms of CPU usage. Any granularity lower than 1 ms is treated as 1 ms.
101+--
102+-- >>> import Control.Concurrent (threadDelay)
103+-- >>> import Streamly.Internal.Data.Stream.Generate as Stream (timesWith)
104+-- >>> Stream.mapM_ (\x -> print x >> threadDelay 1000000) $ Stream.take 3 $ Stream.timesWith 0.01
105+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
106+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
107+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
108+--
109+-- Note: This API is not safe on 32-bit machines.
110+--
111+-- /Pre-release/
112+--
113+{-# INLINE timesWith #-}
114+timesWith :: MonadIO m => Double -> Stream m (AbsTime, RelTime64)
115+timesWith g = fromStreamD $ D.times g
116+
117+-- | @absTimesWith g@ returns a stream of absolute timestamps using a clock of
118+-- granularity @g@ specified in seconds. A low granularity clock is more
119+-- expensive in terms of CPU usage. Any granularity lower than 1 ms is treated
120+-- as 1 ms.
121+--
122+-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimesWith 0.01
123+-- AbsTime (TimeSpec {sec = ..., nsec = ...})
124+-- AbsTime (TimeSpec {sec = ..., nsec = ...})
125+-- AbsTime (TimeSpec {sec = ..., nsec = ...})
126+--
127+-- Note: This API is not safe on 32-bit machines.
128+--
129+-- /Pre-release/
130+--
131+{-# INLINE absTimesWith #-}
132+absTimesWith :: MonadIO m => Double -> Stream m AbsTime
133+absTimesWith = fmap (uncurry addToAbsTime64) . timesWith
134+
135+-- | @relTimesWith g@ returns a stream of relative time values starting from 0,
136+-- using a clock of granularity @g@ specified in seconds. A low granularity
137+-- clock is more expensive in terms of CPU usage. Any granularity lower than 1
138+-- ms is treated as 1 ms.
139+--
140+-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimesWith 0.01
141+-- RelTime64 (NanoSecond64 ...)
142+-- RelTime64 (NanoSecond64 ...)
143+-- RelTime64 (NanoSecond64 ...)
144+--
145+-- Note: This API is not safe on 32-bit machines.
146+--
147+-- /Pre-release/
148+--
149+{-# INLINE relTimesWith #-}
150+relTimesWith :: MonadIO m => Double -> Stream m RelTime64
151+relTimesWith = fmap snd . timesWith
152+
87153------------------------------------------------------------------------------
88154-- Elimination - Running a Fold
89155------------------------------------------------------------------------------
core/src/Streamly/Internal/Data/Stream/Generate.hs+67−1
@@ -21,6 +21,13 @@ module Streamly.Internal.Data.Stream.Generate
2121 , Stream.fromPure
2222 , Stream.fromEffect
2323
24+ -- * Time Enumeration
25+ , times
26+ , absTimes
27+ , absTimesWith
28+ , relTimes
29+ , relTimesWith
30+
2431 -- * Cyclic Elements
2532 , mfix
2633
@@ -40,7 +47,10 @@ import Control.Monad.IO.Class (MonadIO)
4047import Data.Word (Word8)
4148import Foreign.Storable (Storable)
4249import GHC.Exts (Addr#, Ptr (Ptr))
50+import Streamly.Internal.Data.Stream.Bottom
51+ (absTimesWith, relTimesWith, timesWith)
4352import Streamly.Internal.Data.Stream.Type (Stream, fromStreamK, toStreamK)
53+import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64, addToAbsTime64)
4454import Streamly.Internal.Data.Unfold.Type (Unfold)
4555
4656import qualified Streamly.Internal.Data.Stream.StreamD as D
@@ -49,12 +59,14 @@ import qualified Streamly.Internal.Data.Stream.Type as Stream
4959
5060-- $setup
5161-- >>> :m
62+-- >>> import Control.Concurrent (threadDelay)
63+-- >>> import Data.Function (fix)
64+-- >>> import Prelude hiding (take)
5265-- >>> import qualified Streamly.Data.Fold as Fold
5366-- >>> import qualified Streamly.Data.Unfold as Unfold
5467-- >>> import qualified Streamly.Internal.Data.Stream as Stream
5568-- >>> import GHC.Exts (Ptr (Ptr))
5669
57−
5870------------------------------------------------------------------------------
5971-- From Unfold
6072------------------------------------------------------------------------------
@@ -72,6 +84,60 @@ import qualified Streamly.Internal.Data.Stream.Type as Stream
7284unfold :: Monad m => Unfold m a b -> a -> Stream m b
7385unfold unf = Stream.fromStreamD . D.unfold unf
7486
87+------------------------------------------------------------------------------
88+-- Time Enumeration
89+------------------------------------------------------------------------------
90+
91+-- | @times@ returns a stream of time value tuples with clock of 10 ms
92+-- granularity. The first component of the tuple is an absolute time reference
93+-- (epoch) denoting the start of the stream and the second component is a time
94+-- relative to the reference.
95+--
96+-- >>> Stream.mapM_ (\x -> print x >> threadDelay 1000000) $ Stream.take 3 $ Stream.times
97+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
98+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
99+-- (AbsTime (TimeSpec {sec = ..., nsec = ...}),RelTime64 (NanoSecond64 ...))
100+--
101+-- Note: This API is not safe on 32-bit machines.
102+--
103+-- /Pre-release/
104+--
105+{-# INLINE times #-}
106+times :: MonadIO m => Stream m (AbsTime, RelTime64)
107+times = timesWith 0.01
108+
109+-- | @absTimes@ returns a stream of absolute timestamps using a clock of 10 ms
110+-- granularity.
111+--
112+-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.absTimes
113+-- AbsTime (TimeSpec {sec = ..., nsec = ...})
114+-- AbsTime (TimeSpec {sec = ..., nsec = ...})
115+-- AbsTime (TimeSpec {sec = ..., nsec = ...})
116+--
117+-- Note: This API is not safe on 32-bit machines.
118+--
119+-- /Pre-release/
120+--
121+{-# INLINE absTimes #-}
122+absTimes :: MonadIO m => Stream m AbsTime
123+absTimes = fmap (uncurry addToAbsTime64) times
124+
125+-- | @relTimes@ returns a stream of relative time values starting from 0,
126+-- using a clock of granularity 10 ms.
127+--
128+-- >>> Stream.mapM_ print $ Stream.delayPre 1 $ Stream.take 3 $ Stream.relTimes
129+-- RelTime64 (NanoSecond64 ...)
130+-- RelTime64 (NanoSecond64 ...)
131+-- RelTime64 (NanoSecond64 ...)
132+--
133+-- Note: This API is not safe on 32-bit machines.
134+--
135+-- /Pre-release/
136+--
137+{-# INLINE relTimes #-}
138+relTimes :: MonadIO m => Stream m RelTime64
139+relTimes = fmap snd times
140+
75141-- | We can define cyclic structures using @let@:
76142--
77143-- >>> let (a, b) = ([1, b], head a) in (a, b)