62514f2 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Factor out the async clock related code

62514f2Harendra Kumar6 years ago

src/Streamly/Internal/Control/Concurrent.hsnew file+89

compare with master

unifiedsplit

@@ -0,0 +1,89 @@

1+{-# LANGUAGE UnboxedTuples #-}

3+-- |

4+-- Module : Streamly.Internal.Control.Concurrent

6+-- License : BSD-3-Clause

7+-- Maintainer : streamly@composewell.com

8+-- Stability : experimental

9+-- Portability : GHC

10+

11+module Streamly.Internal.Control.Concurrent

12+ (

13+ MonadAsync

14+ , RunInIO(..)

15+ , doFork

16+ , fork

17+ , forkManaged

18+ )

19+where

20+

21+import Control.Concurrent (ThreadId, forkIO, killThread)

22+import Control.Exception (SomeException(..), catch, mask)

23+import Control.Monad.Catch (MonadThrow)

24+import Control.Monad.IO.Class (MonadIO(..))

25+import Control.Monad.Trans.Control

26+ (MonadBaseControl, control, StM, liftBaseDiscard)

27+import Data.Functor (void)

28+import GHC.Conc (ThreadId(..))

29+import GHC.Exts

30+import GHC.IO (IO(..))

31+import System.Mem.Weak (addFinalizer)

32+

33+-- /Since: 0.8.0 ("Streamly.Prelude")/

34+--

35+-- | A monad that can perform concurrent or parallel IO operations. Streams

36+-- that can be composed concurrently require the underlying monad to be

37+-- 'MonadAsync'.

38+--

39+-- /Since: 0.1.0 ("Streamly")/

40+--

41+-- @since 0.8.0

42+type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

43+

44+newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }

45+

46+-- Stolen from the async package. The perf improvement is modest, 2% on a

47+-- thread heavy benchmark (parallel composition using noop computations).

48+-- A version of forkIO that does not include the outer exception

49+-- handler: saves a bit of time when we will be installing our own

50+-- exception handler.

51+{-# INLINE rawForkIO #-}

52+rawForkIO :: IO () -> IO ThreadId

53+rawForkIO action = IO $ \ s ->

54+ case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)

55+

56+-- | Fork a thread to run the given computation, installing the provided

57+-- exception handler. Lifted to any monad with 'MonadBaseControl IO m'

58+-- capability.

59+--

60+-- TODO: the RunInIO argument can be removed, we can directly pass the action

61+-- as "mrun action" instead.

62+{-# INLINE doFork #-}

63+doFork :: MonadBaseControl IO m

64+ => m ()

65+ -> RunInIO m

66+ -> (SomeException -> IO ())

67+ -> m ThreadId

68+doFork action (RunInIO mrun) exHandler =

69+ control $ \run ->

70+ mask $ \restore -> do

71+ tid <- rawForkIO $ catch (restore $ void $ mrun action)

72+ exHandler

73+ run (return tid)

74+

75+-- | 'fork' lifted to any monad with 'MonadBaseControl IO m' capability.

76+--

77+{-# INLINABLE fork #-}

78+fork :: MonadBaseControl IO m => m () -> m ThreadId

79+fork = liftBaseDiscard forkIO

80+

81+-- | Fork a thread that is automatically killed as soon as the reference to the

82+-- returned threadId is garbage collected.

83+--

84+{-# INLINABLE forkManaged #-}

85+forkManaged :: (MonadIO m, MonadBaseControl IO m) => m () -> m ThreadId

86+forkManaged action = do

87+ tid <- fork action

88+ liftIO $ addFinalizer tid (killThread tid)

89+ return tid

src/Streamly/Internal/Data/SVar.hs+8−63

@@ -105,20 +105,17 @@ module Streamly.Internal.Data.SVar

105 )

106where

107

108−import Control.Concurrent

109− (ThreadId, myThreadId, threadDelay, throwTo, forkIO, killThread)

108+import Control.Concurrent (ThreadId, myThreadId, threadDelay, throwTo)

110import Control.Concurrent.MVar

109import Control.Concurrent.MVar

111 (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar,

110 (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar,

112 tryReadMVar)

111 tryReadMVar)

113import Control.Exception

112import Control.Exception

114− (SomeException(..), catch, mask, assert, Exception, catches,

113+ (SomeException(..), assert, Exception, catches,

115 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),

114 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),

116 BlockedIndefinitelyOnSTM(..))

115 BlockedIndefinitelyOnSTM(..))

