Simon Peyton Jones pushed new branch wip/T27910 at Glasgow Haskell Compiler / GHC
--
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/tree/wip/T27910
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
1
0
[Git][ghc/ghc][wip/romes/async-rpc] Asynchronous RPC with external interpreter
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
12 Oct '26
Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC
Commits:
b43fa1a6 by Rodrigo Mesquita at 2026-10-12T19:32:05+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.
- - - - -
9 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/Server.hs
- libraries/ghci/ghci.cabal.in
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 []
=====================================
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,61 @@ 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
+ -> Message a
+ -> (forall b. Binary b => THMessage b -> IO (Bool, b))
+ -- ^ Handle one message. Return True to end the loop.
+ -> IO a
+sendTHComputation i msg reqHandler = mask $ \restore -> do
+ -- inline "withAsync/wait"
+ var <- newEmptyTMVarIO
+ ht <- forkFinally (restore handlerLoop) (atomically . putTMVar var)
+ restore (sendMessage i msg <* atomically (readTMVar var))
+ `finally` cancel ht var
+ where
+ proc = instProcess i
+ handleOne doneRef = do
+ handleRemoteTHCall (interpPipe proc) $ \req -> do
+ (stop, res) <- reqHandler req
+ writeIORef doneRef stop
+ return res
+ handlerLoop = do
+ doneRef <- newIORef False
+ let loop = do
+ handleOne doneRef
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop
+ in loop
+ cancel ht var = uninterruptibleMask_ (throwTo ht CancelTHHandler <* atomically (readTMVar var))
handleInterpProcessFailure :: InterpProcess -> SomeException -> IO a
handleInterpProcessFailure i e = do
@@ -130,3 +86,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 (RunModFinalizers 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 (RunTH 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,10 +23,22 @@ module GHCi.Message
, ResumeContext(..)
, QState(..)
, getMessage, putMessage, getTHMessage, putTHMessage
- , Pipe, mkPipeFromHandles, mkPipeFromContinuations, remoteCall, remoteTHCall, readPipe, writePipe
+
+ , Pipe, mkPipeFromHandles, mkPipeFromContinuations
+ , remoteCall, remoteTHCall
+ , handleRemoteCall, handleRemoteTHCall
+ , RemoteProcessException(..)
+
, BreakModule
, BreakUnitId
, LoadedDLL
+
+ -- * Low-level interface for async bidirectional RPC
+ , withAsyncReq
+ , withAsyncHandler
+ -- ** Consume `reply` exactly once
+ , WriteReply -- abstract
+ , writeReply
) where
import Prelude -- See note [Why do we import Prelude here?]
@@ -45,6 +58,7 @@ 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
@@ -63,6 +77,8 @@ import Data.Dynamic
import Data.Typeable (TypeRep)
import Data.IORef
import Data.Map (Map)
+import Data.IntMap (IntMap)
+import qualified Data.IntMap as IntMap
import Foreign
import GHC.Generics
import GHC.Stack.CCS
@@ -71,6 +87,11 @@ import qualified GHC.Boot.TH.Monad as TH
import System.Exit
import System.IO
import System.IO.Error
+import Control.Monad
+import Control.Concurrent.STM
+import Data.Bifunctor
+import GHC.Base (State#)
+import GHC.Exts (RealWorld)
-- -----------------------------------------------------------------------------
-- The RPC protocol between GHC and the interactive server
@@ -693,9 +714,11 @@ serializeBCOs rbcos = parMap doChunk (chunkList 100 rbcos)
-- | An opaque pipe for bidirectional binary data transmission.
data Pipe = Pipe
- { getSome :: !(IO ByteString)
- , putAll :: !(B.Builder -> IO ())
- , pipeLeftovers :: !(IORef (Maybe ByteString))
+ { getSome :: !(IO ByteString)
+ , putAll :: !(B.Builder -> IO ())
+ , pipeLeftovers :: !(IORef (Maybe ByteString))
+ , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent access to the stream
+ , pipeAsyncMngr :: !AsyncManager
}
-- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write.
@@ -705,30 +728,204 @@ mkPipeFromHandles pipeRead pipeWrite = do
putAll b = do
B.hPutBuilder pipeWrite b
hFlush pipeWrite
- pipeLeftovers <- newIORef Nothing
- pure $ Pipe { getSome, putAll, pipeLeftovers }
+ mkPipeFromContinuations getSome putAll
-- | 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 }
-
+mkPipeFromContinuations getSome putAll = mdo
+ pipeLeftovers <- newIORef Nothing
+ pipeLock <- newMVar ()
+ mdo
+ pipeAsyncMngr <- newAsyncManager p
+ let p = Pipe { .. }
+ pure p
+
+-- | 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
+remoteCall pipe msg = either throwIO (pure . runGet get)
+ =<< withAsyncReq pipe (putMessage msg)
+
+-- | 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 =
+ withAsyncHandler pipe getMessage \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
+remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
+remoteTHCall pipe msg = either throwIO (pure . runGet get)
+ =<< withAsyncReq pipe (putTHMessage msg)
+
+-- | Handle a 'THMessage' request on the host and reply to the interpreter.
+handleRemoteTHCall
+ :: Pipe -> (forall a. (Show a, Binary a) => THMessage a -> IO a) -> IO ()
+handleRemoteTHCall pipe k =
+ withAsyncHandler pipe getTHMessage \writer (THMsg msg) -> IO \s0 ->
+ case k msg |#> s0 of -- unIO for linearity checking
+ (# s1, a #) -> writeReply writer (put a) |#> s1
+
+--------------------------------------------------------------------------------
+-- * Matching requests with async replies
+--------------------------------------------------------------------------------
+
+-- | Send an asynchronous request and block waiting for a matching reply
+withAsyncReq :: Pipe -> Put -> IO (Either SomeException LB.ByteString)
+withAsyncReq pipe msg = do
+
+ -- Mark pending request
+ uq <- freshReqId (pipeAsyncMngr pipe)
+ wv <- newEmptyMVar
+ atomicModifyIORef' (pendingSent (pipeAsyncMngr pipe)) $
+ \m -> (IntMap.insert uq wv m, ())
+
+ -- Write request
+ writePipe pipe (Request uq (runPut msg))
+
+ -- Block waiting for reply
+ takeMVar wv
+
+-- | 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 'withAsyncHandler'.
+withAsyncHandler :: Pipe -> Get a -> (WriteReply Put %1 -> a -> IO r) -> IO r
+withAsyncHandler pipe getit k = do
+ ack <- newEmptyMVar
+
+ -- Block waiting for request
+ (uq, bs) <- atomically (readTQueue (pendingRecv (pipeAsyncMngr pipe)))
+
+ let reply p = do
+ writePipe pipe (Reply uq (bimap toSerializableException runPut p))
+ putMVar ack ()
+
+ -- Run the continuation which must reply exactly once with 'writeReply'
+ k (WriteReply reply) (runGet getit bs)
+
+ -- 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
+
+-- | 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
+
+-- ** Async matching "internals" -----------------------------------------------
+
+data AsyncManager = AsyncManager
+ { pendingSent :: !(IORef (IntMap (MVar (Either SomeException LB.ByteString))))
+ -- ^ When a request is sent, keep track that we're expecting a reply. The
+ -- MVar is signaled when the reply arrives. The exception can originate
+ -- from the host process, or from the remote one (with 'RemoteProcessException').
+ , pendingRecv :: !(TQueue (CorrelationId, LB.ByteString))
+ -- ^ A queue of pending received requests. When a request is received,
+ -- queue it here. A request handler will pop the next request from it.
+ -- (Nothing except the 'drainPipe' thread should read from the pipe).
+ --
+ -- Seq contract: Dequeue from left, enqueue on right
+ , asyncDrainTid :: !ThreadId
+ , asyncNextReq :: !(IORef Int)
+ }
-writePipe :: Pipe -> Put -> IO ()
-writePipe Pipe{..} put = putAll $ execPut put
+type CorrelationId = Int
-remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
-remoteTHCall pipe msg = do
- writePipe pipe (putTHMessage msg)
- readPipe pipe get
+-- | 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
+data IddMsg = Request !CorrelationId LB.ByteString
+ | Reply !CorrelationId (Either SerializableException LB.ByteString)
+
+instance Binary IddMsg where
+ put (Request i bs) = putWord8 0 >> put i >> put bs
+ put (Reply i bs) = putWord8 1 >> put i >> put bs
+ get = do
+ b <- getWord8
+ case b of
+ 0 -> Request <$> get <*> get
+ 1 -> Reply <$> get <*> get
+ _ -> error $ "impossible: Binary IddMsg - " ++ show b
+
+newAsyncManager :: Pipe -> IO AsyncManager
+newAsyncManager pipe = do
+ asyncNextReq <- newIORef 0
+ pendingSent <- newIORef IntMap.empty
+ pendingRecv <- newTQueueIO
+
+ mdo
+ -- Fork thread to drain read end.
+ asyncDrainTid <- forkIO (drainPipe pipe am)
+ -- todo: proper thread clean up and no leaks
-readPipe :: Pipe -> Get a -> IO a
-readPipe Pipe{..} get = do
+ 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:
+drainPipe :: Pipe -> AsyncManager -> IO ()
+drainPipe p am = flip catch cancelPending $ forever $ do
+ readPipe p >>= \case
+ -- Receive a reply to a pending notification
+ Reply uq bs -> do
+ wv <- atomicModifyIORef' (pendingSent am) $ \m -> do
+ case IntMap.lookup uq m of
+ Nothing -> error "drainPipe: Received reply to no pending request"
+ Just wv -> (IntMap.delete uq m, wv)
+ putMVar wv (bimap (toException . RemoteProcessException) id bs) -- signal matching request
+ -- Receive a request and queue it
+ Request uq bs ->
+ atomically $ writeTQueue (pendingRecv am) (uq,bs)
+ where
+ cancelPending (e::SomeException) = do
+ pendings <- readIORef (pendingSent am)
+ mapM_ (\p -> putMVar p (Left e {-host exception-})) pendings
+
+freshReqId :: AsyncManager -> IO CorrelationId
+freshReqId AsyncManager{asyncNextReq} =
+ atomicModifyIORef' asyncNextReq $ \i -> (i+1, i)
+
+--------------------------------------------------------------------------------
+-- * 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'.
+readPipe :: Pipe -> IO IddMsg
+readPipe Pipe{..} = do
leftovers <- readIORef pipeLeftovers
m <- getBin getSome get leftovers
case m of
@@ -758,3 +955,22 @@ getBin getsome get leftover = go leftover (runGetIncremental get)
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
+
+--------------------------------------------------------------------------------
+-- * 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
=====================================
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,26 @@ 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
+ handleRemoteCall pipe hook $ \(msg :: Message a) -> do
- discardCtrlC
+ 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
+ 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 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
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop doneRef
handleCustom tag payload = do
mresp <- customHandler tag payload
@@ -89,17 +89,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 :: forall a. (Binary a, Show a) => IO a -> IO (QResult a)
wrapRunTH 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 RunTHDone
case r of
Left e
| Just (GHCiQException _ err) <- fromException e -> do
=====================================
libraries/ghci/ghci.cabal.in
=====================================
@@ -95,7 +95,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:
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/b43fa1a66df3a9f90e7a59180a086d5…
--
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/b43fa1a66df3a9f90e7a59180a086d5…
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
1
0
[Git][ghc/ghc][wip/romes/async-rpc] Asynchronous RPC with external interpreter
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
12 Oct '26
Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC
Commits:
eba58aed by Rodrigo Mesquita at 2026-10-12T19:23:21+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.
- - - - -
9 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/Server.hs
- libraries/ghci/ghci.cabal.in
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 []
=====================================
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,60 @@ 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
-- -----------------------------------------------------------------------------
-- 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
+ -> Message a
+ -> (forall b. Binary b => THMessage b -> IO (Bool, b))
+ -- ^ Handle one message. Return True to end the loop.
+ -> IO a
+sendTHComputation i msg reqHandler = mask $ \restore -> do
+ -- inline "withAsync/wait"
+ var <- newEmptyMVar
+ ht <- forkFinally (restore handlerLoop) (putMVar var)
+ restore (sendMessage i msg <* takeMVar var)
+ `finally` cancel ht var
+ where
+ proc = instProcess i
+ handleOne doneRef = do
+ handleRemoteTHCall (interpPipe proc) $ \req -> do
+ (stop, res) <- reqHandler req
+ writeIORef doneRef stop
+ return res
+ handlerLoop = do
+ doneRef <- newIORef False
+ let loop = do
+ handleOne doneRef
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop
+ in loop
+ cancel ht var = uninterruptibleMask_ (throwTo ht CancelTHHandler <* takeMVar var)
handleInterpProcessFailure :: InterpProcess -> SomeException -> IO a
handleInterpProcessFailure i e = do
@@ -130,3 +85,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 (RunModFinalizers 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 (RunTH 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,10 +23,22 @@ module GHCi.Message
, ResumeContext(..)
, QState(..)
, getMessage, putMessage, getTHMessage, putTHMessage
- , Pipe, mkPipeFromHandles, mkPipeFromContinuations, remoteCall, remoteTHCall, readPipe, writePipe
+
+ , Pipe, mkPipeFromHandles, mkPipeFromContinuations
+ , remoteCall, remoteTHCall
+ , handleRemoteCall, handleRemoteTHCall
+ , RemoteProcessException(..)
+
, BreakModule
, BreakUnitId
, LoadedDLL
+
+ -- * Low-level interface for async bidirectional RPC
+ , withAsyncReq
+ , withAsyncHandler
+ -- ** Consume `reply` exactly once
+ , WriteReply -- abstract
+ , writeReply
) where
import Prelude -- See note [Why do we import Prelude here?]
@@ -45,6 +58,7 @@ 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
@@ -63,6 +77,8 @@ import Data.Dynamic
import Data.Typeable (TypeRep)
import Data.IORef
import Data.Map (Map)
+import Data.IntMap (IntMap)
+import qualified Data.IntMap as IntMap
import Foreign
import GHC.Generics
import GHC.Stack.CCS
@@ -71,6 +87,11 @@ import qualified GHC.Boot.TH.Monad as TH
import System.Exit
import System.IO
import System.IO.Error
+import Control.Monad
+import Control.Concurrent.STM
+import Data.Bifunctor
+import GHC.Base (State#)
+import GHC.Exts (RealWorld)
-- -----------------------------------------------------------------------------
-- The RPC protocol between GHC and the interactive server
@@ -693,9 +714,11 @@ serializeBCOs rbcos = parMap doChunk (chunkList 100 rbcos)
-- | An opaque pipe for bidirectional binary data transmission.
data Pipe = Pipe
- { getSome :: !(IO ByteString)
- , putAll :: !(B.Builder -> IO ())
- , pipeLeftovers :: !(IORef (Maybe ByteString))
+ { getSome :: !(IO ByteString)
+ , putAll :: !(B.Builder -> IO ())
+ , pipeLeftovers :: !(IORef (Maybe ByteString))
+ , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent access to the stream
+ , pipeAsyncMngr :: !AsyncManager
}
-- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write.
@@ -705,30 +728,204 @@ mkPipeFromHandles pipeRead pipeWrite = do
putAll b = do
B.hPutBuilder pipeWrite b
hFlush pipeWrite
- pipeLeftovers <- newIORef Nothing
- pure $ Pipe { getSome, putAll, pipeLeftovers }
+ mkPipeFromContinuations getSome putAll
-- | 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 }
-
+mkPipeFromContinuations getSome putAll = mdo
+ pipeLeftovers <- newIORef Nothing
+ pipeLock <- newMVar ()
+ mdo
+ pipeAsyncMngr <- newAsyncManager p
+ let p = Pipe { .. }
+ pure p
+
+-- | 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
+remoteCall pipe msg = either throwIO (pure . runGet get)
+ =<< withAsyncReq pipe (putMessage msg)
+
+-- | 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 =
+ withAsyncHandler pipe getMessage \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
+remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
+remoteTHCall pipe msg = either throwIO (pure . runGet get)
+ =<< withAsyncReq pipe (putTHMessage msg)
+
+-- | Handle a 'THMessage' request on the host and reply to the interpreter.
+handleRemoteTHCall
+ :: Pipe -> (forall a. (Show a, Binary a) => THMessage a -> IO a) -> IO ()
+handleRemoteTHCall pipe k =
+ withAsyncHandler pipe getTHMessage \writer (THMsg msg) -> IO \s0 ->
+ case k msg |#> s0 of -- unIO for linearity checking
+ (# s1, a #) -> writeReply writer (put a) |#> s1
+
+--------------------------------------------------------------------------------
+-- * Matching requests with async replies
+--------------------------------------------------------------------------------
+
+-- | Send an asynchronous request and block waiting for a matching reply
+withAsyncReq :: Pipe -> Put -> IO (Either SomeException LB.ByteString)
+withAsyncReq pipe msg = do
+
+ -- Mark pending request
+ uq <- freshReqId (pipeAsyncMngr pipe)
+ wv <- newEmptyMVar
+ atomicModifyIORef' (pendingSent (pipeAsyncMngr pipe)) $
+ \m -> (IntMap.insert uq wv m, ())
+
+ -- Write request
+ writePipe pipe (Request uq (runPut msg))
+
+ -- Block waiting for reply
+ takeMVar wv
+
+-- | 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 'withAsyncHandler'.
+withAsyncHandler :: Pipe -> Get a -> (WriteReply Put %1 -> a -> IO r) -> IO r
+withAsyncHandler pipe getit k = do
+ ack <- newEmptyMVar
+
+ -- Block waiting for request
+ (uq, bs) <- atomically (readTQueue (pendingRecv (pipeAsyncMngr pipe)))
+
+ let reply p = do
+ writePipe pipe (Reply uq (bimap toSerializableException runPut p))
+ putMVar ack ()
+
+ -- Run the continuation which must reply exactly once with 'writeReply'
+ k (WriteReply reply) (runGet getit bs)
+
+ -- 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
+
+-- | 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
+
+-- ** Async matching "internals" -----------------------------------------------
+
+data AsyncManager = AsyncManager
+ { pendingSent :: !(IORef (IntMap (MVar (Either SomeException LB.ByteString))))
+ -- ^ When a request is sent, keep track that we're expecting a reply. The
+ -- MVar is signaled when the reply arrives. The exception can originate
+ -- from the host process, or from the remote one (with 'RemoteProcessException').
+ , pendingRecv :: !(TQueue (CorrelationId, LB.ByteString))
+ -- ^ A queue of pending received requests. When a request is received,
+ -- queue it here. A request handler will pop the next request from it.
+ -- (Nothing except the 'drainPipe' thread should read from the pipe).
+ --
+ -- Seq contract: Dequeue from left, enqueue on right
+ , asyncDrainTid :: !ThreadId
+ , asyncNextReq :: !(IORef Int)
+ }
-writePipe :: Pipe -> Put -> IO ()
-writePipe Pipe{..} put = putAll $ execPut put
+type CorrelationId = Int
-remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
-remoteTHCall pipe msg = do
- writePipe pipe (putTHMessage msg)
- readPipe pipe get
+-- | 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
+data IddMsg = Request !CorrelationId LB.ByteString
+ | Reply !CorrelationId (Either SerializableException LB.ByteString)
+
+instance Binary IddMsg where
+ put (Request i bs) = putWord8 0 >> put i >> put bs
+ put (Reply i bs) = putWord8 1 >> put i >> put bs
+ get = do
+ b <- getWord8
+ case b of
+ 0 -> Request <$> get <*> get
+ 1 -> Reply <$> get <*> get
+ _ -> error $ "impossible: Binary IddMsg - " ++ show b
+
+newAsyncManager :: Pipe -> IO AsyncManager
+newAsyncManager pipe = do
+ asyncNextReq <- newIORef 0
+ pendingSent <- newIORef IntMap.empty
+ pendingRecv <- newTQueueIO
+
+ mdo
+ -- Fork thread to drain read end.
+ asyncDrainTid <- forkIO (drainPipe pipe am)
+ -- todo: proper thread clean up and no leaks
-readPipe :: Pipe -> Get a -> IO a
-readPipe Pipe{..} get = do
+ 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:
+drainPipe :: Pipe -> AsyncManager -> IO ()
+drainPipe p am = flip catch cancelPending $ forever $ do
+ readPipe p >>= \case
+ -- Receive a reply to a pending notification
+ Reply uq bs -> do
+ wv <- atomicModifyIORef' (pendingSent am) $ \m -> do
+ case IntMap.lookup uq m of
+ Nothing -> error "drainPipe: Received reply to no pending request"
+ Just wv -> (IntMap.delete uq m, wv)
+ putMVar wv (bimap (toException . RemoteProcessException) id bs) -- signal matching request
+ -- Receive a request and queue it
+ Request uq bs ->
+ atomically $ writeTQueue (pendingRecv am) (uq,bs)
+ where
+ cancelPending (e::SomeException) = do
+ pendings <- readIORef (pendingSent am)
+ mapM_ (\p -> putMVar p (Left e {-host exception-})) pendings
+
+freshReqId :: AsyncManager -> IO CorrelationId
+freshReqId AsyncManager{asyncNextReq} =
+ atomicModifyIORef' asyncNextReq $ \i -> (i+1, i)
+
+--------------------------------------------------------------------------------
+-- * 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'.
+readPipe :: Pipe -> IO IddMsg
+readPipe Pipe{..} = do
leftovers <- readIORef pipeLeftovers
m <- getBin getSome get leftovers
case m of
@@ -758,3 +955,22 @@ getBin getsome get leftover = go leftover (runGetIncremental get)
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
+
+--------------------------------------------------------------------------------
+-- * 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
=====================================
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,26 @@ 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
+ handleRemoteCall pipe hook $ \(msg :: Message a) -> do
- discardCtrlC
+ 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
+ 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 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
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop doneRef
handleCustom tag payload = do
mresp <- customHandler tag payload
@@ -89,17 +89,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 :: forall a. (Binary a, Show a) => IO a -> IO (QResult a)
wrapRunTH 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 RunTHDone
case r of
Left e
| Just (GHCiQException _ err) <- fromException e -> do
=====================================
libraries/ghci/ghci.cabal.in
=====================================
@@ -95,7 +95,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:
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/eba58aed0c16f21bd23f48fdafa0d5e…
--
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/eba58aed0c16f21bd23f48fdafa0d5e…
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
1
0
[Git][ghc/ghc][wip/az/epa-tidy-locatedxxx-27] 3 commits: Simplify the error-suppression mechanism in GHC.Tc.Errors
by Alan Zimmerman (＠alanz) 12 Oct '26
by Alan Zimmerman (＠alanz) 12 Oct '26
12 Oct '26
Alan Zimmerman pushed to branch wip/az/epa-tidy-locatedxxx-27 at Glasgow Haskell Compiler / GHC
Commits:
41e80f41 by Simon Peyton Jones at 2026-10-11T14:38:51-04:00
Simplify the error-suppression mechanism in GHC.Tc.Errors
In `GHC.Tc.Errors.reportWanteds` we suppress some errors in favour of
others. But the mechanism had grown crufty, and #27331 exposed a bug.
This MR makes it simpler and more uniform, by doing everything via
the `ei_suppress` field of the ErrorItem.
Claude then found another closely-related bug (see the review linked
on !16564). That bug is immortalised in new test T18851d.
I'm still not happy with `ignoreConstraint` but we can worry about
that another day; I have not touched it.
Main changes
* Instead of /predicting/ when an error will be reported, we set
`cec_error_reported` when an error is /actually/ reported. (I renamed
it from `cec_suppress` to reflect this change.)
Now errors are suppressed only when a preceding error has actually been
reported. Simpler and more robust.
See `Note [cec_error_reported: suppressing less-serious errors]`
The big changes are in `reportWanteds` which looks very different.
* New `Note [ei_suppress: suppressing confusing errors]` describes the
(quite separate) `ei_suppress` mechanism for suppressing confusing
errors, mainly arising from non-empty rewriter sets
* In `tryTypeClassDefaulting`, instead of switching off type-class defaulting
if there is an insoluble /anywhere/, we just switch it off locally when
we find an insoluble sub-tree. See Note [When to do type-class defaulting]
and (W5) in Note [ApproximateWC]
* Greatly simplify `insolubleWantedCt`.
* I made type-class-defaulting be a bit more eager, only switched off when
a peer or ancestor (but not child) constraint is insoluble. See
Note [When to do type-class defaulting].
Small refactorings
* Rename `askNoErrs` to `askErrsFound` and flip its polarity
* The code for "check if ei_suppress is true of all items" code is
moved to `mkErrorItems` rather than in the large `reportWanteds`.
* Code for `has_gadt_match` moves to `mkGivenErrorReporter`, rather than
polluting the large `reportWanteds`.
* Many tests with `-fdefer-type-errors` get more warnings, especially
about ambiguous (Num a) constraints. This is by-design: see
Note [Always warn with -fdefer-type-errors]. Previously GHC was not
honouring this Note but now it is, hence the extra errors.
- - - - -
f217fc31 by Cheng Shao at 2026-10-11T14:39:33-04:00
compiler: ensure cross ghc --info output target unit ids
Previously, for cross ghcs, the base/ghc/ghc-internal unit ids in
--info output do not match the target unit ids reported by ghc-pkg,
breaking cross ghc release builds with hash-unit-ids when being used
with cabal 3.18. This patch fixes it by making --info output target
unit ids. Fixes #27919.
Co-authored-by: Codex <codex(a)openai.com>
- - - - -
e8f40ee0 by Alan Zimmerman at 2026-10-12T18:05:10+01:00
EPA: Start using vocurly as basis for AnnListLayout
For the existing cases that populate AnnList.
This uses AnnListBraces when braces are explicitly supplied, and
AnnListLayout when vocurly is used
The change also shows up AnnList usages that can never capture layout,
these are cleared up later.
- - - - -
65 changed files:
- .gitlab/hello.hs
- + changelog.d/T27731
- + changelog.d/fix-cross-ghc-info-unit-id
- compiler/GHC/Builtin/WiredIn/Ids.hs
- compiler/GHC/Driver/Session.hs
- compiler/GHC/Parser.y
- compiler/GHC/Rename/Module.hs
- compiler/GHC/Settings.hs
- compiler/GHC/Settings/IO.hs
- compiler/GHC/Tc/Errors.hs
- compiler/GHC/Tc/Errors/Ppr.hs
- compiler/GHC/Tc/Errors/Types.hs
- compiler/GHC/Tc/Instance/Family.hs
- compiler/GHC/Tc/Solver/Default.hs
- compiler/GHC/Tc/Solver/Equality.hs
- compiler/GHC/Tc/Types/Constraint.hs
- compiler/GHC/Tc/Utils/Monad.hs
- distrib/configure.ac.in
- hadrian/bindist/Makefile
- hadrian/bindist/config.mk.in
- hadrian/src/Rules/BinaryDist.hs
- hadrian/src/Rules/Generate.hs
- testsuite/tests/ghci/prog020/ghci.prog020a.stderr
- testsuite/tests/ghci/prog020/ghci.prog020b.stderr
- testsuite/tests/ghci/prog021/prog021a.stderr
- testsuite/tests/ghci/prog021/prog021b.stderr
- testsuite/tests/ghci/prog022/ghci.prog022a.stderr
- testsuite/tests/ghci/prog022/ghci.prog022b.stderr
- testsuite/tests/ghci/prog022/ghci.prog022c.stderr
- testsuite/tests/ghci/prog022/ghci.prog022d.stderr
- testsuite/tests/ghci/prog022/ghci.prog022e.stderr
- testsuite/tests/ghci/prog022/ghci.prog022f.stderr
- testsuite/tests/ghci/scripts/Defer02.stderr
- testsuite/tests/module/mod130.stderr
- testsuite/tests/overloadedstrings/should_fail/T25926.stderr
- testsuite/tests/overloadedstrings/should_fail/T27124.stderr
- testsuite/tests/parser/should_compile/DumpSemis.stderr
- testsuite/tests/patsyn/should_fail/records-nofieldselectors.stderr
- testsuite/tests/printer/T17697.stderr
- testsuite/tests/rename/should_fail/T15607.stderr
- testsuite/tests/rename/should_fail/T19843b.stderr
- testsuite/tests/safeHaskell/ghci/p4.stderr
- testsuite/tests/safeHaskell/ghci/p6.stderr
- testsuite/tests/typecheck/should_compile/T11254.stderr
- testsuite/tests/typecheck/should_compile/T15370.stderr
- testsuite/tests/typecheck/should_compile/valid_hole_fits.stderr
- testsuite/tests/typecheck/should_fail/ClassOperator.stderr
- testsuite/tests/typecheck/should_fail/DoExpansion2.stderr
- testsuite/tests/typecheck/should_fail/DoExpansion3.stderr
- testsuite/tests/typecheck/should_fail/FunDepOrigin1b.hs
- testsuite/tests/typecheck/should_fail/T18851.stderr
- + testsuite/tests/typecheck/should_fail/T18851e.hs
- + testsuite/tests/typecheck/should_fail/T18851e.stderr
- + testsuite/tests/typecheck/should_fail/T18851f.hs
- + testsuite/tests/typecheck/should_fail/T18851f.stderr
- testsuite/tests/typecheck/should_fail/T22940.stderr
- testsuite/tests/typecheck/should_fail/T26823.stderr
- + testsuite/tests/typecheck/should_fail/T27731.hs
- + testsuite/tests/typecheck/should_fail/T27731.stderr
- testsuite/tests/typecheck/should_fail/T5691.stderr
- testsuite/tests/typecheck/should_fail/T7437.stderr
- testsuite/tests/typecheck/should_fail/all.T
- testsuite/tests/typecheck/should_fail/mc24.stderr
- testsuite/tests/typecheck/should_fail/tcfail019.stderr
- testsuite/tests/typecheck/should_fail/tcfail189.stderr
The diff was not included because it is too large.
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/a3c5ea4e21b23d45965a8a184bb8ec…
--
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/a3c5ea4e21b23d45965a8a184bb8ec…
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
1
0
[Git][ghc/ghc][wip/romes/async-rpc] Asynchronous RPC with external interpreter
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
12 Oct '26
Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC
Commits:
5d576d30 by Rodrigo Mesquita at 2026-10-12T19:00:26+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.
- - - - -
10 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/Server.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 []
=====================================
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,60 @@ 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
-- -----------------------------------------------------------------------------
-- 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
+ -> Message a
+ -> (forall b. Binary b => THMessage b -> IO (Bool, b))
+ -- ^ Handle one message. Return True to end the loop.
+ -> IO a
+sendTHComputation i msg reqHandler = mask $ \restore -> do
+ -- inline "withAsync/wait"
+ var <- newEmptyMVar
+ ht <- forkFinally (restore handlerLoop) (putMVar var)
+ restore (sendMessage i msg <* takeMVar var)
+ `finally` cancel ht var
+ where
+ proc = instProcess i
+ handleOne doneRef = do
+ handleRemoteTHCall (interpPipe proc) $ \req -> do
+ (stop, res) <- reqHandler req
+ writeIORef doneRef stop
+ return res
+ handlerLoop = do
+ doneRef <- newIORef False
+ let loop = do
+ handleOne doneRef
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop
+ in loop
+ cancel ht var = uninterruptibleMask_ (throwTo ht CancelTHHandler <* takeMVar var)
handleInterpProcessFailure :: InterpProcess -> SomeException -> IO a
handleInterpProcessFailure i e = do
@@ -130,3 +85,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 (RunModFinalizers 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 (RunTH 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,10 +23,22 @@ module GHCi.Message
, ResumeContext(..)
, QState(..)
, getMessage, putMessage, getTHMessage, putTHMessage
- , Pipe, mkPipeFromHandles, mkPipeFromContinuations, remoteCall, remoteTHCall, readPipe, writePipe
+
+ , Pipe, mkPipeFromHandles, mkPipeFromContinuations
+ , remoteCall, remoteTHCall
+ , handleRemoteCall, handleRemoteTHCall
+ , RemoteProcessException(..)
+
, BreakModule
, BreakUnitId
, LoadedDLL
+
+ -- * Low-level interface for async bidirectional RPC
+ , withAsyncReq
+ , withAsyncHandler
+ -- ** Consume `reply` exactly once
+ , WriteReply -- abstract
+ , writeReply
) where
import Prelude -- See note [Why do we import Prelude here?]
@@ -45,6 +58,7 @@ 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
@@ -63,6 +77,8 @@ import Data.Dynamic
import Data.Typeable (TypeRep)
import Data.IORef
import Data.Map (Map)
+import Data.IntMap (IntMap)
+import qualified Data.IntMap as IntMap
import Foreign
import GHC.Generics
import GHC.Stack.CCS
@@ -71,6 +87,11 @@ import qualified GHC.Boot.TH.Monad as TH
import System.Exit
import System.IO
import System.IO.Error
+import Control.Monad
+import Control.Concurrent.STM
+import Data.Bifunctor
+import GHC.Base (State#)
+import GHC.Exts (RealWorld)
-- -----------------------------------------------------------------------------
-- The RPC protocol between GHC and the interactive server
@@ -693,9 +714,11 @@ serializeBCOs rbcos = parMap doChunk (chunkList 100 rbcos)
-- | An opaque pipe for bidirectional binary data transmission.
data Pipe = Pipe
- { getSome :: !(IO ByteString)
- , putAll :: !(B.Builder -> IO ())
- , pipeLeftovers :: !(IORef (Maybe ByteString))
+ { getSome :: !(IO ByteString)
+ , putAll :: !(B.Builder -> IO ())
+ , pipeLeftovers :: !(IORef (Maybe ByteString))
+ , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent access to the stream
+ , pipeAsyncMngr :: !AsyncManager
}
-- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write.
@@ -705,30 +728,204 @@ mkPipeFromHandles pipeRead pipeWrite = do
putAll b = do
B.hPutBuilder pipeWrite b
hFlush pipeWrite
- pipeLeftovers <- newIORef Nothing
- pure $ Pipe { getSome, putAll, pipeLeftovers }
+ mkPipeFromContinuations getSome putAll
-- | 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 }
-
+mkPipeFromContinuations getSome putAll = mdo
+ pipeLeftovers <- newIORef Nothing
+ pipeLock <- newMVar ()
+ mdo
+ pipeAsyncMngr <- newAsyncManager p
+ let p = Pipe { .. }
+ pure p
+
+-- | 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
+remoteCall pipe msg = either throwIO (pure . runGet get)
+ =<< withAsyncReq pipe (putMessage msg)
+
+-- | 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 =
+ withAsyncHandler pipe getMessage \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
+remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
+remoteTHCall pipe msg = either throwIO (pure . runGet get)
+ =<< withAsyncReq pipe (putTHMessage msg)
+
+-- | Handle a 'THMessage' request on the host and reply to the interpreter.
+handleRemoteTHCall
+ :: Pipe -> (forall a. (Show a, Binary a) => THMessage a -> IO a) -> IO ()
+handleRemoteTHCall pipe k =
+ withAsyncHandler pipe getTHMessage \writer (THMsg msg) -> IO \s0 ->
+ case k msg |#> s0 of -- unIO for linearity checking
+ (# s1, a #) -> writeReply writer (put a) |#> s1
+
+--------------------------------------------------------------------------------
+-- * Matching requests with async replies
+--------------------------------------------------------------------------------
+
+-- | Send an asynchronous request and block waiting for a matching reply
+withAsyncReq :: Pipe -> Put -> IO (Either SomeException LB.ByteString)
+withAsyncReq pipe msg = do
+
+ -- Mark pending request
+ uq <- freshReqId (pipeAsyncMngr pipe)
+ wv <- newEmptyMVar
+ atomicModifyIORef' (pendingSent (pipeAsyncMngr pipe)) $
+ \m -> (IntMap.insert uq wv m, ())
+
+ -- Write request
+ writePipe pipe (Request uq (runPut msg))
+
+ -- Block waiting for reply
+ takeMVar wv
+
+-- | 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 'withAsyncHandler'.
+withAsyncHandler :: Pipe -> Get a -> (WriteReply Put %1 -> a -> IO r) -> IO r
+withAsyncHandler pipe getit k = do
+ ack <- newEmptyMVar
+
+ -- Block waiting for request
+ (uq, bs) <- atomically (readTQueue (pendingRecv (pipeAsyncMngr pipe)))
+
+ let reply p = do
+ writePipe pipe (Reply uq (bimap toSerializableException runPut p))
+ putMVar ack ()
+
+ -- Run the continuation which must reply exactly once with 'writeReply'
+ k (WriteReply reply) (runGet getit bs)
+
+ -- 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
+
+-- | 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
+
+-- ** Async matching "internals" -----------------------------------------------
+
+data AsyncManager = AsyncManager
+ { pendingSent :: !(IORef (IntMap (MVar (Either SomeException LB.ByteString))))
+ -- ^ When a request is sent, keep track that we're expecting a reply. The
+ -- MVar is signaled when the reply arrives. The exception can originate
+ -- from the host process, or from the remote one (with 'RemoteProcessException').
+ , pendingRecv :: !(TQueue (CorrelationId, LB.ByteString))
+ -- ^ A queue of pending received requests. When a request is received,
+ -- queue it here. A request handler will pop the next request from it.
+ -- (Nothing except the 'drainPipe' thread should read from the pipe).
+ --
+ -- Seq contract: Dequeue from left, enqueue on right
+ , asyncDrainTid :: !ThreadId
+ , asyncNextReq :: !(IORef Int)
+ }
-writePipe :: Pipe -> Put -> IO ()
-writePipe Pipe{..} put = putAll $ execPut put
+type CorrelationId = Int
-remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
-remoteTHCall pipe msg = do
- writePipe pipe (putTHMessage msg)
- readPipe pipe get
+-- | 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
+data IddMsg = Request !CorrelationId LB.ByteString
+ | Reply !CorrelationId (Either SerializableException LB.ByteString)
+
+instance Binary IddMsg where
+ put (Request i bs) = putWord8 0 >> put i >> put bs
+ put (Reply i bs) = putWord8 1 >> put i >> put bs
+ get = do
+ b <- getWord8
+ case b of
+ 0 -> Request <$> get <*> get
+ 1 -> Reply <$> get <*> get
+ _ -> error $ "impossible: Binary IddMsg - " ++ show b
+
+newAsyncManager :: Pipe -> IO AsyncManager
+newAsyncManager pipe = do
+ asyncNextReq <- newIORef 0
+ pendingSent <- newIORef IntMap.empty
+ pendingRecv <- newTQueueIO
+
+ mdo
+ -- Fork thread to drain read end.
+ asyncDrainTid <- forkIO (drainPipe pipe am)
+ -- todo: proper thread clean up and no leaks
-readPipe :: Pipe -> Get a -> IO a
-readPipe Pipe{..} get = do
+ 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:
+drainPipe :: Pipe -> AsyncManager -> IO ()
+drainPipe p am = flip catch cancelPending $ forever $ do
+ readPipe p >>= \case
+ -- Receive a reply to a pending notification
+ Reply uq bs -> do
+ wv <- atomicModifyIORef' (pendingSent am) $ \m -> do
+ case IntMap.lookup uq m of
+ Nothing -> error "drainPipe: Received reply to no pending request"
+ Just wv -> (IntMap.delete uq m, wv)
+ putMVar wv (bimap (toException . RemoteProcessException) id bs) -- signal matching request
+ -- Receive a request and queue it
+ Request uq bs ->
+ atomically $ writeTQueue (pendingRecv am) (uq,bs)
+ where
+ cancelPending (e::SomeException) = do
+ pendings <- readIORef (pendingSent am)
+ mapM_ (\p -> putMVar p (Left e {-host exception-})) pendings
+
+freshReqId :: AsyncManager -> IO CorrelationId
+freshReqId AsyncManager{asyncNextReq} =
+ atomicModifyIORef' asyncNextReq $ \i -> (i+1, i)
+
+--------------------------------------------------------------------------------
+-- * 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'.
+readPipe :: Pipe -> IO IddMsg
+readPipe Pipe{..} = do
leftovers <- readIORef pipeLeftovers
m <- getBin getSome get leftovers
case m of
@@ -758,3 +955,22 @@ getBin getsome get leftover = go leftover (runGetIncremental get)
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
+
+--------------------------------------------------------------------------------
+-- * 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
=====================================
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,26 @@ 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
+ handleRemoteCall pipe hook $ \(msg :: Message a) -> do
- discardCtrlC
+ 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
+ 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 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
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop doneRef
handleCustom tag payload = do
mresp <- customHandler tag payload
@@ -89,17 +89,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 :: forall a. (Binary a, Show a) => IO a -> IO (QResult a)
wrapRunTH 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 RunTHDone
case r of
Left e
| Just (GHCiQException _ err) <- fromException e -> do
=====================================
libraries/ghci/ghci.cabal.in
=====================================
@@ -95,7 +95,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
=====================================
@@ -144,7 +144,7 @@ customMessage :: (Show a, Bin.Binary a) => Pipe -> ClientCommand a -> IO a
customMessage pipe c = do
let payload = encodeLazy (Some @Bin.Binary c)
putStrLn $ "Sending: " ++ show c
- respBytes <- remoteCall pipe (CustomMessage customTag payload)
+ Right respBytes <- remoteCall pipe (CustomMessage customTag payload)
case decodeLazy respBytes of
Left err -> error $ "Decode error: " ++ err
Right res -> pure res
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/5d576d30f7547c62252b4a486ed80ff…
--
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/5d576d30f7547c62252b4a486ed80ff…
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
1
0
[Git][ghc/ghc][wip/romes/async-rpc] Asynchronous RPC with external interpreter
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
12 Oct '26
Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC
Commits:
e8d943b7 by Rodrigo Mesquita at 2026-10-12T18:54:29+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.
- - - - -
10 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/Server.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 []
=====================================
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,9 @@ 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
+ , RemoteCallFailed(..)
)
where
@@ -27,97 +20,67 @@ 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
-- -----------------------------------------------------------------------------
-- 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.
+--
+-- Throws 'RemoteCallFailed' if the remote call threw an exception in the
+-- external interpreter
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 = unwrapRight =<<
+ remoteCall (interpPipe proc) msg
+ `catchException` \(e :: SomeException) -> handleInterpProcessFailure proc e
+ where
+ proc = instProcess i
+ unwrapRight :: Either SomeException b -> IO b
+ unwrapRight (Right a) = pure a
+ unwrapRight (Left er) = do
+ throwIO (RemoteCallFailed er)
+
+-- | 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
+ -> Message a
+ -> (forall b. Binary b => THMessage b -> IO (Bool, b))
+ -- ^ Handle one message. Return True to end the loop.
+ -> IO a
+sendTHComputation i msg reqHandler = mask $ \restore -> do
+ -- inline "withAsync/wait"
+ var <- newEmptyMVar
+ ht <- forkFinally (restore handlerLoop) (putMVar var)
+ restore (sendMessage i msg <* takeMVar var)
+ `finally` cancel ht var
+ where
+ proc = instProcess i
+ handleOne doneRef = do
+ handleRemoteTHCall (interpPipe proc) $ \req -> do
+ (stop, res) <- reqHandler req
+ writeIORef doneRef stop
+ return res
+ handlerLoop = do
+ doneRef <- newIORef False
+ let loop = do
+ handleOne doneRef
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop
+ in loop
+ cancel ht var = uninterruptibleMask_ (throwTo ht CancelTHHandler <* takeMVar var)
handleInterpProcessFailure :: InterpProcess -> SomeException -> IO a
handleInterpProcessFailure i e = do
@@ -130,3 +93,17 @@ handleInterpProcessFailure i e = do
terminateProcess hdl
_ <- waitForProcess hdl
throw e
+
+-- | The external interpreter server threw an exception while executing this
+-- remote call. The ext. interp is still live and ready to take further
+-- commands (such as 'Shutdown').
+data RemoteCallFailed = RemoteCallFailed SomeException deriving Show
+instance Exception RemoteCallFailed where
+ displayException (RemoteCallFailed e)
+ = "A remote call in the external interpreter threw an exception:\n"
+ ++ displayException 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 (RunModFinalizers 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 (RunTH 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,10 +23,22 @@ module GHCi.Message
, ResumeContext(..)
, QState(..)
, getMessage, putMessage, getTHMessage, putTHMessage
- , Pipe, mkPipeFromHandles, mkPipeFromContinuations, remoteCall, remoteTHCall, readPipe, writePipe
+
+ , Pipe, mkPipeFromHandles, mkPipeFromContinuations
+ , remoteCall, remoteTHCall
+ , handleRemoteCall, handleRemoteTHCall
+ , RemoteProcessException(..)
+
, BreakModule
, BreakUnitId
, LoadedDLL
+
+ -- * Low-level interface for async bidirectional RPC
+ , withAsyncReq
+ , withAsyncHandler
+ -- ** Consume `reply` exactly once
+ , WriteReply -- abstract
+ , writeReply
) where
import Prelude -- See note [Why do we import Prelude here?]
@@ -45,6 +58,7 @@ 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
@@ -63,6 +77,8 @@ import Data.Dynamic
import Data.Typeable (TypeRep)
import Data.IORef
import Data.Map (Map)
+import Data.IntMap (IntMap)
+import qualified Data.IntMap as IntMap
import Foreign
import GHC.Generics
import GHC.Stack.CCS
@@ -71,6 +87,11 @@ import qualified GHC.Boot.TH.Monad as TH
import System.Exit
import System.IO
import System.IO.Error
+import Control.Monad
+import Control.Concurrent.STM
+import Data.Bifunctor
+import GHC.Base (State#)
+import GHC.Exts (RealWorld)
-- -----------------------------------------------------------------------------
-- The RPC protocol between GHC and the interactive server
@@ -693,9 +714,11 @@ serializeBCOs rbcos = parMap doChunk (chunkList 100 rbcos)
-- | An opaque pipe for bidirectional binary data transmission.
data Pipe = Pipe
- { getSome :: !(IO ByteString)
- , putAll :: !(B.Builder -> IO ())
- , pipeLeftovers :: !(IORef (Maybe ByteString))
+ { getSome :: !(IO ByteString)
+ , putAll :: !(B.Builder -> IO ())
+ , pipeLeftovers :: !(IORef (Maybe ByteString))
+ , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent access to the stream
+ , pipeAsyncMngr :: !AsyncManager
}
-- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write.
@@ -705,30 +728,204 @@ mkPipeFromHandles pipeRead pipeWrite = do
putAll b = do
B.hPutBuilder pipeWrite b
hFlush pipeWrite
- pipeLeftovers <- newIORef Nothing
- pure $ Pipe { getSome, putAll, pipeLeftovers }
+ mkPipeFromContinuations getSome putAll
-- | 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 }
-
+mkPipeFromContinuations getSome putAll = mdo
+ pipeLeftovers <- newIORef Nothing
+ pipeLock <- newMVar ()
+ mdo
+ pipeAsyncMngr <- newAsyncManager p
+ let p = Pipe { .. }
+ pure p
+
+-- | 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
+remoteCall pipe msg = either throwIO (pure . runGet get)
+ =<< withAsyncReq pipe (putMessage msg)
+
+-- | 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 =
+ withAsyncHandler pipe getMessage \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
+remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
+remoteTHCall pipe msg = either throwIO (pure . runGet get)
+ =<< withAsyncReq pipe (putTHMessage msg)
+
+-- | Handle a 'THMessage' request on the host and reply to the interpreter.
+handleRemoteTHCall
+ :: Pipe -> (forall a. (Show a, Binary a) => THMessage a -> IO a) -> IO ()
+handleRemoteTHCall pipe k =
+ withAsyncHandler pipe getTHMessage \writer (THMsg msg) -> IO \s0 ->
+ case k msg |#> s0 of -- unIO for linearity checking
+ (# s1, a #) -> writeReply writer (put a) |#> s1
+
+--------------------------------------------------------------------------------
+-- * Matching requests with async replies
+--------------------------------------------------------------------------------
+
+-- | Send an asynchronous request and block waiting for a matching reply
+withAsyncReq :: Pipe -> Put -> IO (Either SomeException LB.ByteString)
+withAsyncReq pipe msg = do
+
+ -- Mark pending request
+ uq <- freshReqId (pipeAsyncMngr pipe)
+ wv <- newEmptyMVar
+ atomicModifyIORef' (pendingSent (pipeAsyncMngr pipe)) $
+ \m -> (IntMap.insert uq wv m, ())
+
+ -- Write request
+ writePipe pipe (Request uq (runPut msg))
+
+ -- Block waiting for reply
+ takeMVar wv
+
+-- | 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 'withAsyncHandler'.
+withAsyncHandler :: Pipe -> Get a -> (WriteReply Put %1 -> a -> IO r) -> IO r
+withAsyncHandler pipe getit k = do
+ ack <- newEmptyMVar
+
+ -- Block waiting for request
+ (uq, bs) <- atomically (readTQueue (pendingRecv (pipeAsyncMngr pipe)))
+
+ let reply p = do
+ writePipe pipe (Reply uq (bimap toSerializableException runPut p))
+ putMVar ack ()
+
+ -- Run the continuation which must reply exactly once with 'writeReply'
+ k (WriteReply reply) (runGet getit bs)
+
+ -- 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
+
+-- | 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
+
+-- ** Async matching "internals" -----------------------------------------------
+
+data AsyncManager = AsyncManager
+ { pendingSent :: !(IORef (IntMap (MVar (Either SomeException LB.ByteString))))
+ -- ^ When a request is sent, keep track that we're expecting a reply. The
+ -- MVar is signaled when the reply arrives. The exception can originate
+ -- from the host process, or from the remote one (with 'RemoteProcessException').
+ , pendingRecv :: !(TQueue (CorrelationId, LB.ByteString))
+ -- ^ A queue of pending received requests. When a request is received,
+ -- queue it here. A request handler will pop the next request from it.
+ -- (Nothing except the 'drainPipe' thread should read from the pipe).
+ --
+ -- Seq contract: Dequeue from left, enqueue on right
+ , asyncDrainTid :: !ThreadId
+ , asyncNextReq :: !(IORef Int)
+ }
-writePipe :: Pipe -> Put -> IO ()
-writePipe Pipe{..} put = putAll $ execPut put
+type CorrelationId = Int
-remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
-remoteTHCall pipe msg = do
- writePipe pipe (putTHMessage msg)
- readPipe pipe get
+-- | 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
+data IddMsg = Request !CorrelationId LB.ByteString
+ | Reply !CorrelationId (Either SerializableException LB.ByteString)
+
+instance Binary IddMsg where
+ put (Request i bs) = putWord8 0 >> put i >> put bs
+ put (Reply i bs) = putWord8 1 >> put i >> put bs
+ get = do
+ b <- getWord8
+ case b of
+ 0 -> Request <$> get <*> get
+ 1 -> Reply <$> get <*> get
+ _ -> error $ "impossible: Binary IddMsg - " ++ show b
+
+newAsyncManager :: Pipe -> IO AsyncManager
+newAsyncManager pipe = do
+ asyncNextReq <- newIORef 0
+ pendingSent <- newIORef IntMap.empty
+ pendingRecv <- newTQueueIO
+
+ mdo
+ -- Fork thread to drain read end.
+ asyncDrainTid <- forkIO (drainPipe pipe am)
+ -- todo: proper thread clean up and no leaks
-readPipe :: Pipe -> Get a -> IO a
-readPipe Pipe{..} get = do
+ 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:
+drainPipe :: Pipe -> AsyncManager -> IO ()
+drainPipe p am = flip catch cancelPending $ forever $ do
+ readPipe p >>= \case
+ -- Receive a reply to a pending notification
+ Reply uq bs -> do
+ wv <- atomicModifyIORef' (pendingSent am) $ \m -> do
+ case IntMap.lookup uq m of
+ Nothing -> error "drainPipe: Received reply to no pending request"
+ Just wv -> (IntMap.delete uq m, wv)
+ putMVar wv (bimap (toException . RemoteProcessException) id bs) -- signal matching request
+ -- Receive a request and queue it
+ Request uq bs ->
+ atomically $ writeTQueue (pendingRecv am) (uq,bs)
+ where
+ cancelPending (e::SomeException) = do
+ pendings <- readIORef (pendingSent am)
+ mapM_ (\p -> putMVar p (Left e {-host exception-})) pendings
+
+freshReqId :: AsyncManager -> IO CorrelationId
+freshReqId AsyncManager{asyncNextReq} =
+ atomicModifyIORef' asyncNextReq $ \i -> (i+1, i)
+
+--------------------------------------------------------------------------------
+-- * 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'.
+readPipe :: Pipe -> IO IddMsg
+readPipe Pipe{..} = do
leftovers <- readIORef pipeLeftovers
m <- getBin getSome get leftovers
case m of
@@ -758,3 +955,22 @@ getBin getsome get leftover = go leftover (runGetIncremental get)
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
+
+--------------------------------------------------------------------------------
+-- * 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
=====================================
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,26 @@ 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
+ handleRemoteCall pipe hook $ \(msg :: Message a) -> do
- discardCtrlC
+ 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
+ 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 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
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop doneRef
handleCustom tag payload = do
mresp <- customHandler tag payload
@@ -89,17 +89,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 :: forall a. (Binary a, Show a) => IO a -> IO (QResult a)
wrapRunTH 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 RunTHDone
case r of
Left e
| Just (GHCiQException _ err) <- fromException e -> do
=====================================
libraries/ghci/ghci.cabal.in
=====================================
@@ -95,7 +95,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
=====================================
@@ -144,7 +144,7 @@ customMessage :: (Show a, Bin.Binary a) => Pipe -> ClientCommand a -> IO a
customMessage pipe c = do
let payload = encodeLazy (Some @Bin.Binary c)
putStrLn $ "Sending: " ++ show c
- respBytes <- remoteCall pipe (CustomMessage customTag payload)
+ Right respBytes <- remoteCall pipe (CustomMessage customTag payload)
case decodeLazy respBytes of
Left err -> error $ "Decode error: " ++ err
Right res -> pure res
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/e8d943b729a2471c1f9362bf0906398…
--
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/e8d943b729a2471c1f9362bf0906398…
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
1
0
[Git][ghc/ghc][wip/romes/async-rpc] Asynchronous RPC with external interpreter
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
12 Oct '26
Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC
Commits:
9c5a83ec by Rodrigo Mesquita at 2026-10-12T18:53:49+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.
- - - - -
10 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/Server.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 []
=====================================
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,9 @@ 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
+ , RemoteCallFailed(..)
)
where
@@ -27,97 +20,67 @@ 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
-- -----------------------------------------------------------------------------
-- 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.
+--
+-- Throws 'RemoteCallFailed' if the remote call threw an exception in the
+-- external interpreter
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 = unwrapRight =<<
+ remoteCall (interpPipe proc) msg
+ `catchException` \(e :: SomeException) -> handleInterpProcessFailure proc e
+ where
+ proc = instProcess i
+ unwrapRight :: Either SomeException b -> IO b
+ unwrapRight (Right a) = pure a
+ unwrapRight (Left er) = do
+ throwIO (RemoteCallFailed er)
+
+-- | 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
+ -> Message a
+ -> (forall b. Binary b => THMessage b -> IO (Bool, b))
+ -- ^ Handle one message. Return True to end the loop.
+ -> IO a
+sendTHComputation i msg reqHandler = mask $ \restore -> do
+ -- inline "withAsync/wait"
+ var <- newEmptyMVar
+ ht <- forkFinally (restore handlerLoop) (putMVar var)
+ restore (sendMessage i msg <* takeMVar var)
+ `finally` cancel ht var
+ where
+ proc = instProcess i
+ handleOne doneRef = do
+ handleRemoteTHCall (interpPipe proc) $ \req -> do
+ (stop, res) <- reqHandler req
+ writeIORef doneRef stop
+ return res
+ handlerLoop = do
+ doneRef <- newIORef False
+ let loop = do
+ handleOne doneRef
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop
+ in loop
+ cancel ht var = uninterruptibleMask_ (throwTo ht CancelTHHandler <* takeMVar var)
handleInterpProcessFailure :: InterpProcess -> SomeException -> IO a
handleInterpProcessFailure i e = do
@@ -130,3 +93,17 @@ handleInterpProcessFailure i e = do
terminateProcess hdl
_ <- waitForProcess hdl
throw e
+
+-- | The external interpreter server threw an exception while executing this
+-- remote call. The ext. interp is still live and ready to take further
+-- commands (such as 'Shutdown').
+data RemoteCallFailed = RemoteCallFailed SomeException deriving Show
+instance Exception RemoteCallFailed where
+ displayException (RemoteCallFailed e)
+ = "A remote call in the external interpreter threw an exception:\n"
+ ++ displayException 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 (RunModFinalizers 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 (RunTH 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,10 +23,22 @@ module GHCi.Message
, ResumeContext(..)
, QState(..)
, getMessage, putMessage, getTHMessage, putTHMessage
- , Pipe, mkPipeFromHandles, mkPipeFromContinuations, remoteCall, remoteTHCall, readPipe, writePipe
+
+ , Pipe, mkPipeFromHandles, mkPipeFromContinuations
+ , remoteCall, remoteTHCall
+ , handleRemoteCall, handleRemoteTHCall
+ , RemoteProcessException(..)
+
, BreakModule
, BreakUnitId
, LoadedDLL
+
+ -- * Low-level interface for async bidirectional RPC
+ , withAsyncReq
+ , withAsyncHandler
+ -- ** Consume `reply` exactly once
+ , WriteReply -- abstract
+ , writeReply
) where
import Prelude -- See note [Why do we import Prelude here?]
@@ -45,6 +58,7 @@ 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
@@ -63,6 +77,8 @@ import Data.Dynamic
import Data.Typeable (TypeRep)
import Data.IORef
import Data.Map (Map)
+import Data.IntMap (IntMap)
+import qualified Data.IntMap as IntMap
import Foreign
import GHC.Generics
import GHC.Stack.CCS
@@ -71,6 +87,11 @@ import qualified GHC.Boot.TH.Monad as TH
import System.Exit
import System.IO
import System.IO.Error
+import Control.Monad
+import Control.Concurrent.STM
+import Data.Bifunctor
+import GHC.Base (State#)
+import GHC.Exts (RealWorld)
-- -----------------------------------------------------------------------------
-- The RPC protocol between GHC and the interactive server
@@ -693,9 +714,11 @@ serializeBCOs rbcos = parMap doChunk (chunkList 100 rbcos)
-- | An opaque pipe for bidirectional binary data transmission.
data Pipe = Pipe
- { getSome :: !(IO ByteString)
- , putAll :: !(B.Builder -> IO ())
- , pipeLeftovers :: !(IORef (Maybe ByteString))
+ { getSome :: !(IO ByteString)
+ , putAll :: !(B.Builder -> IO ())
+ , pipeLeftovers :: !(IORef (Maybe ByteString))
+ , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent access to the stream
+ , pipeAsyncMngr :: !AsyncManager
}
-- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write.
@@ -705,30 +728,204 @@ mkPipeFromHandles pipeRead pipeWrite = do
putAll b = do
B.hPutBuilder pipeWrite b
hFlush pipeWrite
- pipeLeftovers <- newIORef Nothing
- pure $ Pipe { getSome, putAll, pipeLeftovers }
+ mkPipeFromContinuations getSome putAll
-- | 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 }
-
+mkPipeFromContinuations getSome putAll = mdo
+ pipeLeftovers <- newIORef Nothing
+ pipeLock <- newMVar ()
+ mdo
+ pipeAsyncMngr <- newAsyncManager p
+ let p = Pipe { .. }
+ pure p
+
+-- | 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
+remoteCall pipe msg = either throwIO (pure . runGet get)
+ =<< withAsyncReq pipe (putMessage msg)
+
+-- | 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 =
+ withAsyncHandler pipe getMessage \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
+remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
+remoteTHCall pipe msg = either throwIO (pure . runGet get)
+ =<< withAsyncReq pipe (putTHMessage msg)
+
+-- | Handle a 'THMessage' request on the host and reply to the interpreter.
+handleRemoteTHCall
+ :: Pipe -> (forall a. (Show a, Binary a) => THMessage a -> IO a) -> IO ()
+handleRemoteTHCall pipe k =
+ withAsyncHandler pipe getTHMessage \writer (THMsg msg) -> IO \s0 ->
+ case k msg |#> s0 of -- unIO for linearity checking
+ (# s1, a #) -> writeReply writer (put a) |#> s1
+
+--------------------------------------------------------------------------------
+-- * Matching requests with async replies
+--------------------------------------------------------------------------------
+
+-- | Send an asynchronous request and block waiting for a matching reply
+withAsyncReq :: Pipe -> Put -> IO (Either SomeException LB.ByteString)
+withAsyncReq pipe msg = do
+
+ -- Mark pending request
+ uq <- freshReqId (pipeAsyncMngr pipe)
+ wv <- newEmptyMVar
+ atomicModifyIORef' (pendingSent (pipeAsyncMngr pipe)) $
+ \m -> (IntMap.insert uq wv m, ())
+
+ -- Write request
+ writePipe pipe (Request uq (runPut msg))
+
+ -- Block waiting for reply
+ takeMVar wv
+
+-- | 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 'withAsyncHandler'.
+withAsyncHandler :: Pipe -> Get a -> (WriteReply Put %1 -> a -> IO r) -> IO r
+withAsyncHandler pipe getit k = do
+ ack <- newEmptyMVar
+
+ -- Block waiting for request
+ (uq, bs) <- atomically (readTQueue (pendingRecv (pipeAsyncMngr pipe)))
+
+ let reply p = do
+ writePipe pipe (Reply uq (bimap toSerializableException runPut p))
+ putMVar ack ()
+
+ -- Run the continuation which must reply exactly once with 'writeReply'
+ k (WriteReply reply) (runGet getit bs)
+
+ -- 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
+
+-- | 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
+
+-- ** Async matching "internals" -----------------------------------------------
+
+data AsyncManager = AsyncManager
+ { pendingSent :: !(IORef (IntMap (MVar (Either SomeException LB.ByteString))))
+ -- ^ When a request is sent, keep track that we're expecting a reply. The
+ -- MVar is signaled when the reply arrives. The exception can originate
+ -- from the host process, or from the remote one (with 'RemoteProcessException').
+ , pendingRecv :: !(TQueue (CorrelationId, LB.ByteString))
+ -- ^ A queue of pending received requests. When a request is received,
+ -- queue it here. A request handler will pop the next request from it.
+ -- (Nothing except the 'drainPipe' thread should read from the pipe).
+ --
+ -- Seq contract: Dequeue from left, enqueue on right
+ , asyncDrainTid :: !ThreadId
+ , asyncNextReq :: !(IORef Int)
+ }
-writePipe :: Pipe -> Put -> IO ()
-writePipe Pipe{..} put = putAll $ execPut put
+type CorrelationId = Int
-remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
-remoteTHCall pipe msg = do
- writePipe pipe (putTHMessage msg)
- readPipe pipe get
+-- | 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
+data IddMsg = Request !CorrelationId LB.ByteString
+ | Reply !CorrelationId (Either SerializableException LB.ByteString)
+
+instance Binary IddMsg where
+ put (Request i bs) = putWord8 0 >> put i >> put bs
+ put (Reply i bs) = putWord8 1 >> put i >> put bs
+ get = do
+ b <- getWord8
+ case b of
+ 0 -> Request <$> get <*> get
+ 1 -> Reply <$> get <*> get
+ _ -> error $ "impossible: Binary IddMsg - " ++ show b
+
+newAsyncManager :: Pipe -> IO AsyncManager
+newAsyncManager pipe = do
+ asyncNextReq <- newIORef 0
+ pendingSent <- newIORef IntMap.empty
+ pendingRecv <- newTQueueIO
+
+ mdo
+ -- Fork thread to drain read end.
+ asyncDrainTid <- forkIO (drainPipe pipe am)
+ -- todo: proper thread clean up and no leaks
-readPipe :: Pipe -> Get a -> IO a
-readPipe Pipe{..} get = do
+ 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:
+drainPipe :: Pipe -> AsyncManager -> IO ()
+drainPipe p am = flip catch cancelPending $ forever $ do
+ readPipe p >>= \case
+ -- Receive a reply to a pending notification
+ Reply uq bs -> do
+ wv <- atomicModifyIORef' (pendingSent am) $ \m -> do
+ case IntMap.lookup uq m of
+ Nothing -> error "drainPipe: Received reply to no pending request"
+ Just wv -> (IntMap.delete uq m, wv)
+ putMVar wv (bimap (toException . RemoteProcessException) id bs) -- signal matching request
+ -- Receive a request and queue it
+ Request uq bs ->
+ atomically $ writeTQueue (pendingRecv am) (uq,bs)
+ where
+ cancelPending (e::SomeException) = do
+ pendings <- readIORef (pendingSent am)
+ mapM_ (\p -> putMVar p (Left e {-host exception-})) pendings
+
+freshReqId :: AsyncManager -> IO CorrelationId
+freshReqId AsyncManager{asyncNextReq} =
+ atomicModifyIORef' asyncNextReq $ \i -> (i+1, i)
+
+--------------------------------------------------------------------------------
+-- * 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'.
+readPipe :: Pipe -> IO IddMsg
+readPipe Pipe{..} = do
leftovers <- readIORef pipeLeftovers
m <- getBin getSome get leftovers
case m of
@@ -758,3 +955,22 @@ getBin getsome get leftover = go leftover (runGetIncremental get)
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
+
+--------------------------------------------------------------------------------
+-- * 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
=====================================
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,26 @@ 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
+ handleRemoteCall pipe hook $ \(msg :: Message a) -> do
- discardCtrlC
+ 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
+ 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 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
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop doneRef
handleCustom tag payload = do
mresp <- customHandler tag payload
@@ -89,17 +89,24 @@ 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 :: forall a. (Binary a, Show a) => IO a -> IO (QResult a)
wrapRunTH 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 RunTHDone >>= \case
+ Right () -> pure ()
+ Left err -> throwIO (GHCiQUncaughtExceptionOnHost err)
case r of
Left e
| Just (GHCiQException _ err) <- fromException e -> do
=====================================
libraries/ghci/ghci.cabal.in
=====================================
@@ -95,7 +95,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
=====================================
@@ -144,7 +144,7 @@ customMessage :: (Show a, Bin.Binary a) => Pipe -> ClientCommand a -> IO a
customMessage pipe c = do
let payload = encodeLazy (Some @Bin.Binary c)
putStrLn $ "Sending: " ++ show c
- respBytes <- remoteCall pipe (CustomMessage customTag payload)
+ Right respBytes <- remoteCall pipe (CustomMessage customTag payload)
case decodeLazy respBytes of
Left err -> error $ "Decode error: " ++ err
Right res -> pure res
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/9c5a83ec5e65bb00ad72aba574c4113…
--
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/9c5a83ec5e65bb00ad72aba574c4113…
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
1
0
[Git][ghc/ghc][wip/romes/async-rpc] Asynchronous RPC with external interpreter
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
12 Oct '26
Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC
Commits:
414a4710 by Rodrigo Mesquita at 2026-10-12T18:41:39+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 reading and writing, it
doesn't need nor should lock the full remote procedure call.
- - - - -
11 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/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 []
=====================================
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,9 @@ 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
+ , RemoteCallFailed(..)
)
where
@@ -27,97 +20,67 @@ 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
-- -----------------------------------------------------------------------------
-- 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.
+--
+-- Throws 'RemoteCallFailed' if the remote call threw an exception in the
+-- external interpreter
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 = unwrapRight =<<
+ remoteCall (interpPipe proc) msg
+ `catchException` \(e :: SomeException) -> handleInterpProcessFailure proc e
+ where
+ proc = instProcess i
+ unwrapRight :: Either SomeException b -> IO b
+ unwrapRight (Right a) = pure a
+ unwrapRight (Left er) = do
+ throwIO (RemoteCallFailed er)
+
+-- | 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
+ -> Message a
+ -> (forall b. Binary b => THMessage b -> IO (Bool, b))
+ -- ^ Handle one message. Return True to end the loop.
+ -> IO a
+sendTHComputation i msg reqHandler = mask $ \restore -> do
+ -- inline "withAsync/wait"
+ var <- newEmptyMVar
+ ht <- forkFinally (restore handlerLoop) (putMVar var)
+ restore (sendMessage i msg <* takeMVar var)
+ `finally` cancel ht var
+ where
+ proc = instProcess i
+ handleOne doneRef = do
+ handleRemoteTHCall (interpPipe proc) $ \req -> do
+ (stop, res) <- reqHandler req
+ writeIORef doneRef stop
+ return res
+ handlerLoop = do
+ doneRef <- newIORef False
+ let loop = do
+ handleOne doneRef
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop
+ in loop
+ cancel ht var = uninterruptibleMask_ (throwTo ht CancelTHHandler <* takeMVar var)
handleInterpProcessFailure :: InterpProcess -> SomeException -> IO a
handleInterpProcessFailure i e = do
@@ -130,3 +93,17 @@ handleInterpProcessFailure i e = do
terminateProcess hdl
_ <- waitForProcess hdl
throw e
+
+-- | The external interpreter server threw an exception while executing this
+-- remote call. The ext. interp is still live and ready to take further
+-- commands (such as 'Shutdown').
+data RemoteCallFailed = RemoteCallFailed SomeException deriving Show
+instance Exception RemoteCallFailed where
+ displayException (RemoteCallFailed e)
+ = "A remote call in the external interpreter threw an exception:\n"
+ ++ displayException 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 (RunModFinalizers 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 (RunTH 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,10 +23,22 @@ module GHCi.Message
, ResumeContext(..)
, QState(..)
, getMessage, putMessage, getTHMessage, putTHMessage
- , Pipe, mkPipeFromHandles, mkPipeFromContinuations, remoteCall, remoteTHCall, readPipe, writePipe
+
+ , Pipe, mkPipeFromHandles, mkPipeFromContinuations
+ , remoteCall, remoteTHCall
+ , handleRemoteCall, handleRemoteTHCall
+ , RemoteProcessException(..)
+
, BreakModule
, BreakUnitId
, LoadedDLL
+
+ -- * Low-level interface for async bidirectional RPC
+ , withAsyncReq
+ , withAsyncHandler
+ -- ** Consume `reply` exactly once
+ , WriteReply -- abstract
+ , writeReply
) where
import Prelude -- See note [Why do we import Prelude here?]
@@ -45,6 +58,7 @@ 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
@@ -63,6 +77,8 @@ import Data.Dynamic
import Data.Typeable (TypeRep)
import Data.IORef
import Data.Map (Map)
+import Data.IntMap (IntMap)
+import qualified Data.IntMap as IntMap
import Foreign
import GHC.Generics
import GHC.Stack.CCS
@@ -71,6 +87,11 @@ import qualified GHC.Boot.TH.Monad as TH
import System.Exit
import System.IO
import System.IO.Error
+import Control.Monad
+import Control.Concurrent.STM
+import Data.Bifunctor
+import GHC.Base (State#)
+import GHC.Exts (RealWorld)
-- -----------------------------------------------------------------------------
-- The RPC protocol between GHC and the interactive server
@@ -693,9 +714,11 @@ serializeBCOs rbcos = parMap doChunk (chunkList 100 rbcos)
-- | An opaque pipe for bidirectional binary data transmission.
data Pipe = Pipe
- { getSome :: !(IO ByteString)
- , putAll :: !(B.Builder -> IO ())
- , pipeLeftovers :: !(IORef (Maybe ByteString))
+ { getSome :: !(IO ByteString)
+ , putAll :: !(B.Builder -> IO ())
+ , pipeLeftovers :: !(IORef (Maybe ByteString))
+ , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent access to the stream
+ , pipeAsyncMngr :: !AsyncManager
}
-- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write.
@@ -705,30 +728,204 @@ mkPipeFromHandles pipeRead pipeWrite = do
putAll b = do
B.hPutBuilder pipeWrite b
hFlush pipeWrite
- pipeLeftovers <- newIORef Nothing
- pure $ Pipe { getSome, putAll, pipeLeftovers }
+ mkPipeFromContinuations getSome putAll
-- | 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 }
+mkPipeFromContinuations getSome putAll = mdo
+ pipeLeftovers <- newIORef Nothing
+ pipeLock <- newMVar ()
+ mdo
+ pipeAsyncMngr <- newAsyncManager p
+ let p = Pipe { .. }
+ pure p
+
+-- | 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 (Either SomeException a)
+remoteCall pipe msg = bimap id (runGet get)
+ <$> withAsyncReq pipe (putMessage msg)
+
+-- | 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 =
+ withAsyncHandler pipe getMessage \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
+remoteTHCall :: Binary a => Pipe -> THMessage a -> IO (Either SomeException a)
+remoteTHCall pipe msg = bimap id (runGet get)
+ <$> withAsyncReq pipe (putTHMessage msg)
+
+-- | Handle a 'THMessage' request on the host and reply to the interpreter.
+handleRemoteTHCall
+ :: Pipe -> (forall a. (Show a, Binary a) => THMessage a -> IO a) -> IO ()
+handleRemoteTHCall pipe k =
+ withAsyncHandler pipe getTHMessage \writer (THMsg msg) -> IO \s0 ->
+ case k msg |#> s0 of -- unIO for linearity checking
+ (# s1, a #) -> writeReply writer (put a) |#> s1
+
+--------------------------------------------------------------------------------
+-- * Matching requests with async replies
+--------------------------------------------------------------------------------
+
+-- | Send an asynchronous request and block waiting for a matching reply
+withAsyncReq :: Pipe -> Put -> IO (Either SomeException LB.ByteString)
+withAsyncReq pipe msg = do
+
+ -- Mark pending request
+ uq <- freshReqId (pipeAsyncMngr pipe)
+ wv <- newEmptyMVar
+ atomicModifyIORef' (pendingSent (pipeAsyncMngr pipe)) $
+ \m -> (IntMap.insert uq wv m, ())
+
+ -- Write request
+ writePipe pipe (Request uq (runPut msg))
+
+ -- Block waiting for reply
+ takeMVar wv
+
+-- | 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 'withAsyncHandler'.
+withAsyncHandler :: Pipe -> Get a -> (WriteReply Put %1 -> a -> IO r) -> IO r
+withAsyncHandler pipe getit k = do
+ ack <- newEmptyMVar
+
+ -- Block waiting for request
+ (uq, bs) <- atomically (readTQueue (pendingRecv (pipeAsyncMngr pipe)))
+
+ let reply p = do
+ writePipe pipe (Reply uq (bimap toSerializableException runPut p))
+ putMVar ack ()
+
+ -- Run the continuation which must reply exactly once with 'writeReply'
+ k (WriteReply reply) (runGet getit bs)
+
+ -- 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
+
+-- | 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
+
+-- ** Async matching "internals" -----------------------------------------------
+
+data AsyncManager = AsyncManager
+ { pendingSent :: !(IORef (IntMap (MVar (Either SomeException LB.ByteString))))
+ -- ^ When a request is sent, keep track that we're expecting a reply. The
+ -- MVar is signaled when the reply arrives. The exception can originate
+ -- from the host process, or from the remote one (with 'RemoteProcessException').
+ , pendingRecv :: !(TQueue (CorrelationId, LB.ByteString))
+ -- ^ A queue of pending received requests. When a request is received,
+ -- queue it here. A request handler will pop the next request from it.
+ -- (Nothing except the 'drainPipe' thread should read from the pipe).
+ --
+ -- Seq contract: Dequeue from left, enqueue on right
+ , asyncDrainTid :: !ThreadId
+ , asyncNextReq :: !(IORef Int)
+ }
+
+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
+data IddMsg = Request !CorrelationId LB.ByteString
+ | Reply !CorrelationId (Either SerializableException LB.ByteString)
+
+instance Binary IddMsg where
+ put (Request i bs) = putWord8 0 >> put i >> put bs
+ put (Reply i bs) = putWord8 1 >> put i >> put bs
+ get = do
+ b <- getWord8
+ case b of
+ 0 -> Request <$> get <*> get
+ 1 -> Reply <$> get <*> get
+ _ -> error $ "impossible: Binary IddMsg - " ++ show b
-remoteCall :: Binary a => Pipe -> Message a -> IO a
-remoteCall pipe msg = do
- writePipe pipe (putMessage msg)
- readPipe pipe get
+newAsyncManager :: Pipe -> IO AsyncManager
+newAsyncManager pipe = do
+ asyncNextReq <- newIORef 0
+ pendingSent <- newIORef IntMap.empty
+ pendingRecv <- newTQueueIO
-writePipe :: Pipe -> Put -> IO ()
-writePipe Pipe{..} put = putAll $ execPut put
+ mdo
+ -- Fork thread to drain read end.
+ asyncDrainTid <- forkIO (drainPipe pipe am)
+ -- todo: proper thread clean up and no leaks
-remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
-remoteTHCall pipe msg = do
- writePipe pipe (putTHMessage msg)
- readPipe pipe get
+ let am = AsyncManager{..}
+ pure am
-readPipe :: Pipe -> Get a -> IO a
-readPipe Pipe{..} get = do
+-- | 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:
+drainPipe :: Pipe -> AsyncManager -> IO ()
+drainPipe p am = flip catch cancelPending $ forever $ do
+ readPipe p >>= \case
+ -- Receive a reply to a pending notification
+ Reply uq bs -> do
+ wv <- atomicModifyIORef' (pendingSent am) $ \m -> do
+ case IntMap.lookup uq m of
+ Nothing -> error "drainPipe: Received reply to no pending request"
+ Just wv -> (IntMap.delete uq m, wv)
+ putMVar wv (bimap (toException . RemoteProcessException) id bs) -- signal matching request
+ -- Receive a request and queue it
+ Request uq bs ->
+ atomically $ writeTQueue (pendingRecv am) (uq,bs)
+ where
+ cancelPending (e::SomeException) = do
+ pendings <- readIORef (pendingSent am)
+ mapM_ (\p -> putMVar p (Left e {-host exception-})) pendings
+
+freshReqId :: AsyncManager -> IO CorrelationId
+freshReqId AsyncManager{asyncNextReq} =
+ atomicModifyIORef' asyncNextReq $ \i -> (i+1, i)
+
+--------------------------------------------------------------------------------
+-- * 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'.
+readPipe :: Pipe -> IO IddMsg
+readPipe Pipe{..} = do
leftovers <- readIORef pipeLeftovers
m <- getBin getSome get leftovers
case m of
@@ -758,3 +955,22 @@ getBin getsome get leftover = go leftover (runGetIncremental get)
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
+
+--------------------------------------------------------------------------------
+-- * 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
=====================================
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,26 @@ 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
+ handleRemoteCall pipe hook $ \(msg :: Message a) -> do
- discardCtrlC
+ 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
+ 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 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
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop doneRef
handleCustom tag payload = do
mresp <- customHandler tag payload
@@ -89,17 +89,24 @@ 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 :: forall a. (Binary a, Show a) => IO a -> IO (QResult a)
wrapRunTH 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 RunTHDone >>= \case
+ Right () -> pure ()
+ Left err -> throwIO (GHCiQUncaughtExceptionOnHost err)
case r of
Left e
| Just (GHCiQException _ err) <- fromException e -> do
=====================================
libraries/ghci/GHCi/TH.hs
=====================================
@@ -10,6 +10,7 @@ module GHCi.TH
, runModFinalizerRefs
, runTH
, GHCiQException(..)
+ , GHCiQUncaughtExceptionOnHost(..)
) where
{- Note [Remote Template Haskell]
@@ -158,8 +159,13 @@ ghcCmd m = GHCiQ $ \sRef -> do
s <- readIORef sRef
r <- remoteTHCall (qsPipe s) m
case r of
- THException str -> throwIO (GHCiQException s str)
- THComplete res -> return res
+ Right (THException str) -> throwIO (GHCiQException s str)
+ Right (THComplete res) -> return res
+ Left e -> liftIO $ throwIO (GHCiQUncaughtExceptionOnHost e)
+
+-- | An exception was thrown on the host while processing a 'remoteTHCall'
+data GHCiQUncaughtExceptionOnHost = GHCiQUncaughtExceptionOnHost SomeException deriving Show
+instance Exception GHCiQUncaughtExceptionOnHost
instance MonadIO GHCiQ where
liftIO m = GHCiQ $ \_ -> m
@@ -189,9 +195,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) StartRecover >>= either (throwIO . GHCiQUncaughtExceptionOnHost) pure
e <- try $ unmask $ runGHCiQ (runQinGHCiQ a <* ghcCmd FailIfErrs) sRef
- remoteTHCall (qsPipe s) (EndRecover (isLeft e))
+ remoteTHCall (qsPipe s) (EndRecover (isLeft e)) >>= either (throwIO . GHCiQUncaughtExceptionOnHost) pure
case e of
Left GHCiQException{} ->
runGHCiQ (runQinGHCiQ h) sRef
=====================================
libraries/ghci/ghci.cabal.in
=====================================
@@ -95,7 +95,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
=====================================
@@ -144,7 +144,7 @@ customMessage :: (Show a, Bin.Binary a) => Pipe -> ClientCommand a -> IO a
customMessage pipe c = do
let payload = encodeLazy (Some @Bin.Binary c)
putStrLn $ "Sending: " ++ show c
- respBytes <- remoteCall pipe (CustomMessage customTag payload)
+ Right respBytes <- remoteCall pipe (CustomMessage customTag payload)
case decodeLazy respBytes of
Left err -> error $ "Decode error: " ++ err
Right res -> pure res
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/414a471021f388de752de7ecc043f06…
--
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/414a471021f388de752de7ecc043f06…
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
1
0
[Git][ghc/ghc][wip/romes/async-rpc] Asynchronous RPC with external interpreter
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
12 Oct '26
Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC
Commits:
11710a3d by Rodrigo Mesquita at 2026-10-12T18:12:00+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 reading and writing, it
doesn't need nor should lock the full remote procedure call.
- - - - -
11 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/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 []
=====================================
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,9 @@ 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
+ , RemoteCallFailed(..)
)
where
@@ -27,97 +20,67 @@ 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
-- -----------------------------------------------------------------------------
-- 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.
+--
+-- Throws 'RemoteCallFailed' if the remote call threw an exception in the
+-- external interpreter
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 = unwrapRight =<<
+ remoteCall (interpPipe proc) msg
+ `catchException` \(e :: SomeException) -> handleInterpProcessFailure proc e
+ where
+ proc = instProcess i
+ unwrapRight :: Either SomeException b -> IO b
+ unwrapRight (Right a) = pure a
+ unwrapRight (Left er) = do
+ throwIO (RemoteCallFailed er)
+
+-- | 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
+ -> Message a
+ -> (forall b. Binary b => THMessage b -> IO (Bool, b))
+ -- ^ Handle one message. Return True to end the loop.
+ -> IO a
+sendTHComputation i msg reqHandler = mask $ \restore -> do
+ -- inline "withAsync/wait"
+ var <- newEmptyMVar
+ ht <- forkFinally (restore handlerLoop) (putMVar var)
+ restore (sendMessage i msg <* takeMVar var)
+ `finally` cancel ht var
+ where
+ proc = instProcess i
+ handleOne doneRef = do
+ handleRemoteTHCall (interpPipe proc) $ \req -> do
+ (stop, res) <- reqHandler req
+ writeIORef doneRef stop
+ return res
+ handlerLoop = do
+ doneRef <- newIORef False
+ let loop = do
+ handleOne doneRef
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop
+ in loop
+ cancel ht var = uninterruptibleMask_ (throwTo ht CancelTHHandler <* takeMVar var)
handleInterpProcessFailure :: InterpProcess -> SomeException -> IO a
handleInterpProcessFailure i e = do
@@ -130,3 +93,17 @@ handleInterpProcessFailure i e = do
terminateProcess hdl
_ <- waitForProcess hdl
throw e
+
+-- | The external interpreter server threw an exception while executing this
+-- remote call. The ext. interp is still live and ready to take further
+-- commands (such as 'Shutdown').
+data RemoteCallFailed = RemoteCallFailed SomeException deriving Show
+instance Exception RemoteCallFailed where
+ displayException (RemoteCallFailed e)
+ = "A remote call in the external interpreter threw an exception:\n"
+ ++ displayException 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 (RunModFinalizers 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 (RunTH 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,10 +23,21 @@ module GHCi.Message
, ResumeContext(..)
, QState(..)
, getMessage, putMessage, getTHMessage, putTHMessage
- , Pipe, mkPipeFromHandles, mkPipeFromContinuations, remoteCall, remoteTHCall, readPipe, writePipe
+
+ , Pipe, mkPipeFromHandles, mkPipeFromContinuations
+ , remoteCall, remoteTHCall
+ , handleRemoteCall, handleRemoteTHCall
+
, BreakModule
, BreakUnitId
, LoadedDLL
+
+ -- * Low-level interface for async bidirectional RPC
+ , withAsyncReq
+ , withAsyncHandler
+ -- ** Consume `reply` exactly once
+ , WriteReply -- abstract
+ , writeReply
) where
import Prelude -- See note [Why do we import Prelude here?]
@@ -45,6 +57,7 @@ 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
@@ -63,6 +76,8 @@ import Data.Dynamic
import Data.Typeable (TypeRep)
import Data.IORef
import Data.Map (Map)
+import Data.IntMap (IntMap)
+import qualified Data.IntMap as IntMap
import Foreign
import GHC.Generics
import GHC.Stack.CCS
@@ -71,6 +86,11 @@ import qualified GHC.Boot.TH.Monad as TH
import System.Exit
import System.IO
import System.IO.Error
+import Control.Monad
+import Control.Concurrent.STM
+import Data.Bifunctor
+import GHC.Base (State#)
+import GHC.Exts (RealWorld)
-- -----------------------------------------------------------------------------
-- The RPC protocol between GHC and the interactive server
@@ -693,9 +713,11 @@ serializeBCOs rbcos = parMap doChunk (chunkList 100 rbcos)
-- | An opaque pipe for bidirectional binary data transmission.
data Pipe = Pipe
- { getSome :: !(IO ByteString)
- , putAll :: !(B.Builder -> IO ())
- , pipeLeftovers :: !(IORef (Maybe ByteString))
+ { getSome :: !(IO ByteString)
+ , putAll :: !(B.Builder -> IO ())
+ , pipeLeftovers :: !(IORef (Maybe ByteString))
+ , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent access to the stream
+ , pipeAsyncMngr :: !AsyncManager
}
-- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write.
@@ -705,30 +727,193 @@ mkPipeFromHandles pipeRead pipeWrite = do
putAll b = do
B.hPutBuilder pipeWrite b
hFlush pipeWrite
- pipeLeftovers <- newIORef Nothing
- pure $ Pipe { getSome, putAll, pipeLeftovers }
+ mkPipeFromContinuations getSome putAll
-- | 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 }
+mkPipeFromContinuations getSome putAll = mdo
+ pipeLeftovers <- newIORef Nothing
+ pipeLock <- newMVar ()
+ mdo
+ pipeAsyncMngr <- newAsyncManager p
+ let p = Pipe { .. }
+ pure p
+
+-- | 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 (Either SomeException a)
+remoteCall pipe msg = bimap fromSerializableException id . runGet get
+ <$> withAsyncReq pipe (putMessage msg)
+
+-- | 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 =
+ withAsyncHandler pipe getMessage \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
+remoteTHCall :: Binary a => Pipe -> THMessage a -> IO (Either SomeException a)
+remoteTHCall pipe msg = bimap fromSerializableException id . runGet get
+ <$> withAsyncReq pipe (putTHMessage msg)
+
+-- | Handle a 'THMessage' request on the host and reply to the interpreter.
+handleRemoteTHCall
+ :: Pipe -> (forall a. (Show a, Binary a) => THMessage a -> IO a) -> IO ()
+handleRemoteTHCall pipe k =
+ withAsyncHandler pipe getTHMessage \writer (THMsg msg) -> IO \s0 ->
+ case k msg |#> s0 of -- unIO for linearity checking
+ (# s1, a #) -> writeReply writer (put a) |#> s1
+
+--------------------------------------------------------------------------------
+-- * Matching requests with async replies
+--------------------------------------------------------------------------------
+
+-- | Send an asynchronous request and block waiting for a matching reply
+withAsyncReq :: Pipe -> Put -> IO LB.ByteString
+withAsyncReq pipe msg = do
+
+ -- Mark pending request
+ uq <- freshReqId (pipeAsyncMngr pipe)
+ wv <- newEmptyMVar
+ atomicModifyIORef' (pendingSent (pipeAsyncMngr pipe)) $
+ \m -> (IntMap.insert uq wv m, ())
+
+ -- Write request
+ writePipe pipe (Request uq (runPut msg))
+
+ -- Block waiting for reply
+ takeMVar wv
+
+-- | 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 'withAsyncHandler'.
+withAsyncHandler :: Pipe -> Get a -> (WriteReply Put %1 -> a -> IO r) -> IO r
+withAsyncHandler pipe getit k = do
+ ack <- newEmptyMVar
+
+ -- Block waiting for request
+ (uq, bs) <- atomically (readTQueue (pendingRecv (pipeAsyncMngr pipe)))
+
+ let reply p = do
+ writePipe pipe (Reply uq (runPut $ either (put . toSerializableException) id p))
+ putMVar ack ()
+
+ -- Run the continuation which must reply exactly once with 'writeReply'
+ k (WriteReply reply) (runGet getit bs)
+
+ -- 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
+
+-- ** Async matching "internals" -----------------------------------------------
+
+data AsyncManager = AsyncManager
+ { pendingSent :: !(IORef (IntMap (MVar LB.ByteString)))
+ -- ^ When a request is sent, keep track that we're expecting a reply. The
+ -- MVar is signaled when the reply arrives.
+ , pendingRecv :: !(TQueue (CorrelationId, LB.ByteString))
+ -- ^ A queue of pending received requests. When a request is received,
+ -- queue it here. A request handler will pop the next request from it.
+ -- (Nothing except the 'drainPipe' thread should read from the pipe).
+ --
+ -- Seq contract: Dequeue from left, enqueue on right
+ , asyncDrainTid :: !ThreadId
+ , asyncNextReq :: !(IORef Int)
+ }
+
+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 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
+data IddMsg = Request !CorrelationId LB.ByteString
+ | Reply !CorrelationId LB.ByteString
+
+instance Binary IddMsg where
+ put (Request i bs) = putWord8 0 >> put i >> put bs
+ put (Reply i bs) = putWord8 1 >> put i >> put bs
+ get = do
+ b <- getWord8
+ case b of
+ 0 -> Request <$> get <*> get
+ 1 -> Reply <$> get <*> get
+ _ -> error $ "impossible: Binary IddMsg - " ++ show b
-remoteCall :: Binary a => Pipe -> Message a -> IO a
-remoteCall pipe msg = do
- writePipe pipe (putMessage msg)
- readPipe pipe get
+newAsyncManager :: Pipe -> IO AsyncManager
+newAsyncManager pipe = do
+ asyncNextReq <- newIORef 0
+ pendingSent <- newIORef IntMap.empty
+ pendingRecv <- newTQueueIO
-writePipe :: Pipe -> Put -> IO ()
-writePipe Pipe{..} put = putAll $ execPut put
+ mdo
+ -- Fork thread to drain read end.
+ asyncDrainTid <- forkIO (drainPipe pipe am)
+ -- todo: proper thread clean up and no leaks
-remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
-remoteTHCall pipe msg = do
- writePipe pipe (putTHMessage msg)
- readPipe pipe get
+ let am = AsyncManager{..}
+ pure am
-readPipe :: Pipe -> Get a -> IO a
-readPipe Pipe{..} get = do
+-- | 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:
+drainPipe :: Pipe -> AsyncManager -> IO ()
+drainPipe p am = forever $ do
+ readPipe p >>= \case
+ -- Receive a reply to a pending notification
+ Reply uq bs -> do
+ wv <- atomicModifyIORef' (pendingSent am) $ \m -> do
+ case IntMap.lookup uq m of
+ Nothing -> error "drainPipe: Received reply to no pending request"
+ Just wv -> (IntMap.delete uq m, wv)
+ putMVar wv bs -- signal matching request
+ -- Receive a request and queue it
+ Request uq bs ->
+ atomically $ writeTQueue (pendingRecv am) (uq,bs)
+
+freshReqId :: AsyncManager -> IO CorrelationId
+freshReqId AsyncManager{asyncNextReq} =
+ atomicModifyIORef' asyncNextReq $ \i -> (i+1, i)
+
+--------------------------------------------------------------------------------
+-- * 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'.
+readPipe :: Pipe -> IO IddMsg
+readPipe Pipe{..} = do
leftovers <- readIORef pipeLeftovers
m <- getBin getSome get leftovers
case m of
@@ -758,3 +943,22 @@ getBin getsome get leftover = go leftover (runGetIncremental get)
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
+
+--------------------------------------------------------------------------------
+-- * 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
=====================================
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,26 @@ 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
+ handleRemoteCall pipe hook $ \(msg :: Message a) -> do
- discardCtrlC
+ 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
+ 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 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
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop doneRef
handleCustom tag payload = do
mresp <- customHandler tag payload
@@ -89,17 +89,24 @@ 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 :: forall a. (Binary a, Show a) => IO a -> IO (QResult a)
wrapRunTH 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 RunTHDone >>= \case
+ Right () -> pure ()
+ Left err -> throwIO (GHCiQUncaughtExceptionOnHost err)
case r of
Left e
| Just (GHCiQException _ err) <- fromException e -> do
=====================================
libraries/ghci/GHCi/TH.hs
=====================================
@@ -10,6 +10,7 @@ module GHCi.TH
, runModFinalizerRefs
, runTH
, GHCiQException(..)
+ , GHCiQUncaughtExceptionOnHost(..)
) where
{- Note [Remote Template Haskell]
@@ -158,8 +159,13 @@ ghcCmd m = GHCiQ $ \sRef -> do
s <- readIORef sRef
r <- remoteTHCall (qsPipe s) m
case r of
- THException str -> throwIO (GHCiQException s str)
- THComplete res -> return res
+ Right (THException str) -> throwIO (GHCiQException s str)
+ Right (THComplete res) -> return res
+ Left e -> liftIO $ throwIO (GHCiQUncaughtExceptionOnHost e)
+
+-- | An exception was thrown on the host while processing a 'remoteTHCall'
+data GHCiQUncaughtExceptionOnHost = GHCiQUncaughtExceptionOnHost SomeException deriving Show
+instance Exception GHCiQUncaughtExceptionOnHost
instance MonadIO GHCiQ where
liftIO m = GHCiQ $ \_ -> m
@@ -189,9 +195,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) StartRecover >>= either (throwIO . GHCiQUncaughtExceptionOnHost) pure
e <- try $ unmask $ runGHCiQ (runQinGHCiQ a <* ghcCmd FailIfErrs) sRef
- remoteTHCall (qsPipe s) (EndRecover (isLeft e))
+ remoteTHCall (qsPipe s) (EndRecover (isLeft e)) >>= either (throwIO . GHCiQUncaughtExceptionOnHost) pure
case e of
Left GHCiQException{} ->
runGHCiQ (runQinGHCiQ h) sRef
=====================================
libraries/ghci/ghci.cabal.in
=====================================
@@ -95,7 +95,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
=====================================
@@ -144,7 +144,7 @@ customMessage :: (Show a, Bin.Binary a) => Pipe -> ClientCommand a -> IO a
customMessage pipe c = do
let payload = encodeLazy (Some @Bin.Binary c)
putStrLn $ "Sending: " ++ show c
- respBytes <- remoteCall pipe (CustomMessage customTag payload)
+ Right respBytes <- remoteCall pipe (CustomMessage customTag payload)
case decodeLazy respBytes of
Left err -> error $ "Decode error: " ++ err
Right res -> pure res
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/11710a3df7bb1f1253741fc10750c2e…
--
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/11710a3df7bb1f1253741fc10750c2e…
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
1
0
[Git][ghc/ghc][wip/romes/async-rpc] Asynchronous RPC with external interpreter
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
by Rodrigo Mesquita (＠alt-romes) 12 Oct '26
12 Oct '26
Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC
Commits:
25ce96e3 by Rodrigo Mesquita at 2026-10-12T18:10:37+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 reading and writing, it
doesn't need nor should lock the full remote procedure call.
- - - - -
10 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/Server.hs
- libraries/ghci/GHCi/TH.hs
- libraries/ghci/ghci.cabal.in
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 []
=====================================
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,9 @@ 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
+ , RemoteCallFailed(..)
)
where
@@ -27,97 +20,67 @@ 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
-- -----------------------------------------------------------------------------
-- 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.
+--
+-- Throws 'RemoteCallFailed' if the remote call threw an exception in the
+-- external interpreter
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 = unwrapRight =<<
+ remoteCall (interpPipe proc) msg
+ `catchException` \(e :: SomeException) -> handleInterpProcessFailure proc e
+ where
+ proc = instProcess i
+ unwrapRight :: Either SomeException b -> IO b
+ unwrapRight (Right a) = pure a
+ unwrapRight (Left er) = do
+ throwIO (RemoteCallFailed er)
+
+-- | 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
+ -> Message a
+ -> (forall b. Binary b => THMessage b -> IO (Bool, b))
+ -- ^ Handle one message. Return True to end the loop.
+ -> IO a
+sendTHComputation i msg reqHandler = mask $ \restore -> do
+ -- inline "withAsync/wait"
+ var <- newEmptyMVar
+ ht <- forkFinally (restore handlerLoop) (putMVar var)
+ restore (sendMessage i msg <* takeMVar var)
+ `finally` cancel ht var
+ where
+ proc = instProcess i
+ handleOne doneRef = do
+ handleRemoteTHCall (interpPipe proc) $ \req -> do
+ (stop, res) <- reqHandler req
+ writeIORef doneRef stop
+ return res
+ handlerLoop = do
+ doneRef <- newIORef False
+ let loop = do
+ handleOne doneRef
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop
+ in loop
+ cancel ht var = uninterruptibleMask_ (throwTo ht CancelTHHandler <* takeMVar var)
handleInterpProcessFailure :: InterpProcess -> SomeException -> IO a
handleInterpProcessFailure i e = do
@@ -130,3 +93,17 @@ handleInterpProcessFailure i e = do
terminateProcess hdl
_ <- waitForProcess hdl
throw e
+
+-- | The external interpreter server threw an exception while executing this
+-- remote call. The ext. interp is still live and ready to take further
+-- commands (such as 'Shutdown').
+data RemoteCallFailed = RemoteCallFailed SomeException deriving Show
+instance Exception RemoteCallFailed where
+ displayException (RemoteCallFailed e)
+ = "A remote call in the external interpreter threw an exception:\n"
+ ++ displayException 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 (RunModFinalizers 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 (RunTH 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,10 +23,21 @@ module GHCi.Message
, ResumeContext(..)
, QState(..)
, getMessage, putMessage, getTHMessage, putTHMessage
- , Pipe, mkPipeFromHandles, mkPipeFromContinuations, remoteCall, remoteTHCall, readPipe, writePipe
+
+ , Pipe, mkPipeFromHandles, mkPipeFromContinuations
+ , remoteCall, remoteTHCall
+ , handleRemoteCall, handleRemoteTHCall
+
, BreakModule
, BreakUnitId
, LoadedDLL
+
+ -- * Low-level interface for async bidirectional RPC
+ , withAsyncReq
+ , withAsyncHandler
+ -- ** Consume `reply` exactly once
+ , WriteReply -- abstract
+ , writeReply
) where
import Prelude -- See note [Why do we import Prelude here?]
@@ -45,6 +57,7 @@ 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
@@ -63,6 +76,8 @@ import Data.Dynamic
import Data.Typeable (TypeRep)
import Data.IORef
import Data.Map (Map)
+import Data.IntMap (IntMap)
+import qualified Data.IntMap as IntMap
import Foreign
import GHC.Generics
import GHC.Stack.CCS
@@ -71,6 +86,11 @@ import qualified GHC.Boot.TH.Monad as TH
import System.Exit
import System.IO
import System.IO.Error
+import Control.Monad
+import Control.Concurrent.STM
+import Data.Bifunctor
+import GHC.Base (State#)
+import GHC.Exts (RealWorld)
-- -----------------------------------------------------------------------------
-- The RPC protocol between GHC and the interactive server
@@ -693,9 +713,11 @@ serializeBCOs rbcos = parMap doChunk (chunkList 100 rbcos)
-- | An opaque pipe for bidirectional binary data transmission.
data Pipe = Pipe
- { getSome :: !(IO ByteString)
- , putAll :: !(B.Builder -> IO ())
- , pipeLeftovers :: !(IORef (Maybe ByteString))
+ { getSome :: !(IO ByteString)
+ , putAll :: !(B.Builder -> IO ())
+ , pipeLeftovers :: !(IORef (Maybe ByteString))
+ , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent access to the stream
+ , pipeAsyncMngr :: !AsyncManager
}
-- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write.
@@ -705,30 +727,193 @@ mkPipeFromHandles pipeRead pipeWrite = do
putAll b = do
B.hPutBuilder pipeWrite b
hFlush pipeWrite
- pipeLeftovers <- newIORef Nothing
- pure $ Pipe { getSome, putAll, pipeLeftovers }
+ mkPipeFromContinuations getSome putAll
-- | 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 }
+mkPipeFromContinuations getSome putAll = mdo
+ pipeLeftovers <- newIORef Nothing
+ pipeLock <- newMVar ()
+ mdo
+ pipeAsyncMngr <- newAsyncManager p
+ let p = Pipe { .. }
+ pure p
+
+-- | 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 (Either SomeException a)
+remoteCall pipe msg = bimap fromSerializableException id . runGet get
+ <$> withAsyncReq pipe (putMessage msg)
+
+-- | 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 =
+ withAsyncHandler pipe getMessage \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
+remoteTHCall :: Binary a => Pipe -> THMessage a -> IO (Either SomeException a)
+remoteTHCall pipe msg = bimap fromSerializableException id . runGet get
+ <$> withAsyncReq pipe (putTHMessage msg)
+
+-- | Handle a 'THMessage' request on the host and reply to the interpreter.
+handleRemoteTHCall
+ :: Pipe -> (forall a. (Show a, Binary a) => THMessage a -> IO a) -> IO ()
+handleRemoteTHCall pipe k =
+ withAsyncHandler pipe getTHMessage \writer (THMsg msg) -> IO \s0 ->
+ case k msg |#> s0 of -- unIO for linearity checking
+ (# s1, a #) -> writeReply writer (put a) |#> s1
+
+--------------------------------------------------------------------------------
+-- * Matching requests with async replies
+--------------------------------------------------------------------------------
+
+-- | Send an asynchronous request and block waiting for a matching reply
+withAsyncReq :: Pipe -> Put -> IO LB.ByteString
+withAsyncReq pipe msg = do
+
+ -- Mark pending request
+ uq <- freshReqId (pipeAsyncMngr pipe)
+ wv <- newEmptyMVar
+ atomicModifyIORef' (pendingSent (pipeAsyncMngr pipe)) $
+ \m -> (IntMap.insert uq wv m, ())
+
+ -- Write request
+ writePipe pipe (Request uq (runPut msg))
+
+ -- Block waiting for reply
+ takeMVar wv
+
+-- | 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 'withAsyncHandler'.
+withAsyncHandler :: Pipe -> Get a -> (WriteReply Put %1 -> a -> IO r) -> IO r
+withAsyncHandler pipe getit k = do
+ ack <- newEmptyMVar
+
+ -- Block waiting for request
+ (uq, bs) <- atomically (readTQueue (pendingRecv (pipeAsyncMngr pipe)))
+
+ let reply p = do
+ writePipe pipe (Reply uq (runPut $ either (put . toSerializableException) id p))
+ putMVar ack ()
+
+ -- Run the continuation which must reply exactly once with 'writeReply'
+ k (WriteReply reply) (runGet getit bs)
+
+ -- 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
+
+-- ** Async matching "internals" -----------------------------------------------
+
+data AsyncManager = AsyncManager
+ { pendingSent :: !(IORef (IntMap (MVar LB.ByteString)))
+ -- ^ When a request is sent, keep track that we're expecting a reply. The
+ -- MVar is signaled when the reply arrives.
+ , pendingRecv :: !(TQueue (CorrelationId, LB.ByteString))
+ -- ^ A queue of pending received requests. When a request is received,
+ -- queue it here. A request handler will pop the next request from it.
+ -- (Nothing except the 'drainPipe' thread should read from the pipe).
+ --
+ -- Seq contract: Dequeue from left, enqueue on right
+ , asyncDrainTid :: !ThreadId
+ , asyncNextReq :: !(IORef Int)
+ }
+
+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 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
+data IddMsg = Request !CorrelationId LB.ByteString
+ | Reply !CorrelationId LB.ByteString
+
+instance Binary IddMsg where
+ put (Request i bs) = putWord8 0 >> put i >> put bs
+ put (Reply i bs) = putWord8 1 >> put i >> put bs
+ get = do
+ b <- getWord8
+ case b of
+ 0 -> Request <$> get <*> get
+ 1 -> Reply <$> get <*> get
+ _ -> error $ "impossible: Binary IddMsg - " ++ show b
-remoteCall :: Binary a => Pipe -> Message a -> IO a
-remoteCall pipe msg = do
- writePipe pipe (putMessage msg)
- readPipe pipe get
+newAsyncManager :: Pipe -> IO AsyncManager
+newAsyncManager pipe = do
+ asyncNextReq <- newIORef 0
+ pendingSent <- newIORef IntMap.empty
+ pendingRecv <- newTQueueIO
-writePipe :: Pipe -> Put -> IO ()
-writePipe Pipe{..} put = putAll $ execPut put
+ mdo
+ -- Fork thread to drain read end.
+ asyncDrainTid <- forkIO (drainPipe pipe am)
+ -- todo: proper thread clean up and no leaks
-remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
-remoteTHCall pipe msg = do
- writePipe pipe (putTHMessage msg)
- readPipe pipe get
+ let am = AsyncManager{..}
+ pure am
-readPipe :: Pipe -> Get a -> IO a
-readPipe Pipe{..} get = do
+-- | 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:
+drainPipe :: Pipe -> AsyncManager -> IO ()
+drainPipe p am = forever $ do
+ readPipe p >>= \case
+ -- Receive a reply to a pending notification
+ Reply uq bs -> do
+ wv <- atomicModifyIORef' (pendingSent am) $ \m -> do
+ case IntMap.lookup uq m of
+ Nothing -> error "drainPipe: Received reply to no pending request"
+ Just wv -> (IntMap.delete uq m, wv)
+ putMVar wv bs -- signal matching request
+ -- Receive a request and queue it
+ Request uq bs ->
+ atomically $ writeTQueue (pendingRecv am) (uq,bs)
+
+freshReqId :: AsyncManager -> IO CorrelationId
+freshReqId AsyncManager{asyncNextReq} =
+ atomicModifyIORef' asyncNextReq $ \i -> (i+1, i)
+
+--------------------------------------------------------------------------------
+-- * 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'.
+readPipe :: Pipe -> IO IddMsg
+readPipe Pipe{..} = do
leftovers <- readIORef pipeLeftovers
m <- getBin getSome get leftovers
case m of
@@ -758,3 +943,22 @@ getBin getsome get leftover = go leftover (runGetIncremental get)
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
+
+--------------------------------------------------------------------------------
+-- * 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
=====================================
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,26 @@ 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
+ handleRemoteCall pipe hook $ \(msg :: Message a) -> do
- discardCtrlC
+ 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
+ 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 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
+ isDone <- readIORef doneRef
+ if isDone then return ()
+ else loop doneRef
handleCustom tag payload = do
mresp <- customHandler tag payload
@@ -89,17 +89,24 @@ 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 :: forall a. (Binary a, Show a) => IO a -> IO (QResult a)
wrapRunTH 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 RunTHDone >>= \case
+ Right () -> pure ()
+ Left err -> throwIO (GHCiQUncaughtExceptionOnHost err)
case r of
Left e
| Just (GHCiQException _ err) <- fromException e -> do
=====================================
libraries/ghci/GHCi/TH.hs
=====================================
@@ -10,6 +10,7 @@ module GHCi.TH
, runModFinalizerRefs
, runTH
, GHCiQException(..)
+ , GHCiQUncaughtExceptionOnHost(..)
) where
{- Note [Remote Template Haskell]
@@ -158,8 +159,13 @@ ghcCmd m = GHCiQ $ \sRef -> do
s <- readIORef sRef
r <- remoteTHCall (qsPipe s) m
case r of
- THException str -> throwIO (GHCiQException s str)
- THComplete res -> return res
+ Right (THException str) -> throwIO (GHCiQException s str)
+ Right (THComplete res) -> return res
+ Left e -> liftIO $ throwIO (GHCiQUncaughtExceptionOnHost e)
+
+-- | An exception was thrown on the host while processing a 'remoteTHCall'
+data GHCiQUncaughtExceptionOnHost = GHCiQUncaughtExceptionOnHost SomeException deriving Show
+instance Exception GHCiQUncaughtExceptionOnHost
instance MonadIO GHCiQ where
liftIO m = GHCiQ $ \_ -> m
@@ -189,9 +195,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) StartRecover >>= either (throwIO . GHCiQUncaughtExceptionOnHost) pure
e <- try $ unmask $ runGHCiQ (runQinGHCiQ a <* ghcCmd FailIfErrs) sRef
- remoteTHCall (qsPipe s) (EndRecover (isLeft e))
+ remoteTHCall (qsPipe s) (EndRecover (isLeft e)) >>= either (throwIO . GHCiQUncaughtExceptionOnHost) pure
case e of
Left GHCiQException{} ->
runGHCiQ (runQinGHCiQ h) sRef
=====================================
libraries/ghci/ghci.cabal.in
=====================================
@@ -95,7 +95,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:
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/25ce96e34f81e573d5be0f7b35c24fb…
--
View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/25ce96e34f81e573d5be0f7b35c24fb…
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
1
0