Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC Commits: 1210c80c by Rodrigo Mesquita at 2026-10-15T12:02:13+01:00 Asynchronous RPC with external interpreter There's a single pipe to connect the host compiler to the external interpreter. On this pipe, we have historically done synchronous request/reply communication to invoke methods on the external interpreter. However, with the advent of multi-threaded debugging, it has become necessary to do asynchronous calls on the external interpreter, where we may block on the host waiting for a reply, but the external interpreter should handle the request asynchronously and reply when ready. For instance, we want to start the evaluation of the debuggee but not block the interpreter waiting for it to finish or hit a breakpoint. Or e.g. we want to resume two threads of the debuggee simulatenously instead of blocking waiting for the resume result of each one sequentially. And it turns out, we were already doing some ad-hoc asynchronous communication to evaluate Template Haskell on the external interpreter. It used to go as an ad-hoc "send request without expecting reply", followed by handling on the host a set of 'THMessage's sent by the interpreter, followed by "expect a response to arrive now". It worked because there was still a fixed order of messages, but it fits an async model in a non-ad-hoc way. Asynchronous communication that unblocks the external interpreter main thread will also make it possible to resolve issues such as #18852 (exceptions leaving the external interpreter in an unrecoverrable state) The new design -------------- The new invariant: every single request should receive a matching reply. The pipe is bidirectional (the ext. interpreter can receive requests, but it can also send requests for TH evaluation; the host can send requests, but it can also receive TH requests from the interpreter). When a request is sent in whichever direction, it always contains a correlation-id header. The Sender places an empty MVar on a pending-reply map keyed by id and waits on it for a response. See 'remoteCall' and 'remoteTHCall'. The pipe is associated to a thread which drains the read end. Because it's bidirectional, two things may be read off the pipe: - A Reply, with a correlation-id header, which we match against the pending requests map to unblock the corresponding pending-reply MVar - A Request, which is queued to the pending-request queue. To handle a message, we take a Request from the pending-request queue and run a handler that produces a Reply. Currently, the handler is run synchronously in the thread, but it is trivial to make the handler run in a forked thread for certain ("interruptible") messages in an upcoming patch (when it is necessary for the multi-threaded debugger). See 'handleRemoteCall' and 'handleRemoteTHCall'. The old interp lock now just protect the pipe writing (reads don't lock bc they are done exclusively by drainPipe), it doesn't need nor should lock the full remote procedure call. - - - - - 01640eec by Rodrigo Mesquita at 2026-10-15T12:02:13+01:00 interp: Add separate mailboxes for TH async RPC When the host sends a `RunTH` command to the interpreter, it expects a few "reverse requests" from the interpreter (e.g. `NewName`) which must be run in the same context that initiated the `RunTH`. When the host is evaluating many splices at once, it will send many `RunTH` commands to the interpreter and start listening for reverse requests in the meantime. However, a received request previously had no way to be associated to the scope of a particular `RunTH` invocation, so the reverse commands could get replied to from incorrect contexts, or e.g. a `RunTHDone` from the interp. could terminate early the wrong `RunTH` computation. In this commit we add independent mailboxes to the interpreter asynchronous RPC infrastructure, allowing a RunTH to announce a particular `InboxId` that all reverse requests for that RunTH should use. On the host, we create that mailbox id and have `handleRemoteTHCall` listen on that specific mailbox rather than for requests without a specific mailbox address. - - - - - ea5bd12e by Rodrigo Mesquita at 2026-10-15T12:02:14+01:00 refactor: Move Async RPC abstraction into GHCi.Message.Async Carves out the Async RPC over the Pipe abstraction to a new module, making the GHCi.Message more about the specific interpreter messages rather than about the IPC protocol. - - - - - 48377e28 by Rodrigo Mesquita at 2026-10-15T12:02:14+01:00 interp: If we have pending reads in the queue we can deliver them We may still be able to reply even if the read end was closed. As long as we have pending messages we should try to deliver them. - - - - - cd423df8 by Andrea Vezzosi at 2026-10-15T12:03:25+01:00 interp: close Handles when done with Pipe Ensures we close the pipe handles (if created using 'mkPipeFromHandles') on `closePipe`, and do `closePipe` on `Shutdown`. The thread draining the read end always `'finally'` closes the read end. And `closePipe` kills the draining thread and closes the write end. When `remoteAsyncReq`, we throw PipeReadClosedException if the pipe read is closed, since we wouldn't be able to receive the reply. It's fine for the drain thread not to close the write end, even if it dies without `closePipe` (which does close the write end), because: - requests pending a reply will get PipeClosedException (we'll not be able to read any further "real" reply) - new calls to `remoteAsyncReq` will throw PipeReadClosedException - pending pipe writes, in response to an already read request, might still succeed And the clean up code should call close pipe to ensure the write end is closed as well. So, if the drain thread dies naturally and the read end is closed, may still be attempted to be used, but it will throw accordingly. The user likely wants to catch this to close the write end as well. - - - - - 12 changed files: - compiler/GHC/Runtime/Interpreter.hs - compiler/GHC/Runtime/Interpreter/JS.hs - compiler/GHC/Runtime/Interpreter/Process.hs - compiler/GHC/Runtime/Interpreter/Types.hs - compiler/GHC/Runtime/Interpreter/Wasm.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/TH.hs - libraries/ghci/ghci.cabal.in - testsuite/tests/ghci/custom-external-interpreter-commands/Main.hs Changes: ===================================== compiler/GHC/Runtime/Interpreter.hs ===================================== @@ -626,11 +626,9 @@ spawnIServ conf = do [] (iservConfOpts conf) interpPipe <- mkPipeFromHandles rh wh - lock <- newMVar () let process = InterpProcess { interpHandle = ph , interpPipe - , interpLock = lock } pending_frees <- newMVar [] @@ -656,6 +654,8 @@ stopInterp interp = case interpInstance interp of if isJust ex then pure () else sendMessage i Shutdown + -- We asked the server to shutdown, so we will also shutdown any communication. + closePipe (interpPipe (instProcess i)) pure InterpPending -- ----------------------------------------------------------------------------- ===================================== compiler/GHC/Runtime/Interpreter/JS.hs ===================================== @@ -127,11 +127,9 @@ startTHRunnerProcess interp_js settings = do std_in <- readIORef interp_in interpPipe <- mkPipeFromHandles rh wh - lock <- newMVar () let proc = InterpProcess { interpHandle = hdl , interpPipe - , interpLock = lock } pure (std_in, proc) @@ -382,7 +380,7 @@ jsLinkPlan logger tmpfs tmp_dir ar_cache link_cfg cfg inst link_plan = do -- Haskell server. server_started <- jsServerStarted <$> readMVar (instJSState (instExtra inst)) if server_started - then sendMessageNoResponse inst $ LoadObj all_js + then sendMessage inst $ LoadObj all_js else jsLoadFile inst all_js ---------------------------------------------------------------- ===================================== compiler/GHC/Runtime/Interpreter/Process.hs ===================================== @@ -2,16 +2,8 @@ module GHC.Runtime.Interpreter.Process ( -- * Message API Message(..) - , DelayedResponse (..) - -- * Top-level message API (these acquire/release a lock) , sendMessage - , sendMessageNoResponse - , sendMessageDelayedResponse - , receiveDelayedResponse - -- * Nested message API (these require the interpreter to already be locked) - , sendAnyValue - , receiveAnyValue - , receiveTHMessage + , sendTHComputation ) where @@ -27,97 +19,63 @@ import GHC.Utils.Exception as Ex import Data.Binary import System.Exit import System.Process -import Control.Concurrent.MVar (MVar, withMVar, takeMVar, putMVar, isEmptyMVar) - -data DelayedResponse a = DelayedResponse +import Control.Concurrent +import Data.IORef +import Control.Concurrent.STM -- ----------------------------------------------------------------------------- -- Top-level Message API --- | Send a message to the interpreter process that doesn't expect a response --- (locks the interpreter while sending) -sendMessageNoResponse :: ExtInterpInstance d -> Message () -> IO () -sendMessageNoResponse i m = - withLock i $ writeInterpProcess (instProcess i) (putMessage m) - --- | Send a message to the interpreter that expects a response --- (locks the interpreter while until the response is received) +-- | Send a message to the interpreter and expect a response, blocking until it arrives. +-- +-- This issues an asynchronous RPC on the external interpreter which does not +-- block the external interpreter thread. We block here on the host because we +-- wait for the asynchronous response to arrive. sendMessage :: Binary a => ExtInterpInstance d -> Message a -> IO a -sendMessage i m = withLock i $ callInterpProcess (instProcess i) m - --- | Send a message to the interpreter process whose response is expected later +sendMessage i msg = + remoteCall (interpPipe proc) msg + `catchException` \(e :: SomeException) -> handleInterpProcessFailure proc e + where + proc = instProcess i + +-- | Send a message with a TH computation to the ext. interpreter (like +-- 'RunTH', or 'RunModFinalizers') and handle 'THMessage' requests from the +-- ext. interpreter to us in the meantime. -- --- This is useful to avoid forgetting to receive the value and to ensure that --- the type of the response isn't lost. Use receiveDelayedResponse to read it. --- (locks the interpreter until the response is received using --- `receiveDelayedResponse`) -sendMessageDelayedResponse :: ExtInterpInstance d -> Message a -> IO (DelayedResponse a) -sendMessageDelayedResponse i m = do - lock i - writeInterpProcess (instProcess i) (putMessage m) - pure DelayedResponse - --- | Expect a delayed result to be received now -receiveDelayedResponse :: Binary a => ExtInterpInstance d -> DelayedResponse a -> IO a -receiveDelayedResponse i DelayedResponse = do - ensureLocked i - r <- readInterpProcess (instProcess i) get - unlock i - pure r - --- ----------------------------------------------------------------------------- --- Nested Message API - --- | Send any value (requires locked interpreter) -sendAnyValue :: Binary a => ExtInterpInstance d -> a -> IO () -sendAnyValue i m = ensureLocked i >> writeInterpProcess (instProcess i) (put m) - --- | Expect a value to be received (requires locked interpreter) -receiveAnyValue :: ExtInterpInstance d -> Get a -> IO a -receiveAnyValue i get = ensureLocked i >> readInterpProcess (instProcess i) get - --- | Wait for a Template Haskell message (requires locked interpreter) -receiveTHMessage :: ExtInterpInstance d -> IO THMsg -receiveTHMessage i = ensureLocked i >> receiveAnyValue i getTHMessage - --- ----------------------------------------------------------------------------- - -getLock :: ExtInterpInstance d -> MVar () -getLock = interpLock . instProcess - -withLock :: ExtInterpInstance d -> IO a -> IO a -withLock i f = withMVar (getLock i) (const f) - -lock :: ExtInterpInstance d -> IO () -lock i = takeMVar (getLock i) - -unlock :: ExtInterpInstance d -> IO () -unlock i = putMVar (getLock i) () - -ensureLocked :: ExtInterpInstance d -> IO () -ensureLocked i = - isEmptyMVar (getLock i) >>= \case - False -> panic "ensureLocked: external interpreter not locked" - _ -> pure () - - --- | Send a 'Message' and receive the response from the interpreter process -callInterpProcess :: Binary a => InterpProcess -> Message a -> IO a -callInterpProcess i msg = - remoteCall (interpPipe i) msg - `catchException` \(e :: SomeException) -> handleInterpProcessFailure i e - --- | Read a value from the interpreter process -readInterpProcess :: InterpProcess -> Get a -> IO a -readInterpProcess i get = - readPipe (interpPipe i) get - `catchException` \(e :: SomeException) -> handleInterpProcessFailure i e - --- | Send a value to the interpreter process -writeInterpProcess :: InterpProcess -> Put -> IO () -writeInterpProcess i put = - writePipe (interpPipe i) put - `catchException` \(e :: SomeException) -> handleInterpProcessFailure i e +-- This function blocks until the TH computation returns a value, but spawns in +-- the background a thread to handle the 'THMessage's until the TH computation +-- is fully done. +sendTHComputation + :: Binary a + => ExtInterpInstance d + -> (InboxId -> Message a) + -> (forall b. Binary b => THMessage b -> IO (Bool, b)) + -- ^ Handle one message. Return True to end the loop. + -- This handler must never uninterrutibly mask an expression which may get + -- stuck. If it does, then 'sendTHComputation' will also get uninterruptibly stuck. + -> IO a +sendTHComputation i msg reqHandler = mask $ \restore -> do + -- inline "withAsync/wait" + var <- newEmptyTMVarIO + mbox <- newMailbox (pipeAsyncMngr (interpPipe proc)) + ht <- forkFinally (restore (handlerLoop mbox)) (atomically . putTMVar var) + restore (sendMessage i (msg mbox) <* atomically (readTMVar var)) + `onException` uninterruptibleMask_ (throwTo ht CancelTHHandler) + where + proc = instProcess i + handleOne mbox doneRef = do + handleRemoteTHCall (interpPipe proc) mbox $ \req -> do + (stop, res) <- reqHandler req + writeIORef doneRef stop + return res + handlerLoop mbox = do + doneRef <- newIORef False + let loop = do + handleOne mbox doneRef + isDone <- readIORef doneRef + if isDone then return () + else loop + in loop handleInterpProcessFailure :: InterpProcess -> SomeException -> IO a handleInterpProcessFailure i e = do @@ -130,3 +88,8 @@ handleInterpProcessFailure i e = do terminateProcess hdl _ <- waitForProcess hdl throw e + +data CancelTHHandler = CancelTHHandler deriving Show +instance Exception CancelTHHandler where + fromException = asyncExceptionFromException + toException = asyncExceptionToException ===================================== compiler/GHC/Runtime/Interpreter/Types.hs ===================================== @@ -111,7 +111,6 @@ type WasmInterp = ExtInterpState WasmInterpConfig () data InterpProcess = InterpProcess { interpPipe :: !Pipe -- ^ Pipe to communicate with the server , interpHandle :: !ProcessHandle -- ^ Process handle of the server - , interpLock :: !(MVar ()) -- ^ Lock to prevent concurrent access to the stream } -- | Status of an external interpreter ===================================== compiler/GHC/Runtime/Interpreter/Wasm.hs ===================================== @@ -73,14 +73,12 @@ spawnWasmInterp WasmInterpConfig {..} = do hSetBuffering rh NoBuffering interpPipe <- mkPipeFromHandles rh wh pending_frees <- newMVar [] - lock <- newMVar () pure $ ExtInterpInstance { instProcess = InterpProcess { interpHandle = ph, - interpPipe, - interpLock = lock + interpPipe }, instPendingFrees = pending_frees, instExtra = () ===================================== compiler/GHC/Tc/Gen/Splice.hs ===================================== @@ -170,6 +170,7 @@ import Data.IORef import GHC.Parser.HaddockLex (lexHsDoc) import GHC.Parser (parseIdentifier) import GHC.Rename.Doc (rnHsDoc) +import Data.Function {- Note [Template Haskell state diagram] @@ -1165,11 +1166,12 @@ runRemoteModFinalizers (ThModFinalizers finRefs) = do case th_state of Nothing -> return () -- TH was not started, nothing to do Just fhv -> do - r <- liftIO $ withForeignRef fhv $ \st -> - withForeignRefs finRefs $ \qrefs -> - sendMessageDelayedResponse inst (RunModFinalizers st qrefs) - () <- runRemoteTH inst [] - qr <- liftIO $ receiveDelayedResponse inst r + qr <- withRunInIO $ \runInIO -> do + recovers <- newIORef [] + withForeignRef fhv $ \st -> + withForeignRefs finRefs $ \qrefs -> + sendTHComputation inst (\ii -> RunModFinalizers ii st qrefs) $ \msg -> + runInIO (runRemoteTH recovers msg) checkQResult qr runQResult @@ -1824,48 +1826,47 @@ runTH ty fhv = do rstate <- getTHState inst loc <- location -- run a remote TH request - r <- liftIO $ + qr <- withRunInIO $ \runInIO -> do + recovers <- newIORef [] withForeignRef rstate $ \state_hv -> - withForeignRef fhv $ \q_hv -> - sendMessageDelayedResponse inst (RunTH state_hv q_hv ty (Just loc)) - -- respond to requests from the interpreter - runRemoteTH inst [] - -- get the final result - qr <- liftIO $ receiveDelayedResponse inst r + withForeignRef fhv $ \q_hv -> + sendTHComputation inst (\ii -> RunTH ii state_hv q_hv ty (Just loc)) $ \msg -> + runInIO (runRemoteTH recovers msg) bs <- checkQResult qr return $! runGet get (LB.fromStrict bs) --- | communicate with a remotely-running TH computation until it finishes. +-- | Communicate with a remotely-running TH computation until it finishes. -- See Note [Remote Template Haskell] in libraries/ghci/GHCi/TH.hs. runRemoteTH - :: ExtInterpInstance d - -> [Messages TcRnMessage] -- saved from nested calls to qRecover - -> TcM () -runRemoteTH inst recovers = do - THMsg msg <- liftIO $ receiveTHMessage inst + :: Binary a + => IORef [Messages TcRnMessage] -- ^ saved from nested calls to qRecover + -> THMessage a + -> TcM (Bool, a) +runRemoteTH recovers msg = do case msg of - RunTHDone -> return () + RunTHDone -> return (True, ()) StartRecover -> do -- Note [TH recover with -fexternal-interpreter] v <- getErrsVar msgs <- readTcRef v writeTcRef v emptyMessages - runRemoteTH inst (msgs : recovers) + modifyIORef recovers (msgs:) & liftIO + return (False, ()) EndRecover caught_error -> do - let (prev_msgs, rest) = case recovers of - [] -> panic "EndRecover" - a : b -> (a,b) + (prev_msgs, rest) <- liftIO $ + readIORef recovers >>= \case + [] -> panic "EndRecover" + a : b -> pure (a,b) v <- getErrsVar warn_msgs <- getWarningMessages <$> readTcRef v -- keep the warnings only if there were no errors writeTcRef v $ if caught_error then prev_msgs else mkMessages warn_msgs `unionMessages` prev_msgs - runRemoteTH inst rest + writeIORef recovers rest & liftIO + return (False, ()) _other -> do - r <- handleTHMessage msg - liftIO $ sendAnyValue inst r - runRemoteTH inst recovers + (False,) <$> handleTHMessage msg -- | Check a QResult checkQResult :: QResult a -> TcM a ===================================== libraries/ghci/GHCi/Message.hs ===================================== @@ -1,6 +1,7 @@ -{-# LANGUAGE GADTs, DeriveGeneric, StandaloneDeriving, ScopedTypeVariables, - GeneralizedNewtypeDeriving, ExistentialQuantification, RecordWildCards, - CPP, NamedFieldPuns #-} +{-# LANGUAGE RecursiveDo, GADTs, DeriveGeneric, StandaloneDeriving, ScopedTypeVariables, + GeneralizedNewtypeDeriving, ExistentialQuantification, RecordWildCards, CPP, + NamedFieldPuns, LambdaCase, TypeAbstractions, TypeApplications, + LinearTypes, RankNTypes, UnboxedTuples, BlockArguments, TupleSections, MagicHash #-} {-# OPTIONS_GHC -fno-warn-name-shadowing -fno-warn-orphans #-} -- | @@ -22,7 +23,16 @@ module GHCi.Message , ResumeContext(..) , QState(..) , getMessage, putMessage, getTHMessage, putTHMessage - , Pipe, mkPipeFromHandles, mkPipeFromContinuations, remoteCall, remoteTHCall, readPipe, writePipe + + , Pipe(..), mkPipeFromHandles, mkPipeFromContinuations, closePipe + , remoteCall, remoteTHCall + , handleRemoteCall, handleRemoteTHCall + , RemoteProcessException(..), PipeClosedException(..) + + , InboxId(..) + , newMailbox + , closeMailbox + , BreakModule , BreakUnitId , LoadedDLL @@ -45,18 +55,14 @@ import qualified GHC.Exts.Heap as Heap import GHC.ForeignSrcLang import GHC.Fingerprint import GHC.Conc (pseq, par) +import GHC.IO (IO(..)) import Control.Concurrent import Control.DeepSeq import Control.Exception -#if MIN_VERSION_base(4,20,0) -import Control.Exception.Context -#endif import Data.Binary import Data.Binary.Get import Data.Binary.Put import Data.ByteString (ByteString) -import qualified Data.ByteString as B -import qualified Data.ByteString.Builder as B import qualified Data.ByteString.Lazy as LB import qualified Data.ByteString.Short as BS import Data.Dynamic @@ -68,9 +74,7 @@ import GHC.Generics import GHC.Stack.CCS import qualified GHC.Boot.TH.Syntax as TH import qualified GHC.Boot.TH.Monad as TH -import System.Exit -import System.IO -import System.IO.Error +import GHCi.Message.Async -- ----------------------------------------------------------------------------- -- The RPC protocol between GHC and the interactive server @@ -214,14 +218,16 @@ data Message a where -- they did, we have to serialize the value anyway, so we might -- as well serialize it to force it. RunTH - :: RemoteRef (IORef QState) + :: InboxId -- ^ announce mailbox for related 'remoteTHCall's + -> RemoteRef (IORef QState) -> HValueRef {- e.g. TH.Q TH.Exp -} -> THResultType -> Maybe TH.Loc -> Message (QResult ByteString) -- | Run the given mod finalizers. - RunModFinalizers :: RemoteRef (IORef QState) + RunModFinalizers :: InboxId -- ^ announce mailbox for related 'remoteTHCall's + -> RemoteRef (IORef QState) -> [RemoteRef (TH.Q ())] -> Message (QResult ()) @@ -458,45 +464,6 @@ data BreakUnitId -- | A dummy type that tags pointers returned by 'LoadDLLs'. data LoadedDLL --- SomeException can't be serialized because it contains dynamic --- types. However, we do very limited things with the exceptions that --- are thrown by interpreted computations: --- --- * We print them, e.g. "*** Exception: <something>" --- * UserInterrupt has a special meaning --- * In ghc -e, exitWith should exit with the appropriate exit code --- --- So all we need to do is distinguish UserInterrupt and ExitCode, and --- all other exceptions can be represented by their 'show' string. --- -data SerializableException - = EUserInterrupt - | EExitCode ExitCode - | EOtherException String - deriving (Generic, Show) - -toSerializableException :: SomeException -> SerializableException -toSerializableException ex - | Just UserInterrupt <- fromException ex = EUserInterrupt - | Just (ec::ExitCode) <- fromException ex = (EExitCode ec) - | otherwise = EOtherException $ -#if MIN_VERSION_base(4,20,0) - -- Exception plus backtrace as seen in `displayExceptionWithInfo` - case displayExceptionContext (someExceptionContext ex) of - "" -> displayException (ex :: SomeException) - cx -> displayException (ex :: SomeException) ++ "\n\n" ++ cx -#else - show (ex :: SomeException) -#endif - -fromSerializableException :: SerializableException -> SomeException -fromSerializableException EUserInterrupt = toException UserInterrupt -fromSerializableException (EExitCode c) = toException c -fromSerializableException (EOtherException str) = toException (ErrorCall str) - -instance Binary ExitCode -instance Binary SerializableException - data THResult a = THException String | THComplete a @@ -513,11 +480,13 @@ instance Binary THResultType -- StartTH message. A new one is created per module that GHC -- typechecks. data QState = QState - { qsMap :: Map TypeRep Dynamic + { qsMap :: Map TypeRep Dynamic -- ^ persistent data between splices in a module - , qsLocation :: Maybe TH.Loc + , qsLocation :: Maybe TH.Loc -- ^ location for current splice, if any - , qsPipe :: Pipe + , qsInboxId :: InboxId + -- ^ the mailbox expecting requests for each RunTH-like run + , qsPipe :: Pipe -- ^ pipe to communicate with GHC } instance Show QState where show _ = "<QState>" @@ -600,9 +569,9 @@ getMessage = do 29 -> Msg <$> (BreakpointStatus <$> get <*> get) 30 -> Msg <$> (GetBreakpointVar <$> get <*> get) 31 -> Msg <$> return StartTH - 32 -> Msg <$> (RunModFinalizers <$> get <*> get) + 32 -> Msg <$> (RunModFinalizers <$> get <*> get <*> get) 33 -> Msg <$> (AddSptEntry <$> get <*> get) - 34 -> Msg <$> (RunTH <$> get <*> get <*> get <*> get) + 34 -> Msg <$> (RunTH <$> get <*> get <*> get <*> get <*> get) 35 -> Msg <$> (GetClosure <$> get) 36 -> Msg <$> (Seq <$> get) 37 -> Msg <$> return RtsRevertCAFs @@ -648,9 +617,9 @@ putMessage m = case m of BreakpointStatus arr ix -> putWord8 29 >> put arr >> put ix GetBreakpointVar a b -> putWord8 30 >> put a >> put b StartTH -> putWord8 31 - RunModFinalizers a b -> putWord8 32 >> put a >> put b + RunModFinalizers a b c -> putWord8 32 >> put a >> put b >> put c AddSptEntry a b -> putWord8 33 >> put a >> put b - RunTH st q loc ty -> putWord8 34 >> put st >> put q >> put loc >> put ty + RunTH ii st q loc ty -> putWord8 34 >> put ii >> put st >> put q >> put loc >> put ty GetClosure a -> putWord8 35 >> put a Seq a -> putWord8 36 >> put a RtsRevertCAFs -> putWord8 37 @@ -688,73 +657,40 @@ serializeBCOs rbcos = parMap doChunk (chunkList 100 rbcos) chunkList _ [] = [] chunkList n xs = as : chunkList n bs where (as,bs) = splitAt n xs --- ----------------------------------------------------------------------------- --- Reading/writing messages - --- | An opaque pipe for bidirectional binary data transmission. -data Pipe = Pipe - { getSome :: !(IO ByteString) - , putAll :: !(B.Builder -> IO ()) - , pipeLeftovers :: !(IORef (Maybe ByteString)) - } - --- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write. -mkPipeFromHandles :: Handle -> Handle -> IO Pipe -mkPipeFromHandles pipeRead pipeWrite = do - let getSome = B.hGetSome pipeRead (32*1024) - putAll b = do - B.hPutBuilder pipeWrite b - hFlush pipeWrite - pipeLeftovers <- newIORef Nothing - pure $ Pipe { getSome, putAll, pipeLeftovers } - --- | Make a 'Pipe' from a reader function and a writer function. -mkPipeFromContinuations :: IO ByteString -> (B.Builder -> IO ()) -> IO Pipe -mkPipeFromContinuations getSome putAll = do - pipeLeftovers <- newIORef Nothing - pure $ Pipe { getSome, putAll, pipeLeftovers } - +-- | Send a 'Message' request to and wait for a reply from the interpreter +-- (blocks on the calling side, doesn't necessarily block the interpreter) remoteCall :: Binary a => Pipe -> Message a -> IO a -remoteCall pipe msg = do - writePipe pipe (putMessage msg) - readPipe pipe get - -writePipe :: Pipe -> Put -> IO () -writePipe Pipe{..} put = putAll $ execPut put - -remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a -remoteTHCall pipe msg = do - writePipe pipe (putTHMessage msg) - readPipe pipe get - -readPipe :: Pipe -> Get a -> IO a -readPipe Pipe{..} get = do - leftovers <- readIORef pipeLeftovers - m <- getBin getSome get leftovers - case m of - Nothing -> throw $ - mkIOError eofErrorType "GHCi.Message.readPipe" Nothing Nothing - Just (result, new_leftovers) -> do - writeIORef pipeLeftovers new_leftovers - return result - -getBin - :: (IO ByteString) -> Get a -> Maybe ByteString - -> IO (Maybe (a, Maybe ByteString)) - -getBin getsome get leftover = go leftover (runGetIncremental get) - where - go Nothing (Done leftover _ msg) = - return (Just (msg, if B.null leftover then Nothing else Just leftover)) - go _ Done{} = throwIO (ErrorCall "getBin: Done with leftovers") - go (Just leftover) (Partial fun) = do - go Nothing (fun (Just leftover)) - go Nothing (Partial fun) = do - -- putStrLn "before hGetSome" - b <- getsome - -- putStrLn $ "hGetSome: " ++ show (B.length b) - if B.null b - then return Nothing - else go Nothing (fun (Just b)) - go _lft (Fail _rest _off str) = - throwIO (ErrorCall ("getBin: " ++ str)) +remoteCall pipe msg = either throwIO (pure . runGet get) + =<< remoteAsyncReq pipe (putMessage msg) Nothing + +-- | 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) => Message a -> IO a) + -> IO () +handleRemoteCall pipe hook k = + handleAsyncReq pipe getMessage Nothing \writer msg0 -> IO \s0 -> + case hook msg0 |#> s0 of -- unIO for linearity checking + (# s1, Msg msg #) -> case k msg |#> s1 of + (# s2, a #) -> writeReply writer (put a) |#> s2 + +-- | Send a 'THMessage' request to the host and wait for a reply from the host +-- to the interpreter. +-- +-- A 'THMessage' request from the interpreter is always associated with a +-- 'RunTH'-like run. Since the host may have many in-flight TH runs, we always +-- want a specific host 'InboxId' (which is announced in the 'RunTH' request) +remoteTHCall :: Binary a => Pipe -> InboxId -> THMessage a -> IO a +remoteTHCall pipe ii msg = either throwIO (pure . runGet get) + =<< remoteAsyncReq pipe (putTHMessage msg) (Just ii) + +-- | Handle a 'THMessage' request on the host and reply to the interpreter. +-- 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 () +handleRemoteTHCall pipe ii k = + handleAsyncReq pipe getTHMessage (Just ii) \writer (THMsg msg) -> IO \s0 -> + case k msg |#> s0 of -- unIO for linearity checking + (# s1, a #) -> writeReply writer (put a) |#> s1 + ===================================== libraries/ghci/GHCi/Message/Async.hs ===================================== @@ -0,0 +1,498 @@ +{-# LANGUAGE RecursiveDo, GADTs, DeriveGeneric, StandaloneDeriving, ScopedTypeVariables, + GeneralizedNewtypeDeriving, ExistentialQuantification, RecordWildCards, CPP, + NamedFieldPuns, LambdaCase, TypeAbstractions, TypeApplications, + LinearTypes, RankNTypes, UnboxedTuples, BlockArguments, TupleSections, MagicHash #-} +{-# OPTIONS_GHC -fno-warn-name-shadowing -fno-warn-orphans #-} +module GHCi.Message.Async + ( + -- * Pipes + Pipe(..), mkPipeFromHandles, mkPipeFromContinuations + , closePipe, PipeReadClosedException(..) + + -- * Interface for async bidirectional RPC + , remoteAsyncReq, handleAsyncReq + , RemoteProcessException(..), DecodeAsyncReqException(..) + + -- ** Mailboxes + , InboxId(..) + , newMailbox + , closeMailbox + + -- ** Async matching internals + , AsyncManager(..), CorrelationId + , IddMsg(..), newAsyncManager + + -- ** Consume `reply` exactly once + , WriteReply(..) + , writeReply, (|#>) + + -- * Serializable exceptions + , SerializableException(..) + , toSerializableException, fromSerializableException + ) where + +import Prelude -- See note [Why do we import Prelude here?] +import GHCi.TH.Binary () -- For Binary instances + +import GHC.IO (IO(..)) +import Control.Concurrent +import Control.Exception +#if MIN_VERSION_base(4,20,0) +import Control.Exception.Context +#endif +import Data.Binary +import Data.Binary.Get +import Data.Binary.Put +import Data.ByteString (ByteString) +import qualified Data.ByteString as B +import qualified Data.ByteString.Builder as B +import qualified Data.ByteString.Lazy as LB +import Data.IORef +import Data.IntMap (IntMap) +import qualified Data.IntMap as IntMap +import GHC.Generics +import System.Exit +import System.IO +import Control.Monad +import Control.Concurrent.STM +import Data.Bifunctor +import GHC.Base (State#) +import GHC.Exts (RealWorld) + +-------------------------------------------------------------------------------- +-- * Reading/writing and Pipes +-------------------------------------------------------------------------------- + +-- | An opaque pipe for bidirectional binary data transmission. +data Pipe = Pipe + { getSome :: !(IO ByteString) + , putAll :: !(B.Builder -> IO ()) + , pipeLeftovers :: !(IORef (Maybe ByteString)) + , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent writes to the stream + , pipeAsyncMngr :: !AsyncManager + , pipeCloseRead :: !(IO ()) + -- ^ closes read end of the pipe + , pipeCloseWrite :: !(IO ()) + -- ^ closes write end of the 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. + } + +-- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write. +-- Must call 'closePipe' at the end to clean up. +mkPipeFromHandles :: Handle -> Handle -> IO Pipe +mkPipeFromHandles pipeRead pipeWrite = do + let getSome = B.hGetSome pipeRead (32*1024) + putAll b = do + B.hPutBuilder pipeWrite b + hFlush pipeWrite + mkPipeFromContinuations getSome putAll (hClose pipeRead) (hClose pipeWrite) + +-- | 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 ()) + -> IO () + -- ^ pipeCloseRead + -> IO () + -- ^ pipeCloseWrite + -> IO Pipe +mkPipeFromContinuations getSome putAll pipeCloseRead pipeCloseWrite = mdo + pipeLeftovers <- newIORef Nothing + pipeLock <- newMVar () + pipeReadClosed <- newTVarIO False + mdo + pipeAsyncMngr <- newAsyncManager p + let p = Pipe { .. } + pure p + +-- | Cleanly close our ends of the Pipe, terminating the async manager's +-- drain-read-end thread. Also calls 'pipeCloseRead' and 'pipeCloseWrite', in that order. +-- +-- If the pipe was created using 'mkPipeFromHandles', the handles are also closed. +-- +-- The requests we've sent that are pending are all canceled with +-- 'PipeReadClosedException' by ourselves. +-- +-- No special notification is sent to the remote process, a message to that +-- effect should be sent ahead of time if one is needed (see 'Shutdown'). +closePipe :: Pipe -> IO () +closePipe Pipe{..} = do + + readAlreadyClosed <- atomically $ swapTVar pipeReadClosed True + + -- this triggers pipeCloseRead + unless readAlreadyClosed $ do + + -- releases handle lock taken by reading, and cleans up by replying to every + -- pending request with 'PipeReadClosedException'. + throwTo (asyncDrainTid pipeAsyncMngr) PipeReadClosedException + + pipeCloseWrite + +-- | We've closed the pipe read end on this process +data PipeReadClosedException = PipeReadClosedException deriving Show +instance Exception PipeReadClosedException + +-------------------------------------------------------------------------------- +-- ** Pipe thread-safe read/write +-------------------------------------------------------------------------------- +-- All messages on the wire are 'IddMsg', which contain a correlation header +-- and identify whether they are requests or replies. + +writePipe :: Pipe -> IddMsg -> IO () +writePipe p@Pipe{..} = withLock p . putAll . execPut . put + +-- | Only the 'drainPipe' thread should read from the pipe directly. Other +-- functions should wait for the matching reply to be delivered asynchronously. +-- (see 'withAsyncReq'). It doesn't lock to read bc it assumes single reader is 'drainPipe'. +-- +-- Returns @Nothing@ on EOF. +readPipe :: Pipe -> IO (Maybe IddMsg) +readPipe Pipe{..} = do + leftovers <- readIORef pipeLeftovers + m <- getBin getSome get leftovers + case m of + Nothing -> return Nothing + Just (result, new_leftovers) -> do + writeIORef pipeLeftovers new_leftovers + return (Just result) + +getBin + :: (IO ByteString) -> Get a -> Maybe ByteString + -> IO (Maybe (a, Maybe ByteString)) + +getBin getsome get leftover = go leftover (runGetIncremental get) + where + go Nothing (Done leftover _ msg) = + return (Just (msg, if B.null leftover then Nothing else Just leftover)) + go _ Done{} = throwIO (ErrorCall "getBin: Done with leftovers") + go (Just leftover) (Partial fun) = do + go Nothing (fun (Just leftover)) + go Nothing (Partial fun) = do + -- putStrLn "before hGetSome" + b <- getsome + -- putStrLn $ "hGetSome: " ++ show (B.length b) + if B.null b + then return Nothing + else go Nothing (fun (Just b)) + go _lft (Fail _rest _off str) = + throwIO (ErrorCall ("getBin: " ++ str)) + +withLock :: Pipe -> IO c -> IO c +withLock Pipe{..} = bracket (takeMVar pipeLock) (putMVar pipeLock) . const + +-------------------------------------------------------------------------------- +-- * Matching requests with async replies +-------------------------------------------------------------------------------- + +-- | Send an asynchronous request and block waiting for a matching reply +remoteAsyncReq :: Pipe -> Put -> (Maybe InboxId) -> IO (Either SomeException LB.ByteString) +remoteAsyncReq pipe msg ii = do + + let pendingReplies = pendingReply (pipeAsyncMngr pipe) + + -- Mark pending request + uq <- freshId (pipeAsyncMngr pipe) + wv <- newEmptyTMVarIO + + isClosed <- atomically $ do + isClosed <- readTVar (pipeReadClosed pipe) + -- don't send the req if we won't get a reply. + if isClosed + then pure True + else do + modifyTVar pendingReplies (IntMap.insert uq wv) + pure False + + when isClosed $ do + throwIO PipeReadClosedException + + -- 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)) + + -- Block waiting for reply + atomically (takeTMVar wv) + `onException` + atomically (modifyTVar pendingReplies (IntMap.delete uq)) + +-- | Receive a single asynchronous request and send a matching reply +-- asynchronously (the handler may fork a computation that eventually replies +-- and can return before actually replying, as long as it eventually does). +-- +-- The handler receives as an argument the function to use to reply to the request. +-- 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 +handleAsyncReq pipe getit mii k = mask $ \restore -> do + let am = pipeAsyncMngr pipe + ack <- newEmptyMVar + + -- Block waiting for request + mbreq <- atomically $ do + 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) + Nothing | isClosed -> pure Nothing -- empty and closed + | otherwise -> retry -- wait for msg + + case mbreq of + Nothing -> throwIO PipeReadClosedException -- can't expect reqs + Just (uq, bs) -> do + let reply p = do + 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 + let exc = DecodeAsyncReqException err + reply (Left (toException exc)) + throwIO exc + Right (_, _, it) -> + restore (k (WriteReply reply) it) + + -- If there's an exception before the reply is sent, reply with an error instead. + `catch` \(e :: SomeException) -> do + + tryTakeMVar ack >>= \case + Just () -> + -- Already replied + throwIO e + Nothing -> do + -- Reply with an error + 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 + + +-- ** Mailboxes ---------------------------------------------------------------- + +-- | A 'Request' may optionally declare an 'InboxId'. If it does, it can only +-- be read by a 'handleAsyncReq' for that specific 'InboxId'. That is, +-- @'handleAsyncReq' p get Nothing@ will not receive messages meant for a +-- specific inbox. +-- +-- This is useful to prevent mixing up reverse requests, where we want to send +-- many reverse requests but have the caller handle them without interleaving +-- them. +newtype InboxId = InboxId Int deriving (Show, Binary) + +-- | Create a new mailbox from where we can read all requests addressed to this 'InboxId'. +newMailbox :: AsyncManager -> IO InboxId +newMailbox am = InboxId <$> freshId am + -- `mailboxQueues` is written on demand, no need to do it here. + +-- | Clean up a created mailbox by 'InboxId'. +closeMailbox :: AsyncManager -> InboxId -> IO () +closeMailbox AsyncManager{..} (InboxId ii) = + atomically (modifyTVar mailboxQueues (IntMap.delete ii)) + +-- | Get or create a new mailbox for this 'InboxId' +getMailbox :: AsyncManager -> InboxId -> STM (TQueue (CorrelationId, LB.ByteString)) +getMailbox am (InboxId ii) = do + mboxes <- readTVar (mailboxQueues am) + case IntMap.lookup ii mboxes of + Nothing -> do + q <- newTQueue + writeTVar (mailboxQueues am) (IntMap.insert ii q mboxes) + pure q + Just q -> + pure q + +-- ** Async matching "internals" ----------------------------------------------- + +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 + -- TMVar is signaled when the reply arrives. The exception can originate + -- from the host process, or from the remote one (with 'RemoteProcessException'). + , requestsQueue :: !(TQueue (CorrelationId, LB.ByteString)) + -- ^ A queue of pending received requests. When a request is received + -- /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@. + , mailboxQueues :: !(TVar (IntMap (TQueue (CorrelationId, LB.ByteString)))) + -- ^ Requests for a specific 'InboxId' that can be read using + -- @'handleAsyncReq' p get (Just <mailboxid>)@. + , asyncDrainTid :: !ThreadId + , asyncFreshId :: !(IORef Int) + } + +-- | Each 'Request' expects a 'Reply' with a matching 'CorrelationId'. +type CorrelationId = Int + +-- | We have a single type for requests and replies because both the host and +-- interpreter may send requests *and* replies on the write end *and* on recv +-- both on the read end. +-- +-- For instance: +-- 1. The host sends a RunTH request on its Host Write End (HWE) +-- 2. The interpreter sends a NewName TH request on its Interpreter Write End (IWE) +-- 3. The host sends a reply to the TH request on its HWE +-- 4. The interpreter sends the reply to the RunTH request on its IWE +-- +-- Additionally, we want to keep concurrent TH requests for independent RunTH +-- calls separated. Thus the interpreter requests additionally specify a +-- mailbox announced by RunTH: +-- 1. To interpreter: @Request _ Nothing (RunTH <mailid>)@ +-- 2. From interpreter: @Request _ (Just <mailid>) NewName@ +-- And the host will relate the <mailid> reverse requests to that RunTH run +-- exclusively. +data IddMsg = Request !CorrelationId !(Maybe InboxId) LB.ByteString + | Reply !CorrelationId (Either SerializableException LB.ByteString) + +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 + get = do + b <- getWord8 + case b of + 0 -> Request <$> get <*> get <*> get + 1 -> Reply <$> get <*> get + _ -> fail $ "impossible: Binary IddMsg - " ++ show b + +newAsyncManager :: Pipe -> IO AsyncManager +newAsyncManager pipe = do + asyncFreshId <- newIORef 0 + pendingReply <- newTVarIO IntMap.empty + requestsQueue <- newTQueueIO + mailboxQueues <- newTVarIO IntMap.empty + + -- Fork thread to drain read end. + -- (cleaned up on 'closePipe') + asyncDrainTid <- forkIOWithUnmask (drainPipe pipe) + + let am = AsyncManager{..} + pure am + +-- | Drains the read end of the Pipe and match 'Reply'es against pending 'Request's. +-- +-- In pratice, we'll have both a thread draining the read end of the host's +-- 'Pipe' and a thread draining the read end of the interpreter's 'Pipe', since +-- both processes construct a 'Pipe' and send and receive messages +-- +-- Closes the Pipe read end when done reading +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) + loop + -- EOF on read handle + Nothing -> do + -- clean up pipe then exit loop + replyToPendings (toException PipeReadClosedException) + pure () + + replyToPendings e = atomically $ do + -- reqs are only added if pipe not closed; + -- thus we close and take all pending atomically. + writeTVar (pipeReadClosed p) True + pendings <- swapTVar (pendingReply am) IntMap.empty -- take pending + mapM_ (\p -> putTMVar p (Left e {-host exception-})) pendings + + am = pipeAsyncMngr p + +-- | An exception that can be thrown on the host process representing an +-- expection that happened on the remote one. +newtype RemoteProcessException = RemoteProcessException SerializableException deriving Show +instance Exception RemoteProcessException + +-- | Come up with a fresh id for 'CorrelationId' or 'InboxId' +freshId :: AsyncManager -> IO Int +freshId AsyncManager{asyncFreshId} = + atomicModifyIORef' asyncFreshId $ \i -> (i+1, i) + +-------------------------------------------------------------------------------- +-- * Consume `reply` exactly once +-------------------------------------------------------------------------------- + +-- | Every request should get exactly one reply. +data WriteReply a where + WriteReply :: (Either SomeException a -> IO ()) %1 -> WriteReply a + +-- | Write a reply exactly once (consuming the ability to write more replies) +writeReply :: WriteReply a %1 -> a -> IO () +writeReply (WriteReply r) = \x -> r (Right x) + +-- | Infix, linear, 'unIO' +(|#>) :: IO a %1 -> State# RealWorld -> (# State# RealWorld, a #) +(|#>) (IO fs) s = fs s + +-------------------------------------------------------------------------------- +-- * Serializable Exceptions +-------------------------------------------------------------------------------- + +-- SomeException can't be serialized because it contains dynamic +-- types. However, we do very limited things with the exceptions that +-- are thrown by interpreted computations: +-- +-- * We print them, e.g. "*** Exception: <something>" +-- * UserInterrupt has a special meaning +-- * In ghc -e, exitWith should exit with the appropriate exit code +-- +-- So all we need to do is distinguish UserInterrupt and ExitCode, and +-- all other exceptions can be represented by their 'show' string. +-- +data SerializableException + = EUserInterrupt + | EExitCode ExitCode + | EOtherException String + deriving (Generic, Show) + +toSerializableException :: SomeException -> SerializableException +toSerializableException ex + | Just UserInterrupt <- fromException ex = EUserInterrupt + | Just (ec::ExitCode) <- fromException ex = (EExitCode ec) + | otherwise = EOtherException $ +#if MIN_VERSION_base(4,20,0) + -- Exception plus backtrace as seen in `displayExceptionWithInfo` + case displayExceptionContext (someExceptionContext ex) of + "" -> displayException (ex :: SomeException) + cx -> displayException (ex :: SomeException) ++ "\n\n" ++ cx +#else + show (ex :: SomeException) +#endif + +fromSerializableException :: SerializableException -> SomeException +fromSerializableException EUserInterrupt = toException UserInterrupt +fromSerializableException (EExitCode c) = toException c +fromSerializableException (EOtherException str) = toException (ErrorCall str) + +instance Binary ExitCode +instance Binary SerializableException ===================================== libraries/ghci/GHCi/Server.hs ===================================== @@ -1,4 +1,4 @@ -{-# LANGUAGE CPP, RankNTypes, RecordWildCards, GADTs, ScopedTypeVariables #-} +{-# LANGUAGE LambdaCase, CPP, RankNTypes, RecordWildCards, GADTs, ScopedTypeVariables #-} module GHCi.Server ( MessageHook , CustomMessageHandler @@ -15,6 +15,7 @@ import GHCi.Signals import GHCi.TH import GHCi.Message import Data.ByteString (ByteString) +import Data.IORef #if defined(wasm32_HOST_ARCH) import qualified Data.ByteString.Builder as B import qualified Data.ByteString.Internal as B @@ -60,27 +61,29 @@ servWithCustom -> (forall a .IO a -> IO a) -> CustomMessageHandler -> IO () -servWithCustom verbose hook pipe restore customHandler = loop +servWithCustom verbose hook pipe restore customHandler = newIORef False >>= loop where - loop = do + loop doneRef = do when verbose $ trace "reading pipe..." - Msg msg <- readPipe pipe getMessage >>= hook - - discardCtrlC - - when verbose $ trace ("msg: " ++ (show msg)) - case msg of - CustomMessage tag payload -> handleCustom tag payload - Shutdown -> return () - RunTH st q ty loc -> wrapRunTH $ runTH pipe st q ty loc - RunModFinalizers st qrefs -> wrapRunTH $ runModFinalizerRefs pipe st qrefs - _other -> run msg >>= reply - - reply :: forall a. (Binary a, Show a) => a -> IO () - reply r = do - when verbose $ trace ("writing pipe: " ++ show r) - writePipe pipe (put r) - loop + handleRemoteCall pipe hook $ \(msg :: Message a) -> do + + discardCtrlC + + when verbose $ trace ("msg: " ++ (show msg)) + case msg of + CustomMessage tag payload -> handleCustom tag payload + Shutdown -> writeIORef doneRef True >> reply () + -- TODO: These two should probably be done off the main thread + RunTH ii st q ty loc -> wrapRunTH ii $ runTH pipe ii st q ty loc + RunModFinalizers ii st qrefs -> wrapRunTH ii $ runModFinalizerRefs pipe ii st qrefs + _other -> run msg >>= reply + + -- closePipe here will wait until every active pipeWrite pipe exits, but + -- they should never block for long, as the host process is either reading + -- or has called pipeCloseRead. + isDone <- readIORef doneRef + if isDone then closePipe pipe + else loop doneRef handleCustom tag payload = do mresp <- customHandler tag payload @@ -89,17 +92,22 @@ servWithCustom verbose hook pipe restore customHandler = loop Nothing -> error $ "GHCi.Server: unhandled CustomMessage with tag " ++ show tag + reply :: forall a. Show a => a -> IO a + reply r = do + when verbose $ trace ("writing pipe: " ++ show r) + return r -- the result of the handler gets sent + -- Run some TH code, which may interact with GHC by sending -- THMessage requests, and then finally send RunTHDone followed by a -- QResult. For an overview of how TH works with Remote GHCi, see -- Note [Remote Template Haskell] in libraries/ghci/GHCi/TH.hs. - wrapRunTH :: forall a. (Binary a, Show a) => IO a -> IO () - wrapRunTH io = do + wrapRunTH :: forall a. (Binary a, Show a) => InboxId -> IO a -> IO (QResult a) + wrapRunTH ii io = do when verbose $ trace "wrapRunTH..." r <- try io when verbose $ trace "wrapRunTH done." when verbose $ trace "writing RunTHDone." - writePipe pipe (putTHMessage RunTHDone) + remoteTHCall pipe ii RunTHDone case r of Left e | Just (GHCiQException _ err) <- fromException e -> do @@ -179,7 +187,7 @@ defaultServerWithCustom customHandler = do #if defined(wasm32_HOST_ARCH) -- See Note [wasm ghci signal handlers] for details installSignalHandlers $ js_register_signal_handler cb_sig - pipe <- mkPipeFromContinuations (recv_buf cb_recv) (send_buf cb_send) + pipe <- mkPipeFromContinuations (recv_buf cb_recv) (send_buf cb_send) (pure ()) (pure ()) #else when verbose $ printf "GHC iserv starting (in: %s; out: %s)\n" (show inh) (show outh) ===================================== libraries/ghci/GHCi/TH.hs ===================================== @@ -115,8 +115,8 @@ import qualified GHC.Boot.TH.Monad as TH import Unsafe.Coerce -- | Create a new instance of 'QState' -initQState :: Pipe -> QState -initQState p = QState M.empty Nothing p +initQState :: Pipe -> InboxId -> QState +initQState p ii = QState M.empty Nothing ii p -- | The monad in which we run TH computations on the server newtype GHCiQ a = GHCiQ { runGHCiQ :: IORef QState -> IO a } @@ -156,7 +156,7 @@ noLoc = TH.Loc "<no file>" "<no package>" "<no module>" (0,0) (0,0) ghcCmd :: Binary a => THMessage (THResult a) -> GHCiQ a ghcCmd m = GHCiQ $ \sRef -> do s <- readIORef sRef - r <- remoteTHCall (qsPipe s) m + r <- remoteTHCall (qsPipe s) (qsInboxId s) m case r of THException str -> throwIO (GHCiQException s str) THComplete res -> return res @@ -189,9 +189,9 @@ metaHandlersGHCiQ runInIO = TH.MetaHandlers { -- See Note [TH recover with -fexternal-interpreter] in GHC.Tc.Gen.Splice , mRecover = \h a -> runInIO $ GHCiQ $ \sRef -> mask $ \unmask -> do s <- readIORef sRef - remoteTHCall (qsPipe s) StartRecover + remoteTHCall (qsPipe s) (qsInboxId s) StartRecover e <- try $ unmask $ runGHCiQ (runQinGHCiQ a <* ghcCmd FailIfErrs) sRef - remoteTHCall (qsPipe s) (EndRecover (isLeft e)) + remoteTHCall (qsPipe s) (qsInboxId s) (EndRecover (isLeft e)) case e of Left GHCiQException{} -> runGHCiQ (runQinGHCiQ h) sRef @@ -233,26 +233,28 @@ metaHandlersGHCiQ runInIO = TH.MetaHandlers { -- a new IORef QState, and return a RemoteRef to it. startTH :: IO (RemoteRef (IORef QState)) startTH = do - r <- newIORef (initQState (error "startTH: no pipe")) + r <- newIORef (initQState (error "startTH: no pipe") (error "startTH: no inbox id")) mkRemoteRef r -- | Runs the mod finalizers. -- -- The references must be created on the caller process. -runModFinalizerRefs :: Pipe -> RemoteRef (IORef QState) +runModFinalizerRefs :: Pipe -> InboxId + -> RemoteRef (IORef QState) -> [RemoteRef (TH.Q ())] -> IO () -runModFinalizerRefs pipe rstate qrefs = do +runModFinalizerRefs pipe ii rstate qrefs = do qs <- mapM localRef qrefs qstateref <- localRef rstate qstate <- readIORef qstateref - qstate' <- newIORef $ qstate { qsPipe = pipe } + qstate' <- newIORef $ qstate { qsInboxId = ii, qsPipe = pipe } _ <- runGHCiQ (runQinGHCiQ $ sequence_ qs) qstate' return () -- | The implementation of the 'RunTH' message runTH :: Pipe + -> InboxId -> RemoteRef (IORef QState) -- ^ The TH state, created by 'startTH' -> HValueRef @@ -264,13 +266,13 @@ runTH -> IO ByteString -- ^ Returns an (encoded) result that depends on the THResultType -runTH pipe rstate rhv ty mb_loc = do +runTH pipe ii rstate rhv ty mb_loc = do hv <- localRef rhv case ty of - THExp -> runTHQ pipe rstate mb_loc (unsafeCoerce hv :: TH.Q TH.Exp) - THPat -> runTHQ pipe rstate mb_loc (unsafeCoerce hv :: TH.Q TH.Pat) - THType -> runTHQ pipe rstate mb_loc (unsafeCoerce hv :: TH.Q TH.Type) - THDec -> runTHQ pipe rstate mb_loc (unsafeCoerce hv :: TH.Q [TH.Dec]) + THExp -> runTHQ pipe ii rstate mb_loc (unsafeCoerce hv :: TH.Q TH.Exp) + THPat -> runTHQ pipe ii rstate mb_loc (unsafeCoerce hv :: TH.Q TH.Pat) + THType -> runTHQ pipe ii rstate mb_loc (unsafeCoerce hv :: TH.Q TH.Type) + THDec -> runTHQ pipe ii rstate mb_loc (unsafeCoerce hv :: TH.Q [TH.Dec]) THAnnWrapper -> do hv <- unsafeCoerce <$> localRef rhv case hv :: AnnotationWrapper of @@ -279,10 +281,10 @@ runTH pipe rstate rhv ty mb_loc = do -- | Run a Q computation. runTHQ - :: Binary a => Pipe -> RemoteRef (IORef QState) -> Maybe TH.Loc -> TH.Q a + :: Binary a => Pipe -> InboxId -> RemoteRef (IORef QState) -> Maybe TH.Loc -> TH.Q a -> IO ByteString -runTHQ pipe rstate mb_loc ghciq = do +runTHQ pipe ii rstate mb_loc ghciq = do qstateref <- localRef rstate - modifyIORef' qstateref (\qstate -> qstate { qsLocation = mb_loc, qsPipe = pipe }) + modifyIORef' qstateref (\qstate -> qstate { qsLocation = mb_loc, qsInboxId = ii, qsPipe = pipe }) r <- runGHCiQ (runQinGHCiQ ghciq) qstateref return $! LB.toStrict (runPut (put r)) ===================================== libraries/ghci/ghci.cabal.in ===================================== @@ -76,6 +76,7 @@ library exposed-modules: GHCi.BreakArray GHCi.Message + GHCi.Message.Async GHCi.ResolvedBCO GHCi.RemoteTypes GHCi.FFI @@ -95,7 +96,8 @@ library filepath >= 1.4 && < 1.6, ghc-boot == @ProjectVersionMunged@, ghc-heap >= 9.10.1 && <=@ProjectVersionMunged@, - transformers >= 0.5 && < 0.7 + transformers >= 0.5 && < 0.7, + stm >= 2.4 && < 2.6 if flag(bootstrap) build-depends: ===================================== testsuite/tests/ghci/custom-external-interpreter-commands/Main.hs ===================================== @@ -12,6 +12,7 @@ import GHCi.Message , mkPipeFromHandles , remoteCall , Pipe + , closePipe ) import GHCi.Server ( CustomMessageHandler @@ -153,14 +154,13 @@ customMessage pipe c = do runClient :: Int -> IO () runClient input = do serverExe <- getExecutablePath - withServer serverExe $ \hFromServer hToServer -> do - pipe <- mkPipeFromHandles hFromServer hToServer + withServer serverExe $ \pipe -> do res <- customMessage pipe (SquareCommand input) putStrLn $ "Square returned: " ++ show res res2 <- customMessage pipe (MulCommand 2 res) putStrLn $ "Mul returned: " ++ show res2 -withServer :: FilePath -> (Handle -> Handle -> IO a) -> IO a +withServer :: FilePath -> (Pipe -> IO a) -> IO a withServer serverExe action = do (ghcRead, serverWrite) <- createPipe (serverRead, ghcWrite) <- createPipe @@ -171,13 +171,14 @@ withServer serverExe action = do , std_out = Inherit , std_err = Inherit } - bracket (mkHandles ghcRead ghcWrite) - (\(hFromServer, hToServer) -> do + bracket (mkHandles ghcRead ghcWrite >>= \hds -> (hds,) <$> uncurry mkPipeFromHandles hds) + (\((hFromServer, hToServer), pipe) -> do + closePipe pipe hClose hFromServer hClose hToServer terminateProcess ph void (waitForProcess ph)) - (\(hFromServer, hToServer) -> action hFromServer hToServer) + (\(_, pipe) -> action pipe) where mkHandles r w = do hR <- fdToHandle r View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/c2c42e7a8e8584caab528cfce1122e9... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/c2c42e7a8e8584caab528cfce1122e9... 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