git.iko.soy/streamly
on 83d00da help sign in
streamly/src/Streamly/Internal/Control/Concurrent.hs
89 lines 2.9 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 , doFork
16 , fork
17 , forkManaged
18 )
19where
21import Control.Concurrent (ThreadId, forkIO, killThread)
22import Control.Exception (SomeException(..), catch, mask)
23import Control.Monad.Catch (MonadThrow)
24import Control.Monad.IO.Class (MonadIO(..))
25import Control.Monad.Trans.Control
26 (MonadBaseControl, control, StM, liftBaseDiscard)
27import Data.Functor (void)
28import GHC.Conc (ThreadId(..))
29import GHC.Exts
30import GHC.IO (IO(..))
31import System.Mem.Weak (addFinalizer)
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
42type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
44newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }
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 #-}
52rawForkIO :: IO () -> IO ThreadId
53rawForkIO action = IO $ \ s ->
54 case fork# action s of (# s1, tid #) -> (# s1, ThreadId tid #)
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 #-}
63doFork :: MonadBaseControl IO m
64 => m ()
65 -> RunInIO m
66 -> (SomeException -> IO ())
67 -> m ThreadId
68doFork 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)
75-- | 'fork' lifted to any monad with 'MonadBaseControl IO m' capability.
76--
77{-# INLINABLE fork #-}
78fork :: MonadBaseControl IO m => m () -> m ThreadId
79fork = liftBaseDiscard forkIO
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 #-}
85forkManaged :: (MonadIO m, MonadBaseControl IO m) => m () -> m ThreadId
86forkManaged action = do
87 tid <- fork action
88 liftIO $ addFinalizer tid (killThread tid)
89 return tid