GitLab

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)