3cba8fa · streamly · git.iko.soy

src/Streamly/Internal/Control/Concurrent.hs+5−59

compare with master

unifiedsplit

@@ -1,5 +1,3 @@

1−{-# LANGUAGE UnboxedTuples #-}

2−

3-- |

1-- |

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

2-- Module : Streamly.Internal.Control.Concurrent

@@ -7,29 +5,22 @@

7-- Maintainer : streamly@composewell.com

5-- Maintainer : streamly@composewell.com

8-- Stability : experimental

6-- Stability : experimental

9-- Portability : GHC

7-- Portability : GHC

8+--

9+-- Note: This module is primarily for abstractions related to MonadBaseControl.

10+-- Please do not add any general routines in this. It should be renamed

11+-- appropriately.

10

12

11module Streamly.Internal.Control.Concurrent

13module Streamly.Internal.Control.Concurrent

12 (

14 (

13 MonadAsync

15 MonadAsync

14 , RunInIO(..)

16 , RunInIO(..)

15 , captureMonadState

17 , captureMonadState

16− , doFork

17− , fork

18− , forkManaged

19 )

18 )

20where

19where

21

20

22−import Control.Concurrent (ThreadId, forkIO, killThread)

23−import Control.Exception (SomeException(..), catch, mask)

24import Control.Monad.Catch (MonadThrow)

21import Control.Monad.Catch (MonadThrow)

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

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

26−import Control.Monad.Trans.Control

23+import Control.Monad.Trans.Control (MonadBaseControl, control, StM)

27− (MonadBaseControl, control, StM, liftBaseDiscard)

28−import Data.Functor (void)

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

30−import GHC.Exts

31−import GHC.IO (IO(..))

32−import System.Mem.Weak (addFinalizer)

33

24

34-- /Since: 0.8.0 ("Streamly.Prelude")/

25-- /Since: 0.8.0 ("Streamly.Prelude")/

35--

26--

@@ -53,48 +44,3 @@ newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }

53-- execution.

44-- execution.

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

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

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

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

56−

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

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

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

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

61−-- exception handler.