117import Control.Monad (when)

116import Control.Monad (when)

118−import Control.Monad.Catch (MonadThrow)

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

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

120−import Control.Monad.Trans.Control

118+import Control.Monad.Trans.Control (MonadBaseControl, control)

121− (MonadBaseControl, control, StM, liftBaseDiscard)

122import Streamly.Internal.Data.Atomics

119import Streamly.Internal.Data.Atomics

123 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,

120 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,

124 storeLoadBarrier)

121 storeLoadBarrier)

@@ -126,21 +123,19 @@ import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)

126import Data.Functor (void)

123import Data.Functor (void)

127import Data.Heap (Heap, Entry(..))

124import Data.Heap (Heap, Entry(..))

128import Data.Int (Int64)

125import Data.Int (Int64)

129−import Data.Kind (Type)

130import Data.IORef

126import Data.IORef

131 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)

127 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)

128+import Data.Kind (Type)

132import Data.Maybe (fromJust, fromMaybe)

129import Data.Maybe (fromJust, fromMaybe)

133#if __GLASGOW_HASKELL__ < 808

130#if __GLASGOW_HASKELL__ < 808

134import Data.Semigroup ((<>))

131import Data.Semigroup ((<>))

135#endif

132#endif

136import Data.Set (Set)

133import Data.Set (Set)

137−import GHC.Conc (ThreadId(..))

138−import GHC.Exts

139−import GHC.IO (IO(..))

140import System.IO (hPutStrLn, stderr)

134import System.IO (hPutStrLn, stderr)

141−import System.Mem.Weak (addFinalizer)

142

135

143−import Streamly.Internal.Data.Time.Clock (Clock(..), getTime)

136+import Streamly.Internal.Control.Concurrent

137+ (MonadAsync, RunInIO(..), doFork, fork, forkManaged)

138+import Streamly.Internal.Data.Time.System (Clock(..), getTime)

144import Streamly.Internal.Data.Time.Units

139import Streamly.Internal.Data.Time.Units

145 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,

140 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,

146 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)

141 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)

@@ -898,66 +893,16 @@ withDiagMVar sv label action =

898-- Spawning threads

893-- Spawning threads

899------------------------------------------------------------------------------

894------------------------------------------------------------------------------

900

895

901−-- /Since: 0.8.0 ("Streamly.Prelude")/

902−--

903−-- | A monad that can perform concurrent or parallel IO operations. Streams

904−-- that can be composed concurrently require the underlying monad to be

905−-- 'MonadAsync'.

906−--

907−-- /Since: 0.1.0 ("Streamly")/

908−--

909−-- @since 0.8.0

910−type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

911−

912−-- When we run computations concurrently, we completely isolate the state of

896+-- | When we run computations concurrently, we completely isolate the state of

913-- the concurrent computations from the parent computation. The invariant is

897-- the concurrent computations from the parent computation. The invariant is

914-- that we should never be running two concurrent computations in the same

898-- that we should never be running two concurrent computations in the same

915-- thread without using the runInIO function. Also, we should never be running

899-- thread without using the runInIO function. Also, we should never be running

916-- a concurrent computation in the parent thread, otherwise it may affect the

900-- a concurrent computation in the parent thread, otherwise it may affect the

917-- state of the parent which is against the defined semantics of concurrent

901-- state of the parent which is against the defined semantics of concurrent

918-- execution.

902-- execution.

919−newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }

920−

921captureMonadState :: MonadBaseControl IO m => m (RunInIO m)

903captureMonadState :: MonadBaseControl IO m => m (RunInIO m)

922captureMonadState = control $ \run -> run (return $ RunInIO run)

904captureMonadState = control $ \run -> run (return $ RunInIO run)

923

905

924−-- Stolen from the async package. The perf improvement is modest, 2% on a

925−-- thread heavy benchmark (parallel composition using noop computations).

926−-- A version of forkIO that does not include the outer exception

927−-- handler: saves a bit of time when we will be installing our own

928−-- exception handler.

929−{-# INLINE rawForkIO #-}

930−rawForkIO :: IO () -> IO ThreadId

931−rawForkIO action = IO $ \ s ->

932− case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)

933−

