Rodrigo Mesquita pushed to branch wip/romes/async-rpc at Glasgow Haskell Compiler / GHC Commits: 86f0d70c by Rodrigo Mesquita at 2026-10-15T11:33:44+01:00 we may still reply if we had some queued msgs read already - - - - - c2c42e7a by Andrea Vezzosi at 2026-10-15T11:42:27+01:00 interp: close Handles when done with Pipe Ensures we close the pipe handles (if created using 'mkPipeFromHandles') on `closePipe`, and do `closePipe` on `Shutdown`. The thread draining the read end always `'finally'` closes the read end. And `closePipe` kills the draining thread and closes the write end. When `remoteAsyncReq`, we throw PipeReadClosedException if the pipe read is closed, since we wouldn't be able to receive the reply. It's fine for the drain thread not to close the write end, even if it dies without `closePipe` (which does close the write end), because: - requests pending a reply will get PipeClosedException (we'll not be able to read any further "real" reply) - new calls to `remoteAsyncReq` will throw PipeReadClosedException - pending pipe writes, in response to an already read request, might still succeed And the clean up code should call close pipe to ensure the write end is closed as well. So, if the drain thread dies naturally and the read end is closed, may still be attempted to be used, but it will throw accordingly. The user likely wants to catch this to close the write end as well. - - - - - 3 changed files: - compiler/GHC/Runtime/Interpreter.hs - libraries/ghci/GHCi/Message/Async.hs - libraries/ghci/GHCi/Server.hs Changes: ===================================== compiler/GHC/Runtime/Interpreter.hs ===================================== @@ -654,6 +654,8 @@ stopInterp interp = case interpInstance interp of if isJust ex then pure () else sendMessage i Shutdown + -- We asked the server to shutdown, so we will also shutdown any communication. + closePipe (interpPipe (instProcess i)) pure InterpPending -- ----------------------------------------------------------------------------- ===================================== libraries/ghci/GHCi/Message/Async.hs ===================================== @@ -70,8 +70,13 @@ data Pipe = Pipe , 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 + , pipeCloseRead :: !(IO ()) + -- ^ closes read end of the pipe + , pipeCloseWrite :: !(IO ()) + -- ^ closes write end of the pipe + , pipeReadClosed :: !(TVar Bool) + -- ^ After the pipe is closed, messages can no longer be received from the + -- other end, but the 'AsyncManager' queues might not be empty yet. } -- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write. @@ -82,43 +87,49 @@ mkPipeFromHandles pipeRead pipeWrite = do putAll b = do B.hPutBuilder pipeWrite b hFlush pipeWrite - mkPipeFromContinuations getSome putAll + mkPipeFromContinuations getSome putAll (hClose pipeRead) (hClose pipeWrite) -- | Make a 'Pipe' from a reader function and a writer function. --- The pipe should be closed with 'closePipe'. -mkPipeFromContinuations :: IO ByteString -> (B.Builder -> IO ()) -> IO Pipe -mkPipeFromContinuations getSome putAll = mdo +-- The pipe should be closed with 'closePipe', which will call 'pipeCloseRead' and 'pipeCloseWrite'. +mkPipeFromContinuations :: IO ByteString -> (B.Builder -> IO ()) + -> IO () + -- ^ pipeCloseRead + -> IO () + -- ^ pipeCloseWrite + -> IO Pipe +mkPipeFromContinuations getSome putAll pipeCloseRead pipeCloseWrite = mdo pipeLeftovers <- newIORef Nothing pipeLock <- newMVar () - pipeIsClosed <- newTVarIO False + pipeReadClosed <- 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. +-- | Cleanly close our ends of the Pipe, terminating the async manager's +-- drain-read-end thread. Also calls 'pipeCloseRead' and 'pipeCloseWrite', in that order. -- --- If the pipe was created using 'mkPipeFromHandles', the handles must only by --- closed after calling 'closePipe'. +-- If the pipe was created using 'mkPipeFromHandles', the handles are also closed. -- -- 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. +-- No special notification is sent to the remote process, a message to that +-- effect should be sent ahead of time if one is needed (see 'Shutdown'). closePipe :: Pipe -> IO () closePipe Pipe{..} = do - isAlreadyClosed <- atomically $ swapTVar pipeIsClosed True + readAlreadyClosed <- atomically $ swapTVar pipeReadClosed True - unless isAlreadyClosed $ do + -- this triggers pipeCloseRead + unless readAlreadyClosed $ do -- releases handle lock taken by reading, and cleans up by replying to every -- pending request with 'PipeClosedException'. throwTo (asyncDrainTid pipeAsyncMngr) PipeClosedException + pipeCloseWrite + -- | We've closed the pipe on our end and we terminate all pending requests with this. data PipeClosedException = PipeClosedException deriving Show instance Exception PipeClosedException @@ -186,7 +197,8 @@ remoteAsyncReq pipe msg ii = do wv <- newEmptyTMVarIO isClosed <- atomically $ do - isClosed <- readTVar (pipeIsClosed pipe) + isClosed <- readTVar (pipeReadClosed pipe) + -- don't send the req if we won't get a reply. if isClosed then pure True else do @@ -223,15 +235,17 @@ handleAsyncReq pipe getit mii k = mask $ \restore -> do -- 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 + isClosed <- readTVar (pipeReadClosed pipe) + taken <- case mii of + Nothing -> tryReadTQueue $ requestsQueue am + Just ii -> tryReadTQueue =<< getMailbox am ii + case taken of + Just req -> pure (Just req) + Nothing | isClosed -> pure Nothing -- empty and closed + | otherwise -> retry -- wait for msg case mbreq of - Nothing -> throwIO PipeClosedException + Nothing -> throwIO PipeClosedException -- can't expect reqs Just (uq, bs) -> do let reply p = do writePipe pipe (Reply uq (bimap toSerializableException runPut p)) @@ -368,8 +382,10 @@ newAsyncManager pipe = do -- In pratice, we'll have both a thread draining the read end of the host's -- 'Pipe' and a thread draining the read end of the interpreter's 'Pipe', since -- both processes construct a 'Pipe' and send and receive messages +-- +-- Closes the Pipe read end when done reading drainPipe :: Pipe -> (forall x. IO x -> IO x) -> IO () -drainPipe p unmask = catch (unmask loop) replyToPendings where +drainPipe p unmask = catch (unmask $ loop `finally` pipeCloseRead p) replyToPendings where loop = do readPipe p >>= \case -- Receive a reply to a pending notification @@ -404,7 +420,9 @@ drainPipe p unmask = catch (unmask loop) replyToPendings where pure () replyToPendings e = atomically $ do - writeTVar (pipeIsClosed p) True + -- reqs are only added if pipe not closed; + -- thus we close and take all pending atomically. + writeTVar (pipeReadClosed p) True pendings <- swapTVar (pendingReply am) IntMap.empty -- take pending mapM_ (\p -> putTMVar p (Left e {-host exception-})) pendings ===================================== libraries/ghci/GHCi/Server.hs ===================================== @@ -78,6 +78,9 @@ servWithCustom verbose hook pipe restore customHandler = newIORef False >>= loop RunModFinalizers ii st qrefs -> wrapRunTH ii $ runModFinalizerRefs pipe ii st qrefs _other -> run msg >>= reply + -- closePipe here will wait until every active pipeWrite pipe exits, but + -- they should never block for long, as the host process is either reading + -- or has called pipeCloseRead. isDone <- readIORef doneRef if isDone then closePipe pipe else loop doneRef @@ -184,7 +187,7 @@ defaultServerWithCustom customHandler = do #if defined(wasm32_HOST_ARCH) -- See Note [wasm ghci signal handlers] for details installSignalHandlers $ js_register_signal_handler cb_sig - pipe <- mkPipeFromContinuations (recv_buf cb_recv) (send_buf cb_send) + pipe <- mkPipeFromContinuations (recv_buf cb_recv) (send_buf cb_send) (pure ()) (pure ()) #else when verbose $ printf "GHC iserv starting (in: %s; out: %s)\n" (show inh) (show outh) View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/7d5fc67e1bbc53107ae639eaf613c2e... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/7d5fc67e1bbc53107ae639eaf613c2e... 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