Andrea Vezzosi pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC Commits: f4cd1fff by Andrea Vezzosi at 2026-10-27T11:37:54+01:00 interp: spawn fillPipe thread, writePipe now safely interruptible Previously an async exception during a call to `writePipe` could have caused a partial write, making any further write impossible to parse from the other end. Now `writePipe` merely pushes the message onto a queue, and the actual write happens in a separate `fillPipe` thread. As a result of this and previous changes, `interpCmd` no longer needs `interruptibleMask` to preserve the state of the protocol. `writePipe` waits for the actual write to complete, and rethrows exceptions raised in the write attempt, if any. - - - - - 90ceb49f by Andrea Vezzosi at 2026-10-27T11:38:07+01:00 Follow `Note [Exceptions in TH]` in `qReport` implementation `qReport b msg` should use `seqList msg` to trigger any exceptions in the `String`, as explained in the note. Through refactors the `seqList msg` call followed the bulk of the implementation in `GHC.Tc.Gen.Splice.report`, but the call should be in `qReport` itself, so that, even with -fexternal-interpreter, `msg` gets forced right away and any exceptions get reported as a compilation error. Follow up to 'interp: spawn fillPipe thread, writePipe now safely interruptible'. - - - - - 9fc81afa by Rodrigo Mesquita at 2026-10-27T11:38:07+01:00 test: make T3171 not fragile and add version for ext interp too - actually test the sleep of the user-given expression gets interrupted - the ext interp version tests two things: - Requests can be cancelled without terminating the ext interp (#18852) - The stmt being evaluated gets interrupted in the ext interp (that is, ^C gets sent to the thread running the user-given expression and doesn't affect the ext interp main thread) - testing ext interp requires sending ^C to whole process group Fixes #15383 - - - - - a5f19ce3 by Andrea Vezzosi at 2026-10-27T11:38:07+01:00 interp: cancel requests if async exceptions while waiting for reply. - - - - - d16a4f36 by Andrea Vezzosi at 2026-10-27T11:38:08+01:00 interp: do not terminate ext. interp. process on exception `sendMessage` throwing an exception does not warrant terminating the external interpreter process anymore, because communication can keep going no matter what. `stopInterp` now resets the whole `Interp` state, so that a new ExtInterpInstance can be later created and work correctly. `withExtInterp` now calls `stopInterp` on a `PipeClosedException` and rethrows. Fixes #18852 - - - - - 16 changed files: - compiler/GHC/Runtime/Interpreter.hs - compiler/GHC/Runtime/Interpreter/Process.hs - compiler/GHC/Tc/Errors/Types.hs - compiler/GHC/Tc/Gen/Splice.hs - libraries/ghci/GHCi/Message.hs - libraries/ghci/GHCi/Message/Async.hs - libraries/ghci/GHCi/Server.hs - libraries/ghci/GHCi/Server/Threads.hs - libraries/template-haskell/Language/Haskell/TH/Syntax.hs - testsuite/tests/ghci/should_run/Makefile - testsuite/tests/ghci/should_run/T3171.stdout - + testsuite/tests/ghci/should_run/T3171ext.stdout - testsuite/tests/ghci/should_run/all.T - + testsuite/tests/th/T8987b.hs - + testsuite/tests/th/T8987b.stderr - testsuite/tests/th/all.T Changes: ===================================== compiler/GHC/Runtime/Interpreter.hs ===================================== @@ -105,7 +105,7 @@ import GHCi.Run import Control.Concurrent import Control.Monad import Control.Monad.IO.Class -import Control.Monad.Catch as MC (mask) +import Control.Monad.Catch as MC (mask, handle, MonadThrow (..)) import Data.Binary import Data.ByteString (ByteString) import qualified Data.ByteString.Short as SBS @@ -196,16 +196,16 @@ interpCmd interp msg = case interpInstance interp of #if defined(HAVE_INTERNAL_INTERPRETER) InternalInterp -> run msg -- Just run it directly #endif - ExternalInterp ext -> withExtInterp ext $ \inst -> - uninterruptibleMask_ $ -- Note [uninterruptibleMask_ and interpCmd] - sendMessage inst msg + ExternalInterp ext -> withExtInterp interp ext $ \inst -> + sendMessage inst msg - -withExtInterp :: ExceptionMonad m => ExtInterp -> (forall d. ExtInterpInstance d -> m a) -> m a -withExtInterp ext action = case ext of +withExtInterp :: ExceptionMonad m => Interp -> ExtInterp -> (forall d. ExtInterpInstance d -> m a) -> m a +withExtInterp interp ext action = stopInterpOnPipeClosed $ case ext of ExtJS i -> withJSInterp i action ExtWasm i -> withWasmInterp i action ExtIServ i -> withIServ i action + where + stopInterpOnPipeClosed = MC.handle $ \ (e :: PipeClosedException) -> liftIO (stopInterp interp) >> MC.throwM e withExtInterpStatus :: ExtInterp -> (forall d. ExtInterpStatusVar d -> m a) -> m a withExtInterpStatus ext action = case ext of @@ -213,15 +213,6 @@ withExtInterpStatus ext action = case ext of ExtWasm i -> action $ interpStatus i ExtIServ i -> action (interpStatus i) --- Note [uninterruptibleMask_ and interpCmd] --- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ --- If we receive an async exception, such as ^C, while communicating --- with the iserv process then we will be out-of-sync and not be able --- to recover. Thus we use uninterruptibleMask_ during --- communication. A ^C will be delivered to the iserv process (because --- signals get sent to the whole process group) which will interrupt --- the running computation and return an EvalException result. - -- | Grab a lock on the 'IServ' and do something with it. -- Overloaded because this is used from TcM as well as IO. withIServ @@ -480,11 +471,9 @@ lookupSymbol interp str = withSymbolCache interp str $ #endif ExternalInterp ext -> case ext of ExtIServ i -> withIServ i $ \inst -> do - uninterruptibleMask_ $ sendMessage inst (LookupSymbol (fastStringToShortByteString (interpSymbolToCLabel str))) ExtJS {} -> pprPanic "lookupSymbol not supported by the JS interpreter" (ppr str) ExtWasm i -> withWasmInterp i $ \inst -> do - uninterruptibleMask_ $ sendMessage inst (LookupSymbol (fastStringToShortByteString (interpSymbolToCLabel str))) lookupSymbolInDLL :: Interp -> RemotePtr LoadedDLL -> InterpSymbol s -> IO (Maybe (RemotePtr ())) @@ -495,7 +484,6 @@ lookupSymbolInDLL interp dll str = withSymbolCache interp str $ #endif ExternalInterp ext -> case ext of ExtIServ i -> withIServ i $ \inst -> do - uninterruptibleMask_ $ sendMessage inst (LookupSymbolInDLL dll (fastStringToShortByteString (interpSymbolToCLabel str))) ExtJS {} -> pprPanic "lookupSymbol not supported by the JS interpreter" (ppr str) -- wasm dyld doesn't track which symbol comes from which .so @@ -656,6 +644,9 @@ stopInterp interp = case interpInstance interp of else sendMessage i Shutdown -- We asked the server to shutdown, so we will also shutdown any communication. closePipe (interpPipe (instProcess i)) + modifyMVar_ (loader_state (interpLoader interp)) (const $ pure Nothing) + purgeInterpSymbolCache (interpSymbolCache interp) + modifyMVar_ (interpStringCache interp) (const $ pure emptyFsEnv) pure InterpPending -- ----------------------------------------------------------------------------- ===================================== compiler/GHC/Runtime/Interpreter/Process.hs ===================================== @@ -12,13 +12,10 @@ import GHC.Prelude import GHC.Runtime.Interpreter.Types import GHCi.Message -import GHC.IO (catchException) import GHC.Utils.Panic import GHC.Utils.Exception as Ex import Data.Binary -import System.Exit -import System.Process import Control.Concurrent import Data.IORef import Control.Concurrent.STM @@ -34,7 +31,6 @@ import Control.Concurrent.STM sendMessage :: Binary a => ExtInterpInstance d -> Message a -> IO a sendMessage i msg = remoteCall (interpPipe proc) msg - `catchException` \(e :: SomeException) -> handleInterpProcessFailure proc e where proc = instProcess i @@ -64,7 +60,7 @@ sendTHComputation i msg reqHandler = mask $ \restore -> do where proc = instProcess i handleOne mbox doneRef = do - handleRemoteTHCall (interpPipe proc) mbox $ \req -> do + handleRemoteTHCall (interpPipe proc) mbox $ \_setCancel req -> do (stop, res) <- reqHandler req writeIORef doneRef stop return res @@ -77,18 +73,6 @@ sendTHComputation i msg reqHandler = mask $ \restore -> do else loop in loop -handleInterpProcessFailure :: InterpProcess -> SomeException -> IO a -handleInterpProcessFailure i e = do - let hdl = interpHandle i - ex <- getProcessExitCode hdl - case ex of - Just (ExitFailure n) -> - throwIO (InstallationError ("External interpreter terminated (" ++ show n ++ ")")) - _ -> do - terminateProcess hdl - _ <- waitForProcess hdl - throw e - data CancelTHHandler = CancelTHHandler deriving Show instance Exception CancelTHHandler where fromException = asyncExceptionFromException ===================================== compiler/GHC/Tc/Errors/Types.hs ===================================== @@ -7091,6 +7091,7 @@ data SpliceFailReason th/T5976 th/T7276a th/T8987 + th/T8987b th/TH_exn1 th/TH_exn2 th/TH_runIO ===================================== compiler/GHC/Tc/Gen/Splice.hs ===================================== @@ -1160,7 +1160,7 @@ runRemoteModFinalizers (ThModFinalizers finRefs) = do runQinTcM $ sequence_ qs #endif - ExternalInterp ext -> withExtInterp ext $ \inst -> do + ExternalInterp ext -> withExtInterp interp ext $ \inst -> do tcg <- getGblEnv th_state <- readTcRef (tcg_th_remote_state tcg) case th_state of @@ -1472,11 +1472,9 @@ when showing an error message. To call runQ in the Tc monad, we need to make TcM an instance of Quasi: -} --- 'msg' is forced to ensure exceptions don't escape, --- see Note [Exceptions in TH] report :: Bool -> [Char] -> TcM () -report True msg = seqList msg $ addErr $ TcRnTHError $ ReportCustomQuasiError True msg -report False msg = seqList msg $ addDiagnostic $ TcRnTHError $ ReportCustomQuasiError False msg +report True msg = addErr $ TcRnTHError $ ReportCustomQuasiError True msg +report False msg = addDiagnostic $ TcRnTHError $ ReportCustomQuasiError False msg addTopDecls :: [TH.Dec] -> TcM () addTopDecls thds = do @@ -1575,7 +1573,7 @@ metaHandlersTcM runInIO = TH.MetaHandlers { ; let i = toInteger (getKey u) ; return (TH.mkNameU s i) } - , mReport = fmap runInIO . report + , mReport = \ b msg -> runInIO (report b msg) , mLocation = runInIO location @@ -1819,7 +1817,7 @@ runTH ty fhv = do return r #endif - ExternalInterp ext -> withExtInterp ext $ \inst -> do + ExternalInterp ext -> withExtInterp interp ext $ \inst -> do -- Run it on the server. For an overview of how TH works with -- Remote GHCi, see Note [Remote Template Haskell] in -- libraries/ghci/GHCi/TH.hs. ===================================== libraries/ghci/GHCi/Message.hs ===================================== @@ -27,7 +27,7 @@ module GHCi.Message , Pipe(..), mkPipeFromHandles, mkPipeFromContinuations, closePipe , remoteCall, remoteTHCall , handleRemoteCall, handleRemoteTHCall, writeReply - , RemoteProcessException(..), PipeReadClosedException(..) + , RemoteProcessException(..), PipeClosedException(..) , InboxId(..) , newMailbox @@ -666,12 +666,12 @@ remoteCall pipe msg = either throwIO (pure . runGet get) -- | Handle a single 'Message' request on the interpreter and reply to the host handleRemoteCall :: Pipe -> (Msg -> IO Msg) {-^ Msg hook -} - -> (forall a. (Show a, Binary a) => WriteReply a %1 -> Message a -> IO r) + -> (forall a. (Show a, Binary a) => WriteReply a %1 -> (IO () -> IO ()) -> Message a -> IO r) -> IO r handleRemoteCall pipe hook k = - handleAsyncReq pipe getMessage Nothing \(WriteReply writer) msg0 -> IO \s0 -> + handleAsyncReq pipe getMessage Nothing \(WriteReply writer) sa msg0 -> IO \s0 -> case hook msg0 |#> s0 of -- unIO for linearity checking - (# s1, Msg msg #) -> k (WriteReply \x -> writer (fmap put x)) msg |#> s1 + (# s1, Msg msg #) -> k (WriteReply \x -> writer (fmap put x)) sa msg |#> s1 -- | Send a 'THMessage' request to the host and wait for a reply from the host -- to the interpreter. @@ -687,9 +687,9 @@ remoteTHCall pipe ii msg = either throwIO (pure . runGet get) -- Symmterically to 'remoteTHCall', we always want to specify the 'InboxId' for -- RunTH "reverse requests" handleRemoteTHCall - :: Pipe -> InboxId -> (forall a. (Show a, Binary a) => THMessage a -> IO a) -> IO () + :: Pipe -> InboxId -> (forall a. (Show a, Binary a) => (IO () -> IO ()) -> THMessage a -> IO a) -> IO () handleRemoteTHCall pipe ii k = - handleAsyncReq pipe getTHMessage (Just ii) \writer (THMsg msg) -> IO \s0 -> - case k msg |#> s0 of -- unIO for linearity checking + handleAsyncReq pipe getTHMessage (Just ii) \writer sa (THMsg msg) -> IO \s0 -> + case k sa msg |#> s0 of -- unIO for linearity checking (# s1, a #) -> writeReply writer (put a) |#> s1 ===================================== libraries/ghci/GHCi/Message/Async.hs ===================================== @@ -7,7 +7,7 @@ module GHCi.Message.Async ( -- * Pipes Pipe(..), mkPipeFromHandles, mkPipeFromContinuations - , closePipe, PipeReadClosedException(..) + , closePipe, PipeClosedException(..) -- * Interface for async bidirectional RPC , remoteAsyncReq, handleAsyncReq @@ -77,6 +77,10 @@ data Pipe = Pipe , pipeReadClosed :: !(TVar Bool) -- ^ After the pipe is closed, messages can no longer be received from the -- other end, but the 'AsyncManager' queues might not be empty yet. + , pipeWriteClosed :: !(TVar Bool) + -- ^ After the write end of the pipe is closed we can no longer submit + -- requests or respond to ones we receive, but we might still receive + -- replies. } -- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write. @@ -91,7 +95,13 @@ mkPipeFromHandles pipeRead pipeWrite = do -- | Make a 'Pipe' from a reader function and a writer function. -- The pipe should be closed with 'closePipe', which will call 'pipeCloseRead' and 'pipeCloseWrite'. -mkPipeFromContinuations :: IO ByteString -> (B.Builder -> IO ()) +-- +-- Arguments 'getSome' and 'putAll' should throw an `IOException` when communication fails +-- or is no longer possible. +mkPipeFromContinuations :: IO ByteString + -- ^ getSome + -> (B.Builder -> IO ()) + -- ^ putAll -> IO () -- ^ pipeCloseRead -> IO () @@ -101,6 +111,7 @@ mkPipeFromContinuations getSome putAll pipeCloseRead pipeCloseWrite = mdo pipeLeftovers <- newIORef Nothing pipeLock <- newMVar () pipeReadClosed <- newTVarIO False + pipeWriteClosed <- newTVarIO False mdo pipeAsyncMngr <- newAsyncManager p let p = Pipe { .. } @@ -114,19 +125,33 @@ mkPipeFromContinuations getSome putAll pipeCloseRead pipeCloseWrite = mdo -- write. Only then do we stop the drain thread, which closes the read end -- itself once its read has finished. On Windows its read can't be -- interrupted, so this waits for the other side to close its write end too. +-- Finally we stop the write thread, which could be blocked waiting for more to write. closePipe :: Pipe -> IO () closePipe p@Pipe{..} = do readAlreadyClosed <- atomically $ do closed <- readTVar pipeReadClosed failPendingReplies p (toException PipeReadClosedException) pure closed - withLock p pipeCloseWrite `finally` - unless readAlreadyClosed - (throwTo (asyncDrainTid pipeAsyncMngr) PipeReadClosedException) --- | We've closed the pipe read end on this process -data PipeReadClosedException = PipeReadClosedException deriving Show -instance Exception PipeReadClosedException + writeAlreadyClosed <- atomically $ do + closed <- readTVar pipeWriteClosed + failPendingWrites p (toException PipeWriteClosedException) + pure closed + + unless writeAlreadyClosed (withLock p pipeCloseWrite) + `finally` do + unless readAlreadyClosed + (throwTo (asyncDrainTid pipeAsyncMngr) PipeReadClosedException) + unless writeAlreadyClosed + (throwTo (asyncFillTid pipeAsyncMngr) PipeWriteClosedException) + +data PipeClosedException + = PipeReadClosedException -- ^ We've closed the pipe read end on this process + + | PipeWriteClosedException -- ^ We've closed the pipe write end on this process + deriving Show +instance Exception PipeClosedException + -------------------------------------------------------------------------------- -- ** Pipe thread-safe read/write @@ -135,7 +160,19 @@ instance Exception PipeReadClosedException -- and identify whether they are requests or replies. writePipe :: Pipe -> IddMsg -> IO () -writePipe p@Pipe{..} = withLock p . putAll . execPut . put +writePipe p msg = do + var <- atomically $ writePipeAsync p msg + either throwIO pure =<< atomically (takeTMVar var) + +writePipeAsync :: Pipe -> IddMsg -> STM (TMVar (Either SomeException ())) +writePipeAsync Pipe{..} msg = do + var <- newEmptyTMVar + readTVar pipeWriteClosed >>= \case + True -> throwSTM PipeWriteClosedException + False -> pure () + writeTQueue (writeQueue pipeAsyncMngr) (msg, var) + pure var + -- | Only the 'drainPipe' thread should read from the pipe directly. Other -- functions should wait for the matching reply to be delivered asynchronously. @@ -181,6 +218,8 @@ withLock Pipe{..} = bracket (takeMVar pipeLock) (putMVar pipeLock) . const -------------------------------------------------------------------------------- -- | Send an asynchronous request and block waiting for a matching reply +-- +-- See Note [Exceptions during serialization] about exceptions the 'Put' argument might raise. remoteAsyncReq :: Pipe -> Put -> (Maybe InboxId) -> IO (Either SomeException LB.ByteString) remoteAsyncReq pipe msg ii = do @@ -205,14 +244,18 @@ remoteAsyncReq pipe msg ii = do -- Write request writePipe pipe (Request uq ii (runPut msg)) `onException` - -- the req wasn't done, or no one will be reading the reply: - -- in either case, drop it from the pending map. - atomically (modifyTVar pendingReplies (IntMap.delete uq)) + -- the req wasn't done, or no one will be reading the reply: + -- in either case, drop it from the pending map. + atomically (modifyTVar pendingReplies (IntMap.delete uq)) -- Block waiting for reply atomically (takeTMVar wv) `onException` - atomically (modifyTVar pendingReplies (IntMap.delete uq)) + atomically (do + modifyTVar pendingReplies (IntMap.delete uq) + -- we don't wait for Cancel to be written on the pipe, + -- and won't get a reply: nowhere to report it anyway. + writePipeAsync pipe (Cancel uq)) -- | Receive a single asynchronous request and send a matching reply -- asynchronously (the handler may fork a computation that eventually replies @@ -222,19 +265,25 @@ remoteAsyncReq pipe msg ii = do -- It must be used exactly once (all requests expect a reply, and you can't -- reply more than once). Linear types enforce the handler replies in every -- branch, and if the handler otherwise throws, it is handled by 'handleAsyncReq'. -handleAsyncReq :: Pipe -> Get a -> (Maybe InboxId) -> (WriteReply Put %1 -> a -> IO r) -> IO r +-- +-- See Note [Exceptions during serialization] about exceptions the 'Put' argument might raise. +handleAsyncReq :: Pipe -> Get a -> (Maybe InboxId) -> (WriteReply Put %1 -> (IO () -> IO ()) -> a -> IO r) -> IO r handleAsyncReq pipe getit mii k = mask $ \restore -> do let am = pipeAsyncMngr pipe ack <- newEmptyMVar -- Block waiting for request mbreq <- atomically $ do + readTVar (pipeWriteClosed pipe) >>= \case + -- We will not be able to write the reply, so no point in continuiting. + True -> throwSTM PipeWriteClosedException + False -> pure () isClosed <- readTVar (pipeReadClosed pipe) taken <- case mii of Nothing -> tryReadTQueue $ requestsQueue am Just ii -> tryReadTQueue =<< getMailbox am ii case taken of - Just req -> pure (Just req) + Just req@(uq,_) -> fetchedRequest am uq >> pure (Just req) Nothing | isClosed -> pure Nothing -- empty and closed | otherwise -> retry -- wait for msg @@ -242,9 +291,9 @@ handleAsyncReq pipe getit mii k = mask $ \restore -> do Nothing -> throwIO PipeReadClosedException -- can't expect reqs Just (uq, bs) -> do let reply p = do - writePipe pipe (Reply uq (bimap toSerializableException runPut p)) + b <- atomically $ doneHandling am uq + when b $ writePipe pipe (Reply uq (bimap toSerializableException runPut p)) putMVar ack () - -- Run the continuation which must reply exactly once with 'writeReply' case runGetOrFail getit bs of Left (_, _, err) -> do @@ -252,7 +301,7 @@ handleAsyncReq pipe getit mii k = mask $ \restore -> do reply (Left (toException exc)) throwIO exc Right (_, _, it) -> - restore (k (WriteReply reply) it) + restore (k (WriteReply reply) (setActive am uq) it) -- If there's an exception before the reply is sent, reply with an error instead. `catch` \(e :: SomeException) -> do @@ -266,6 +315,7 @@ handleAsyncReq pipe getit mii k = mask $ \restore -> do reply (Left e) throwIO e + -- | 'handleSyncReq' failed to decode an incoming request payload using the given @'Get' a@ newtype DecodeAsyncReqException = DecodeAsyncReqException String deriving Show instance Exception DecodeAsyncReqException @@ -307,6 +357,28 @@ getMailbox am (InboxId ii) = do -- ** Async matching "internals" ----------------------------------------------- +-- | Tracks handling status of a received 'Request'. The values are states of a +-- tiny state machine used to coordinate cancelations. +-- +-- 'handleAsyncReq' will pop a request from the appropriate queue, setting the +-- status to `Fetched`. Then one of two things can happen first: +-- - we receive a 'Cancel', then status is set to `Canceling` so the cancel +-- action will run when set. +-- - handler sets a cancel action, then reception of a `Cancel` runs that +-- action and deletes the RequestStatus. +-- +-- If RequestStatus is `Canceling` or absent, then a `Reply` won't be sent. +-- +-- The two possible transition paths: +-- requestsQueue -> Fetched -> Active cancel -> gone +-- requestsQueue -> Fetched -> Canceling -> gone + +data RequestStatus + = Fetched -- ^ 'Fetched' from a request queue, to pass to handler. + | Active (IO ()) -- ^ Handler set a cancel action, and has not sent a reply + -- yet. The cancel action must not block. + | Canceling -- ^ `Cancel` received while status is still `Fetched`. + data AsyncManager = AsyncManager { pendingReply :: !(TVar (IntMap (TMVar (Either SomeException LB.ByteString)))) -- ^ When a request is sent, keep track that we're expecting a reply. The @@ -317,10 +389,16 @@ data AsyncManager = AsyncManager -- /without/ a specific 'InboxId', queue it here. A request handler will -- pop the next request from it. (Nothing except the 'drainPipe' thread -- should read from the pipe). Read with @'handleAsyncReq' p get Nothing@. + , cancelStatus :: !(TVar (IntMap RequestStatus)) + -- ^ When a `Cancel uq` is received, we lookup `uq` in this map and call the + -- callback if present. , mailboxQueues :: !(TVar (IntMap (TQueue (CorrelationId, LB.ByteString)))) -- ^ Requests for a specific 'InboxId' that can be read using -- @'handleAsyncReq' p get (Just <mailboxid>)@. + , writeQueue :: !(TQueue (IddMsg, TMVar (Either SomeException ()))) + -- ^ Messages to write on the pipe. , asyncDrainTid :: !ThreadId + , asyncFillTid :: !ThreadId , asyncFreshId :: !(IORef Int) } @@ -346,15 +424,21 @@ type CorrelationId = Int -- exclusively. data IddMsg = Request !CorrelationId !(Maybe InboxId) LB.ByteString | Reply !CorrelationId (Either SerializableException LB.ByteString) + | Cancel !CorrelationId -- ^ Technically a Request, but no Reply + -- is expected, the CorrelationId is the + -- one of the Request being canceled. instance Binary IddMsg where put (Request i ii bs) = putWord8 0 >> put i >> put ii >> put bs put (Reply i bs) = putWord8 1 >> put i >> put bs + put (Cancel i) = putWord8 2 >> put i + get = do b <- getWord8 case b of 0 -> Request <$> get <*> get <*> get 1 -> Reply <$> get <*> get + 2 -> Cancel <$> get _ -> fail $ "impossible: Binary IddMsg - " ++ show b newAsyncManager :: Pipe -> IO AsyncManager @@ -362,11 +446,17 @@ newAsyncManager pipe = do asyncFreshId <- newIORef 0 pendingReply <- newTVarIO IntMap.empty requestsQueue <- newTQueueIO + cancelStatus <- newTVarIO IntMap.empty mailboxQueues <- newTVarIO IntMap.empty + writeQueue <- newTQueueIO -- Fork thread to drain read end. -- (cleaned up on 'closePipe') - asyncDrainTid <- forkIOWithUnmask (drainPipe pipe) + asyncDrainTid <- mask_ $ forkIOWithUnmask (drainPipe pipe) + + -- Fork thread to fill write end. + -- (cleaned up on 'closePipe') + asyncFillTid <- mask_ $ forkIOWithUnmask (fillPipe pipe) let am = AsyncManager{..} pure am @@ -382,41 +472,134 @@ drainPipe :: Pipe -> (forall x. IO x -> IO x) -> IO () drainPipe p unmask = catch (unmask $ loop `finally` pipeCloseRead p) replyToPendings where loop = do readPipe p >>= \case - -- Receive a reply to a pending notification - Just (Reply uq bs) -> do - atomically $ do - m <- readTVar (pendingReply am) - case IntMap.lookup uq m of - Nothing -> - -- the req was dropped from pending bc there was an exception, and no - -- one's listening any more. the reply is ignored - pure () - Just wv -> do - writeTVar (pendingReply am) (IntMap.delete uq m) - -- signal matching request - putTMVar wv (bimap (toException . RemoteProcessException) id bs) - loop - -- Receive a request and queue it - Just (Request uq mbi bs) -> do - atomically $ case mbi of - Nothing -> - -- "Global" request - writeTQueue (requestsQueue am) (uq,bs) - Just ii -> do - -- "Specific mailbox" request - q <- getMailbox am ii - writeTQueue q (uq,bs) + Just msg -> do + handleMsg msg loop -- EOF on read handle Nothing -> do -- clean up pipe then exit loop replyToPendings (toException PipeReadClosedException) pure () + handleMsg = \case + -- Receive a reply to a pending notification + Reply uq bs -> do + atomically $ do + m <- readTVar (pendingReply am) + case IntMap.lookup uq m of + Nothing -> + -- the req was dropped from pending bc there was an exception, and no + -- one's listening any more. the reply is ignored + pure () + Just wv -> do + writeTVar (pendingReply am) (IntMap.delete uq m) + -- signal matching request + putTMVar wv (bimap (toException . RemoteProcessException) id bs) + -- Receive a request and queue it + Request uq mbi bs -> do + atomically $ case mbi of + Nothing -> + -- "Global" request + writeTQueue (requestsQueue am) (uq,bs) + Just ii -> do + -- "Specific mailbox" request + q <- getMailbox am ii + writeTQueue q (uq,bs) + Cancel uq -> cancelRequest am uq replyToPendings e = atomically $ failPendingReplies p e am = pipeAsyncMngr p +cancelRequest :: AsyncManager -> CorrelationId -> IO () +cancelRequest am uq = join . atomically $ do + cs <- readTVar (cancelStatus am) + case IntMap.lookup uq cs of + Nothing -> pure (pure ()) + Just Fetched -> writeTVar (cancelStatus am) (IntMap.insert uq Canceling cs) >> pure (pure ()) + Just (Active cancel) -> writeTVar (cancelStatus am) (IntMap.delete uq cs) >> pure cancel + Just Canceling -> pure (pure ()) + +-- | Returns @True@ when the handler should write its 'Reply'. +doneHandling :: AsyncManager -> CorrelationId -> STM Bool +doneHandling am uq = do + cs <- readTVar (cancelStatus am) + case IntMap.lookup uq cs of + Nothing -> pure False + Just s -> do + writeTVar (cancelStatus am) $ IntMap.delete uq cs + case s of + Canceling -> pure False + _ -> pure True + +setActive :: AsyncManager -> CorrelationId -> IO () -> IO () +setActive am uq cancel = join . atomically $ do + cs <- readTVar (cancelStatus am) + case IntMap.lookup uq cs of + Nothing -> pure (pure ()) -- reply already received. + Just Fetched -> writeTVar (cancelStatus am) (IntMap.insert uq (Active cancel) cs) >> pure (pure ()) + Just (Active _) -> writeTVar (cancelStatus am) (IntMap.insert uq (Active cancel) cs) >> pure (pure ()) + Just Canceling -> pure cancel + +fetchedRequest :: AsyncManager -> CorrelationId -> STM () +fetchedRequest am uq = do + cs <- readTVar (cancelStatus am) + case IntMap.lookup uq cs of + Nothing -> writeTVar (cancelStatus am) (IntMap.insert uq Fetched cs) + Just Fetched -> error "impossible" + Just (Active _) -> error "impossible" + Just Canceling -> error "impossible" + + + +{- +Note [Exceptions during serialization] +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +For efficiency reasons, any `IddMsg` to be sent is forced only when already +writing onto the Pipe, so we cannot discern whether any raised exception +originated due to forcing during serialization or from interacting with the +Pipe. + +If the exceptions happen to be `SomeAsyncException` or `IOException` the write +end of the `Pipe` will be closed while rethrowing them, as those are likely to +indicate something is wrong with the `Pipe` itself. Same for +PipeWriteClosedException, used by `closePipe`. + +It is recommended to fully force messages at the source, to avoid needlessly +interrupting communication. +-} + + +-- | "Fill"s write end of the Pipe with queued messages. +-- +-- Because Pipe is written to from this thread rather than 'writePipe', calls to +-- the latter can be interrupted without compromising the protocol. +-- +-- Closes the Pipe write end when stopping, notifying any messages still queued to write. +fillPipe :: Pipe -> (forall x. IO x -> IO x) -> IO () +fillPipe p unmask = catch (loop `finally` pipeCloseWrite p) (atomically . failPendingWrites p) where + loop = do + (msg,var) <- atomically $ readTQueue (writeQueue am) + success <- handle (\ e -> atomically (putTMVar var (Left e)) >> rethrowIfPipeFatal e >> pure False) $ do + unmask $ withLock p (putAll p . execPut . put $ msg) + pure True + when success $ atomically $ putTMVar var (Right ()) + loop + + -- See Note [Exceptions during serialization] + rethrowIfPipeFatal e + | Just SomeAsyncException{} <- fromException e = throwIO e + | Just (_ :: IOException) <- fromException e = throwIO e + | Just PipeWriteClosedException <- fromException e = throwIO e + | otherwise = pure () + am = pipeAsyncMngr p + +failPendingWrites :: Pipe -> SomeException -> STM () +failPendingWrites p e = do + writeTVar (pipeWriteClosed p) True + writes <- flushTQueue (writeQueue (pipeAsyncMngr p)) + mapM_ (`writeTMVar` Left e) (map snd writes) + failPendingReplies :: Pipe -> SomeException -> STM () failPendingReplies p e = do -- reqs are only added if pipe not closed; @@ -456,7 +639,7 @@ runWriteReply (WriteReply r) = r (|#>) (IO fs) s = fs s (%>>=) :: (IO a) -> (a -> IO b) %1 -> IO b -(%>>=) = \ m f -> IO (\s0 -> +(%>>=) = \ m -> \ f -> IO (\s0 -> case m |#> s0 of -- unIO for linearity checking (# s1, x #) -> f x |#> s1) infixl 1 %>>= ===================================== libraries/ghci/GHCi/Server.hs ===================================== @@ -66,13 +66,13 @@ servWithCustom -> IO () servWithCustom verbose hook pipe restore customHandler = do ref <- newIORef False - withThreadManager $ \mngr -> + withThreadManager $ \mngr -> do loop ref mngr `finally` closePipe pipe where loop doneRef mngr = do when verbose $ trace "reading pipe..." - handleRemoteCall pipe hook $ \wr (msg :: Message a) -> + handleRemoteCall pipe hook $ \wr setCancel (msg :: Message a) -> discardCtrlC %>>= \_ -> when verbose (trace ("msg: " ++ (show msg))) %>>= \_ -> @@ -83,8 +83,10 @@ servWithCustom verbose hook pipe restore customHandler = do reply () %>>= writeReply wr RunTH ii st q ty loc -> spawnFinally mngr (runWriteReply wr) (wrapRunTH ii (runTH pipe ii st q ty loc)) + (setCancel . void . forkIO . killThread) RunModFinalizers ii st qs -> spawnFinally mngr (runWriteReply wr) (wrapRunTH ii (runModFinalizerRefs pipe ii st qs)) + (setCancel . void . forkIO . killThread) _other -> run msg >>= reply %>>= writeReply wr isDone <- readIORef doneRef ===================================== libraries/ghci/GHCi/Server/Threads.hs ===================================== @@ -41,8 +41,8 @@ withThreadManager k = do -- | Linear and resource-safe because 'ThreadManager' is guaranteed to wait or -- cancel all threads. -spawnFinally :: ThreadManager -> (Either SomeException a -> IO ()) %1 -> IO a -> IO () -spawnFinally ThreadManager{..} = unsafe_linear \fin -> \action -> void $ +spawnFinally :: ThreadManager -> (Either SomeException a -> IO ()) %1 -> IO a -> (ThreadId -> IO b) -> IO b +spawnFinally ThreadManager{..} = unsafe_linear \fin -> \action -> \k -> forkFinally ( mask $ \unmask -> do tid <- myThreadId @@ -57,6 +57,7 @@ spawnFinally ThreadManager{..} = unsafe_linear \fin -> \action -> void $ else throwIO ThreadKilled ) ( \ea -> fin ea ) + >>= k unsafe_linear :: (a -> b) %1 -> (a %1 -> b) unsafe_linear = Unsafe.unsafeCoerce (Unsafe.unsafeCoerce :: (a -> b) -> (a %1 -> b)) ===================================== libraries/template-haskell/Language/Haskell/TH/Syntax.hs ===================================== @@ -528,7 +528,9 @@ class (MonadIO m, MonadFail m) => Quasi m where -- | Report an error (True) or warning (False) -- ...but carry on; use 'fail' to stop. See 'report'. qReport :: Bool -> String -> m () - qReport b s = qRunQ $ report b s + -- 'msg' is forced to ensure exceptions don't escape, + -- see Note [Exceptions in TH] + qReport b msg = seqList msg $ qRunQ $ Internal.report b msg -- | See 'recover'. qRecover :: m a -- ^ the error handler @@ -665,13 +667,18 @@ instance Quasi Q where -- | Report an error (True) or warning (False), -- but carry on; use 'fail' to stop. report :: Bool -> String -> Q () -report = Internal.report +report = qReport {-# DEPRECATED report "Use reportError or reportWarning instead" #-} -- deprecated in 7.6 -- | Report an error to the user, but allow the current splice's computation to carry on. To abort the computation, use 'fail'. reportError :: String -> Q () -reportError = report True +reportError = qReport True -- | Report a warning to the user, and carry on. reportWarning :: String -> Q () -reportWarning = report False +reportWarning = qReport False + +-- | copied from GHC.Utils.Misc +seqList :: [a] -> b -> b +seqList [] b = b +seqList (x:xs) b = x `seq` seqList xs b ===================================== testsuite/tests/ghci/should_run/Makefile ===================================== @@ -3,10 +3,20 @@ include $(TOP)/mk/boilerplate.mk include $(TOP)/mk/test.mk # Test that threadDelay can be interrupted by ^C. -T3171: - echo "do Control.Concurrent.threadDelay 3000000; putStrLn \"threadDelay was not interrupted\"" | \ - "$(TEST_HC)" $(TEST_HC_OPTS_INTERACTIVE) & \ - sleep 2; kill -INT $$!; wait +# Wait for the statement to start before interrupting (#15383), and signal +# the external interpreter too (the `pgrep -P $!`), as ^C in a terminal would. +# +# The ext interp tests both: +# - interrupting a remote request doesn't terminate the ext interpreter (#18852) +# - a ^C gets propagated to the thread evaluating the user request on the ext +# interpreter so that it gets interrupted +T3171ext: T3171_OPTS = -fexternal-interpreter +T3171 T3171ext: + rm -f $@.started + echo 'do { writeFile "$@.started" ""; (Control.Concurrent.threadDelay 100000000 >> putStrLn "threadDelay was not interrupted.") `Control.Exception.catch` \(e :: Control.Exception.SomeException) -> System.IO.hPutStrLn System.IO.stderr ("threadDelay was interrupted: " ++ show e) >> Control.Exception.throwIO e }' | \ + "$(TEST_HC)" $(TEST_HC_OPTS_INTERACTIVE) $(T3171_OPTS) & \ + while [ ! -e $@.started ] && kill -0 $$! 2>/dev/null; do sleep 0.1; done; \ + kill -INT $$! $$(pgrep -P $$!); wait TopEnvIface: "$(TEST_HC)" $(TEST_HC_OPTS_INTERACTIVE) TopEnvIface -v1 -fno-hide-source-paths -e "a" -e ":q" -fwrite-if-simplified-core -fwrite-interface ===================================== testsuite/tests/ghci/should_run/T3171.stdout ===================================== @@ -1 +1,2 @@ +threadDelay was interrupted: user interrupt Interrupted. ===================================== testsuite/tests/ghci/should_run/T3171ext.stdout ===================================== @@ -0,0 +1,2 @@ +threadDelay was interrupted: user interrupt +Interrupted. ===================================== testsuite/tests/ghci/should_run/all.T ===================================== @@ -12,11 +12,17 @@ test('T27287', [just_ghci, combined_output], makefile_test, []) test('T3171', [when(opsys('mingw32'),skip), - fragile(15383), req_interp, combined_output], makefile_test, []) +test('T3171ext', + [when(opsys('mingw32'),skip), + req_interp, + unless(config.have_ext_interp, skip), + combined_output], + makefile_test, []) + test('ghcirun004', just_ghci, compile_and_run, ['']) test('T8377', just_ghci, compile_and_run, ['']) test('T9914', just_ghci, ghci_script, ['T9914.script']) ===================================== testsuite/tests/th/T8987b.hs ===================================== @@ -0,0 +1,8 @@ +{-# LANGUAGE TemplateHaskell #-} + +module T8987b where +import Language.Haskell.TH +import System.IO.Error +import Control.Exception + +$(reportWarning ['1', throw (userError "IOError in warning")] >> return []) ===================================== testsuite/tests/th/T8987b.stderr ===================================== @@ -0,0 +1,6 @@ +T8987b.hs:1:1: error: [GHC-87897] + Exception when trying to run compile-time code: + user error (IOError in warning) + Code: (reportWarning ['1', throw (userError "IOError in warning")] + >> return []) + ===================================== testsuite/tests/th/all.T ===================================== @@ -306,6 +306,7 @@ test('T8884', normal, compile, ['-v0']) test('T8954', normal, compile, ['-v0']) test('T8932', normal, compile_fail, ['-v0']) test('T8987', normal, compile_fail, ['-v0']) +test('T8987b', normal, compile_fail, ['-v0']) test('T7241', normal, compile_fail, ['-v0']) test('T9262', normal, compile, ['-v0']) test('T9199', normal, compile, ['-v0']) View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/5be020df64ac8df323a83a0043fc425... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/5be020df64ac8df323a83a0043fc425... You're receiving this email because of your account on gitlab.haskell.org. Manage all notifications: https://gitlab.haskell.org/-/profile/notifications | Help: https://gitlab.haskell.org/help