@@ -105,20 +105,17 @@ module Streamly.Internal.Data.SVar
105105 )
106106where
107107
108−import Control.Concurrent
109− (ThreadId, myThreadId, threadDelay, throwTo, forkIO, killThread)
108+import Control.Concurrent (ThreadId, myThreadId, threadDelay, throwTo)
110109import Control.Concurrent.MVar
111110 (MVar, newEmptyMVar, tryPutMVar, takeMVar, tryTakeMVar, newMVar,
112111 tryReadMVar)
113112import Control.Exception
114− (SomeException(..), catch, mask, assert, Exception, catches,
113+ (SomeException(..), assert, Exception, catches,
115114 throwIO, Handler(..), BlockedIndefinitelyOnMVar(..),
116115 BlockedIndefinitelyOnSTM(..))
117116import Control.Monad (when)
118−import Control.Monad.Catch (MonadThrow)
119117import Control.Monad.IO.Class (MonadIO(..))
120−import Control.Monad.Trans.Control
121− (MonadBaseControl, control, StM, liftBaseDiscard)
118+import Control.Monad.Trans.Control (MonadBaseControl, control)
122119import Streamly.Internal.Data.Atomics
123120 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,
124121 storeLoadBarrier)
@@ -126,21 +123,19 @@ import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)
126123import Data.Functor (void)
127124import Data.Heap (Heap, Entry(..))
128125import Data.Int (Int64)
129−import Data.Kind (Type)
130126import Data.IORef
131127 (IORef, modifyIORef, newIORef, readIORef, writeIORef, atomicModifyIORef)
128+import Data.Kind (Type)
132129import Data.Maybe (fromJust, fromMaybe)
133130#if __GLASGOW_HASKELL__ < 808
134131import Data.Semigroup ((<>))
135132#endif
136133import Data.Set (Set)
137−import GHC.Conc (ThreadId(..))
138−import GHC.Exts
139−import GHC.IO (IO(..))
140134import System.IO (hPutStrLn, stderr)
141−import System.Mem.Weak (addFinalizer)
142135
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)
144139import Streamly.Internal.Data.Time.Units
145140 (AbsTime, NanoSecond64(..), MicroSecond64(..), diffAbsTime64,
146141 fromRelTime64, toRelTime64, showNanoSecond64, showRelTime64)
@@ -898,66 +893,16 @@ withDiagMVar sv label action =
898893-- Spawning threads
899894------------------------------------------------------------------------------
900895
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
913897-- the concurrent computations from the parent computation. The invariant is
914898-- that we should never be running two concurrent computations in the same
915899-- thread without using the runInIO function. Also, we should never be running
916900-- a concurrent computation in the parent thread, otherwise it may affect the
917901-- state of the parent which is against the defined semantics of concurrent
918902-- execution.
919−newtype RunInIO m = RunInIO { runInIO :: forall b. m b -> IO (StM m b) }
920−
921903captureMonadState :: MonadBaseControl IO m => m (RunInIO m)
922904captureMonadState = control $ \run -> run (return $ RunInIO run)
923905
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−
961906------------------------------------------------------------------------------
962907-- Collecting results from child workers in a streamed fashion
963908------------------------------------------------------------------------------