git.iko.soy/streamly
on cfcf26d help sign in
streamly/src/Streamly/Internal/Control/Concurrent.hs
100 lines 3.5 kB history raw
1{-# LANGUAGE UnboxedTuples #-}
3-- |
4-- Module : Streamly.Internal.Control.Concurrent
5-- Copyright : (c) 2017 Composewell Technologies
6-- License : BSD-3-Clause
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
9-- Portability : GHC
11module Streamly.Internal.Control.Concurrent
12 (
13 MonadAsync
14 , RunInIO(..)
15 , captureMonadState
16 , doFork
17 , fork
18 , forkManaged
19 )
20where
22import Control.Concurrent (ThreadId, forkIO, killThread)
23import Control.Exception (SomeException(..), catch, mask)
24import Control.Monad.Catch (MonadThrow)
25import Control.Monad.IO.Class (MonadIO(..))
26import Control.Monad.Trans.Control
27 (MonadBaseControl, control, StM, liftBaseDiscard)
28import Data.Functor (void)
29import GHC.Conc (ThreadId(..))
30import GHC.Exts
31import GHC.IO (IO(..))
32import System.Mem.Weak (addFinalizer)
34-- /Since: 0.8.0 ("Streamly.Prelude")/
35--
36-- | A monad that can perform concurrent or parallel IO operations. Streams
37-- that can be composed concurrently require the underlying monad to be
38-- 'MonadAsync'.
39--
40-- /Since: 0.1.0 ("Streamly")/
41--
42-- @since 0.8.0
43type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
45newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }
47-- | When we run computations concurrently, we completely isolate the state of
48-- the concurrent computations from the parent computation. The invariant is
49-- that we should never be running two concurrent computations in the same
50-- thread without using the runInIO function. Also, we should never be running
51-- a concurrent computation in the parent thread, otherwise it may affect the
52-- state of the parent which is against the defined semantics of concurrent
53-- execution.
54captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
55captureMonadState = control $ \run -> run (return $ RunInIO run)
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 #-}
63rawForkIO :: IO () -> IO ThreadId
64rawForkIO action = IO $ \ s ->
65 case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)
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 #-}
74doFork :: MonadBaseControl IO m
75 => m ()
76 -> RunInIO m
77 -> (SomeException -> IO ())
78 -> m ThreadId
79doFork 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)
86-- | 'fork' lifted to any monad with 'MonadBaseControl IO m' capability.
87--
88{-# INLINABLE fork #-}
89fork :: MonadBaseControl IO m => m () -> m ThreadId
90fork = liftBaseDiscard forkIO
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 #-}
96forkManaged :: (MonadIO m, MonadBaseControl IO m) => m () -> m ThreadId
97forkManaged action = do
98 tid <- fork action
99 liftIO $ addFinalizer tid (killThread tid)
100 return tid