X-Git-Url: http://git.megacz.com/?a=blobdiff_plain;f=GHC%2FConc.lhs;h=d676a1a7ae7c1f6ec672d7a24e68492ee8956ec9;hb=c497471bb7fc204afb546f3116ad116f8569e452;hp=57500f4a3f0899ef81be2731045692d58103a706;hpb=bb31de5a29cc746d6140c1b425b6c264af7c9de5;p=ghc-base.git diff --git a/GHC/Conc.lhs b/GHC/Conc.lhs index 57500f4..d676a1a 100644 --- a/GHC/Conc.lhs +++ b/GHC/Conc.lhs @@ -29,7 +29,9 @@ module GHC.Conc -- * Forking and suchlike , forkIO -- :: IO a -> IO ThreadId + , forkIOUnmasked , forkOnIO -- :: Int -> IO a -> IO ThreadId + , forkOnIOUnmasked , numCapabilities -- :: Int , childHandler -- :: Exception -> IO () , myThreadId -- :: IO ThreadId @@ -211,11 +213,11 @@ thread. The new thread will be a lightweight thread; if you want to use a foreign library that uses thread-local storage, use 'Control.Concurrent.forkOS' instead. -GHC note: the new thread inherits the /blocked/ state of the parent -(see 'Control.Exception.block'). +GHC note: the new thread inherits the /masked/ state of the parent +(see 'Control.Exception.mask'). The newly created thread has an exception handler that discards the -exceptions 'BlockedOnDeadMVar', 'BlockedIndefinitely', and +exceptions 'BlockedIndefinitelyOnMVar', 'BlockedIndefinitelyOnSTM', and 'ThreadKilled', and passes all other exceptions to the uncaught exception handler (see 'setUncaughtExceptionHandler'). -} @@ -225,6 +227,11 @@ forkIO action = IO $ \ s -> where action_plus = catchException action childHandler +-- | Like 'forkIO', but the child thread is created with asynchronous exceptions +-- unmasked (see 'Control.Exception.mask'). +forkIOUnmasked :: IO () -> IO ThreadId +forkIOUnmasked io = forkIO (unsafeUnmask io) + {- | Like 'forkIO', but lets you specify on which CPU the thread is created. Unlike a `forkIO` thread, a thread created by `forkOnIO` @@ -244,6 +251,11 @@ forkOnIO (I# cpu) action = IO $ \ s -> where action_plus = catchException action childHandler +-- | Like 'forkOnIO', but the child thread is created with +-- asynchronous exceptions unmasked (see 'Control.Exception.mask'). +forkOnIOUnmasked :: Int -> IO () -> IO ThreadId +forkOnIOUnmasked cpu io = forkOnIO cpu (unsafeUnmask io) + -- | the value passed to the @+RTS -N@ flag. This is the number of -- Haskell threads that can run truly simultaneously at any given -- time, and is typically set to the number of physical CPU cores on @@ -275,16 +287,11 @@ real_handler se@(SomeException ex) = Just StackOverflow -> reportStackOverflow _ -> reportError se -{- | 'killThread' terminates the given thread (GHC only). -Any work already done by the thread isn\'t -lost: the computation is suspended until required by another thread. -The memory used by the thread will be garbage collected if it isn\'t -referenced from anywhere. The 'killThread' function is defined in -terms of 'throwTo': +{- | 'killThread' raises the 'ThreadKilled' exception in the given +thread (GHC only). > killThread tid = throwTo tid ThreadKilled -Killthread is a no-op if the target thread has already completed. -} killThread :: ThreadId -> IO () killThread tid = throwTo tid ThreadKilled @@ -299,10 +306,14 @@ when dealing with race conditions: eg. if there are two threads that can kill each other, it is guaranteed that only one of the threads will get to kill the other. +Whatever work the target thread was doing when the exception was +raised is not lost: the computation is suspended until required by +another thread. + If the target thread is currently making a foreign call, then the exception will not be raised (and hence 'throwTo' will not return) until the call has completed. This is the case regardless of whether -the call is inside a 'block' or not. +the call is inside a 'mask' or not. Important note: the behaviour of 'throwTo' differs from that described in the paper \"Asynchronous exceptions in Haskell\" @@ -311,14 +322,20 @@ In the paper, 'throwTo' is non-blocking; but the library implementation adopts a more synchronous design in which 'throwTo' does not return until the exception is received by the target thread. The trade-off is discussed in Section 9 of the paper. Like any blocking operation, 'throwTo' is therefore interruptible (see Section 5.3 of -the paper). +the paper). Unlike other interruptible operations, however, 'throwTo' +is /always/ interruptible, even if it does not actually block. -There is currently no guarantee that the exception delivered by 'throwTo' will be -delivered at the first possible opportunity. In particular, a thread may -unblock and then re-block exceptions (using 'unblock' and 'block') without receiving -a pending 'throwTo'. This is arguably undesirable behaviour. +There is no guarantee that the exception will be delivered promptly, +although the runtime will endeavour to ensure that arbitrary +delays don't occur. In GHC, an exception can only be raised when a +thread reaches a /safe point/, where a safe point is where memory +allocation occurs. Some loops do not perform any memory allocation +inside the loop and therefore cannot be interrupted by a 'throwTo'. - -} +Blocked 'throwTo' is fair: if multiple threads are trying to throw an +exception to the same target thread, they will succeed in FIFO order. + + -} throwTo :: Exception e => ThreadId -> e -> IO () throwTo (ThreadId tid) ex = IO $ \ s -> case (killThread# tid (toException ex) s) of s1 -> (# s1, () #) @@ -472,6 +489,10 @@ thenSTM (STM m) k = STM ( \s -> returnSTM :: a -> STM a returnSTM x = STM (\s -> (# s, x #)) +instance MonadPlus STM where + mzero = retry + mplus = orElse + -- | Unsafely performs IO in the STM monad. Beware: this is a highly -- dangerous thing to do. -- @@ -602,18 +623,18 @@ MVar utilities \begin{code} withMVar :: MVar a -> (a -> IO b) -> IO b withMVar m io = - block $ do + mask $ \restore -> do a <- takeMVar m - b <- catchAny (unblock (io a)) + b <- catchAny (restore (io a)) (\e -> do putMVar m a; throw e) putMVar m a return b modifyMVar_ :: MVar a -> (a -> IO a) -> IO () modifyMVar_ m io = - block $ do + mask $ \restore -> do a <- takeMVar m - a' <- catchAny (unblock (io a)) + a' <- catchAny (restore (io a)) (\e -> do putMVar m a; throw e) putMVar m a' return () @@ -837,10 +858,11 @@ foreign import ccall unsafe "getOrSetGHCConcProddingStore" prodServiceThread :: IO () prodServiceThread = do - was_set <- readIORef prodding - writeIORef prodding True - -- no need for atomicModifyIORef, extra prods are harmless. - if (not (was_set)) then wakeupIOManager else return () + -- NB. use atomicModifyIORef here, otherwise there are race + -- conditions in which prodding is left at True but the server is + -- blocked in select(). + was_set <- atomicModifyIORef prodding $ \b -> (True,b) + unless was_set wakeupIOManager -- Machinery needed to ensure that we only have one copy of certain -- CAFs in this module even when the base package is present twice, as @@ -851,7 +873,7 @@ prodServiceThread = do -- sharedCAF :: a -> (Ptr a -> IO (Ptr a)) -> IO a sharedCAF a get_or_set = - block $ do + mask_ $ do stable_ref <- newStablePtr a let ref = castPtr (castStablePtrToPtr stable_ref) ref2 <- get_or_set ref @@ -1014,6 +1036,17 @@ service_loop -> IO () service_loop wakeup readfds writefds ptimeval old_reqs old_delays = do + -- reset prodding before we look at the new requests. If a new + -- client arrives after this point they will send a wakup which will + -- cause the server to loop around again, so we can be sure to not + -- miss any requests. + -- + -- NB. it's important to do this in the *first* iteration of + -- service_loop, rather than after calling select(), since a client + -- may have set prodding to True without sending a wakeup byte down + -- the pipe, because the pipe wasn't set up. + atomicModifyIORef prodding (\_ -> (False, ())) + -- pick up new IO requests new_reqs <- atomicModifyIORef pendingEvents (\a -> ([],a)) let reqs = new_reqs ++ old_reqs @@ -1084,8 +1117,6 @@ service_loop wakeup readfds writefds ptimeval old_reqs old_delays = do unless exit $ do - atomicModifyIORef prodding (\_ -> (False, ())) - reqs' <- if wakeup_all then do wakeupAll reqs; return [] else completeRequests reqs readfds writefds []