git.iko.soy/streamly
on f310bed help sign in
streamly/src/Streamly/Time.hs
125 lines 4.5 kB history raw
1{-# LANGUAGE FlexibleContexts #-}
3-- |
4-- Module : Streamly.Time
5-- Copyright : (c) 2017 Harendra Kumar
6--
7-- License : BSD3
8-- Maintainer : harendra.kumar@gmail.com
9-- Stability : experimental
10-- Portability : GHC
11--
12-- Time utilities for reactive programming.
14module Streamly.Time
15 ( intervalsOf
17 -- * Deprecated
18 , periodic
19 , withClock
20 )
21where
23import Control.Concurrent (threadDelay)
24import Control.Monad (when)
25import Control.Monad.IO.Class (MonadIO(liftIO))
26import Data.Maybe (fromJust, isJust)
28import Streamly (parallel)
29import Streamly.SVar (MonadAsync)
30import Streamly.Streams.StreamK (IsStream(..))
31import Streamly.Fold (Fold)
33import qualified Streamly.Fold as FL
34import qualified Streamly.Prelude as S
36-- | Run an action forever periodically at the given frequency specified in per
37-- second (Hz).
38--
39-- @since 0.1.0
40{-# DEPRECATED periodic "Please use the \"rate\" combinator instead" #-}
41periodic :: Int -> IO () -> IO ()
42periodic freq action = do
43 action
44 threadDelay (1000000 `div` freq)
45 periodic freq action
47-- | Run a computation on every clock tick, the clock runs at the specified
48-- frequency. It allows running a computation at high frequency efficiently by
49-- maintaining a local clock and adjusting it with the provided base clock at
50-- longer intervals. The first argument is a base clock returning some notion
51-- of time in microseconds. The second argument is the frequency in per second
52-- (Hz). The third argument is the action to run, the action is provided the
53-- local time as an argument.
54--
55-- @since 0.1.0
56{-# DEPRECATED withClock "Please use the \"rate\" combinator instead" #-}
57withClock :: IO Int -> Int -> (Int -> IO ()) -> IO ()
58withClock clock freq action = do
59 t <- clock
60 go t period period t 0
62 where
64 period = 1000000 `div` freq
66 -- Note that localTime is roughly but not exactly equal to (lastAdj + tick
67 -- * n). That is because we do not abruptly adjust the clock skew instead
68 -- we adjust the tick size.
69 go lastAdj delay tick localTime n = do
70 action localTime
71 when (delay > 0) $ threadDelay delay
73 if n == freq
74 then do
75 (t, newTick, newDelay) <- adjustClock lastAdj localTime delay
76 go t newDelay newTick (localTime + newTick) 0
77 else go lastAdj delay tick (localTime + tick) (n + 1)
79 -- Adjust the tick size rather than the clock to avoid abrupt changes
80 -- resulting in jittery behavior at the end of every interval.
81 adjustClock lastAdj localTime delay = do
82 baseTime <- clock
83 let newTick = period + (baseTime - localTime) `div` freq
84 lastPeriod = (baseTime - lastAdj) `div` freq
85 newDelay = max 0 (delay + period - lastPeriod)
86 return (baseTime, newTick, newDelay)
88-------------------------------------------------------------------------------
89-- Stream element unaware: Time based buffering
90-------------------------------------------------------------------------------
92-- There are two ways in which time based buffering can be implemented.
93--
94-- 1) One is to use an API like groupsOf to group elements by time and then
95-- flush when the group is complete. This will require checking time on each
96-- element yield.
97--
98-- 2) Use an async thread to write. The producer thread would just queue the
99-- elements to be written on a queue associated with the writer thread. When
100-- the write thread is scheduled elements that have been buffered will be
101-- written based on a buffering policy. This will require the producer to
102-- synchronize on the queue, send a doorbell when the queue has items in it.
103-- The queue can be an array which can be written directly to the IO device.
104--
105-- We can try both and see which one performs better.
107-- XXX add this example after fixing the serial stream rate control
108-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
109-- > [3,7,11,15,19]
110--
111-- | Group the input stream into windows of @n@ second each and then fold each
112-- group using the provided fold function.
113--
114-- @since 0.7.0
115{-# INLINE intervalsOf #-}
116intervalsOf
117 :: (IsStream t, MonadAsync m)
118 => Double -> Fold m a b -> t m a -> t m b
119intervalsOf n f m = FL.grouped (catMaybes f) s
120 where
121 s = S.map (\x -> (Just x, False)) m `parallel` S.repeatM timeout
122 timeout = do
123 liftIO $ threadDelay (round $ n * 1000000)
124 return (Nothing, True)
125 catMaybes = FL.lfilter isJust . FL.lmap fromJust