62−{-# INLINE rawForkIO #-}

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

64−rawForkIO (IO action) = IO $ \ s ->

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

66−

67−-- | Fork a thread to run the given computation, installing the provided

68−-- exception handler. Lifted to any monad with 'MonadBaseControl IO m'

69−-- capability.

70−--

71−-- TODO: the RunInIO argument can be removed, we can directly pass the action

72−-- as "mrun action" instead.

73−{-# INLINE doFork #-}

74−doFork :: MonadBaseControl IO m

75− => m ()

76− -> RunInIO m

77− -> (SomeException -> IO ())

78− -> m ThreadId

79−doFork action (RunInIO mrun) exHandler =

80− control $ \run ->

81− mask $ \restore -> do

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

83− exHandler

84− run (return tid)

85−

86−-- | 'fork' lifted to any monad with 'MonadBaseControl IO m' capability.

87−--

88−{-# INLINABLE fork #-}

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

90−fork = liftBaseDiscard forkIO

91−

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

93−-- returned threadId is garbage collected.

94−--

95−{-# INLINABLE forkManaged #-}

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

97−forkManaged action = do

98− tid <- fork action

99− liftIO $ addFinalizer tid (killThread tid)

100− return tid

src/Streamly/Internal/Control/ForkIO.hsnew file+50

@@ -0,0 +1,50 @@

1+{-# LANGUAGE UnboxedTuples #-}

2+

3+-- |

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

6+-- License : BSD-3-Clause

7+-- Maintainer : streamly@composewell.com

8+-- Stability : experimental

9+-- Portability : GHC

10+

11+module Streamly.Internal.Control.ForkIO

12+ ( rawForkIO

13+ , forkIOManaged

14+ , forkManagedWith

15+ )

16+where

17+

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

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

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

21+import GHC.Exts

22+import GHC.IO (IO(..))

23+import System.Mem.Weak (addFinalizer)

24+

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

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

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

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

29+-- exception handler.

30+{-# INLINE rawForkIO #-}

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

32+rawForkIO (IO action) = IO $ \ s ->

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

34+

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

36+-- returned threadId is garbage collected.

37+--

38+{-# INLINABLE forkManagedWith #-}

39+forkManagedWith :: MonadIO m => (m () -> m ThreadId) -> m () -> m ThreadId

40+forkManagedWith fork action = do

41+ tid <- fork action

42+ liftIO $ addFinalizer tid (killThread tid)

43+ return tid

44+

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

46+-- returned threadId is garbage collected.

47+--

48+{-# INLINABLE forkIOManaged #-}

49+forkIOManaged :: IO () -> IO ThreadId

50+forkIOManaged = forkManagedWith forkIO

src/Streamly/Internal/Control/ForkLifted.hsnew file+55

@@ -0,0 +1,55 @@

1+-- |

2+-- Module : Streamly.Internal.Control.ForkLifted

4+-- License : BSD-3-Clause

5+-- Maintainer : streamly@composewell.com

6+-- Stability : experimental

7+-- Portability : GHC

8+

9+module Streamly.Internal.Control.ForkLifted

10+ (

11+ doFork

12+ , fork

13+ , forkManaged

14+ )

15+where

16+

17+import Control.Concurrent (ThreadId, forkIO)

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

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

20+import Control.Monad.Trans.Control (MonadBaseControl, control, liftBaseDiscard)

21+import Data.Functor (void)

22+import Streamly.Internal.Control.Concurrent (RunInIO(..))

23+import Streamly.Internal.Control.ForkIO (rawForkIO, forkManagedWith)

24+

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

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

27+-- capability.

28+--

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

30+-- as "mrun action" instead.

31+{-# INLINE doFork #-}

32+doFork :: MonadBaseControl IO m

33+ => m ()

34+ -> RunInIO m

35+ -> (SomeException -> IO ())

36+ -> m ThreadId

37+doFork action (RunInIO mrun) exHandler =

38+ control $ \run ->

39+ mask $ \restore -> do

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

41+ exHandler

42+ run (return tid)

43+

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

45+--

46+{-# INLINABLE fork #-}

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

48+fork = liftBaseDiscard forkIO

49+

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

51+-- returned threadId is garbage collected.

52+--

53+{-# INLINABLE forkManaged #-}

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

55+forkManaged = forkManagedWith fork

src/Streamly/Internal/Data/SVar/Dispatch.hs+2−2

@@ -49,8 +49,8 @@ import Data.IORef (IORef, modifyIORef, newIORef, readIORef, writeIORef)

49#if __GLASGOW_HASKELL__ < 804

50import Data.Semigroup ((<>))

51#endif

52−import Streamly.Internal.Control.Concurrent

52+import Streamly.Internal.Control.Concurrent (MonadAsync, captureMonadState)

53− (MonadAsync, captureMonadState, doFork)

53+import Streamly.Internal.Control.ForkLifted (doFork)

54import Streamly.Internal.Data.Atomics

55 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,

56 storeLoadBarrier)

src/Streamly/Internal/Data/Stream/SVar/Eliminate.hs+2−1

@@ -31,7 +31,8 @@ import Control.Monad (when, void)

31import Control.Monad.Catch (throwM)

32import Control.Monad.IO.Class (MonadIO(liftIO))

33import Data.IORef (newIORef, readIORef, writeIORef)

34−import Streamly.Internal.Control.Concurrent (MonadAsync, doFork)

34+import Streamly.Internal.Control.Concurrent (MonadAsync)

35+import Streamly.Internal.Control.ForkLifted (doFork)

35import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)

36import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)

36import Streamly.Internal.Data.Fold.SVar (write, writeLimited)

37import Streamly.Internal.Data.Fold.SVar (write, writeLimited)

37import Streamly.Internal.Data.Fold.Type (Fold(..))

38import Streamly.Internal.Data.Fold.Type (Fold(..))

src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+2−1

@@ -133,7 +133,8 @@ import Data.Maybe (fromJust, isJust)

133import GHC.Types (SPEC(..))

134import qualified Control.Monad.Catch as MC

135

136−import Streamly.Internal.Control.Concurrent (MonadAsync, fork, forkManaged)

136+import Streamly.Internal.Control.Concurrent (MonadAsync)

137+import Streamly.Internal.Control.ForkLifted (fork, forkManaged)

137import Streamly.Internal.Data.Fold.Type (Fold(..))

138import Streamly.Internal.Data.Fold.Type (Fold(..))

138import Streamly.Internal.Data.Pipe.Type (Pipe(..), PipeState(..))

139import Streamly.Internal.Data.Pipe.Type (Pipe(..), PipeState(..))

139import Streamly.Internal.Data.SVar.Type (defState, adaptState)

140import Streamly.Internal.Data.SVar.Type (defState, adaptState)

src/Streamly/Internal/Data/Time/Clock.hs+3−3

@@ -33,7 +33,7 @@ import Control.Monad (forever, when, void)

33import Streamly.Internal.Data.Time.Clock.Type (Clock(..), getTime)

34import Streamly.Internal.Data.Time.Units

35 (MicroSecond64(..), fromAbsTime, addToAbsTime, toRelTime)

36−import Streamly.Internal.Control.Concurrent (forkManaged)

36+import Streamly.Internal.Control.ForkIO (forkIOManaged)

37

38import qualified Streamly.Internal.Data.IORef.Prim as Prim

39

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

82asyncClock clock g = do

83 timeVar <- Prim.newIORef 0

84 updateTimeVar clock timeVar

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

85+ tid <- forkIOManaged $ forever (updateWithDelay clock g timeVar)

86 return (tid, timeVar)

87

88{-# INLINE readClock #-}

@@ -145,7 +145,7 @@ timer clock g period = do

145 reset = resetTimerExpiry clock p1 timeVar

146 process = processTimerTick clock g timeVar mvar reset

147 reset

148− tid <- forkManaged $ forever process

148+ tid <- forkIOManaged $ forever process

149 return $ Timer tid mvar reset

150

151-- | Blocking wait for a timer tick.

src/Streamly/Internal/Network/Inet/TCP.hs+2−1

@@ -103,7 +103,8 @@ import Network.Socket

103 socket)

104import Prelude hiding (read)

105

106−import Streamly.Internal.Control.Concurrent (MonadAsync, fork)

106+import Streamly.Internal.Control.Concurrent (MonadAsync)

107+import Streamly.Internal.Control.ForkLifted (fork)

107import Streamly.Internal.Data.Array.Foreign.Type (Array(..), writeNUnsafe)

108import Streamly.Internal.Data.Array.Foreign.Type (Array(..), writeNUnsafe)

108import Streamly.Internal.Data.Fold.Type (Fold(..))

109import Streamly.Internal.Data.Fold.Type (Fold(..))

109import Streamly.Internal.Data.Stream.IsStream.Type (IsStream)

110import Streamly.Internal.Data.Stream.IsStream.Type (IsStream)

streamly.cabal+2

@@ -384,6 +384,8 @@ library

384 , Streamly.Internal.BaseCompat

385 , Streamly.Internal.Control.Exception

386 , Streamly.Internal.Control.Monad

387+ , Streamly.Internal.Control.ForkIO

388+ , Streamly.Internal.Control.ForkLifted

387 , Streamly.Internal.Control.Concurrent

389 , Streamly.Internal.Control.Concurrent

388 , Streamly.Internal.Data.Cont

390 , Streamly.Internal.Data.Cont

389 , Streamly.Internal.Data.Tuple.Strict

391 , Streamly.Internal.Data.Tuple.Strict