Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC Commits: 1c251824 by Rodrigo Mesquita at 2026-10-14T15:35:42+01:00 Close pipe cleans up pending requests and the other process can handle EOF and raise correct exceptions when pipe is read or written when closed More race fixes - - - - - 3 changed files: - libraries/ghci/GHCi/Message.hs - libraries/ghci/GHCi/Server.hs - testsuite/tests/ghci/custom-external-interpreter-commands/Main.hs Changes: ===================================== libraries/ghci/GHCi/Message.hs ===================================== @@ -27,7 +27,7 @@ module GHCi.Message , Pipe(..), mkPipeFromHandles, mkPipeFromContinuations, closePipe , remoteCall, remoteTHCall , handleRemoteCall, handleRemoteTHCall - , RemoteProcessException(..) + , RemoteProcessException(..), PipeClosedException(..) , BreakModule , BreakUnitId @@ -89,7 +89,6 @@ import qualified GHC.Boot.TH.Syntax as TH import qualified GHC.Boot.TH.Monad as TH import System.Exit import System.IO -import System.IO.Error import Control.Monad import Control.Concurrent.STM import Data.Bifunctor @@ -722,38 +721,54 @@ data Pipe = Pipe , pipeLeftovers :: !(IORef (Maybe ByteString)) , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent writes to the stream , pipeAsyncMngr :: !AsyncManager - , pipeCleanupRW :: !(IO ()) - -- ^ Clean up the read/write resources, e.g. closes the handles if 'mkPipeFromHandles' + , 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 and close the handles. +-- Must call 'closePipe' at the end to clean up. mkPipeFromHandles :: Handle -> Handle -> IO Pipe mkPipeFromHandles pipeRead pipeWrite = do let getSome = B.hGetSome pipeRead (32*1024) putAll b = do B.hPutBuilder pipeWrite b hFlush pipeWrite - mkPipeFromContinuations getSome putAll (hClose pipeRead >> hClose pipeWrite) + mkPipeFromContinuations getSome putAll --- | Make a 'Pipe' from a reader function and a writer function, plus an action --- to clean up the reader and writer resources. The pipe should be closed with --- 'closePipe', which will call this clean up action too. -mkPipeFromContinuations :: IO ByteString -> (B.Builder -> IO ()) -> IO () -> IO Pipe -mkPipeFromContinuations getSome putAll pipeCleanupRW = mdo +-- | 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 this pipe, terminating the async manager's drain-read-end thread. --- This doesn't close the handles +-- | 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 - killThread (asyncDrainTid pipeAsyncMngr) -- releases handle lock taken by reading - pipeCleanupRW -- hClose won't deadlock now that the drainThread is killed + + 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) @@ -798,20 +813,28 @@ remoteAsyncReq pipe msg = do -- Mark pending request uq <- freshReqId (pipeAsyncMngr pipe) - wv <- newEmptyMVar - atomicModifyIORef' pendingReplies $ - \m -> (IntMap.insert uq wv m, ()) + 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 (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. - atomicModifyIORef' pendingReplies - (\m -> (IntMap.delete uq m, ())) + atomically (modifyTVar pendingReplies (IntMap.delete uq)) -- Block waiting for reply - takeMVar wv + atomically (takeTMVar wv) -- | Receive a single asynchronous request and send a matching reply -- asynchronously (the handler may fork a computation that eventually replies @@ -826,32 +849,40 @@ handleAsyncReq pipe getit k = do ack <- newEmptyMVar -- Block waiting for request - (uq, bs) <- atomically (readTQueue (requestsQueue (pipeAsyncMngr pipe))) - - let reply p = do - writePipe pipe (Reply uq (bimap toSerializableException runPut p)) - putMVar ack () - - -- Run the continuation which must reply exactly once with 'writeReply' - case runGetOrFail getit bs of - Left (_, _, err) -> do - let exc = DecodeAsyncReqException err - reply (Left (toException exc)) - throwIO exc - Right (_, _, it) -> - 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 + mbreq <- atomically $ do + isClosed <- readTVar (pipeIsClosed pipe) + isEmpty <- isEmptyTQueue (requestsQueue (pipeAsyncMngr pipe)) + if isClosed && isEmpty + then return Nothing + else Just <$> readTQueue (requestsQueue (pipeAsyncMngr pipe)) + + 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) -> + 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. @@ -862,12 +893,16 @@ instance Exception RemoteProcessException 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 :: !(IORef (IntMap (MVar (Either SomeException LB.ByteString)))) + { pendingReply :: !(TVar (IntMap (TMVar (Either SomeException LB.ByteString)))) -- ^ When a request is sent, keep track that we're expecting a reply. The - -- MVar is signaled when the reply arrives. The exception can originate + -- 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, @@ -904,45 +939,58 @@ instance Binary IddMsg where newAsyncManager :: Pipe -> IO AsyncManager newAsyncManager pipe = do asyncNextReq <- newIORef 0 - pendingReply <- newIORef IntMap.empty + pendingReply <- newTVarIO IntMap.empty requestsQueue <- newTQueueIO - mdo - -- Fork thread to drain read end. - -- (cleaned up on 'closePipe') - asyncDrainTid <- forkIO (drainPipe pipe am) + -- Fork thread to drain read end. + -- (cleaned up on 'closePipe') + asyncDrainTid <- forkIO (drainPipe pipe) - let am = AsyncManager{..} - pure am + let am = AsyncManager{..} + pure am -- | Drains the read end of the Pipe and match 'Reply'es against pending 'Request's. -- -- In pratice, we'll have both a thread draining the read end of the host's -- 'Pipe' and a thread draining the read end of the interpreter's 'Pipe', since --- both processes construct a 'Pipe' and send and receive messages: -drainPipe :: Pipe -> AsyncManager -> IO () -drainPipe p am = flip catch cancelPending $ forever $ do - readPipe p >>= \case - -- Receive a reply to a pending notification - Reply uq bs -> do - mwv <- atomicModifyIORef' (pendingReply am) $ \m -> do - case IntMap.lookup uq m of - Nothing -> (m, Nothing) - Just wv -> (IntMap.delete uq m, Just wv) - case mwv of - Just wv -> - putMVar wv (bimap (toException . RemoteProcessException) id bs) -- signal matching request - Nothing -> - -- the req was dropped from pending bc there was an exception, and no - -- one's listening any more. the reply is ignored - pure () - -- Receive a request and queue it - Request uq bs -> - atomically $ writeTQueue (requestsQueue am) (uq,bs) - where - cancelPending (e::SomeException) = do - pendings <- readIORef (pendingReply am) - mapM_ (\p -> putMVar p (Left e {-host exception-})) pendings +-- both processes construct a 'Pipe' and send and receive messages +drainPipe :: Pipe -> IO () +drainPipe p = catch loop replyToPendings where + loop = do + readPipe p >>= \case + -- Receive a reply to a pending notification + Just (Reply uq bs) -> do + mwv <- atomically $ do + m <- readTVar (pendingReply am) + case IntMap.lookup uq m of + Nothing -> pure Nothing + Just wv -> do + writeTVar (pendingReply am) (IntMap.delete uq m) + pure (Just wv) + case mwv of + Just wv -> atomically $ + putTMVar wv (bimap (toException . RemoteProcessException) id bs) -- signal matching request + Nothing -> + -- the req was dropped from pending bc there was an exception, and no + -- one's listening any more. the reply is ignored + pure () + loop + -- Receive a request and queue it + Just (Request uq bs) -> do + atomically $ writeTQueue (requestsQueue am) (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 freshReqId :: AsyncManager -> IO CorrelationId freshReqId AsyncManager{asyncNextReq} = @@ -960,16 +1008,17 @@ writePipe p@Pipe{..} = withLock p . putAll . execPut . put -- | Only the 'drainPipe' thread should read from the pipe directly. Other -- functions should wait for the matching reply to be delivered asynchronously. -- (see 'withAsyncReq'). It doesn't lock to read bc it assumes single reader is 'drainPipe'. -readPipe :: Pipe -> IO IddMsg +-- +-- Returns @Nothing@ on EOF. +readPipe :: Pipe -> IO (Maybe IddMsg) readPipe Pipe{..} = do leftovers <- readIORef pipeLeftovers m <- getBin getSome get leftovers case m of - Nothing -> throw $ - mkIOError eofErrorType "GHCi.Message.readPipe" Nothing Nothing + Nothing -> return Nothing Just (result, new_leftovers) -> do writeIORef pipeLeftovers new_leftovers - return result + return (Just result) getBin :: (IO ByteString) -> Get a -> Maybe ByteString ===================================== libraries/ghci/GHCi/Server.hs ===================================== @@ -79,7 +79,7 @@ servWithCustom verbose hook pipe restore customHandler = newIORef False >>= loop _other -> run msg >>= reply isDone <- readIORef doneRef - if isDone then return () + if isDone then closePipe pipe else loop doneRef handleCustom tag payload = do @@ -184,7 +184,7 @@ defaultServerWithCustom customHandler = do #if defined(wasm32_HOST_ARCH) -- See Note [wasm ghci signal handlers] for details installSignalHandlers $ js_register_signal_handler cb_sig - pipe <- mkPipeFromContinuations (recv_buf cb_recv) (send_buf cb_send) (pure ()) + pipe <- mkPipeFromContinuations (recv_buf cb_recv) (send_buf cb_send) #else when verbose $ printf "GHC iserv starting (in: %s; out: %s)\n" (show inh) (show outh) ===================================== testsuite/tests/ghci/custom-external-interpreter-commands/Main.hs ===================================== @@ -171,12 +171,14 @@ withServer serverExe action = do , std_out = Inherit , std_err = Inherit } - bracket (mkHandles ghcRead ghcWrite >>= uncurry mkPipeFromHandles) - (\pipe -> do + bracket (mkHandles ghcRead ghcWrite >>= \hds -> (hds,) <$> uncurry mkPipeFromHandles hds) + (\((hFromServer, hToServer), pipe) -> do closePipe pipe + hClose hFromServer + hClose hToServer terminateProcess ph void (waitForProcess ph)) - (\pipe -> action pipe) + (\(_, pipe) -> action pipe) where mkHandles r w = do hR <- fdToHandle r View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/1c2518249550c0cf00e8a3fa61ef282a... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/1c2518249550c0cf00e8a3fa61ef282a... 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