Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC

Commits:

12 changed files:

Changes:

  • compiler/GHC/Runtime/Interpreter.hs
    ... ... @@ -626,11 +626,9 @@ spawnIServ conf = do
    626 626 
                                               []
    627 627 
                                               (iservConfOpts    conf)
    628 628 
       interpPipe <- mkPipeFromHandles rh wh
    629 
    -  lock <- newMVar ()
    630 629 
       let process = InterpProcess
    631 630 
                       { interpHandle = ph
    632 631 
                       , interpPipe
    633 
    -                  , interpLock   = lock
    634 632 
                       }
    635 633
    636 634 
       pending_frees <- newMVar []
    ... ... @@ -656,6 +654,8 @@ stopInterp interp = case interpInstance interp of
    656 654 
                 if isJust ex
    657 655 
                    then pure ()
    658 656 
                    else sendMessage i Shutdown
    657 
    +            -- We asked the server to shutdown, so we will also shutdown any communication.
    658 
    +            closePipe (interpPipe (instProcess i))
    659 659 
                 pure InterpPending
    660 660
    661 661 
     -- -----------------------------------------------------------------------------

  • compiler/GHC/Runtime/Interpreter/JS.hs
    ... ... @@ -127,11 +127,9 @@ startTHRunnerProcess interp_js settings = do
    127 127 
       std_in <- readIORef interp_in
    128 128
    129 129 
       interpPipe <- mkPipeFromHandles rh wh
    130 
    -  lock <- newMVar ()
    131 130 
       let proc = InterpProcess
    132 131 
                   { interpHandle = hdl
    133 132 
                   , interpPipe
    134 
    -              , interpLock   = lock
    135 133 
                   }
    136 134 
       pure (std_in, proc)
    137 135
    ... ... @@ -382,7 +380,7 @@ jsLinkPlan logger tmpfs tmp_dir ar_cache link_cfg cfg inst link_plan = do
    382 380 
       --   Haskell server.
    383 381 
       server_started <- jsServerStarted <$> readMVar (instJSState (instExtra inst))
    384 382 
       if server_started
    385 
    -    then sendMessageNoResponse inst $ LoadObj all_js
    383 
    +    then sendMessage inst $ LoadObj all_js
    386 384 
         else jsLoadFile            inst all_js
    387 385
    388 386 
       ----------------------------------------------------------------

  • compiler/GHC/Runtime/Interpreter/Process.hs
    ... ... @@ -2,16 +2,8 @@ module GHC.Runtime.Interpreter.Process
    2 2 
       (
    3 3 
       -- * Message API
    4 4 
         Message(..)
    5 
    -  , DelayedResponse (..)
    6 
    -  -- * Top-level message API (these acquire/release a lock)
    7 5 
       , sendMessage
    8 
    -  , sendMessageNoResponse
    9 
    -  , sendMessageDelayedResponse
    10 
    -  , receiveDelayedResponse
    11 
    -  -- * Nested message API (these require the interpreter to already be locked)
    12 
    -  , sendAnyValue
    13 
    -  , receiveAnyValue
    14 
    -  , receiveTHMessage
    6 
    +  , sendTHComputation
    15 7 
       )
    16 8 
     where
    17 9
    ... ... @@ -27,97 +19,63 @@ import GHC.Utils.Exception as Ex
    27 19 
     import Data.Binary
    28 20 
     import System.Exit
    29 21 
     import System.Process
    30 
    -import Control.Concurrent.MVar (MVar, withMVar, takeMVar, putMVar, isEmptyMVar)
    31 
    -
    32 
    -data DelayedResponse a = DelayedResponse
    22 
    +import Control.Concurrent
    23 
    +import Data.IORef
    24 
    +import Control.Concurrent.STM
    33 25
    34 26 
     -- -----------------------------------------------------------------------------
    35 27 
     -- Top-level Message API
    36 28
    37 
    --- | Send a message to the interpreter process that doesn't expect a response
    38 
    ---   (locks the interpreter while sending)
    39 
    -sendMessageNoResponse :: ExtInterpInstance d -> Message () -> IO ()
    40 
    -sendMessageNoResponse i m =
    41 
    -  withLock i $ writeInterpProcess (instProcess i) (putMessage m)
    42 
    -
    43 
    --- | Send a message to the interpreter that expects a response
    44 
    ---   (locks the interpreter while until the response is received)
    29 
    +-- | Send a message to the interpreter and expect a response, blocking until it arrives.
    30 
    +--
    31 
    +-- This issues an asynchronous RPC on the external interpreter which does not
    32 
    +-- block the external interpreter thread. We block here on the host because we
    33 
    +-- wait for the asynchronous response to arrive.
    45 34 
     sendMessage :: Binary a => ExtInterpInstance d -> Message a -> IO a
    46 
    -sendMessage i m = withLock i $ callInterpProcess (instProcess i) m
    47 
    -
    48 
    --- | Send a message to the interpreter process whose response is expected later
    35 
    +sendMessage i msg =
    36 
    +  remoteCall (interpPipe proc) msg
    37 
    +    `catchException` \(e :: SomeException) -> handleInterpProcessFailure proc e
    38 
    +  where
    39 
    +    proc = instProcess i
    40 
    +
    41 
    +-- | Send a message with a TH computation to the ext. interpreter (like
    42 
    +-- 'RunTH', or 'RunModFinalizers') and handle 'THMessage' requests from the
    43 
    +-- ext. interpreter to us in the meantime.
    49 44 
     --
    50 
    --- This is useful to avoid forgetting to receive the value and to ensure that
    51 
    --- the type of the response isn't lost. Use receiveDelayedResponse to read it.
    52 
    --- (locks the interpreter until the response is received using
    53 
    --- `receiveDelayedResponse`)
    54 
    -sendMessageDelayedResponse :: ExtInterpInstance d -> Message a -> IO (DelayedResponse a)
    55 
    -sendMessageDelayedResponse i m = do
    56 
    -  lock i
    57 
    -  writeInterpProcess (instProcess i) (putMessage m)
    58 
    -  pure DelayedResponse
    59 
    -
    60 
    --- | Expect a delayed result to be received now
    61 
    -receiveDelayedResponse :: Binary a => ExtInterpInstance d -> DelayedResponse a -> IO a
    62 
    -receiveDelayedResponse i DelayedResponse = do
    63 
    -  ensureLocked i
    64 
    -  r <- readInterpProcess (instProcess i) get
    65 
    -  unlock i
    66 
    -  pure r
    67 
    -
    68 
    --- -----------------------------------------------------------------------------
    69 
    --- Nested Message API
    70 
    -
    71 
    --- | Send any value (requires locked interpreter)
    72 
    -sendAnyValue :: Binary a => ExtInterpInstance d -> a -> IO ()
    73 
    -sendAnyValue i m = ensureLocked i >> writeInterpProcess (instProcess i) (put m)
    74 
    -
    75 
    --- | Expect a value to be received (requires locked interpreter)
    76 
    -receiveAnyValue :: ExtInterpInstance d -> Get a -> IO a
    77 
    -receiveAnyValue i get = ensureLocked i >> readInterpProcess (instProcess i) get
    78 
    -
    79 
    --- | Wait for a Template Haskell message (requires locked interpreter)
    80 
    -receiveTHMessage :: ExtInterpInstance d -> IO THMsg
    81 
    -receiveTHMessage i = ensureLocked i >> receiveAnyValue i getTHMessage
    82 
    -
    83 
    --- -----------------------------------------------------------------------------
    84 
    -
    85 
    -getLock :: ExtInterpInstance d -> MVar ()
    86 
    -getLock = interpLock . instProcess
    87 
    -
    88 
    -withLock :: ExtInterpInstance d -> IO a -> IO a
    89 
    -withLock i f = withMVar (getLock i) (const f)
    90 
    -
    91 
    -lock :: ExtInterpInstance d -> IO ()
    92 
    -lock i = takeMVar (getLock i)
    93 
    -
    94 
    -unlock :: ExtInterpInstance d -> IO ()
    95 
    -unlock i = putMVar (getLock i) ()
    96 
    -
    97 
    -ensureLocked :: ExtInterpInstance d -> IO ()
    98 
    -ensureLocked i =
    99 
    -  isEmptyMVar (getLock i) >>= \case
    100 
    -    False -> panic "ensureLocked: external interpreter not locked"
    101 
    -    _     -> pure ()
    102 
    -
    103 
    -
    104 
    --- | Send a 'Message' and receive the response from the interpreter process
    105 
    -callInterpProcess :: Binary a => InterpProcess -> Message a -> IO a
    106 
    -callInterpProcess i msg =
    107 
    -  remoteCall (interpPipe i) msg
    108 
    -    `catchException` \(e :: SomeException) -> handleInterpProcessFailure i e
    109 
    -
    110 
    --- | Read a value from the interpreter process
    111 
    -readInterpProcess :: InterpProcess -> Get a -> IO a
    112 
    -readInterpProcess i get =
    113 
    -  readPipe (interpPipe i) get
    114 
    -    `catchException` \(e :: SomeException) -> handleInterpProcessFailure i e
    115 
    -
    116 
    --- | Send a value to the interpreter process
    117 
    -writeInterpProcess :: InterpProcess -> Put -> IO ()
    118 
    -writeInterpProcess i put =
    119 
    -  writePipe (interpPipe i) put
    120 
    -    `catchException` \(e :: SomeException) -> handleInterpProcessFailure i e
    45 
    +-- This function blocks until the TH computation returns a value, but spawns in
    46 
    +-- the background a thread to handle the 'THMessage's until the TH computation
    47 
    +-- is fully done.
    48 
    +sendTHComputation
    49 
    +  :: Binary a
    50 
    +  => ExtInterpInstance d
    51 
    +  -> (InboxId -> Message a)
    52 
    +  -> (forall b. Binary b => THMessage b -> IO (Bool, b))
    53 
    +  -- ^ Handle one message. Return True to end the loop.
    54 
    +  -- This handler must never uninterrutibly mask an expression which may get
    55 
    +  -- stuck. If it does, then 'sendTHComputation' will also get uninterruptibly stuck.
    56 
    +  -> IO a
    57 
    +sendTHComputation i msg reqHandler = mask $ \restore -> do
    58 
    +  -- inline "withAsync/wait"
    59 
    +  var  <- newEmptyTMVarIO
    60 
    +  mbox <- newMailbox (pipeAsyncMngr (interpPipe proc))
    61 
    +  ht   <- forkFinally (restore (handlerLoop mbox)) (atomically . putTMVar var)
    62 
    +  restore (sendMessage i (msg mbox) <* atomically (readTMVar var))
    63 
    +    `onException` uninterruptibleMask_ (throwTo ht CancelTHHandler)
    64 
    +  where
    65 
    +    proc = instProcess i
    66 
    +    handleOne mbox doneRef = do
    67 
    +      handleRemoteTHCall (interpPipe proc) mbox $ \req -> do
    68 
    +        (stop, res) <- reqHandler req
    69 
    +        writeIORef doneRef stop
    70 
    +        return res
    71 
    +    handlerLoop mbox = do
    72 
    +      doneRef <- newIORef False
    73 
    +      let loop = do
    74 
    +            handleOne mbox doneRef
    75 
    +            isDone <- readIORef doneRef
    76 
    +            if isDone then return ()
    77 
    +                      else loop
    78 
    +       in loop
    121 79
    122 80 
     handleInterpProcessFailure :: InterpProcess -> SomeException -> IO a
    123 81 
     handleInterpProcessFailure i e = do
    ... ... @@ -130,3 +88,8 @@ handleInterpProcessFailure i e = do
    130 88 
           terminateProcess hdl
    131 89 
           _ <- waitForProcess hdl
    132 90 
           throw e
    91 
    +
    92 
    +data CancelTHHandler = CancelTHHandler deriving Show
    93 
    +instance Exception CancelTHHandler where
    94 
    +  fromException = asyncExceptionFromException
    95 
    +  toException = asyncExceptionToException

  • compiler/GHC/Runtime/Interpreter/Types.hs
    ... ... @@ -111,7 +111,6 @@ type WasmInterp = ExtInterpState WasmInterpConfig ()
    111 111 
     data InterpProcess = InterpProcess
    112 112 
       { interpPipe   :: !Pipe           -- ^ Pipe to communicate with the server
    113 113 
       , interpHandle :: !ProcessHandle  -- ^ Process handle of the server
    114 
    -  , interpLock   :: !(MVar ())      -- ^ Lock to prevent concurrent access to the stream
    115 114 
       }
    116 115
    117 116 
     -- | Status of an external interpreter

  • compiler/GHC/Runtime/Interpreter/Wasm.hs
    ... ... @@ -73,14 +73,12 @@ spawnWasmInterp WasmInterpConfig {..} = do
    73 73 
       hSetBuffering rh NoBuffering
    74 74 
       interpPipe <- mkPipeFromHandles rh wh
    75 75 
       pending_frees <- newMVar []
    76 
    -  lock <- newMVar ()
    77 76 
       pure
    78 77 
         $ ExtInterpInstance
    79 78 
           { instProcess =
    80 79 
               InterpProcess
    81 80 
                 { interpHandle = ph,
    82 
    -              interpPipe,
    83 
    -              interpLock = lock
    81 
    +              interpPipe
    84 82 
                 },
    85 83 
             instPendingFrees = pending_frees,
    86 84 
             instExtra = ()

  • compiler/GHC/Tc/Gen/Splice.hs
    ... ... @@ -170,6 +170,7 @@ import Data.IORef
    170 170 
     import GHC.Parser.HaddockLex (lexHsDoc)
    171 171 
     import GHC.Parser (parseIdentifier)
    172 172 
     import GHC.Rename.Doc (rnHsDoc)
    173 
    +import Data.Function
    173 174
    174 175 
     {-
    175 176 
     Note [Template Haskell state diagram]
    ... ... @@ -1165,11 +1166,12 @@ runRemoteModFinalizers (ThModFinalizers finRefs) = do
    1165 1166 
           case th_state of
    1166 1167 
             Nothing -> return () -- TH was not started, nothing to do
    1167 1168 
             Just fhv -> do
    1168 
    -          r <- liftIO $ withForeignRef fhv $ \st ->
    1169 
    -            withForeignRefs finRefs $ \qrefs ->
    1170 
    -              sendMessageDelayedResponse inst (RunModFinalizers st qrefs)
    1171 
    -          () <- runRemoteTH inst []
    1172 
    -          qr <- liftIO $ receiveDelayedResponse inst r
    1169 
    +          qr <- withRunInIO $ \runInIO -> do
    1170 
    +            recovers <- newIORef []
    1171 
    +            withForeignRef fhv $ \st ->
    1172 
    +              withForeignRefs finRefs $ \qrefs ->
    1173 
    +                sendTHComputation inst (\ii -> RunModFinalizers ii st qrefs) $ \msg ->
    1174 
    +                  runInIO (runRemoteTH recovers msg)
    1173 1175 
               checkQResult qr
    1174 1176
    1175 1177 
     runQResult
    ... ... @@ -1824,48 +1826,47 @@ runTH ty fhv = do
    1824 1826 
           rstate <- getTHState inst
    1825 1827 
           loc <- location
    1826 1828 
           -- run a remote TH request
    1827 
    -      r <- liftIO $
    1829 
    +      qr <- withRunInIO $ \runInIO -> do
    1830 
    +        recovers <- newIORef []
    1828 1831 
             withForeignRef rstate $ \state_hv ->
    1829 
    -        withForeignRef fhv $ \q_hv ->
    1830 
    -          sendMessageDelayedResponse inst (RunTH state_hv q_hv ty (Just loc))
    1831 
    -      -- respond to requests from the interpreter
    1832 
    -      runRemoteTH inst []
    1833 
    -      -- get the final result
    1834 
    -      qr <- liftIO $ receiveDelayedResponse inst r
    1832 
    +          withForeignRef fhv $ \q_hv ->
    1833 
    +            sendTHComputation inst (\ii -> RunTH ii state_hv q_hv ty (Just loc)) $ \msg ->
    1834 
    +              runInIO (runRemoteTH recovers msg)
    1835 1835 
           bs <- checkQResult qr
    1836 1836 
           return $! runGet get (LB.fromStrict bs)
    1837 1837
    1838 1838
    1839 
    --- | communicate with a remotely-running TH computation until it finishes.
    1839 
    +-- | Communicate with a remotely-running TH computation until it finishes.
    1840 1840 
     -- See Note [Remote Template Haskell] in libraries/ghci/GHCi/TH.hs.
    1841 1841 
     runRemoteTH
    1842 
    -  :: ExtInterpInstance d
    1843 
    -  -> [Messages TcRnMessage]   --  saved from nested calls to qRecover
    1844 
    -  -> TcM ()
    1845 
    -runRemoteTH inst recovers = do
    1846 
    -  THMsg msg <- liftIO $ receiveTHMessage inst
    1842 
    +  :: Binary a
    1843 
    +  => IORef [Messages TcRnMessage] -- ^ saved from nested calls to qRecover
    1844 
    +  -> THMessage a
    1845 
    +  -> TcM (Bool, a)
    1846 
    +runRemoteTH recovers msg = do
    1847 1847 
       case msg of
    1848 
    -    RunTHDone -> return ()
    1848 
    +    RunTHDone -> return (True, ())
    1849 1849 
         StartRecover -> do -- Note [TH recover with -fexternal-interpreter]
    1850 1850 
           v <- getErrsVar
    1851 1851 
           msgs <- readTcRef v
    1852 1852 
           writeTcRef v emptyMessages
    1853 
    -      runRemoteTH inst (msgs : recovers)
    1853 
    +      modifyIORef recovers (msgs:) & liftIO
    1854 
    +      return (False, ())
    1854 1855 
         EndRecover caught_error -> do
    1855 
    -      let (prev_msgs, rest) = case recovers of
    1856 
    -             [] -> panic "EndRecover"
    1857 
    -             a : b -> (a,b)
    1856 
    +      (prev_msgs, rest) <- liftIO $
    1857 
    +        readIORef recovers >>= \case
    1858 
    +          [] -> panic "EndRecover"
    1859 
    +          a : b -> pure (a,b)
    1858 1860 
           v <- getErrsVar
    1859 1861 
           warn_msgs <- getWarningMessages <$> readTcRef v
    1860 1862 
           -- keep the warnings only if there were no errors
    1861 1863 
           writeTcRef v $ if caught_error
    1862 1864 
             then prev_msgs
    1863 1865 
             else mkMessages warn_msgs `unionMessages` prev_msgs
    1864 
    -      runRemoteTH inst rest
    1866 
    +      writeIORef recovers rest & liftIO
    1867 
    +      return (False, ())
    1865 1868 
         _other -> do
    1866 
    -      r <- handleTHMessage msg
    1867 
    -      liftIO $ sendAnyValue inst r
    1868 
    -      runRemoteTH inst recovers
    1869 
    +      (False,) <$> handleTHMessage msg
    1869 1870
    1870 1871 
     -- | Check a QResult
    1871 1872 
     checkQResult :: QResult a -> TcM a

  • libraries/ghci/GHCi/Message.hs
    1 
    -{-# LANGUAGE GADTs, DeriveGeneric, StandaloneDeriving, ScopedTypeVariables,
    2 
    -    GeneralizedNewtypeDeriving, ExistentialQuantification, RecordWildCards,
    3 
    -    CPP, NamedFieldPuns #-}
    1 
    +{-# LANGUAGE RecursiveDo, GADTs, DeriveGeneric, StandaloneDeriving, ScopedTypeVariables,
    2 
    +    GeneralizedNewtypeDeriving, ExistentialQuantification, RecordWildCards, CPP,
    3 
    +    NamedFieldPuns, LambdaCase, TypeAbstractions, TypeApplications,
    4 
    +    LinearTypes, RankNTypes, UnboxedTuples, BlockArguments, TupleSections, MagicHash #-}
    4 5 
     {-# OPTIONS_GHC -fno-warn-name-shadowing -fno-warn-orphans #-}
    5 6
    6 7 
     -- |
    ... ... @@ -22,7 +23,16 @@ module GHCi.Message
    22 23 
       , ResumeContext(..)
    23 24 
       , QState(..)
    24 25 
       , getMessage, putMessage, getTHMessage, putTHMessage
    25 
    -  , Pipe, mkPipeFromHandles, mkPipeFromContinuations, remoteCall, remoteTHCall, readPipe, writePipe
    26 
    +
    27 
    +  , Pipe(..), mkPipeFromHandles, mkPipeFromContinuations, closePipe
    28 
    +  , remoteCall, remoteTHCall
    29 
    +  , handleRemoteCall, handleRemoteTHCall
    30 
    +  , RemoteProcessException(..), PipeClosedException(..)
    31 
    +
    32 
    +  , InboxId(..)
    33 
    +  , newMailbox
    34 
    +  , closeMailbox
    35 
    +
    26 36 
       , BreakModule
    27 37 
       , BreakUnitId
    28 38 
       , LoadedDLL
    ... ... @@ -45,18 +55,14 @@ import qualified GHC.Exts.Heap as Heap
    45 55 
     import GHC.ForeignSrcLang
    46 56 
     import GHC.Fingerprint
    47 57 
     import GHC.Conc (pseq, par)
    58 
    +import GHC.IO (IO(..))
    48 59 
     import Control.Concurrent
    49 60 
     import Control.DeepSeq
    50 61 
     import Control.Exception
    51 
    -#if MIN_VERSION_base(4,20,0)
    52 
    -import Control.Exception.Context
    53 
    -#endif
    54 62 
     import Data.Binary
    55 63 
     import Data.Binary.Get
    56 64 
     import Data.Binary.Put
    57 65 
     import Data.ByteString (ByteString)
    58 
    -import qualified Data.ByteString as B
    59 
    -import qualified Data.ByteString.Builder as B
    60 66 
     import qualified Data.ByteString.Lazy as LB
    61 67 
     import qualified Data.ByteString.Short as BS
    62 68 
     import Data.Dynamic
    ... ... @@ -68,9 +74,7 @@ import GHC.Generics
    68 74 
     import GHC.Stack.CCS
    69 75 
     import qualified GHC.Boot.TH.Syntax        as TH
    70 76 
     import qualified GHC.Boot.TH.Monad         as TH
    71 
    -import System.Exit
    72 
    -import System.IO
    73 
    -import System.IO.Error
    77 
    +import GHCi.Message.Async
    74 78
    75 79 
     -- -----------------------------------------------------------------------------
    76 80 
     -- The RPC protocol between GHC and the interactive server
    ... ... @@ -214,14 +218,16 @@ data Message a where
    214 218 
       -- they did, we have to serialize the value anyway, so we might
    215 219 
       -- as well serialize it to force it.
    216 220 
       RunTH
    217 
    -   :: RemoteRef (IORef QState)
    221 
    +   :: InboxId -- ^ announce mailbox for related 'remoteTHCall's
    222 
    +   -> RemoteRef (IORef QState)
    218 223 
        -> HValueRef {- e.g. TH.Q TH.Exp -}
    219 224 
        -> THResultType
    220 225 
        -> Maybe TH.Loc
    221 226 
        -> Message (QResult ByteString)
    222 227
    223 228 
       -- | Run the given mod finalizers.
    224 
    -  RunModFinalizers :: RemoteRef (IORef QState)
    229 
    +  RunModFinalizers :: InboxId -- ^ announce mailbox for related 'remoteTHCall's
    230 
    +                   -> RemoteRef (IORef QState)
    225 231 
                        -> [RemoteRef (TH.Q ())]
    226 232 
                        -> Message (QResult ())
    227 233
    ... ... @@ -458,45 +464,6 @@ data BreakUnitId
    458 464 
     -- | A dummy type that tags pointers returned by 'LoadDLLs'.
    459 465 
     data LoadedDLL
    460 466
    461 
    --- SomeException can't be serialized because it contains dynamic
    462 
    --- types.  However, we do very limited things with the exceptions that
    463 
    --- are thrown by interpreted computations:
    464 
    ---
    465 
    --- * We print them, e.g. "*** Exception: <something>"
    466 
    --- * UserInterrupt has a special meaning
    467 
    --- * In ghc -e, exitWith should exit with the appropriate exit code
    468 
    ---
    469 
    --- So all we need to do is distinguish UserInterrupt and ExitCode, and
    470 
    --- all other exceptions can be represented by their 'show' string.
    471 
    ---
    472 
    -data SerializableException
    473 
    -  = EUserInterrupt
    474 
    -  | EExitCode ExitCode
    475 
    -  | EOtherException String
    476 
    -  deriving (Generic, Show)
    477 
    -
    478 
    -toSerializableException :: SomeException -> SerializableException
    479 
    -toSerializableException ex
    480 
    -  | Just UserInterrupt <- fromException ex  = EUserInterrupt
    481 
    -  | Just (ec::ExitCode) <- fromException ex = (EExitCode ec)
    482 
    -  | otherwise = EOtherException $
    483 
    -#if MIN_VERSION_base(4,20,0)
    484 
    -      -- Exception plus backtrace as seen in `displayExceptionWithInfo`
    485 
    -      case displayExceptionContext (someExceptionContext ex) of
    486 
    -        "" -> displayException (ex :: SomeException)
    487 
    -        cx -> displayException (ex :: SomeException) ++ "\n\n" ++ cx
    488 
    -#else
    489 
    -      show (ex :: SomeException)
    490 
    -#endif
    491 
    -
    492 
    -fromSerializableException :: SerializableException -> SomeException
    493 
    -fromSerializableException EUserInterrupt = toException UserInterrupt
    494 
    -fromSerializableException (EExitCode c) = toException c
    495 
    -fromSerializableException (EOtherException str) = toException (ErrorCall str)
    496 
    -
    497 
    -instance Binary ExitCode
    498 
    -instance Binary SerializableException
    499 
    -
    500 467 
     data THResult a
    501 468 
       = THException String
    502 469 
       | THComplete a
    ... ... @@ -513,11 +480,13 @@ instance Binary THResultType
    513 480 
     -- StartTH message.  A new one is created per module that GHC
    514 481 
     -- typechecks.
    515 482 
     data QState = QState
    516 
    -  { qsMap        :: Map TypeRep Dynamic
    483 
    +  { qsMap      :: Map TypeRep Dynamic
    517 484 
            -- ^ persistent data between splices in a module
    518 
    -  , qsLocation   :: Maybe TH.Loc
    485 
    +  , qsLocation :: Maybe TH.Loc
    519 486 
            -- ^ location for current splice, if any
    520 
    -  , qsPipe :: Pipe
    487 
    +  , qsInboxId  :: InboxId
    488 
    +      -- ^ the mailbox expecting requests for each RunTH-like run
    489 
    +  , qsPipe     :: Pipe
    521 490 
            -- ^ pipe to communicate with GHC
    522 491 
       }
    523 492 
     instance Show QState where show _ = "<QState>"
    ... ... @@ -600,9 +569,9 @@ getMessage = do
    600 569 
           29 -> Msg <$> (BreakpointStatus <$> get <*> get)
    601 570 
           30 -> Msg <$> (GetBreakpointVar <$> get <*> get)
    602 571 
           31 -> Msg <$> return StartTH
    603 
    -      32 -> Msg <$> (RunModFinalizers <$> get <*> get)
    572 
    +      32 -> Msg <$> (RunModFinalizers <$> get <*> get <*> get)
    604 573 
           33 -> Msg <$> (AddSptEntry <$> get <*> get)
    605 
    -      34 -> Msg <$> (RunTH <$> get <*> get <*> get <*> get)
    574 
    +      34 -> Msg <$> (RunTH <$> get <*> get <*> get <*> get <*> get)
    606 575 
           35 -> Msg <$> (GetClosure <$> get)
    607 576 
           36 -> Msg <$> (Seq <$> get)
    608 577 
           37 -> Msg <$> return RtsRevertCAFs
    ... ... @@ -648,9 +617,9 @@ putMessage m = case m of
    648 617 
       BreakpointStatus arr ix     -> putWord8 29 >> put arr >> put ix
    649 618 
       GetBreakpointVar a b        -> putWord8 30 >> put a >> put b
    650 619 
       StartTH                     -> putWord8 31
    651 
    -  RunModFinalizers a b        -> putWord8 32 >> put a >> put b
    620 
    +  RunModFinalizers a b c      -> putWord8 32 >> put a >> put b >> put c
    652 621 
       AddSptEntry a b             -> putWord8 33 >> put a >> put b
    653 
    -  RunTH st q loc ty           -> putWord8 34 >> put st >> put q >> put loc >> put ty
    622 
    +  RunTH ii st q loc ty        -> putWord8 34 >> put ii >> put st >> put q >> put loc >> put ty
    654 623 
       GetClosure a                -> putWord8 35 >> put a
    655 624 
       Seq a                       -> putWord8 36 >> put a
    656 625 
       RtsRevertCAFs               -> putWord8 37
    ... ... @@ -688,73 +657,40 @@ serializeBCOs rbcos = parMap doChunk (chunkList 100 rbcos)
    688 657 
       chunkList _ [] = []
    689 658 
       chunkList n xs = as : chunkList n bs where (as,bs) = splitAt n xs
    690 659
    691 
    --- -----------------------------------------------------------------------------
    692 
    --- Reading/writing messages
    693 
    -
    694 
    --- | An opaque pipe for bidirectional binary data transmission.
    695 
    -data Pipe = Pipe
    696 
    -  { getSome :: !(IO ByteString)
    697 
    -  , putAll :: !(B.Builder -> IO ())
    698 
    -  , pipeLeftovers :: !(IORef (Maybe ByteString))
    699 
    -  }
    700 
    -
    701 
    --- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write.
    702 
    -mkPipeFromHandles :: Handle -> Handle -> IO Pipe
    703 
    -mkPipeFromHandles pipeRead pipeWrite = do
    704 
    -  let getSome = B.hGetSome pipeRead (32*1024)
    705 
    -      putAll b = do
    706 
    -        B.hPutBuilder pipeWrite b
    707 
    -        hFlush pipeWrite
    708 
    -  pipeLeftovers <- newIORef Nothing
    709 
    -  pure $ Pipe { getSome, putAll, pipeLeftovers }
    710 
    -
    711 
    --- | Make a 'Pipe' from a reader function and a writer function.
    712 
    -mkPipeFromContinuations :: IO ByteString -> (B.Builder -> IO ()) -> IO Pipe
    713 
    -mkPipeFromContinuations getSome putAll = do
    714 
    -  pipeLeftovers <- newIORef Nothing
    715 
    -  pure $ Pipe { getSome, putAll, pipeLeftovers }
    716 
    -
    660 
    +-- | Send a 'Message' request to and wait for a reply from the interpreter
    661 
    +-- (blocks on the calling side, doesn't necessarily block the interpreter)
    717 662 
     remoteCall :: Binary a => Pipe -> Message a -> IO a
    718 
    -remoteCall pipe msg = do
    719 
    -  writePipe pipe (putMessage msg)
    720 
    -  readPipe pipe get
    721 
    -
    722 
    -writePipe :: Pipe -> Put -> IO ()
    723 
    -writePipe Pipe{..} put = putAll $ execPut put
    724 
    -
    725 
    -remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a
    726 
    -remoteTHCall pipe msg = do
    727 
    -  writePipe pipe (putTHMessage msg)
    728 
    -  readPipe pipe get
    729 
    -
    730 
    -readPipe :: Pipe -> Get a -> IO a
    731 
    -readPipe Pipe{..} get = do
    732 
    -  leftovers <- readIORef pipeLeftovers
    733 
    -  m <- getBin getSome get leftovers
    734 
    -  case m of
    735 
    -    Nothing -> throw $
    736 
    -      mkIOError eofErrorType "GHCi.Message.readPipe" Nothing Nothing
    737 
    -    Just (result, new_leftovers) -> do
    738 
    -      writeIORef pipeLeftovers new_leftovers
    739 
    -      return result
    740 
    -
    741 
    -getBin
    742 
    -  :: (IO ByteString) -> Get a -> Maybe ByteString
    743 
    -  -> IO (Maybe (a, Maybe ByteString))
    744 
    -
    745 
    -getBin getsome get leftover = go leftover (runGetIncremental get)
    746 
    - where
    747 
    -   go Nothing (Done leftover _ msg) =
    748 
    -     return (Just (msg, if B.null leftover then Nothing else Just leftover))
    749 
    -   go _ Done{} = throwIO (ErrorCall "getBin: Done with leftovers")
    750 
    -   go (Just leftover) (Partial fun) = do
    751 
    -     go Nothing (fun (Just leftover))
    752 
    -   go Nothing (Partial fun) = do
    753 
    -     -- putStrLn "before hGetSome"
    754 
    -     b <- getsome
    755 
    -     -- putStrLn $ "hGetSome: " ++ show (B.length b)
    756 
    -     if B.null b
    757 
    -        then return Nothing
    758 
    -        else go Nothing (fun (Just b))
    759 
    -   go _lft (Fail _rest _off str) =
    760 
    -     throwIO (ErrorCall ("getBin: " ++ str))
    663 
    +remoteCall pipe msg = either throwIO (pure . runGet get)
    664 
    +                  =<< remoteAsyncReq pipe (putMessage msg) Nothing
    665 
    +
    666 
    +-- | Handle a single 'Message' request on the interpreter and reply to the host
    667 
    +handleRemoteCall :: Pipe
    668 
    +                 -> (Msg -> IO Msg) {-^ Msg hook -}
    669 
    +                 -> (forall a. (Show a, Binary a) => Message a -> IO a)
    670 
    +                 -> IO ()
    671 
    +handleRemoteCall pipe hook k =
    672 
    +  handleAsyncReq pipe getMessage Nothing \writer msg0 -> IO \s0 ->
    673 
    +    case hook msg0 |#> s0 of -- unIO for linearity checking
    674 
    +      (# s1, Msg msg #) -> case k msg |#> s1 of
    675 
    +        (# s2, a #) -> writeReply writer (put a) |#> s2
    676 
    +
    677 
    +-- | Send a 'THMessage' request to the host and wait for a reply from the host
    678 
    +-- to the interpreter.
    679 
    +--
    680 
    +-- A 'THMessage' request from the interpreter is always associated with a
    681 
    +-- 'RunTH'-like run. Since the host may have many in-flight TH runs, we always
    682 
    +-- want a specific host 'InboxId' (which is announced in the 'RunTH' request)
    683 
    +remoteTHCall :: Binary a => Pipe -> InboxId -> THMessage a -> IO a
    684 
    +remoteTHCall pipe ii msg = either throwIO (pure . runGet get)
    685 
    +                       =<< remoteAsyncReq pipe (putTHMessage msg) (Just ii)
    686 
    +
    687 
    +-- | Handle a 'THMessage' request on the host and reply to the interpreter.
    688 
    +-- Symmterically to 'remoteTHCall', we always want to specify the 'InboxId' for
    689 
    +-- RunTH "reverse requests"
    690 
    +handleRemoteTHCall
    691 
    +  :: Pipe -> InboxId -> (forall a. (Show a, Binary a) => THMessage a -> IO a) -> IO ()
    692 
    +handleRemoteTHCall pipe ii k =
    693 
    +  handleAsyncReq pipe getTHMessage (Just ii) \writer (THMsg msg) -> IO \s0 ->
    694 
    +    case k msg |#> s0 of -- unIO for linearity checking
    695 
    +      (# s1, a #) -> writeReply writer (put a) |#> s1
    696 
    +

  • libraries/ghci/GHCi/Message/Async.hs
    1 
    +{-# LANGUAGE RecursiveDo, GADTs, DeriveGeneric, StandaloneDeriving, ScopedTypeVariables,
    2 
    +    GeneralizedNewtypeDeriving, ExistentialQuantification, RecordWildCards, CPP,
    3 
    +    NamedFieldPuns, LambdaCase, TypeAbstractions, TypeApplications,
    4 
    +    LinearTypes, RankNTypes, UnboxedTuples, BlockArguments, TupleSections, MagicHash #-}
    5 
    +{-# OPTIONS_GHC -fno-warn-name-shadowing -fno-warn-orphans #-}
    6 
    +module GHCi.Message.Async
    7 
    +  (
    8 
    +    -- * Pipes
    9 
    +    Pipe(..), mkPipeFromHandles, mkPipeFromContinuations
    10 
    +  , closePipe, PipeReadClosedException(..)
    11 
    +
    12 
    +    -- * Interface for async bidirectional RPC
    13 
    +  , remoteAsyncReq, handleAsyncReq
    14 
    +  , RemoteProcessException(..), DecodeAsyncReqException(..)
    15 
    +
    16 
    +    -- ** Mailboxes
    17 
    +  , InboxId(..)
    18 
    +  , newMailbox
    19 
    +  , closeMailbox
    20 
    +
    21 
    +    -- ** Async matching internals
    22 
    +  , AsyncManager(..), CorrelationId
    23 
    +  , IddMsg(..), newAsyncManager
    24 
    +
    25 
    +    -- ** Consume `reply` exactly once
    26 
    +  , WriteReply(..)
    27 
    +  , writeReply, (|#>)
    28 
    +
    29 
    +    -- * Serializable exceptions
    30 
    +  , SerializableException(..)
    31 
    +  , toSerializableException, fromSerializableException
    32 
    +  ) where
    33 
    +
    34 
    +import Prelude -- See note [Why do we import Prelude here?]
    35 
    +import GHCi.TH.Binary () -- For Binary instances
    36 
    +
    37 
    +import GHC.IO (IO(..))
    38 
    +import Control.Concurrent
    39 
    +import Control.Exception
    40 
    +#if MIN_VERSION_base(4,20,0)
    41 
    +import Control.Exception.Context
    42 
    +#endif
    43 
    +import Data.Binary
    44 
    +import Data.Binary.Get
    45 
    +import Data.Binary.Put
    46 
    +import Data.ByteString (ByteString)
    47 
    +import qualified Data.ByteString as B
    48 
    +import qualified Data.ByteString.Builder as B
    49 
    +import qualified Data.ByteString.Lazy as LB
    50 
    +import Data.IORef
    51 
    +import Data.IntMap (IntMap)
    52 
    +import qualified Data.IntMap as IntMap
    53 
    +import GHC.Generics
    54 
    +import System.Exit
    55 
    +import System.IO
    56 
    +import Control.Monad
    57 
    +import Control.Concurrent.STM
    58 
    +import Data.Bifunctor
    59 
    +import GHC.Base (State#)
    60 
    +import GHC.Exts (RealWorld)
    61 
    +
    62 
    +--------------------------------------------------------------------------------
    63 
    +-- * Reading/writing and Pipes
    64 
    +--------------------------------------------------------------------------------
    65 
    +
    66 
    +-- | An opaque pipe for bidirectional binary data transmission.
    67 
    +data Pipe = Pipe
    68 
    +  { getSome         :: !(IO ByteString)
    69 
    +  , putAll          :: !(B.Builder -> IO ())
    70 
    +  , pipeLeftovers   :: !(IORef (Maybe ByteString))
    71 
    +  , pipeLock        :: !(MVar ()) -- ^ Lock to prevent concurrent writes to the stream
    72 
    +  , pipeAsyncMngr   :: !AsyncManager
    73 
    +  , pipeCloseRead   :: !(IO ())
    74 
    +  -- ^ closes read end of the pipe
    75 
    +  , pipeCloseWrite  :: !(IO ())
    76 
    +  -- ^ closes write end of the pipe
    77 
    +  , pipeReadClosed  :: !(TVar Bool)
    78 
    +    -- ^ After the pipe is closed, messages can no longer be received from the
    79 
    +    -- other end, but the 'AsyncManager' queues might not be empty yet.
    80 
    +  }
    81 
    +
    82 
    +-- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write.
    83 
    +-- Must call 'closePipe' at the end to clean up.
    84 
    +mkPipeFromHandles :: Handle -> Handle -> IO Pipe
    85 
    +mkPipeFromHandles pipeRead pipeWrite = do
    86 
    +  let getSome = B.hGetSome pipeRead (32*1024)
    87 
    +      putAll b = do
    88 
    +        B.hPutBuilder pipeWrite b
    89 
    +        hFlush pipeWrite
    90 
    +  mkPipeFromContinuations getSome putAll (hClose pipeRead) (hClose pipeWrite)
    91 
    +
    92 
    +-- | Make a 'Pipe' from a reader function and a writer function.
    93 
    +-- The pipe should be closed with 'closePipe', which will call 'pipeCloseRead' and 'pipeCloseWrite'.
    94 
    +mkPipeFromContinuations :: IO ByteString -> (B.Builder -> IO ())
    95 
    +                        -> IO ()
    96 
    +                        -- ^ pipeCloseRead
    97 
    +                        -> IO ()
    98 
    +                        -- ^ pipeCloseWrite
    99 
    +                        -> IO Pipe
    100 
    +mkPipeFromContinuations getSome putAll pipeCloseRead pipeCloseWrite = mdo
    101 
    +  pipeLeftovers   <- newIORef Nothing
    102 
    +  pipeLock        <- newMVar ()
    103 
    +  pipeReadClosed  <- newTVarIO False
    104 
    +  mdo
    105 
    +    pipeAsyncMngr <- newAsyncManager p
    106 
    +    let p = Pipe { .. }
    107 
    +    pure p
    108 
    +
    109 
    +-- | Cleanly close our ends of the Pipe, terminating the async manager's
    110 
    +-- drain-read-end thread. Also calls 'pipeCloseRead' and 'pipeCloseWrite', in that order.
    111 
    +--
    112 
    +-- If the pipe was created using 'mkPipeFromHandles', the handles are also closed.
    113 
    +--
    114 
    +-- The requests we've sent that are pending are all canceled with
    115 
    +-- 'PipeReadClosedException' by ourselves.
    116 
    +--
    117 
    +-- No special notification is sent to the remote process, a message to that
    118 
    +-- effect should be sent ahead of time if one is needed (see 'Shutdown').
    119 
    +closePipe :: Pipe -> IO ()
    120 
    +closePipe Pipe{..} = do
    121 
    +
    122 
    +  readAlreadyClosed <- atomically $ swapTVar pipeReadClosed True
    123 
    +
    124 
    +  -- this triggers pipeCloseRead
    125 
    +  unless readAlreadyClosed $ do
    126 
    +
    127 
    +    -- releases handle lock taken by reading, and cleans up by replying to every
    128 
    +    -- pending request with 'PipeReadClosedException'.
    129 
    +    throwTo (asyncDrainTid pipeAsyncMngr) PipeReadClosedException
    130 
    +
    131 
    +  pipeCloseWrite
    132 
    +
    133 
    +-- | We've closed the pipe read end on this process
    134 
    +data PipeReadClosedException = PipeReadClosedException deriving Show
    135 
    +instance Exception PipeReadClosedException
    136 
    +
    137 
    +--------------------------------------------------------------------------------
    138 
    +-- ** Pipe thread-safe read/write
    139 
    +--------------------------------------------------------------------------------
    140 
    +-- All messages on the wire are 'IddMsg', which contain a correlation header
    141 
    +-- and identify whether they are requests or replies.
    142 
    +
    143 
    +writePipe :: Pipe -> IddMsg -> IO ()
    144 
    +writePipe p@Pipe{..} = withLock p . putAll . execPut . put
    145 
    +
    146 
    +-- | Only the 'drainPipe' thread should read from the pipe directly. Other
    147 
    +-- functions should wait for the matching reply to be delivered asynchronously.
    148 
    +-- (see 'withAsyncReq'). It doesn't lock to read bc it assumes single reader is 'drainPipe'.
    149 
    +--
    150 
    +-- Returns @Nothing@ on EOF.
    151 
    +readPipe :: Pipe -> IO (Maybe IddMsg)
    152 
    +readPipe Pipe{..} = do
    153 
    +  leftovers <- readIORef pipeLeftovers
    154 
    +  m <- getBin getSome get leftovers
    155 
    +  case m of
    156 
    +    Nothing -> return Nothing
    157 
    +    Just (result, new_leftovers) -> do
    158 
    +      writeIORef pipeLeftovers new_leftovers
    159 
    +      return (Just result)
    160 
    +
    161 
    +getBin
    162 
    +  :: (IO ByteString) -> Get a -> Maybe ByteString
    163 
    +  -> IO (Maybe (a, Maybe ByteString))
    164 
    +
    165 
    +getBin getsome get leftover = go leftover (runGetIncremental get)
    166 
    + where
    167 
    +   go Nothing (Done leftover _ msg) =
    168 
    +     return (Just (msg, if B.null leftover then Nothing else Just leftover))
    169 
    +   go _ Done{} = throwIO (ErrorCall "getBin: Done with leftovers")
    170 
    +   go (Just leftover) (Partial fun) = do
    171 
    +     go Nothing (fun (Just leftover))
    172 
    +   go Nothing (Partial fun) = do
    173 
    +     -- putStrLn "before hGetSome"
    174 
    +     b <- getsome
    175 
    +     -- putStrLn $ "hGetSome: " ++ show (B.length b)
    176 
    +     if B.null b
    177 
    +        then return Nothing
    178 
    +        else go Nothing (fun (Just b))
    179 
    +   go _lft (Fail _rest _off str) =
    180 
    +     throwIO (ErrorCall ("getBin: " ++ str))
    181 
    +
    182 
    +withLock :: Pipe -> IO c -> IO c
    183 
    +withLock Pipe{..} = bracket (takeMVar pipeLock) (putMVar pipeLock) . const
    184 
    +
    185 
    +--------------------------------------------------------------------------------
    186 
    +-- * Matching requests with async replies
    187 
    +--------------------------------------------------------------------------------
    188 
    +
    189 
    +-- | Send an asynchronous request and block waiting for a matching reply
    190 
    +remoteAsyncReq :: Pipe -> Put -> (Maybe InboxId) -> IO (Either SomeException LB.ByteString)
    191 
    +remoteAsyncReq pipe msg ii = do
    192 
    +
    193 
    +  let pendingReplies = pendingReply (pipeAsyncMngr pipe)
    194 
    +
    195 
    +  -- Mark pending request
    196 
    +  uq <- freshId (pipeAsyncMngr pipe)
    197 
    +  wv <- newEmptyTMVarIO
    198 
    +
    199 
    +  isClosed <- atomically $ do
    200 
    +    isClosed <- readTVar (pipeReadClosed pipe)
    201 
    +                -- don't send the req if we won't get a reply.
    202 
    +    if isClosed
    203 
    +      then pure True
    204 
    +      else do
    205 
    +        modifyTVar pendingReplies (IntMap.insert uq wv)
    206 
    +        pure False
    207 
    +
    208 
    +  when isClosed $ do
    209 
    +    throwIO PipeReadClosedException
    210 
    +
    211 
    +  -- Write request
    212 
    +  writePipe pipe (Request uq ii (runPut msg))
    213 
    +    `onException`
    214 
    +      -- the req wasn't done, or no one will be reading the reply:
    215 
    +      -- in either case, drop it from the pending map.
    216 
    +      atomically (modifyTVar pendingReplies (IntMap.delete uq))
    217 
    +
    218 
    +  -- Block waiting for reply
    219 
    +  atomically (takeTMVar wv)
    220 
    +    `onException`
    221 
    +      atomically (modifyTVar pendingReplies (IntMap.delete uq))
    222 
    +
    223 
    +-- | Receive a single asynchronous request and send a matching reply
    224 
    +-- asynchronously (the handler may fork a computation that eventually replies
    225 
    +-- and can return before actually replying, as long as it eventually does).
    226 
    +--
    227 
    +-- The handler receives as an argument the function to use to reply to the request.
    228 
    +-- It must be used exactly once (all requests expect a reply, and you can't
    229 
    +-- reply more than once). Linear types enforce the handler replies in every
    230 
    +-- branch, and if the handler otherwise throws, it is handled by 'handleAsyncReq'.
    231 
    +handleAsyncReq :: Pipe -> Get a -> (Maybe InboxId) -> (WriteReply Put %1 -> a -> IO r) -> IO r
    232 
    +handleAsyncReq pipe getit mii k = mask $ \restore -> do
    233 
    +  let am = pipeAsyncMngr pipe
    234 
    +  ack <- newEmptyMVar
    235 
    +
    236 
    +  -- Block waiting for request
    237 
    +  mbreq <- atomically $ do
    238 
    +    isClosed <- readTVar (pipeReadClosed pipe)
    239 
    +    taken <- case mii of
    240 
    +      Nothing -> tryReadTQueue  $  requestsQueue am
    241 
    +      Just ii -> tryReadTQueue =<< getMailbox am ii
    242 
    +    case taken of
    243 
    +      Just req -> pure (Just req)
    244 
    +      Nothing  | isClosed  -> pure Nothing -- empty and closed
    245 
    +               | otherwise -> retry -- wait for msg
    246 
    +
    247 
    +  case mbreq of
    248 
    +    Nothing -> throwIO PipeReadClosedException -- can't expect reqs
    249 
    +    Just (uq, bs) -> do
    250 
    +      let reply p = do
    251 
    +            writePipe pipe (Reply uq (bimap toSerializableException runPut p))
    252 
    +            putMVar ack ()
    253 
    +
    254 
    +      -- Run the continuation which must reply exactly once with 'writeReply'
    255 
    +      case runGetOrFail getit bs of
    256 
    +        Left (_, _, err) -> do
    257 
    +          let exc = DecodeAsyncReqException err
    258 
    +          reply (Left (toException exc))
    259 
    +          throwIO exc
    260 
    +        Right (_, _, it) ->
    261 
    +          restore (k (WriteReply reply) it)
    262 
    +
    263 
    +            -- If there's an exception before the reply is sent, reply with an error instead.
    264 
    +            `catch` \(e :: SomeException) -> do
    265 
    +
    266 
    +                tryTakeMVar ack >>= \case
    267 
    +                  Just () ->
    268 
    +                    -- Already replied
    269 
    +                    throwIO e
    270 
    +                  Nothing -> do
    271 
    +                    -- Reply with an error
    272 
    +                    reply (Left e)
    273 
    +                    throwIO e
    274 
    +
    275 
    +-- | 'handleSyncReq' failed to decode an incoming request payload using the given @'Get' a@
    276 
    +newtype DecodeAsyncReqException = DecodeAsyncReqException String deriving Show
    277 
    +instance Exception DecodeAsyncReqException
    278 
    +
    279 
    +
    280 
    +-- ** Mailboxes ----------------------------------------------------------------
    281 
    +
    282 
    +-- | A 'Request' may optionally declare an 'InboxId'. If it does, it can only
    283 
    +-- be read by a 'handleAsyncReq' for that specific 'InboxId'. That is,
    284 
    +-- @'handleAsyncReq' p get Nothing@ will not receive messages meant for a
    285 
    +-- specific inbox.
    286 
    +--
    287 
    +-- This is useful to prevent mixing up reverse requests, where we want to send
    288 
    +-- many reverse requests but have the caller handle them without interleaving
    289 
    +-- them.
    290 
    +newtype InboxId = InboxId Int deriving (Show, Binary)
    291 
    +
    292 
    +-- | Create a new mailbox from where we can read all requests addressed to this 'InboxId'.
    293 
    +newMailbox :: AsyncManager -> IO InboxId
    294 
    +newMailbox am = InboxId <$> freshId am
    295 
    +  -- `mailboxQueues` is written on demand, no need to do it here.
    296 
    +
    297 
    +-- | Clean up a created mailbox by 'InboxId'.
    298 
    +closeMailbox :: AsyncManager -> InboxId -> IO ()
    299 
    +closeMailbox AsyncManager{..} (InboxId ii) =
    300 
    +  atomically (modifyTVar mailboxQueues (IntMap.delete ii))
    301 
    +
    302 
    +-- | Get or create a new mailbox for this 'InboxId'
    303 
    +getMailbox :: AsyncManager -> InboxId -> STM (TQueue (CorrelationId, LB.ByteString))
    304 
    +getMailbox am (InboxId ii) = do
    305 
    +  mboxes <- readTVar (mailboxQueues am)
    306 
    +  case IntMap.lookup ii mboxes of
    307 
    +    Nothing -> do
    308 
    +      q <- newTQueue
    309 
    +      writeTVar (mailboxQueues am) (IntMap.insert ii q mboxes)
    310 
    +      pure q
    311 
    +    Just q ->
    312 
    +      pure q
    313 
    +
    314 
    +-- ** Async matching "internals" -----------------------------------------------
    315 
    +
    316 
    +data AsyncManager = AsyncManager
    317 
    +  { pendingReply :: !(TVar (IntMap (TMVar (Either SomeException LB.ByteString))))
    318 
    +    -- ^ When a request is sent, keep track that we're expecting a reply. The
    319 
    +    -- TMVar is signaled when the reply arrives. The exception can originate
    320 
    +    -- from the host process, or from the remote one (with 'RemoteProcessException').
    321 
    +  , requestsQueue :: !(TQueue (CorrelationId, LB.ByteString))
    322 
    +    -- ^ A queue of pending received requests. When a request is received
    323 
    +    -- /without/ a specific 'InboxId', queue it here. A request handler will
    324 
    +    -- pop the next request from it. (Nothing except the 'drainPipe' thread
    325 
    +    -- should read from the pipe). Read with @'handleAsyncReq' p get Nothing@.
    326 
    +  , mailboxQueues :: !(TVar (IntMap (TQueue (CorrelationId, LB.ByteString))))
    327 
    +    -- ^ Requests for a specific 'InboxId' that can be read using
    328 
    +    -- @'handleAsyncReq' p get (Just <mailboxid>)@.
    329 
    +  , asyncDrainTid :: !ThreadId
    330 
    +  , asyncFreshId  :: !(IORef Int)
    331 
    +  }
    332 
    +
    333 
    +-- | Each 'Request' expects a 'Reply' with a matching 'CorrelationId'.
    334 
    +type CorrelationId = Int
    335 
    +
    336 
    +-- | We have a single type for requests and replies because both the host and
    337 
    +-- interpreter may send requests *and* replies on the write end *and* on recv
    338 
    +-- both on the read end.
    339 
    +--
    340 
    +-- For instance:
    341 
    +--  1. The host sends a RunTH request on its Host Write End (HWE)
    342 
    +--  2. The interpreter sends a NewName TH request on its Interpreter Write End (IWE)
    343 
    +--  3. The host sends a reply to the TH request on its HWE
    344 
    +--  4. The interpreter sends the reply to the RunTH request on its IWE
    345 
    +--
    346 
    +-- Additionally, we want to keep concurrent TH requests for independent RunTH
    347 
    +-- calls separated. Thus the interpreter requests additionally specify a
    348 
    +-- mailbox announced by RunTH:
    349 
    +--  1. To interpreter:   @Request _ Nothing (RunTH <mailid>)@
    350 
    +--  2. From interpreter: @Request _ (Just <mailid>) NewName@
    351 
    +-- And the host will relate the <mailid> reverse requests to that RunTH run
    352 
    +-- exclusively.
    353 
    +data IddMsg  = Request !CorrelationId !(Maybe InboxId) LB.ByteString
    354 
    +             | Reply   !CorrelationId (Either SerializableException LB.ByteString)
    355 
    +
    356 
    +instance Binary IddMsg where
    357 
    +  put (Request i ii bs) = putWord8 0 >> put i >> put ii >> put bs
    358 
    +  put (Reply i bs)      = putWord8 1 >> put i >> put bs
    359 
    +  get = do
    360 
    +    b <- getWord8
    361 
    +    case b of
    362 
    +      0 -> Request <$> get <*> get <*> get
    363 
    +      1 -> Reply   <$> get <*> get
    364 
    +      _ -> fail $ "impossible: Binary IddMsg - " ++ show b
    365 
    +
    366 
    +newAsyncManager :: Pipe -> IO AsyncManager
    367 
    +newAsyncManager pipe = do
    368 
    +  asyncFreshId  <- newIORef 0
    369 
    +  pendingReply  <- newTVarIO IntMap.empty
    370 
    +  requestsQueue <- newTQueueIO
    371 
    +  mailboxQueues <- newTVarIO IntMap.empty
    372 
    +
    373 
    +  -- Fork thread to drain read end.
    374 
    +  -- (cleaned up on 'closePipe')
    375 
    +  asyncDrainTid <- forkIOWithUnmask (drainPipe pipe)
    376 
    +
    377 
    +  let am = AsyncManager{..}
    378 
    +  pure am
    379 
    +
    380 
    +-- | Drains the read end of the Pipe and match 'Reply'es against pending 'Request's.
    381 
    +--
    382 
    +-- In pratice, we'll have both a thread draining the read end of the host's
    383 
    +-- 'Pipe' and a thread draining the read end of the interpreter's 'Pipe', since
    384 
    +-- both processes construct a 'Pipe' and send and receive messages
    385 
    +--
    386 
    +-- Closes the Pipe read end when done reading
    387 
    +drainPipe :: Pipe -> (forall x. IO x -> IO x) -> IO ()
    388 
    +drainPipe p unmask = catch (unmask $ loop `finally` pipeCloseRead p) replyToPendings  where
    389 
    +  loop = do
    390 
    +    readPipe p >>= \case
    391 
    +      -- Receive a reply to a pending notification
    392 
    +      Just (Reply uq bs) -> do
    393 
    +        atomically $ do
    394 
    +          m <- readTVar (pendingReply am)
    395 
    +          case IntMap.lookup uq m of
    396 
    +            Nothing ->
    397 
    +              -- the req was dropped from pending bc there was an exception, and no
    398 
    +              -- one's listening any more. the reply is ignored
    399 
    +              pure ()
    400 
    +            Just wv -> do
    401 
    +              writeTVar (pendingReply am) (IntMap.delete uq m)
    402 
    +              -- signal matching request
    403 
    +              putTMVar wv (bimap (toException . RemoteProcessException) id bs)
    404 
    +        loop
    405 
    +      -- Receive a request and queue it
    406 
    +      Just (Request uq mbi bs) -> do
    407 
    +        atomically $ case mbi of
    408 
    +          Nothing ->
    409 
    +            -- "Global" request
    410 
    +            writeTQueue (requestsQueue am) (uq,bs)
    411 
    +          Just ii -> do
    412 
    +            -- "Specific mailbox" request
    413 
    +            q <- getMailbox am ii
    414 
    +            writeTQueue q (uq,bs)
    415 
    +        loop
    416 
    +      -- EOF on read handle
    417 
    +      Nothing -> do
    418 
    +        -- clean up pipe then exit loop
    419 
    +        replyToPendings (toException PipeReadClosedException)
    420 
    +        pure ()
    421 
    +
    422 
    +  replyToPendings e = atomically $ do
    423 
    +    -- reqs are only added if pipe not closed;
    424 
    +    -- thus we close and take all pending atomically.
    425 
    +    writeTVar (pipeReadClosed p) True
    426 
    +    pendings <- swapTVar (pendingReply am) IntMap.empty -- take pending
    427 
    +    mapM_ (\p -> putTMVar p (Left e {-host exception-})) pendings
    428 
    +
    429 
    +  am = pipeAsyncMngr p
    430 
    +
    431 
    +-- | An exception that can be thrown on the host process representing an
    432 
    +-- expection that happened on the remote one.
    433 
    +newtype RemoteProcessException = RemoteProcessException SerializableException deriving Show
    434 
    +instance Exception RemoteProcessException
    435 
    +
    436 
    +-- | Come up with a fresh id for 'CorrelationId' or 'InboxId'
    437 
    +freshId :: AsyncManager -> IO Int
    438 
    +freshId AsyncManager{asyncFreshId} =
    439 
    +  atomicModifyIORef' asyncFreshId $ \i -> (i+1, i)
    440 
    +
    441 
    +--------------------------------------------------------------------------------
    442 
    +-- * Consume `reply` exactly once
    443 
    +--------------------------------------------------------------------------------
    444 
    +
    445 
    +-- | Every request should get exactly one reply.
    446 
    +data WriteReply a where
    447 
    +  WriteReply :: (Either SomeException a -> IO ()) %1 -> WriteReply a
    448 
    +
    449 
    +-- | Write a reply exactly once (consuming the ability to write more replies)
    450 
    +writeReply :: WriteReply a %1 -> a -> IO ()
    451 
    +writeReply (WriteReply r) = \x -> r (Right x)
    452 
    +
    453 
    +-- | Infix, linear, 'unIO'
    454 
    +(|#>) :: IO a %1 -> State# RealWorld -> (# State# RealWorld, a #)
    455 
    +(|#>) (IO fs) s = fs s
    456 
    +
    457 
    +--------------------------------------------------------------------------------
    458 
    +-- * Serializable Exceptions
    459 
    +--------------------------------------------------------------------------------
    460 
    +
    461 
    +-- SomeException can't be serialized because it contains dynamic
    462 
    +-- types.  However, we do very limited things with the exceptions that
    463 
    +-- are thrown by interpreted computations:
    464 
    +--
    465 
    +-- * We print them, e.g. "*** Exception: <something>"
    466 
    +-- * UserInterrupt has a special meaning
    467 
    +-- * In ghc -e, exitWith should exit with the appropriate exit code
    468 
    +--
    469 
    +-- So all we need to do is distinguish UserInterrupt and ExitCode, and
    470 
    +-- all other exceptions can be represented by their 'show' string.
    471 
    +--
    472 
    +data SerializableException
    473 
    +  = EUserInterrupt
    474 
    +  | EExitCode ExitCode
    475 
    +  | EOtherException String
    476 
    +  deriving (Generic, Show)
    477 
    +
    478 
    +toSerializableException :: SomeException -> SerializableException
    479 
    +toSerializableException ex
    480 
    +  | Just UserInterrupt <- fromException ex  = EUserInterrupt
    481 
    +  | Just (ec::ExitCode) <- fromException ex = (EExitCode ec)
    482 
    +  | otherwise = EOtherException $
    483 
    +#if MIN_VERSION_base(4,20,0)
    484 
    +      -- Exception plus backtrace as seen in `displayExceptionWithInfo`
    485 
    +      case displayExceptionContext (someExceptionContext ex) of
    486 
    +        "" -> displayException (ex :: SomeException)
    487 
    +        cx -> displayException (ex :: SomeException) ++ "\n\n" ++ cx
    488 
    +#else
    489 
    +      show (ex :: SomeException)
    490 
    +#endif
    491 
    +
    492 
    +fromSerializableException :: SerializableException -> SomeException
    493 
    +fromSerializableException EUserInterrupt = toException UserInterrupt
    494 
    +fromSerializableException (EExitCode c) = toException c
    495 
    +fromSerializableException (EOtherException str) = toException (ErrorCall str)
    496 
    +
    497 
    +instance Binary ExitCode
    498 
    +instance Binary SerializableException

  • libraries/ghci/GHCi/Server.hs
    1 
    -{-# LANGUAGE CPP, RankNTypes, RecordWildCards, GADTs, ScopedTypeVariables #-}
    1 
    +{-# LANGUAGE LambdaCase, CPP, RankNTypes, RecordWildCards, GADTs, ScopedTypeVariables #-}
    2 2 
     module GHCi.Server
    3 3 
       ( MessageHook
    4 4 
       , CustomMessageHandler
    ... ... @@ -15,6 +15,7 @@ import GHCi.Signals
    15 15 
     import GHCi.TH
    16 16 
     import GHCi.Message
    17 17 
     import Data.ByteString (ByteString)
    18 
    +import Data.IORef
    18 19 
     #if defined(wasm32_HOST_ARCH)
    19 20 
     import qualified Data.ByteString.Builder as B
    20 21 
     import qualified Data.ByteString.Internal as B
    ... ... @@ -60,27 +61,29 @@ servWithCustom
    60 61 
       -> (forall a .IO a -> IO a)
    61 62 
       -> CustomMessageHandler
    62 63 
       -> IO ()
    63 
    -servWithCustom verbose hook pipe restore customHandler = loop
    64 
    +servWithCustom verbose hook pipe restore customHandler = newIORef False >>= loop
    64 65 
      where
    65 
    -  loop = do
    66 
    +  loop doneRef = do
    66 67 
         when verbose $ trace "reading pipe..."
    67 
    -    Msg msg <- readPipe pipe getMessage >>= hook
    68 
    -
    69 
    -    discardCtrlC
    70 
    -
    71 
    -    when verbose $ trace ("msg: " ++ (show msg))
    72 
    -    case msg of
    73 
    -      CustomMessage tag payload -> handleCustom tag payload
    74 
    -      Shutdown -> return ()
    75 
    -      RunTH st q ty loc -> wrapRunTH $ runTH pipe st q ty loc
    76 
    -      RunModFinalizers st qrefs -> wrapRunTH $ runModFinalizerRefs pipe st qrefs
    77 
    -      _other -> run msg >>= reply
    78 
    -
    79 
    -  reply :: forall a. (Binary a, Show a) => a -> IO ()
    80 
    -  reply r = do
    81 
    -    when verbose $ trace ("writing pipe: " ++ show r)
    82 
    -    writePipe pipe (put r)
    83 
    -    loop
    68 
    +    handleRemoteCall pipe hook $ \(msg :: Message a) -> do
    69 
    +
    70 
    +      discardCtrlC
    71 
    +
    72 
    +      when verbose $ trace ("msg: " ++ (show msg))
    73 
    +      case msg of
    74 
    +        CustomMessage tag payload -> handleCustom tag payload
    75 
    +        Shutdown -> writeIORef doneRef True >> reply ()
    76 
    +        -- TODO: These two should probably be done off the main thread
    77 
    +        RunTH ii st q ty loc -> wrapRunTH ii $ runTH pipe ii st q ty loc
    78 
    +        RunModFinalizers ii st qrefs -> wrapRunTH ii $ runModFinalizerRefs pipe ii st qrefs
    79 
    +        _other -> run msg >>= reply
    80 
    +
    81 
    +    -- closePipe here will wait until every active pipeWrite pipe exits, but
    82 
    +    -- they should never block for long, as the host process is either reading
    83 
    +    -- or has called pipeCloseRead.
    84 
    +    isDone <- readIORef doneRef
    85 
    +    if isDone then closePipe pipe
    86 
    +              else loop doneRef
    84 87
    85 88 
       handleCustom tag payload = do
    86 89 
         mresp <- customHandler tag payload
    ... ... @@ -89,17 +92,22 @@ servWithCustom verbose hook pipe restore customHandler = loop
    89 92 
           Nothing ->
    90 93 
             error $ "GHCi.Server: unhandled CustomMessage with tag " ++ show tag
    91 94
    95 
    +  reply :: forall a. Show a => a -> IO a
    96 
    +  reply r = do
    97 
    +    when verbose $ trace ("writing pipe: " ++ show r)
    98 
    +    return r -- the result of the handler gets sent
    99 
    +
    92 100 
       -- Run some TH code, which may interact with GHC by sending
    93 101 
       -- THMessage requests, and then finally send RunTHDone followed by a
    94 102 
       -- QResult.  For an overview of how TH works with Remote GHCi, see
    95 103 
       -- Note [Remote Template Haskell] in libraries/ghci/GHCi/TH.hs.
    96 
    -  wrapRunTH :: forall a. (Binary a, Show a) => IO a -> IO ()
    97 
    -  wrapRunTH io = do
    104 
    +  wrapRunTH :: forall a. (Binary a, Show a) => InboxId -> IO a -> IO (QResult a)
    105 
    +  wrapRunTH ii io = do
    98 106 
         when verbose $ trace "wrapRunTH..."
    99 107 
         r <- try io
    100 108 
         when verbose $ trace "wrapRunTH done."
    101 109 
         when verbose $ trace "writing RunTHDone."
    102 
    -    writePipe pipe (putTHMessage RunTHDone)
    110 
    +    remoteTHCall pipe ii RunTHDone
    103 111 
         case r of
    104 112 
           Left e
    105 113 
             | Just (GHCiQException _ err) <- fromException e  -> do
    ... ... @@ -179,7 +187,7 @@ defaultServerWithCustom customHandler = do
    179 187 
     #if defined(wasm32_HOST_ARCH)
    180 188 
       -- See Note [wasm ghci signal handlers] for details
    181 189 
       installSignalHandlers $ js_register_signal_handler cb_sig
    182 
    -  pipe <- mkPipeFromContinuations (recv_buf cb_recv) (send_buf cb_send)
    190 
    +  pipe <- mkPipeFromContinuations (recv_buf cb_recv) (send_buf cb_send) (pure ()) (pure ())
    183 191 
     #else
    184 192 
       when verbose $
    185 193 
         printf "GHC iserv starting (in: %s; out: %s)\n" (show inh) (show outh)

  • libraries/ghci/GHCi/TH.hs
    ... ... @@ -115,8 +115,8 @@ import qualified GHC.Boot.TH.Monad as TH
    115 115 
     import Unsafe.Coerce
    116 116
    117 117 
     -- | Create a new instance of 'QState'
    118 
    -initQState :: Pipe -> QState
    119 
    -initQState p = QState M.empty Nothing p
    118 
    +initQState :: Pipe -> InboxId -> QState
    119 
    +initQState p ii = QState M.empty Nothing ii p
    120 120
    121 121 
     -- | The monad in which we run TH computations on the server
    122 122 
     newtype GHCiQ a = GHCiQ { runGHCiQ :: IORef QState -> IO a }
    ... ... @@ -156,7 +156,7 @@ noLoc = TH.Loc "<no file>" "<no package>" "<no module>" (0,0) (0,0)
    156 156 
     ghcCmd :: Binary a => THMessage (THResult a) -> GHCiQ a
    157 157 
     ghcCmd m = GHCiQ $ \sRef -> do
    158 158 
       s <- readIORef sRef
    159 
    -  r <- remoteTHCall (qsPipe s) m
    159 
    +  r <- remoteTHCall (qsPipe s) (qsInboxId s) m
    160 160 
       case r of
    161 161 
         THException str -> throwIO (GHCiQException s str)
    162 162 
         THComplete res -> return res
    ... ... @@ -189,9 +189,9 @@ metaHandlersGHCiQ runInIO = TH.MetaHandlers {
    189 189 
       -- See Note [TH recover with -fexternal-interpreter] in GHC.Tc.Gen.Splice
    190 190 
       , mRecover = \h a -> runInIO $ GHCiQ $ \sRef -> mask $ \unmask -> do
    191 191 
           s <- readIORef sRef
    192 
    -      remoteTHCall (qsPipe s) StartRecover
    192 
    +      remoteTHCall (qsPipe s) (qsInboxId s) StartRecover
    193 193 
           e <- try $ unmask $ runGHCiQ (runQinGHCiQ a <* ghcCmd FailIfErrs) sRef
    194 
    -      remoteTHCall (qsPipe s) (EndRecover (isLeft e))
    194 
    +      remoteTHCall (qsPipe s) (qsInboxId s) (EndRecover (isLeft e))
    195 195 
           case e of
    196 196 
             Left GHCiQException{} ->
    197 197 
               runGHCiQ (runQinGHCiQ h) sRef
    ... ... @@ -233,26 +233,28 @@ metaHandlersGHCiQ runInIO = TH.MetaHandlers {
    233 233 
     -- a new IORef QState, and return a RemoteRef to it.
    234 234 
     startTH :: IO (RemoteRef (IORef QState))
    235 235 
     startTH = do
    236 
    -  r <- newIORef (initQState (error "startTH: no pipe"))
    236 
    +  r <- newIORef (initQState (error "startTH: no pipe") (error "startTH: no inbox id"))
    237 237 
       mkRemoteRef r
    238 238
    239 239 
     -- | Runs the mod finalizers.
    240 240 
     --
    241 241 
     -- The references must be created on the caller process.
    242 
    -runModFinalizerRefs :: Pipe -> RemoteRef (IORef QState)
    242 
    +runModFinalizerRefs :: Pipe -> InboxId
    243 
    +                    -> RemoteRef (IORef QState)
    243 244 
                         -> [RemoteRef (TH.Q ())]
    244 245 
                         -> IO ()
    245 
    -runModFinalizerRefs pipe rstate qrefs = do
    246 
    +runModFinalizerRefs pipe ii rstate qrefs = do
    246 247 
       qs <- mapM localRef qrefs
    247 248 
       qstateref <- localRef rstate
    248 249 
       qstate <- readIORef qstateref
    249 
    -  qstate' <- newIORef $ qstate { qsPipe = pipe }
    250 
    +  qstate' <- newIORef $ qstate { qsInboxId = ii, qsPipe = pipe }
    250 251 
       _ <- runGHCiQ (runQinGHCiQ $ sequence_ qs) qstate'
    251 252 
       return ()
    252 253
    253 254 
     -- | The implementation of the 'RunTH' message
    254 255 
     runTH
    255 256 
       :: Pipe
    257 
    +  -> InboxId
    256 258 
       -> RemoteRef (IORef QState)
    257 259 
           -- ^ The TH state, created by 'startTH'
    258 260 
       -> HValueRef
    ... ... @@ -264,13 +266,13 @@ runTH
    264 266 
       -> IO ByteString
    265 267 
           -- ^ Returns an (encoded) result that depends on the THResultType
    266 268
    267 
    -runTH pipe rstate rhv ty mb_loc = do
    269 
    +runTH pipe ii rstate rhv ty mb_loc = do
    268 270 
       hv <- localRef rhv
    269 271 
       case ty of
    270 
    -    THExp -> runTHQ pipe rstate mb_loc (unsafeCoerce hv :: TH.Q TH.Exp)
    271 
    -    THPat -> runTHQ pipe rstate mb_loc (unsafeCoerce hv :: TH.Q TH.Pat)
    272 
    -    THType -> runTHQ pipe rstate mb_loc (unsafeCoerce hv :: TH.Q TH.Type)
    273 
    -    THDec -> runTHQ pipe rstate mb_loc (unsafeCoerce hv :: TH.Q [TH.Dec])
    272 
    +    THExp -> runTHQ pipe ii rstate mb_loc (unsafeCoerce hv :: TH.Q TH.Exp)
    273 
    +    THPat -> runTHQ pipe ii rstate mb_loc (unsafeCoerce hv :: TH.Q TH.Pat)
    274 
    +    THType -> runTHQ pipe ii rstate mb_loc (unsafeCoerce hv :: TH.Q TH.Type)
    275 
    +    THDec -> runTHQ pipe ii rstate mb_loc (unsafeCoerce hv :: TH.Q [TH.Dec])
    274 276 
         THAnnWrapper -> do
    275 277 
           hv <- unsafeCoerce <$> localRef rhv
    276 278 
           case hv :: AnnotationWrapper of
    ... ... @@ -279,10 +281,10 @@ runTH pipe rstate rhv ty mb_loc = do
    279 281
    280 282 
     -- | Run a Q computation.
    281 283 
     runTHQ
    282 
    -  :: Binary a => Pipe -> RemoteRef (IORef QState) -> Maybe TH.Loc -> TH.Q a
    284 
    +  :: Binary a => Pipe -> InboxId -> RemoteRef (IORef QState) -> Maybe TH.Loc -> TH.Q a
    283 285 
       -> IO ByteString
    284 
    -runTHQ pipe rstate mb_loc ghciq = do
    286 
    +runTHQ pipe ii rstate mb_loc ghciq = do
    285 287 
       qstateref <- localRef rstate
    286 
    -  modifyIORef' qstateref (\qstate -> qstate { qsLocation = mb_loc, qsPipe = pipe })
    288 
    +  modifyIORef' qstateref (\qstate -> qstate { qsLocation = mb_loc, qsInboxId = ii, qsPipe = pipe })
    287 289 
       r <- runGHCiQ (runQinGHCiQ ghciq) qstateref
    288 290 
       return $! LB.toStrict (runPut (put r))

  • libraries/ghci/ghci.cabal.in
    ... ... @@ -76,6 +76,7 @@ library
    76 76 
         exposed-modules:
    77 77 
             GHCi.BreakArray
    78 78 
             GHCi.Message
    79 
    +        GHCi.Message.Async
    79 80 
             GHCi.ResolvedBCO
    80 81 
             GHCi.RemoteTypes
    81 82 
             GHCi.FFI
    ... ... @@ -95,7 +96,8 @@ library
    95 96 
             filepath         >= 1.4 && < 1.6,
    96 97 
             ghc-boot         == @ProjectVersionMunged@,
    97 98 
             ghc-heap         >= 9.10.1 && <=@ProjectVersionMunged@,
    98 
    -        transformers     >= 0.5 && < 0.7
    99 
    +        transformers     >= 0.5 && < 0.7,
    100 
    +        stm              >= 2.4 && < 2.6
    99 101
    100 102 
         if flag(bootstrap)
    101 103 
           build-depends:

  • testsuite/tests/ghci/custom-external-interpreter-commands/Main.hs
    ... ... @@ -12,6 +12,7 @@ import GHCi.Message
    12 12 
       , mkPipeFromHandles
    13 13 
       , remoteCall
    14 14 
       , Pipe
    15 
    +  , closePipe
    15 16 
       )
    16 17 
     import GHCi.Server
    17 18 
       ( CustomMessageHandler
    ... ... @@ -153,14 +154,13 @@ customMessage pipe c = do
    153 154 
     runClient :: Int -> IO ()
    154 155 
     runClient input = do
    155 156 
       serverExe <- getExecutablePath
    156 
    -  withServer serverExe $ \hFromServer hToServer -> do
    157 
    -    pipe <- mkPipeFromHandles hFromServer hToServer
    157 
    +  withServer serverExe $ \pipe -> do
    158 158 
         res  <- customMessage pipe (SquareCommand input)
    159 159 
         putStrLn $ "Square returned: " ++ show res
    160 160 
         res2  <- customMessage pipe (MulCommand 2 res)
    161 161 
         putStrLn $ "Mul returned: " ++ show res2
    162 162
    163 
    -withServer :: FilePath -> (Handle -> Handle -> IO a) -> IO a
    163 
    +withServer :: FilePath -> (Pipe -> IO a) -> IO a
    164 164 
     withServer serverExe action = do
    165 165 
       (ghcRead, serverWrite) <- createPipe
    166 166 
       (serverRead, ghcWrite) <- createPipe
    ... ... @@ -171,13 +171,14 @@ withServer serverExe action = do
    171 171 
         , std_out = Inherit
    172 172 
         , std_err = Inherit
    173 173 
         }
    174 
    -  bracket (mkHandles ghcRead ghcWrite)
    175 
    -          (\(hFromServer, hToServer) -> do
    174 
    +  bracket (mkHandles ghcRead ghcWrite >>= \hds -> (hds,) <$> uncurry mkPipeFromHandles hds)
    175 
    +          (\((hFromServer, hToServer), pipe) -> do
    176 
    +             closePipe pipe
    176 177 
                  hClose hFromServer
    177 178 
                  hClose hToServer
    178 179 
                  terminateProcess ph
    179 180 
                  void (waitForProcess ph))
    180 
    -          (\(hFromServer, hToServer) -> action hFromServer hToServer)
    181 
    +          (\(_, pipe) -> action pipe)
    181 182 
       where
    182 183 
         mkHandles r w = do
    183 184 
           hR <- fdToHandle r

    • —
    View it on GitLab.
    You're receiving this email because of your account on gitlab.haskell.org. Manage all notifications · Help Notification message regarding https://gitlab.haskell.org/ghc/ghc/-/compare/c2c42e7a8e8584caab528cfce1122e9508e03f75...cd423df81aefcb3214d880a35e3d011b9236fda4 at 1792062244