934−{-# INLINE doFork #-}

935−doFork :: MonadBaseControl IO m

936− => m ()

937− -> RunInIO m

938− -> (SomeException -> IO ())

939− -> m ThreadId

940−doFork action (RunInIO mrun) exHandler =

941− control $ \run ->

942− mask $ \restore -> do

943− tid <- rawForkIO $ catch (restore $ void $ mrun action)

944− exHandler

945− run (return tid)

946−

947−{-# INLINABLE fork #-}

948−fork :: MonadBaseControl IO m => m () -> m ThreadId

949−fork = liftBaseDiscard forkIO

950−

951−-- | Fork a thread that is automatically killed as soon as the reference to the

952−-- returned threadId is garbage collected.

953−--

954−{-# INLINABLE forkManaged #-}

955−forkManaged :: (MonadIO m, MonadBaseControl IO m) => m () -> m ThreadId

956−forkManaged action = do

957− tid <- fork action

958− liftIO $ addFinalizer tid (killThread tid)

959− return tid

960−

961------------------------------------------------------------------------------

906------------------------------------------------------------------------------

962-- Collecting results from child workers in a streamed fashion

907-- Collecting results from child workers in a streamed fashion

963------------------------------------------------------------------------------

908------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/StreamD/Generate.hs+3−57

@@ -89,15 +89,11 @@ where

90#include "inline.hs"

92−import Control.Concurrent (threadDelay, ThreadId)

93−import Control.Monad (forever)

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

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

95−import Streamly.Internal.Data.Time.Units (toRelTime64, RelTime64)

93+import Streamly.Internal.Data.Time.Clock

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

94+ (Clock(Monotonic), asyncClock, readClock)

97import Streamly.Internal.Data.Time.Units

95import Streamly.Internal.Data.Time.Units

98− (MicroSecond64(..), fromAbsTime, toAbsTime, AbsTime)

96+ (toAbsTime, AbsTime, toRelTime64, RelTime64)

99−

100−import qualified Streamly.Internal.Data.IORef.Prim as Prim

101

102import Prelude hiding (iterate, repeat, replicate, takeWhile)

98import Prelude hiding (iterate, repeat, replicate, takeWhile)

103import Streamly.Internal.Data.Stream.StreamD.Type

99import Streamly.Internal.Data.Stream.StreamD.Type

@@ -324,56 +320,6 @@ numFromThen from next = enumerateFromStepNum from (next - from)

324-- Time Enumeration

320-- Time Enumeration

325------------------------------------------------------------------------------

321------------------------------------------------------------------------------

326

322

327−{-# INLINE updateTimeVar #-}

328−updateTimeVar :: Clock -> Prim.IORef MicroSecond64 -> IO ()

329−updateTimeVar clock timeVar = do

330− t <- fromAbsTime <$> getTime clock

331− Prim.modifyIORef' timeVar (const t)

332−

333−{-# INLINE updateWithDelay #-}

334−updateWithDelay :: RealFrac a =>

335− Clock -> a -> Prim.IORef MicroSecond64 -> IO ()

336−updateWithDelay clock precision timeVar = do

337− threadDelay (delayTime precision)

338− updateTimeVar clock timeVar

339−

340− where

341−

342− -- Keep the minimum at least a millisecond to avoid high CPU usage

343− {-# INLINE delayTime #-}

344− delayTime g

345− | g' >= fromIntegral (maxBound :: Int) = maxBound

346− | g' < 1000 = 1000

347− | otherwise = round g'

348−

349− where

350−

351− g' = g * 10 ^ (6 :: Int)

352−

353−-- XXX Move this to the time/clock module?

354−--

355−-- | @asyncClock g@ starts a clock thread that updates an IORef with current

356−-- time as a 64-bit value in microseconds, every 'g' seconds. The IORef can be

357−-- read asynchronously. The thread exits automatically when the reference to

358−-- the returned 'ThreadId' is lost.

359−--

360−-- Minimum granularity of clock update is 1 ms. Higher is better for

361−-- performance.

362−--

363−-- CAUTION! This is safe only on a 64-bit machine. On a 32-bit machine a 64-bit

364−-- 'Var' cannot be read consistently without a lock while another thread is

365−-- writing to it.

366−asyncClock :: Clock -> Double -> IO (ThreadId, Prim.IORef MicroSecond64)

367−asyncClock clock g = do

368− timeVar <- Prim.newIORef undefined

369− updateTimeVar clock timeVar

370− tid <- forkManaged $ forever (updateWithDelay clock g timeVar)

371− return (tid, timeVar)

372−

373−{-# INLINE readClock #-}

374−readClock :: (ThreadId, Prim.IORef MicroSecond64) -> IO MicroSecond64

375−readClock (_, timeVar) = Prim.readIORef timeVar

376−

377{-# INLINE_NORMAL times #-}

323{-# INLINE_NORMAL times #-}

378times :: MonadAsync m => Double -> Stream m (AbsTime, RelTime64)

324times :: MonadAsync m => Double -> Stream m (AbsTime, RelTime64)

379times g = Stream step Nothing

325times g = Stream step Nothing

src/Streamly/Internal/Data/Time/Clock.hs+64−1

@@ -11,7 +11,70 @@ module Streamly.Internal.Data.Time.Clock

11 -- * System clock

12 Clock(..)

13 , getTime

14+

15+ -- * Async clock

16+ , asyncClock

17+ , readClock

14 )

18 )

15where

19where

21+import Control.Concurrent (threadDelay, ThreadId)

22+import Control.Monad (forever)

17−import Streamly.Internal.Data.Time.System

23+import Streamly.Internal.Data.Time.System (Clock(..), getTime)

24+import Streamly.Internal.Data.Time.Units (MicroSecond64(..), fromAbsTime)

25+import Streamly.Internal.Control.Concurrent (forkManaged)

26+

27+import qualified Streamly.Internal.Data.IORef.Prim as Prim

28+

29+

30+------------------------------------------------------------------------------

31+-- Async clock

32+------------------------------------------------------------------------------

33+

34+{-# INLINE updateTimeVar #-}

35+updateTimeVar :: Clock -> Prim.IORef MicroSecond64 -> IO ()

36+updateTimeVar clock timeVar = do

37+ t <- fromAbsTime <$> getTime clock

38+ Prim.modifyIORef' timeVar (const t)

39+

40+{-# INLINE updateWithDelay #-}

41+updateWithDelay :: RealFrac a =>

42+ Clock -> a -> Prim.IORef MicroSecond64 -> IO ()

43+updateWithDelay clock precision timeVar = do

44+ threadDelay (delayTime precision)

45+ updateTimeVar clock timeVar

46+

47+ where

48+

49+ -- Keep the minimum at least a millisecond to avoid high CPU usage

50+ {-# INLINE delayTime #-}

51+ delayTime g

52+ | g' >= fromIntegral (maxBound :: Int) = maxBound

53+ | g' < 1000 = 1000

54+ | otherwise = round g'

55+

56+ where

57+

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

59+

60+-- | @asyncClock g@ starts a clock thread that updates an IORef with current

61+-- time as a 64-bit value in microseconds, every 'g' seconds. The IORef can be

62+-- read asynchronously. The thread exits automatically when the reference to

63+-- the returned 'ThreadId' is lost.

64+--

65+-- Minimum granularity of clock update is 1 ms. Higher is better for

66+-- performance.

67+--

68+-- CAUTION! This is safe only on a 64-bit machine. On a 32-bit machine a 64-bit

69+-- 'Var' cannot be read consistently without a lock while another thread is

70+-- writing to it.

71+asyncClock :: Clock -> Double -> IO (ThreadId, Prim.IORef MicroSecond64)

72+asyncClock clock g = do

73+ timeVar <- Prim.newIORef undefined

74+ updateTimeVar clock timeVar

75+ tid <- forkManaged $ forever (updateWithDelay clock g timeVar)

76+ return (tid, timeVar)

77+

78+{-# INLINE readClock #-}

79+readClock :: (ThreadId, Prim.IORef MicroSecond64) -> IO MicroSecond64

80+readClock (_, timeVar) = Prim.readIORef timeVar

src/Streamly/Internal/Data/Time/Units.hs+1−1

@@ -1,4 +1,4 @@

1−{-# LANGUAGE DataKinds #-}

1+{-# LANGUAGE TypeInType #-}

2{-# LANGUAGE UnboxedTuples #-}

4-- |

streamly.cabal+1

@@ -407,6 +407,7 @@ library

407 , Streamly.Internal.BaseCompat

408 , Streamly.Internal.Control.Exception

409 , Streamly.Internal.Control.Monad

410+ , Streamly.Internal.Control.Concurrent

410 , Streamly.Internal.Data.Cont

411 , Streamly.Internal.Data.Cont

411 , Streamly.Internal.Data.Tuple.Strict

412 , Streamly.Internal.Data.Tuple.Strict

412 , Streamly.Internal.Data.Maybe.Strict

413 , Streamly.Internal.Data.Maybe.Strict