Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC Commits: f8c724a7 by Rodrigo Mesquita at 2026-10-14T18:34:12+01:00 refactor: Move Async RPC abstraction into GHCi.Message.Async Carves out the Async RPC over the Pipe abstraction to a new module, making the GHCi.Message more about the specific interpreter messages rather than about the IPC protocol. - - - - - 3 changed files: - libraries/ghci/GHCi/Message.hs - + libraries/ghci/GHCi/Message/Async.hs - libraries/ghci/ghci.cabal.in Changes: ===================================== libraries/ghci/GHCi/Message.hs ===================================== @@ -69,35 +69,22 @@ import GHC.IO (IO(..)) import Control.Concurrent import Control.DeepSeq import Control.Exception -#if MIN_VERSION_base(4,20,0) -import Control.Exception.Context -#endif import Data.Binary import Data.Binary.Get import Data.Binary.Put import Data.ByteString (ByteString) -import qualified Data.ByteString as B -import qualified Data.ByteString.Builder as B import qualified Data.ByteString.Lazy as LB import qualified Data.ByteString.Short as BS import Data.Dynamic 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 import qualified GHC.Boot.TH.Syntax as TH import qualified GHC.Boot.TH.Monad as TH -import System.Exit -import System.IO -import Control.Monad -import Control.Concurrent.STM -import Data.Bifunctor -import GHC.Base (State#) -import GHC.Exts (RealWorld) +import GHCi.Message.Async -- ----------------------------------------------------------------------------- -- The RPC protocol between GHC and the interactive server @@ -487,45 +474,6 @@ data BreakUnitId -- | A dummy type that tags pointers returned by 'LoadDLLs'. data LoadedDLL --- SomeException can't be serialized because it contains dynamic --- types. However, we do very limited things with the exceptions that --- are thrown by interpreted computations: --- --- * We print them, e.g. "*** Exception: <something>" --- * UserInterrupt has a special meaning --- * In ghc -e, exitWith should exit with the appropriate exit code --- --- So all we need to do is distinguish UserInterrupt and ExitCode, and --- all other exceptions can be represented by their 'show' string. --- -data SerializableException - = EUserInterrupt - | EExitCode ExitCode - | EOtherException String - deriving (Generic, Show) - -toSerializableException :: SomeException -> SerializableException -toSerializableException ex - | Just UserInterrupt <- fromException ex = EUserInterrupt - | Just (ec::ExitCode) <- fromException ex = (EExitCode ec) - | otherwise = EOtherException $ -#if MIN_VERSION_base(4,20,0) - -- Exception plus backtrace as seen in `displayExceptionWithInfo` - case displayExceptionContext (someExceptionContext ex) of - "" -> displayException (ex :: SomeException) - cx -> displayException (ex :: SomeException) ++ "\n\n" ++ cx -#else - show (ex :: SomeException) -#endif - -fromSerializableException :: SerializableException -> SomeException -fromSerializableException EUserInterrupt = toException UserInterrupt -fromSerializableException (EExitCode c) = toException c -fromSerializableException (EOtherException str) = toException (ErrorCall str) - -instance Binary ExitCode -instance Binary SerializableException - data THResult a = THException String | THComplete a @@ -719,65 +667,6 @@ serializeBCOs rbcos = parMap doChunk (chunkList 100 rbcos) chunkList _ [] = [] chunkList n xs = as : chunkList n bs where (as,bs) = splitAt n xs --- ----------------------------------------------------------------------------- --- Reading/writing messages - --- | An opaque pipe for bidirectional binary data transmission. -data Pipe = Pipe - { getSome :: !(IO ByteString) - , putAll :: !(B.Builder -> IO ()) - , pipeLeftovers :: !(IORef (Maybe ByteString)) - , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent writes to the stream - , pipeAsyncMngr :: !AsyncManager - , pipeIsClosed :: !(TVar Bool) - -- ^ After the pipe is closed, requests can no longer be received nor sent - } - --- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write. --- Must call 'closePipe' at the end to clean up. -mkPipeFromHandles :: Handle -> Handle -> IO Pipe -mkPipeFromHandles pipeRead pipeWrite = do - let getSome = B.hGetSome pipeRead (32*1024) - putAll b = do - B.hPutBuilder pipeWrite b - hFlush pipeWrite - mkPipeFromContinuations getSome putAll - --- | Make a 'Pipe' from a reader function and a writer function. --- The pipe should be closed with 'closePipe'. -mkPipeFromContinuations :: IO ByteString -> (B.Builder -> IO ()) -> IO Pipe -mkPipeFromContinuations getSome putAll = mdo - pipeLeftovers <- newIORef Nothing - pipeLock <- newMVar () - pipeIsClosed <- newTVarIO False - mdo - pipeAsyncMngr <- newAsyncManager p - let p = Pipe { .. } - pure p - --- | Cleanly close our end of the Pipe, terminating the async manager's --- drain-read-end thread. Calling 'closePipe' on an already closed pipe is a no-op. --- --- If the pipe was created using 'mkPipeFromHandles', the handles must only by --- closed after calling 'closePipe'. --- --- The requests we've sent that are pending are all canceled with --- 'PipeClosedException' by ourselves. --- --- The remote process does not necessarily detect that this end of the pipe was --- closed, but it must handle cleanly and clean up after exceptions caused from --- it such as a handle being closed. -closePipe :: Pipe -> IO () -closePipe Pipe{..} = do - - isAlreadyClosed <- atomically $ swapTVar pipeIsClosed True - - unless isAlreadyClosed $ do - - -- releases handle lock taken by reading, and cleans up by replying to every - -- pending request with 'PipeClosedException'. - throwTo (asyncDrainTid pipeAsyncMngr) PipeClosedException - -- | 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 @@ -815,314 +704,3 @@ handleRemoteTHCall pipe ii k = 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 -remoteAsyncReq :: Pipe -> Put -> (Maybe InboxId) -> IO (Either SomeException LB.ByteString) -remoteAsyncReq pipe msg ii = do - - let pendingReplies = pendingReply (pipeAsyncMngr pipe) - - -- Mark pending request - uq <- freshId (pipeAsyncMngr pipe) - wv <- newEmptyTMVarIO - - isClosed <- atomically $ do - isClosed <- readTVar (pipeIsClosed pipe) - if isClosed - then pure True - else do - modifyTVar pendingReplies (IntMap.insert uq wv) - pure False - - when isClosed $ do - throwIO PipeClosedException - - -- Write request - writePipe pipe (Request uq ii (runPut msg)) - `onException` - -- the req wasn't done, or no one will be reading the reply: - -- in either case, drop it from the pending map. - atomically (modifyTVar pendingReplies (IntMap.delete uq)) - - -- Block waiting for reply - atomically (takeTMVar wv) - `onException` - atomically (modifyTVar pendingReplies (IntMap.delete uq)) - --- | Receive a single asynchronous request and send a matching reply --- asynchronously (the handler may fork a computation that eventually replies --- and can return before actually replying, as long as it eventually does). --- --- The handler receives as an argument the function to use to reply to the request. --- It must be used exactly once (all requests expect a reply, and you can't --- reply more than once). Linear types enforce the handler replies in every --- branch, and if the handler otherwise throws, it is handled by 'handleAsyncReq'. -handleAsyncReq :: Pipe -> Get a -> (Maybe InboxId) -> (WriteReply Put %1 -> a -> IO r) -> IO r -handleAsyncReq pipe getit mii k = mask $ \restore -> do - let am = pipeAsyncMngr pipe - ack <- newEmptyMVar - - -- Block waiting for request - mbreq <- atomically $ do - isClosed <- readTVar (pipeIsClosed pipe) - if isClosed - then return Nothing - else Just <$> case mii of - Nothing -> readTQueue $ requestsQueue am - Just ii -> readTQueue =<< getMailbox am ii - - case mbreq of - Nothing -> throwIO PipeClosedException - Just (uq, bs) -> do - let reply p = do - writePipe pipe (Reply uq (bimap toSerializableException runPut p)) - putMVar ack () - - -- Run the continuation which must reply exactly once with 'writeReply' - case runGetOrFail getit bs of - Left (_, _, err) -> do - let exc = DecodeAsyncReqException err - reply (Left (toException exc)) - throwIO exc - Right (_, _, it) -> - restore (k (WriteReply reply) it) - - -- If there's an exception before the reply is sent, reply with an error instead. - `catch` \(e :: SomeException) -> do - - tryTakeMVar ack >>= \case - Just () -> - -- Already replied - throwIO e - Nothing -> do - -- Reply with an error - reply (Left e) - throwIO e - --- | 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 - --- | 'handleSyncReq' failed to decode an incoming request payload using the given @'Get' a@ -newtype DecodeAsyncReqException = DecodeAsyncReqException String deriving Show -instance Exception DecodeAsyncReqException - --- | We've closed the pipe on our end and we terminate all pending requests with this. -data PipeClosedException = PipeClosedException deriving Show -instance Exception PipeClosedException - --- ** Async matching "internals" ----------------------------------------------- - -data AsyncManager = AsyncManager - { pendingReply :: !(TVar (IntMap (TMVar (Either SomeException LB.ByteString)))) - -- ^ When a request is sent, keep track that we're expecting a reply. The - -- TMVar is signaled when the reply arrives. The exception can originate - -- from the host process, or from the remote one (with 'RemoteProcessException'). - , requestsQueue :: !(TQueue (CorrelationId, LB.ByteString)) - -- ^ A queue of pending received requests. When a request is received - -- /without/ a specific 'InboxId', queue it here. A request handler will - -- pop the next request from it. (Nothing except the 'drainPipe' thread - -- should read from the pipe). Read with @'handleAsyncReq' p get Nothing@. - , mailboxQueues :: !(TVar (IntMap (TQueue (CorrelationId, LB.ByteString)))) - -- ^ Requests for a specific 'InboxId' that can be read using - -- @'handleAsyncReq' p get (Just <mailboxid>)@. - , asyncDrainTid :: !ThreadId - , asyncFreshId :: !(IORef Int) - } - --- | Each 'Request' expects a 'Reply' with a matching 'CorrelationId'. -type CorrelationId = Int - --- | A 'Request' may optionally declare an 'InboxId'. If it does, it can only --- be read by a 'handleAsyncReq' for that specific 'InboxId'. That is, --- @'handleAsyncReq' p get Nothing@ will not receive messages meant for a --- specific inbox. --- --- This is useful to prevent mixing up reverse requests, where we want to send --- many reverse requests but have the caller handle them without interleaving --- them. -newtype InboxId = InboxId Int deriving (Show, Binary) - --- | Create a new mailbox from where we can read all requests addressed to this 'InboxId'. -newMailbox :: AsyncManager -> IO InboxId -newMailbox am = InboxId <$> freshId am - -- `mailboxQueues` is written on demand, no need to do it here. - --- | Clean up a created mailbox by 'InboxId'. -closeMailbox :: AsyncManager -> InboxId -> IO () -closeMailbox AsyncManager{..} (InboxId ii) = - atomically (modifyTVar mailboxQueues (IntMap.delete ii)) - --- | We have a single type for requests and replies because both the host and --- interpreter may send requests *and* replies on the write end *and* on recv --- both on the read end. --- --- For instance: --- 1. The host sends a RunTH request on its Host Write End (HWE) --- 2. The interpreter sends a NewName TH request on its Interpreter Write End (IWE) --- 3. The host sends a reply to the TH request on its HWE --- 4. The interpreter sends the reply to the RunTH request on its IWE --- --- Additionally, we want to keep concurrent TH requests for independent RunTH --- calls separated. Thus the interpreter requests additionally specify a --- mailbox announced by RunTH: --- 1. To interpreter: @Request _ Nothing (RunTH <mailid>)@ --- 2. From interpreter: @Request _ (Just <mailid>) NewName@ --- And the host will relate the <mailid> reverse requests to that RunTH run --- exclusively. -data IddMsg = Request !CorrelationId !(Maybe InboxId) LB.ByteString - | Reply !CorrelationId (Either SerializableException LB.ByteString) - -instance Binary IddMsg where - put (Request i ii bs) = putWord8 0 >> put i >> put ii >> put bs - put (Reply i bs) = putWord8 1 >> put i >> put bs - get = do - b <- getWord8 - case b of - 0 -> Request <$> get <*> get <*> get - 1 -> Reply <$> get <*> get - _ -> fail $ "impossible: Binary IddMsg - " ++ show b - -newAsyncManager :: Pipe -> IO AsyncManager -newAsyncManager pipe = do - asyncFreshId <- newIORef 0 - pendingReply <- newTVarIO IntMap.empty - requestsQueue <- newTQueueIO - mailboxQueues <- newTVarIO IntMap.empty - - -- Fork thread to drain read end. - -- (cleaned up on 'closePipe') - asyncDrainTid <- forkIOWithUnmask (drainPipe pipe) - - let am = AsyncManager{..} - pure am - --- | Drains the read end of the Pipe and match 'Reply'es against pending 'Request's. --- --- In pratice, we'll have both a thread draining the read end of the host's --- 'Pipe' and a thread draining the read end of the interpreter's 'Pipe', since --- both processes construct a 'Pipe' and send and receive messages -drainPipe :: Pipe -> (forall x. IO x -> IO x) -> IO () -drainPipe p unmask = catch (unmask loop) replyToPendings where - loop = do - readPipe p >>= \case - -- Receive a reply to a pending notification - Just (Reply uq bs) -> do - atomically $ do - m <- readTVar (pendingReply am) - case IntMap.lookup uq m of - Nothing -> - -- the req was dropped from pending bc there was an exception, and no - -- one's listening any more. the reply is ignored - pure () - Just wv -> do - writeTVar (pendingReply am) (IntMap.delete uq m) - -- signal matching request - putTMVar wv (bimap (toException . RemoteProcessException) id bs) - loop - -- Receive a request and queue it - Just (Request uq mbi bs) -> do - atomically $ case mbi of - Nothing -> - -- "Global" request - writeTQueue (requestsQueue am) (uq,bs) - Just ii -> do - -- "Specific mailbox" request - q <- getMailbox am ii - writeTQueue q (uq,bs) - loop - -- EOF on read handle - Nothing -> do - -- clean up pipe then exit loop - replyToPendings (toException PipeClosedException) - pure () - - replyToPendings e = atomically $ do - writeTVar (pipeIsClosed p) True - pendings <- swapTVar (pendingReply am) IntMap.empty -- take pending - mapM_ (\p -> putTMVar p (Left e {-host exception-})) pendings - - am = pipeAsyncMngr p - -freshId :: AsyncManager -> IO Int -freshId AsyncManager{asyncFreshId} = - atomicModifyIORef' asyncFreshId $ \i -> (i+1, i) - -getMailbox :: AsyncManager -> InboxId -> STM (TQueue (CorrelationId, LB.ByteString)) -getMailbox am (InboxId ii) = do - mboxes <- readTVar (mailboxQueues am) - case IntMap.lookup ii mboxes of - Nothing -> do - q <- newTQueue - writeTVar (mailboxQueues am) (IntMap.insert ii q mboxes) - pure q - Just q -> - pure q - --------------------------------------------------------------------------------- --- * Pipe thread-safe read/write --------------------------------------------------------------------------------- --- All messages on the wire are 'IddMsg', which contain a correlation header --- and identify whether they are requests or replies. - -writePipe :: Pipe -> IddMsg -> IO () -writePipe p@Pipe{..} = withLock p . putAll . execPut . put - --- | Only the 'drainPipe' thread should read from the pipe directly. Other --- functions should wait for the matching reply to be delivered asynchronously. --- (see 'withAsyncReq'). It doesn't lock to read bc it assumes single reader is 'drainPipe'. --- --- Returns @Nothing@ on EOF. -readPipe :: Pipe -> IO (Maybe IddMsg) -readPipe Pipe{..} = do - leftovers <- readIORef pipeLeftovers - m <- getBin getSome get leftovers - case m of - Nothing -> return Nothing - Just (result, new_leftovers) -> do - writeIORef pipeLeftovers new_leftovers - return (Just result) - -getBin - :: (IO ByteString) -> Get a -> Maybe ByteString - -> IO (Maybe (a, Maybe ByteString)) - -getBin getsome get leftover = go leftover (runGetIncremental get) - where - go Nothing (Done leftover _ msg) = - return (Just (msg, if B.null leftover then Nothing else Just leftover)) - go _ Done{} = throwIO (ErrorCall "getBin: Done with leftovers") - go (Just leftover) (Partial fun) = do - go Nothing (fun (Just leftover)) - go Nothing (Partial fun) = do - -- putStrLn "before hGetSome" - b <- getsome - -- putStrLn $ "hGetSome: " ++ show (B.length b) - if B.null b - then return Nothing - else go Nothing (fun (Just b)) - go _lft (Fail _rest _off str) = - throwIO (ErrorCall ("getBin: " ++ str)) - -withLock :: Pipe -> IO c -> IO c -withLock Pipe{..} = bracket (takeMVar pipeLock) (putMVar pipeLock) . const - --------------------------------------------------------------------------------- --- * 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/Message/Async.hs ===================================== @@ -0,0 +1,480 @@ +{-# LANGUAGE RecursiveDo, GADTs, DeriveGeneric, StandaloneDeriving, ScopedTypeVariables, + GeneralizedNewtypeDeriving, ExistentialQuantification, RecordWildCards, CPP, + NamedFieldPuns, LambdaCase, TypeAbstractions, TypeApplications, + LinearTypes, RankNTypes, UnboxedTuples, BlockArguments, TupleSections, MagicHash #-} +{-# OPTIONS_GHC -fno-warn-name-shadowing -fno-warn-orphans #-} +module GHCi.Message.Async + ( + -- * Pipes + Pipe(..), mkPipeFromHandles, mkPipeFromContinuations + , closePipe, PipeClosedException(..) + + -- * Interface for async bidirectional RPC + , remoteAsyncReq, handleAsyncReq + , RemoteProcessException(..), DecodeAsyncReqException(..) + + -- ** Mailboxes + , InboxId(..) + , newMailbox + , closeMailbox + + -- ** Async matching internals + , AsyncManager(..), CorrelationId + , IddMsg(..), newAsyncManager + + -- ** Consume `reply` exactly once + , WriteReply(..) + , writeReply, (|#>) + + -- * Serializable exceptions + , SerializableException(..) + , toSerializableException, fromSerializableException + ) where + +import Prelude -- See note [Why do we import Prelude here?] +import GHCi.TH.Binary () -- For Binary instances + +import GHC.IO (IO(..)) +import Control.Concurrent +import Control.Exception +#if MIN_VERSION_base(4,20,0) +import Control.Exception.Context +#endif +import Data.Binary +import Data.Binary.Get +import Data.Binary.Put +import Data.ByteString (ByteString) +import qualified Data.ByteString as B +import qualified Data.ByteString.Builder as B +import qualified Data.ByteString.Lazy as LB +import Data.IORef +import Data.IntMap (IntMap) +import qualified Data.IntMap as IntMap +import GHC.Generics +import System.Exit +import System.IO +import Control.Monad +import Control.Concurrent.STM +import Data.Bifunctor +import GHC.Base (State#) +import GHC.Exts (RealWorld) + +-------------------------------------------------------------------------------- +-- * Reading/writing and Pipes +-------------------------------------------------------------------------------- + +-- | An opaque pipe for bidirectional binary data transmission. +data Pipe = Pipe + { getSome :: !(IO ByteString) + , putAll :: !(B.Builder -> IO ()) + , pipeLeftovers :: !(IORef (Maybe ByteString)) + , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent writes to the stream + , pipeAsyncMngr :: !AsyncManager + , pipeIsClosed :: !(TVar Bool) + -- ^ After the pipe is closed, requests can no longer be received nor sent + } + +-- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write. +-- Must call 'closePipe' at the end to clean up. +mkPipeFromHandles :: Handle -> Handle -> IO Pipe +mkPipeFromHandles pipeRead pipeWrite = do + let getSome = B.hGetSome pipeRead (32*1024) + putAll b = do + B.hPutBuilder pipeWrite b + hFlush pipeWrite + mkPipeFromContinuations getSome putAll + +-- | Make a 'Pipe' from a reader function and a writer function. +-- The pipe should be closed with 'closePipe'. +mkPipeFromContinuations :: IO ByteString -> (B.Builder -> IO ()) -> IO Pipe +mkPipeFromContinuations getSome putAll = mdo + pipeLeftovers <- newIORef Nothing + pipeLock <- newMVar () + pipeIsClosed <- newTVarIO False + mdo + pipeAsyncMngr <- newAsyncManager p + let p = Pipe { .. } + pure p + +-- | Cleanly close our end of the Pipe, terminating the async manager's +-- drain-read-end thread. Calling 'closePipe' on an already closed pipe is a no-op. +-- +-- If the pipe was created using 'mkPipeFromHandles', the handles must only by +-- closed after calling 'closePipe'. +-- +-- The requests we've sent that are pending are all canceled with +-- 'PipeClosedException' by ourselves. +-- +-- The remote process does not necessarily detect that this end of the pipe was +-- closed, but it must handle cleanly and clean up after exceptions caused from +-- it such as a handle being closed. +closePipe :: Pipe -> IO () +closePipe Pipe{..} = do + + isAlreadyClosed <- atomically $ swapTVar pipeIsClosed True + + unless isAlreadyClosed $ do + + -- releases handle lock taken by reading, and cleans up by replying to every + -- pending request with 'PipeClosedException'. + throwTo (asyncDrainTid pipeAsyncMngr) PipeClosedException + +-- | We've closed the pipe on our end and we terminate all pending requests with this. +data PipeClosedException = PipeClosedException deriving Show +instance Exception PipeClosedException + +-------------------------------------------------------------------------------- +-- ** Pipe thread-safe read/write +-------------------------------------------------------------------------------- +-- All messages on the wire are 'IddMsg', which contain a correlation header +-- and identify whether they are requests or replies. + +writePipe :: Pipe -> IddMsg -> IO () +writePipe p@Pipe{..} = withLock p . putAll . execPut . put + +-- | Only the 'drainPipe' thread should read from the pipe directly. Other +-- functions should wait for the matching reply to be delivered asynchronously. +-- (see 'withAsyncReq'). It doesn't lock to read bc it assumes single reader is 'drainPipe'. +-- +-- Returns @Nothing@ on EOF. +readPipe :: Pipe -> IO (Maybe IddMsg) +readPipe Pipe{..} = do + leftovers <- readIORef pipeLeftovers + m <- getBin getSome get leftovers + case m of + Nothing -> return Nothing + Just (result, new_leftovers) -> do + writeIORef pipeLeftovers new_leftovers + return (Just result) + +getBin + :: (IO ByteString) -> Get a -> Maybe ByteString + -> IO (Maybe (a, Maybe ByteString)) + +getBin getsome get leftover = go leftover (runGetIncremental get) + where + go Nothing (Done leftover _ msg) = + return (Just (msg, if B.null leftover then Nothing else Just leftover)) + go _ Done{} = throwIO (ErrorCall "getBin: Done with leftovers") + go (Just leftover) (Partial fun) = do + go Nothing (fun (Just leftover)) + go Nothing (Partial fun) = do + -- putStrLn "before hGetSome" + b <- getsome + -- putStrLn $ "hGetSome: " ++ show (B.length b) + if B.null b + then return Nothing + else go Nothing (fun (Just b)) + go _lft (Fail _rest _off str) = + throwIO (ErrorCall ("getBin: " ++ str)) + +withLock :: Pipe -> IO c -> IO c +withLock Pipe{..} = bracket (takeMVar pipeLock) (putMVar pipeLock) . const + +-------------------------------------------------------------------------------- +-- * Matching requests with async replies +-------------------------------------------------------------------------------- + +-- | Send an asynchronous request and block waiting for a matching reply +remoteAsyncReq :: Pipe -> Put -> (Maybe InboxId) -> IO (Either SomeException LB.ByteString) +remoteAsyncReq pipe msg ii = do + + let pendingReplies = pendingReply (pipeAsyncMngr pipe) + + -- Mark pending request + uq <- freshId (pipeAsyncMngr pipe) + wv <- newEmptyTMVarIO + + isClosed <- atomically $ do + isClosed <- readTVar (pipeIsClosed pipe) + if isClosed + then pure True + else do + modifyTVar pendingReplies (IntMap.insert uq wv) + pure False + + when isClosed $ do + throwIO PipeClosedException + + -- Write request + writePipe pipe (Request uq ii (runPut msg)) + `onException` + -- the req wasn't done, or no one will be reading the reply: + -- in either case, drop it from the pending map. + atomically (modifyTVar pendingReplies (IntMap.delete uq)) + + -- Block waiting for reply + atomically (takeTMVar wv) + `onException` + atomically (modifyTVar pendingReplies (IntMap.delete uq)) + +-- | Receive a single asynchronous request and send a matching reply +-- asynchronously (the handler may fork a computation that eventually replies +-- and can return before actually replying, as long as it eventually does). +-- +-- The handler receives as an argument the function to use to reply to the request. +-- It must be used exactly once (all requests expect a reply, and you can't +-- reply more than once). Linear types enforce the handler replies in every +-- branch, and if the handler otherwise throws, it is handled by 'handleAsyncReq'. +handleAsyncReq :: Pipe -> Get a -> (Maybe InboxId) -> (WriteReply Put %1 -> a -> IO r) -> IO r +handleAsyncReq pipe getit mii k = mask $ \restore -> do + let am = pipeAsyncMngr pipe + ack <- newEmptyMVar + + -- Block waiting for request + mbreq <- atomically $ do + isClosed <- readTVar (pipeIsClosed pipe) + if isClosed + then return Nothing + else Just <$> case mii of + Nothing -> readTQueue $ requestsQueue am + Just ii -> readTQueue =<< getMailbox am ii + + case mbreq of + Nothing -> throwIO PipeClosedException + Just (uq, bs) -> do + let reply p = do + writePipe pipe (Reply uq (bimap toSerializableException runPut p)) + putMVar ack () + + -- Run the continuation which must reply exactly once with 'writeReply' + case runGetOrFail getit bs of + Left (_, _, err) -> do + let exc = DecodeAsyncReqException err + reply (Left (toException exc)) + throwIO exc + Right (_, _, it) -> + restore (k (WriteReply reply) it) + + -- If there's an exception before the reply is sent, reply with an error instead. + `catch` \(e :: SomeException) -> do + + tryTakeMVar ack >>= \case + Just () -> + -- Already replied + throwIO e + Nothing -> do + -- Reply with an error + reply (Left e) + throwIO e + +-- | 'handleSyncReq' failed to decode an incoming request payload using the given @'Get' a@ +newtype DecodeAsyncReqException = DecodeAsyncReqException String deriving Show +instance Exception DecodeAsyncReqException + + +-- ** Mailboxes ---------------------------------------------------------------- + +-- | A 'Request' may optionally declare an 'InboxId'. If it does, it can only +-- be read by a 'handleAsyncReq' for that specific 'InboxId'. That is, +-- @'handleAsyncReq' p get Nothing@ will not receive messages meant for a +-- specific inbox. +-- +-- This is useful to prevent mixing up reverse requests, where we want to send +-- many reverse requests but have the caller handle them without interleaving +-- them. +newtype InboxId = InboxId Int deriving (Show, Binary) + +-- | Create a new mailbox from where we can read all requests addressed to this 'InboxId'. +newMailbox :: AsyncManager -> IO InboxId +newMailbox am = InboxId <$> freshId am + -- `mailboxQueues` is written on demand, no need to do it here. + +-- | Clean up a created mailbox by 'InboxId'. +closeMailbox :: AsyncManager -> InboxId -> IO () +closeMailbox AsyncManager{..} (InboxId ii) = + atomically (modifyTVar mailboxQueues (IntMap.delete ii)) + +-- | Get or create a new mailbox for this 'InboxId' +getMailbox :: AsyncManager -> InboxId -> STM (TQueue (CorrelationId, LB.ByteString)) +getMailbox am (InboxId ii) = do + mboxes <- readTVar (mailboxQueues am) + case IntMap.lookup ii mboxes of + Nothing -> do + q <- newTQueue + writeTVar (mailboxQueues am) (IntMap.insert ii q mboxes) + pure q + Just q -> + pure q + +-- ** Async matching "internals" ----------------------------------------------- + +data AsyncManager = AsyncManager + { pendingReply :: !(TVar (IntMap (TMVar (Either SomeException LB.ByteString)))) + -- ^ When a request is sent, keep track that we're expecting a reply. The + -- TMVar is signaled when the reply arrives. The exception can originate + -- from the host process, or from the remote one (with 'RemoteProcessException'). + , requestsQueue :: !(TQueue (CorrelationId, LB.ByteString)) + -- ^ A queue of pending received requests. When a request is received + -- /without/ a specific 'InboxId', queue it here. A request handler will + -- pop the next request from it. (Nothing except the 'drainPipe' thread + -- should read from the pipe). Read with @'handleAsyncReq' p get Nothing@. + , mailboxQueues :: !(TVar (IntMap (TQueue (CorrelationId, LB.ByteString)))) + -- ^ Requests for a specific 'InboxId' that can be read using + -- @'handleAsyncReq' p get (Just <mailboxid>)@. + , asyncDrainTid :: !ThreadId + , asyncFreshId :: !(IORef Int) + } + +-- | Each 'Request' expects a 'Reply' with a matching 'CorrelationId'. +type CorrelationId = Int + +-- | We have a single type for requests and replies because both the host and +-- interpreter may send requests *and* replies on the write end *and* on recv +-- both on the read end. +-- +-- For instance: +-- 1. The host sends a RunTH request on its Host Write End (HWE) +-- 2. The interpreter sends a NewName TH request on its Interpreter Write End (IWE) +-- 3. The host sends a reply to the TH request on its HWE +-- 4. The interpreter sends the reply to the RunTH request on its IWE +-- +-- Additionally, we want to keep concurrent TH requests for independent RunTH +-- calls separated. Thus the interpreter requests additionally specify a +-- mailbox announced by RunTH: +-- 1. To interpreter: @Request _ Nothing (RunTH <mailid>)@ +-- 2. From interpreter: @Request _ (Just <mailid>) NewName@ +-- And the host will relate the <mailid> reverse requests to that RunTH run +-- exclusively. +data IddMsg = Request !CorrelationId !(Maybe InboxId) LB.ByteString + | Reply !CorrelationId (Either SerializableException LB.ByteString) + +instance Binary IddMsg where + put (Request i ii bs) = putWord8 0 >> put i >> put ii >> put bs + put (Reply i bs) = putWord8 1 >> put i >> put bs + get = do + b <- getWord8 + case b of + 0 -> Request <$> get <*> get <*> get + 1 -> Reply <$> get <*> get + _ -> fail $ "impossible: Binary IddMsg - " ++ show b + +newAsyncManager :: Pipe -> IO AsyncManager +newAsyncManager pipe = do + asyncFreshId <- newIORef 0 + pendingReply <- newTVarIO IntMap.empty + requestsQueue <- newTQueueIO + mailboxQueues <- newTVarIO IntMap.empty + + -- Fork thread to drain read end. + -- (cleaned up on 'closePipe') + asyncDrainTid <- forkIOWithUnmask (drainPipe pipe) + + let am = AsyncManager{..} + pure am + +-- | Drains the read end of the Pipe and match 'Reply'es against pending 'Request's. +-- +-- In pratice, we'll have both a thread draining the read end of the host's +-- 'Pipe' and a thread draining the read end of the interpreter's 'Pipe', since +-- both processes construct a 'Pipe' and send and receive messages +drainPipe :: Pipe -> (forall x. IO x -> IO x) -> IO () +drainPipe p unmask = catch (unmask loop) replyToPendings where + loop = do + readPipe p >>= \case + -- Receive a reply to a pending notification + Just (Reply uq bs) -> do + atomically $ do + m <- readTVar (pendingReply am) + case IntMap.lookup uq m of + Nothing -> + -- the req was dropped from pending bc there was an exception, and no + -- one's listening any more. the reply is ignored + pure () + Just wv -> do + writeTVar (pendingReply am) (IntMap.delete uq m) + -- signal matching request + putTMVar wv (bimap (toException . RemoteProcessException) id bs) + loop + -- Receive a request and queue it + Just (Request uq mbi bs) -> do + atomically $ case mbi of + Nothing -> + -- "Global" request + writeTQueue (requestsQueue am) (uq,bs) + Just ii -> do + -- "Specific mailbox" request + q <- getMailbox am ii + writeTQueue q (uq,bs) + loop + -- EOF on read handle + Nothing -> do + -- clean up pipe then exit loop + replyToPendings (toException PipeClosedException) + pure () + + replyToPendings e = atomically $ do + writeTVar (pipeIsClosed p) True + pendings <- swapTVar (pendingReply am) IntMap.empty -- take pending + mapM_ (\p -> putTMVar p (Left e {-host exception-})) pendings + + am = pipeAsyncMngr p + +-- | An exception that can be thrown on the host process representing an +-- expection that happened on the remote one. +newtype RemoteProcessException = RemoteProcessException SerializableException deriving Show +instance Exception RemoteProcessException + +-- | Come up with a fresh id for 'CorrelationId' or 'InboxId' +freshId :: AsyncManager -> IO Int +freshId AsyncManager{asyncFreshId} = + atomicModifyIORef' asyncFreshId $ \i -> (i+1, i) + +-------------------------------------------------------------------------------- +-- * Consume `reply` exactly once +-------------------------------------------------------------------------------- + +-- | Every request should get exactly one reply. +data WriteReply a where + WriteReply :: (Either SomeException a -> IO ()) %1 -> WriteReply a + +-- | Write a reply exactly once (consuming the ability to write more replies) +writeReply :: WriteReply a %1 -> a -> IO () +writeReply (WriteReply r) = \x -> r (Right x) + +-- | Infix, linear, 'unIO' +(|#>) :: IO a %1 -> State# RealWorld -> (# State# RealWorld, a #) +(|#>) (IO fs) s = fs s + +-------------------------------------------------------------------------------- +-- * Serializable Exceptions +-------------------------------------------------------------------------------- + +-- SomeException can't be serialized because it contains dynamic +-- types. However, we do very limited things with the exceptions that +-- are thrown by interpreted computations: +-- +-- * We print them, e.g. "*** Exception: <something>" +-- * UserInterrupt has a special meaning +-- * In ghc -e, exitWith should exit with the appropriate exit code +-- +-- So all we need to do is distinguish UserInterrupt and ExitCode, and +-- all other exceptions can be represented by their 'show' string. +-- +data SerializableException + = EUserInterrupt + | EExitCode ExitCode + | EOtherException String + deriving (Generic, Show) + +toSerializableException :: SomeException -> SerializableException +toSerializableException ex + | Just UserInterrupt <- fromException ex = EUserInterrupt + | Just (ec::ExitCode) <- fromException ex = (EExitCode ec) + | otherwise = EOtherException $ +#if MIN_VERSION_base(4,20,0) + -- Exception plus backtrace as seen in `displayExceptionWithInfo` + case displayExceptionContext (someExceptionContext ex) of + "" -> displayException (ex :: SomeException) + cx -> displayException (ex :: SomeException) ++ "\n\n" ++ cx +#else + show (ex :: SomeException) +#endif + +fromSerializableException :: SerializableException -> SomeException +fromSerializableException EUserInterrupt = toException UserInterrupt +fromSerializableException (EExitCode c) = toException c +fromSerializableException (EOtherException str) = toException (ErrorCall str) + +instance Binary ExitCode +instance Binary SerializableException ===================================== libraries/ghci/ghci.cabal.in ===================================== @@ -76,6 +76,7 @@ library exposed-modules: GHCi.BreakArray GHCi.Message + GHCi.Message.Async GHCi.ResolvedBCO GHCi.RemoteTypes GHCi.FFI View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/f8c724a7e68b95a7883e466621e4a7ce... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/f8c724a7e68b95a7883e466621e4a7ce... 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