Rodrigo Mesquita pushed to branch wip/romes/multi-threaded-debugger at Glasgow Haskell Compiler / GHC Commits: 5f600346 by Rodrigo Mesquita at 2026-10-06T17:45:16+01:00 Asynchronous RPC with external interpreter - Kill InterpProcess.interpLock to Pipe.pipeLock - - - - - 1c15fcb9 by Rodrigo Mesquita at 2026-10-06T19:27:25+01:00 Async more work - - - - - 2540e67a by Rodrigo Mesquita at 2026-10-06T19:28:18+01:00 WIP: Asynchronous eval/resume - - - - - 9 changed files: - compiler/GHC/Runtime/Eval.hs - compiler/GHC/Runtime/Interpreter.hs - compiler/GHC/Runtime/Interpreter/JS.hs - compiler/GHC/Runtime/Interpreter/Process.hs - compiler/GHC/Runtime/Interpreter/Types.hs - compiler/GHC/Runtime/Interpreter/Wasm.hs - libraries/ghci/GHCi/Message.hs - libraries/ghci/GHCi/Run.hs - libraries/ghci/GHCi/Server.hs Changes: ===================================== compiler/GHC/Runtime/Eval.hs ===================================== @@ -188,12 +188,13 @@ execOptions = ExecOptions -- by default we just care about breakpoints hit in this eval thread } --- | Run a statement in the current interactive context. +-- | Run a statement in the current interactive context, as a new thread in the interpreter. +-- Returns an action which blocks waiting for the execution to finish and yield a result. execStmt :: GhcMonad m => String -- ^ a statement (bind or expression) -> ExecOptions - -> m ExecResult + -> m (m ExecResult) execStmt input exec_opts@ExecOptions{..} = do hsc_env <- getSession @@ -204,13 +205,15 @@ execStmt input exec_opts@ExecOptions{..} = do case mb_stmt of -- empty statement / comment - Nothing -> return (ExecComplete (Right []) 0) + Nothing -> pure (pure (ExecComplete (Right []) 0)) Just stmt -> execStmt' stmt input exec_opts -- | Like `execStmt`, but takes a parsed statement as argument. Useful when -- doing preprocessing on the AST before execution, e.g. in GHCi (see -- GHCi.UI.runStmt). -execStmt' :: GhcMonad m => GhciLStmt GhcPs -> String -> ExecOptions -> m ExecResult +-- +-- Returns an action which blocks waiting for the execution to finish and yield a result. +execStmt' :: GhcMonad m => GhciLStmt GhcPs -> String -> ExecOptions -> m (m ExecResult) execStmt' stmt stmt_text ExecOptions{..} = do hsc_env <- getSession let interp = hscInterp hsc_env @@ -226,21 +229,22 @@ execStmt' stmt stmt_text ExecOptions{..} = do case r of Nothing -> -- empty statement / comment - return (ExecComplete (Right []) 0) + pure (pure (ExecComplete (Right []) 0)) Just (ids, hval, fix_env) -> do updateFixityEnv fix_env let eval_opts = initEvalOpts idflags' (enableGhcStepMode execSingleStep) (enableIsolateThreadBreaks execIsolateMode) - status <- liftIO $ evalStmt interp eval_opts (execWrap hval) + rtid <- liftIO $ evalStmt interp eval_opts (execWrap hval) let ic = hsc_IC hsc_env bindings = (ic_tythings ic, ic_gre_cache ic) size = ghciHistSize idflags' - - handleRunStatus execSingleStep eval_opts stmt_text bindings ids - status (emptyHistory size) + pure $ do + status <- liftIO (waitThreadResult interp rtid) + handleRunStatus execSingleStep eval_opts stmt_text bindings ids + status (emptyHistory size) runDecls :: GhcMonad m => String -> m [TyThing] runDecls = runDeclsWithLocation "<interactive>" 1 @@ -326,16 +330,16 @@ handleRunStatus step eval_opts expr bindings final_ids status history0 = do return (ExecComplete (Left (fromSerializableException e)) alloc) -- Nothing case: we stopped when an exception was raised, not at a breakpoint. - EvalBreak apStack_ref Nothing resume_ctxt ccs -> do - resume_ctxt_fhv <- liftIO $ mkFinalizedHValue interp resume_ctxt - apStack_fhv <- liftIO $ mkFinalizedHValue interp apStack_ref + EvalBreak apStack_ref Nothing resume_tid ccs -> do + resume_tid_fhv <- liftIO $ mkFinalizedHValue interp resume_tid + apStack_fhv <- liftIO $ mkFinalizedHValue interp apStack_ref let span = mkGeneralSrcSpan (fsLit "<unknown>") (hsc_env1, break_ids) <- liftIO $ bindLocalsAtBreakpoint hsc_env apStack_fhv span Nothing let resume = Resume { resumeStmt = expr - , resumeContext = resume_ctxt_fhv + , resumeContext = resume_tid_fhv , resumeBindings = bindings , resumeFinalIds = final_ids , resumeApStack = apStack_fhv @@ -356,7 +360,7 @@ handleRunStatus step eval_opts expr bindings final_ids status history0 = do -- The interpreter yields on a breakpoint if: -- - the breakpoint was explicitly enabled (in @BreakArray@) -- - or one of the stepping options in @EvalOpts@ caused us to stop at one - EvalBreak apStack_ref (Just eval_break) resume_ctxt ccs -> do + EvalBreak apStack_ref (Just eval_break) resume_tid ccs -> do let ibi = evalBreakpointToId eval_break let hug = hsc_HUG hsc_env info_brks <- liftIO $ readIModBreaks hug ibi @@ -368,8 +372,8 @@ handleRunStatus step eval_opts expr bindings final_ids status history0 = do breakArray <- getBreakArray interp ibi info_brks breakpointStatus interp breakArray (ibi_info_index ibi) - apStack_fhv <- liftIO $ mkFinalizedHValue interp apStack_ref - resume_ctxt_fhv <- liftIO $ mkFinalizedHValue interp resume_ctxt + apStack_fhv <- liftIO $ mkFinalizedHValue interp apStack_ref + resume_tid_fhv <- liftIO $ mkFinalizedHValue interp resume_tid -- This breakpoint is enabled or we mean to break here; -- we want to stop instead of just logging it. @@ -382,7 +386,7 @@ handleRunStatus step eval_opts expr bindings final_ids status history0 = do let resume = Resume { resumeStmt = expr - , resumeContext = resume_ctxt_fhv + , resumeContext = resume_tid_fhv , resumeBindings = bindings , resumeFinalIds = final_ids , resumeApStack = apStack_fhv @@ -398,7 +402,9 @@ handleRunStatus step eval_opts expr bindings final_ids status history0 = do return (ExecBreak break_ids (Just ibi) resume) else do -- resume with the same step type - status <- liftIO $ GHCi.resumeStmt interp eval_opts resume_ctxt_fhv + status <- liftIO $ do + GHCi.resumeStmt interp eval_opts resume_tid_fhv + GHCi.waitThreadResult interp resume_tid_fhv history <- if not tracing then pure history0 else do history1 <- liftIO $ mkHistory hug apStack_fhv ibi let !history' = history1 `consBL` history0 @@ -414,7 +420,9 @@ resumeExec :: GhcMonad m -> ThreadBreaksIsolationMode -> Maybe Int -> Resume - -> m ExecResult + -> m (m ExecResult) + -- ^ Returns an action which blocks waiting for the resumed thread + -- to break again or yield a result resumeExec step isolateMode mbCnt r = do hsc_env <- getSession let ic = hsc_IC hsc_env @@ -438,7 +446,7 @@ resumeExec step isolateMode mbCnt r = do case r of Resume { resumeStmt = expr - , resumeContext = fhv + , resumeContext = rtid , resumeBindings = bindings , resumeFinalIds = final_ids , resumeApStack = apStack @@ -454,17 +462,19 @@ resumeExec step isolateMode mbCnt r = do let eval_opts = initEvalOpts dflags (enableGhcStepMode step) (enableIsolateThreadBreaks isolateMode) - status <- liftIO $ GHCi.resumeStmt interp eval_opts fhv - let prevHistoryLst = fromListBL 50 hist - hug = hsc_HUG hsc_env - hist' = case mb_brkpt of - Nothing -> pure prevHistoryLst - Just bi - | breakHere False step span -> do - hist1 <- liftIO (mkHistory hug apStack bi) - return $ hist1 `consBL` fromListBL 50 hist - | otherwise -> pure prevHistoryLst - handleRunStatus step eval_opts expr bindings final_ids status =<< hist' + liftIO $ GHCi.resumeStmt interp eval_opts rtid + pure $ do + status <- liftIO $ GHCi.waitThreadResult interp rtid + let prevHistoryLst = fromListBL 50 hist + hug = hsc_HUG hsc_env + hist' = case mb_brkpt of + Nothing -> pure prevHistoryLst + Just bi + | breakHere False step span -> do + hist1 <- liftIO (mkHistory hug apStack bi) + return $ hist1 `consBL` fromListBL 50 hist + | otherwise -> pure prevHistoryLst + handleRunStatus step eval_opts expr bindings final_ids status =<< hist' setupBreakpoint :: GhcMonad m => Interp -> InternalBreakpointId -> Int -> m () -- #19157 setupBreakpoint interp ibi cnt = do @@ -1275,7 +1285,9 @@ compileParsedExprRemote expr@(L loc _) = withSession $ \hsc_env -> do updateFixityEnv fix_env let eval_opts = initEvalOpts dflags EvalStepNone True - status <- liftIO $ evalStmt interp eval_opts (EvalThis hvals_io) + status <- liftIO $ do + rtid <- evalStmt interp eval_opts (EvalThis hvals_io) + waitThreadResult interp rtid case status of EvalComplete _ (EvalSuccess [hval]) -> return hval EvalComplete _ (EvalException e) -> ===================================== compiler/GHC/Runtime/Interpreter.hs ===================================== @@ -11,6 +11,7 @@ module GHC.Runtime.Interpreter , evalStmt, EvalStatus_(..), EvalStatus, EvalResult(..), EvalExpr(..) , resumeStmt , abandonStmt + , waitThreadResult , evalIO , evalString , evalStringToIOString @@ -286,11 +287,11 @@ evalStmt :: Interp -> EvalOpts -> EvalExpr ForeignHValue - -> IO (EvalStatus_ [ForeignHValue] [HValueRef]) + -> IO (ForeignRef ThreadId) evalStmt interp opts foreign_expr = do - status <- withExpr foreign_expr $ \expr -> + r <- withExpr foreign_expr $ \expr -> interpCmd interp (EvalStmt opts expr) - handleEvalStatus interp status + mkFinalizedHValue interp r where withExpr :: EvalExpr ForeignHValue -> (EvalExpr HValueRef -> IO a) -> IO a withExpr (EvalThis fhv) cont = @@ -304,17 +305,25 @@ resumeStmt :: Interp -> EvalOpts -> ForeignRef ThreadId - -> IO (EvalStatus_ [ForeignHValue] [HValueRef]) + -> IO () resumeStmt interp opts resume_tid = do - status <- withForeignRef resume_tid $ \rhv -> + withForeignRef resume_tid $ \rhv -> interpCmd interp (ResumeStmt opts rhv) - handleEvalStatus interp status abandonStmt :: Interp -> ForeignRef ThreadId -> IO () abandonStmt interp resume_tid = withForeignRef resume_tid $ \rhv -> interpCmd interp (AbandonStmt rhv) +waitThreadResult + :: Interp + -> ForeignRef ThreadId + -> IO (EvalStatus_ [ForeignHValue] [HValueRef]) +waitThreadResult interp tid = do + status <- withForeignRef tid $ \rhv -> do + interpCmd interp (WaitThreadResult rhv) + handleEvalStatus interp status + handleEvalStatus :: Interp -> EvalStatus [HValueRef] @@ -417,8 +426,9 @@ whereFrom interp ref = seqHValue :: Interp -> UnitEnv -> Logger -> ForeignHValue -> IO (EvalResult ()) seqHValue interp unit_env logger ref = withForeignRef ref $ \hval -> do - status <- interpCmd interp (Seq hval) - handleSeqHValueStatus interp unit_env logger status + rtid <- interpCmd interp (Seq hval) + status <- interpCmd interp (WaitThreadResult rtid) + handleSeqHValueStatus interp unit_env logger (clearEvalStatus status) evalBreakpointToId :: EvalBreakpoint -> InternalBreakpointId evalBreakpointToId eval_break = @@ -460,9 +470,10 @@ handleSeqHValueStatus interp unit_env logger eval_status = Just modbreaks -> put =<< getBreakLoc (readIModModBreaks hug) ibi modbreaks -- resume the seq (:force) processing in the iserv process - withForeignRef resume_ctxt_fhv $ \hval -> do - status <- interpCmd interp (ResumeSeq hval) - handleSeqHValueStatus interp unit_env logger status + withForeignRef resume_ctxt_fhv $ \rtid -> do + interpCmd interp (ResumeSeq rtid) + status <- interpCmd interp (WaitThreadResult rtid) + handleSeqHValueStatus interp unit_env logger (clearEvalStatus status) (EvalComplete _ r) -> return r @@ -626,11 +637,9 @@ spawnIServ conf = do [] (iservConfOpts conf) interpPipe <- mkPipeFromHandles rh wh - lock <- newMVar () let process = InterpProcess { interpHandle = ph , interpPipe - , interpLock = lock } pending_frees <- newMVar [] @@ -771,3 +780,9 @@ readIModModBreaks hug mod = imodBreaks_modBreaks . expectJust <$> readIModBreaks fromEvalResult :: EvalResult a -> IO a fromEvalResult (EvalException e) = throwIO (fromSerializableException e) fromEvalResult (EvalSuccess a) = return a + +clearEvalStatus :: EvalStatus a -> EvalStatus () +clearEvalStatus = \case + EvalComplete w (EvalException se) -> EvalComplete w (EvalException se) + EvalComplete w (EvalSuccess _) -> EvalComplete w (EvalSuccess ()) + EvalBreak ap mb rt rccs -> EvalBreak ap mb rt rccs ===================================== compiler/GHC/Runtime/Interpreter/JS.hs ===================================== @@ -127,11 +127,9 @@ startTHRunnerProcess interp_js settings = do std_in <- readIORef interp_in interpPipe <- mkPipeFromHandles rh wh - lock <- newMVar () let proc = InterpProcess { interpHandle = hdl , interpPipe - , interpLock = lock } pure (std_in, proc) @@ -382,7 +380,7 @@ jsLinkPlan logger tmpfs tmp_dir ar_cache link_cfg cfg inst link_plan = do -- Haskell server. server_started <- jsServerStarted <$> readMVar (instJSState (instExtra inst)) if server_started - then sendMessageNoResponse inst $ LoadObj all_js + then sendMessage inst $ LoadObj all_js else jsLoadFile inst all_js ---------------------------------------------------------------- ===================================== compiler/GHC/Runtime/Interpreter/Process.hs ===================================== @@ -2,16 +2,7 @@ module GHC.Runtime.Interpreter.Process ( -- * Message API Message(..) - , DelayedResponse (..) - -- * Top-level message API (these acquire/release a lock) , sendMessage - , sendMessageNoResponse - , sendMessageDelayedResponse - , receiveDelayedResponse - -- * Nested message API (these require the interpreter to already be locked) - , sendAnyValue - , receiveAnyValue - , receiveTHMessage ) where @@ -27,79 +18,13 @@ import GHC.Utils.Exception as Ex import Data.Binary import System.Exit import System.Process -import Control.Concurrent.MVar (MVar, withMVar, takeMVar, putMVar, isEmptyMVar) - -data DelayedResponse a = DelayedResponse -- ----------------------------------------------------------------------------- -- Top-level Message API --- | Send a message to the interpreter process that doesn't expect a response --- (locks the interpreter while sending) -sendMessageNoResponse :: ExtInterpInstance d -> Message () -> IO () -sendMessageNoResponse i m = - withLock i $ writeInterpProcess (instProcess i) (putMessage m) - --- | Send a message to the interpreter that expects a response --- (locks the interpreter while until the response is received) +-- | Send a message to the interpreter and expect a response, blocking until it arrives. sendMessage :: Binary a => ExtInterpInstance d -> Message a -> IO a -sendMessage i m = withLock i $ callInterpProcess (instProcess i) m - --- | Send a message to the interpreter process whose response is expected later --- --- This is useful to avoid forgetting to receive the value and to ensure that --- the type of the response isn't lost. Use receiveDelayedResponse to read it. --- (locks the interpreter until the response is received using --- `receiveDelayedResponse`) -sendMessageDelayedResponse :: ExtInterpInstance d -> Message a -> IO (DelayedResponse a) -sendMessageDelayedResponse i m = do - lock i - writeInterpProcess (instProcess i) (putMessage m) - pure DelayedResponse - --- | Expect a delayed result to be received now -receiveDelayedResponse :: Binary a => ExtInterpInstance d -> DelayedResponse a -> IO a -receiveDelayedResponse i DelayedResponse = do - ensureLocked i - r <- readInterpProcess (instProcess i) get - unlock i - pure r - --- ----------------------------------------------------------------------------- --- Nested Message API - --- | Send any value (requires locked interpreter) -sendAnyValue :: Binary a => ExtInterpInstance d -> a -> IO () -sendAnyValue i m = ensureLocked i >> writeInterpProcess (instProcess i) (put m) - --- | Expect a value to be received (requires locked interpreter) -receiveAnyValue :: ExtInterpInstance d -> Get a -> IO a -receiveAnyValue i get = ensureLocked i >> readInterpProcess (instProcess i) get - --- | Wait for a Template Haskell message (requires locked interpreter) -receiveTHMessage :: ExtInterpInstance d -> IO THMsg -receiveTHMessage i = ensureLocked i >> receiveAnyValue i getTHMessage - --- ----------------------------------------------------------------------------- - -getLock :: ExtInterpInstance d -> MVar () -getLock = interpLock . instProcess - -withLock :: ExtInterpInstance d -> IO a -> IO a -withLock i f = withMVar (getLock i) (const f) - -lock :: ExtInterpInstance d -> IO () -lock i = takeMVar (getLock i) - -unlock :: ExtInterpInstance d -> IO () -unlock i = putMVar (getLock i) () - -ensureLocked :: ExtInterpInstance d -> IO () -ensureLocked i = - isEmptyMVar (getLock i) >>= \case - False -> panic "ensureLocked: external interpreter not locked" - _ -> pure () - +sendMessage i m = callInterpProcess (instProcess i) m -- | Send a 'Message' and receive the response from the interpreter process callInterpProcess :: Binary a => InterpProcess -> Message a -> IO a @@ -107,18 +32,6 @@ callInterpProcess i msg = remoteCall (interpPipe i) msg `catchException` \(e :: SomeException) -> handleInterpProcessFailure i e --- | Read a value from the interpreter process -readInterpProcess :: InterpProcess -> Get a -> IO a -readInterpProcess i get = - readPipe (interpPipe i) get - `catchException` \(e :: SomeException) -> handleInterpProcessFailure i e - --- | Send a value to the interpreter process -writeInterpProcess :: InterpProcess -> Put -> IO () -writeInterpProcess i put = - writePipe (interpPipe i) put - `catchException` \(e :: SomeException) -> handleInterpProcessFailure i e - handleInterpProcessFailure :: InterpProcess -> SomeException -> IO a handleInterpProcessFailure i e = do let hdl = interpHandle i ===================================== compiler/GHC/Runtime/Interpreter/Types.hs ===================================== @@ -111,7 +111,6 @@ type WasmInterp = ExtInterpState WasmInterpConfig () data InterpProcess = InterpProcess { interpPipe :: !Pipe -- ^ Pipe to communicate with the server , interpHandle :: !ProcessHandle -- ^ Process handle of the server - , interpLock :: !(MVar ()) -- ^ Lock to prevent concurrent access to the stream } -- | Status of an external interpreter ===================================== compiler/GHC/Runtime/Interpreter/Wasm.hs ===================================== @@ -73,14 +73,12 @@ spawnWasmInterp WasmInterpConfig {..} = do hSetBuffering rh NoBuffering interpPipe <- mkPipeFromHandles rh wh pending_frees <- newMVar [] - lock <- newMVar () pure $ ExtInterpInstance { instProcess = InterpProcess { interpHandle = ph, - interpPipe, - interpLock = lock + interpPipe }, instPendingFrees = pending_frees, instExtra = () ===================================== libraries/ghci/GHCi/Message.hs ===================================== @@ -12,7 +12,7 @@ module GHCi.Message ( Message(..), Msg(..) , ConInfoTable(..) - , THMessage(..), THMsg(..) + , THMessage(..) -- , THMsg(..) , QResult(..) , EvalStatus_(..,EvalBreak), EvalStatus, EvalResult(..), EvalOpts(..), EvalExpr(..) , EvalBreak(..), EvalBreakpoint (..) @@ -20,7 +20,7 @@ module GHCi.Message , toSerializableException, fromSerializableException , THResult(..), THResultType(..) , QState(..) - , getMessage, putMessage, getTHMessage, putTHMessage + , getMessage, putMessage, {- getTHMessage, -} putTHMessage , Pipe, mkPipeFromHandles, mkPipeFromContinuations, remoteCall, remoteTHCall, readPipe, writePipe , BreakModule , BreakUnitId @@ -62,6 +62,8 @@ import Data.Dynamic import Data.Typeable (TypeRep) import Data.IORef import Data.Map (Map) +import Data.IntMap (IntMap) +import qualified Data.IntMap as IntMap import Foreign import GHC.Generics import GHC.Stack.CCS @@ -70,6 +72,7 @@ import qualified GHC.Boot.TH.Monad as TH import System.Exit import System.IO import System.IO.Error +import Control.Monad -- ----------------------------------------------------------------------------- -- The RPC protocol between GHC and the interactive server @@ -132,13 +135,13 @@ data Message a where EvalStmt :: EvalOpts -> EvalExpr HValueRef {- IO [a] -} - -> Message (EvalStatus [HValueRef]) {- [a] -} + -> Message (RemoteRef ThreadId) {- [a] -} -- | Resume evaluation of a statement after a breakpoint ResumeStmt :: EvalOpts -> RemoteRef ThreadId - -> Message (EvalStatus [HValueRef]) + -> Message () -- | Abandon evaluation of a statement after a breakpoint AbandonStmt @@ -240,12 +243,23 @@ data Message a where -- | Evaluate something. This is used to support :force in GHCi. Seq :: HValueRef - -> Message (EvalStatus ()) + -> Message (RemoteRef ThreadId) -- | Resume forcing a free variable in a breakpoint (#2950) ResumeSeq :: RemoteRef ThreadId - -> Message (EvalStatus ()) + -> Message () + + WaitThreadResult + :: RemoteRef ThreadId + -> Message (EvalStatus [HValueRef]) + + WaitAnyThreadBreak + :: Message EvalBreak + + WaitThreadResultOrAnyThreadBreak + :: RemoteRef ThreadId + -> Message (EvalStatus [HValueRef]) -- | User-defined request encoded as a tag/payload pair. This is left -- uninterpreted by GHC and is meant for GHC API applications to be able to supply @@ -323,40 +337,40 @@ data THMessage a where deriving instance Show (THMessage a) -data THMsg = forall a . (Binary a, Show a) => THMsg (THMessage a) - -getTHMessage :: Get THMsg -getTHMessage = do - b <- getWord8 - case b of - 0 -> THMsg <$> NewName <$> get - 1 -> THMsg <$> (Report <$> get <*> get) - 2 -> THMsg <$> (LookupName <$> get <*> get) - 3 -> THMsg <$> Reify <$> get - 4 -> THMsg <$> ReifyFixity <$> get - 5 -> THMsg <$> (ReifyInstances <$> get <*> get) - 6 -> THMsg <$> ReifyRoles <$> get - 7 -> THMsg <$> (ReifyAnnotations <$> get <*> get) - 8 -> THMsg <$> ReifyModule <$> get - 9 -> THMsg <$> ReifyConStrictness <$> get - 10 -> THMsg <$> AddDependentFile <$> get - 11 -> THMsg <$> AddTempFile <$> get - 12 -> THMsg <$> AddTopDecls <$> get - 13 -> THMsg <$> (IsExtEnabled <$> get) - 14 -> THMsg <$> return ExtsEnabled - 15 -> THMsg <$> return StartRecover - 16 -> THMsg <$> EndRecover <$> get - 17 -> THMsg <$> return FailIfErrs - 18 -> return (THMsg RunTHDone) - 19 -> THMsg <$> AddModFinalizer <$> get - 20 -> THMsg <$> (AddForeignFilePath <$> get <*> get) - 21 -> THMsg <$> AddCorePlugin <$> get - 22 -> THMsg <$> ReifyType <$> get - 23 -> THMsg <$> (PutDoc <$> get <*> get) - 24 -> THMsg <$> GetDoc <$> get - 25 -> THMsg <$> return GetPackageRoot - 26 -> THMsg <$> AddDependentDirectory <$> get - n -> error ("getTHMessage: unknown message " ++ show n) +-- data THMsg = forall a . (Binary a, Show a) => THMsg (THMessage a) +-- +-- getTHMessage :: Get THMsg +-- getTHMessage = do +-- b <- getWord8 +-- case b of +-- 0 -> THMsg <$> NewName <$> get +-- 1 -> THMsg <$> (Report <$> get <*> get) +-- 2 -> THMsg <$> (LookupName <$> get <*> get) +-- 3 -> THMsg <$> Reify <$> get +-- 4 -> THMsg <$> ReifyFixity <$> get +-- 5 -> THMsg <$> (ReifyInstances <$> get <*> get) +-- 6 -> THMsg <$> ReifyRoles <$> get +-- 7 -> THMsg <$> (ReifyAnnotations <$> get <*> get) +-- 8 -> THMsg <$> ReifyModule <$> get +-- 9 -> THMsg <$> ReifyConStrictness <$> get +-- 10 -> THMsg <$> AddDependentFile <$> get +-- 11 -> THMsg <$> AddTempFile <$> get +-- 12 -> THMsg <$> AddTopDecls <$> get +-- 13 -> THMsg <$> (IsExtEnabled <$> get) +-- 14 -> THMsg <$> return ExtsEnabled +-- 15 -> THMsg <$> return StartRecover +-- 16 -> THMsg <$> EndRecover <$> get +-- 17 -> THMsg <$> return FailIfErrs +-- 18 -> return (THMsg RunTHDone) +-- 19 -> THMsg <$> AddModFinalizer <$> get +-- 20 -> THMsg <$> (AddForeignFilePath <$> get <*> get) +-- 21 -> THMsg <$> AddCorePlugin <$> get +-- 22 -> THMsg <$> ReifyType <$> get +-- 23 -> THMsg <$> (PutDoc <$> get <*> get) +-- 24 -> THMsg <$> GetDoc <$> get +-- 25 -> THMsg <$> return GetPackageRoot +-- 26 -> THMsg <$> AddDependentDirectory <$> get +-- n -> error ("getTHMessage: unknown message " ++ show n) putTHMessage :: THMessage a -> Put putTHMessage m = case m of @@ -719,11 +733,27 @@ serializeBCOs rbcos = parMap doChunk (chunkList 100 rbcos) -- ----------------------------------------------------------------------------- -- Reading/writing messages +type CorrelationId = Int +data Request = Request !CorrelationId Msg +data Reply = Reply !CorrelationId LB.ByteString + +instance Binary Request where + put (Request i (Msg m)) = put i >> putMessage m + get = Request <$> get <*> getMessage + +instance Binary Reply where + put (Reply i bs) = put i >> put bs + get = Reply <$> get <*> get + -- | An opaque pipe for bidirectional binary data transmission. data Pipe = Pipe - { getSome :: !(IO ByteString) - , putAll :: !(B.Builder -> IO ()) - , pipeLeftovers :: !(IORef (Maybe ByteString)) + { getSome :: !(IO ByteString) + , putAll :: !(B.Builder -> IO ()) + , pipeLeftovers :: !(IORef (Maybe ByteString)) + , pipeLock :: !(MVar ()) -- ^ Lock to prevent concurrent access to the stream + , pipePending :: !(IORef (IntMap (MVar LB.ByteString))) + , pipeDrainThread :: !(MVar ThreadId) + , pipeNextReq :: !(IORef Int) } -- | Make a 'Pipe' from a 'Handle' to read and a 'Handle' to write. @@ -733,30 +763,59 @@ mkPipeFromHandles pipeRead pipeWrite = do putAll b = do B.hPutBuilder pipeWrite b hFlush pipeWrite - pipeLeftovers <- newIORef Nothing - pure $ Pipe { getSome, putAll, pipeLeftovers } + mkPipeFromContinuations getSome putAll -- | Make a 'Pipe' from a reader function and a writer function. mkPipeFromContinuations :: IO ByteString -> (B.Builder -> IO ()) -> IO Pipe mkPipeFromContinuations getSome putAll = do - pipeLeftovers <- newIORef Nothing - pure $ Pipe { getSome, putAll, pipeLeftovers } + pipeLeftovers <- newIORef Nothing + pipeLock <- newMVar () + pipePending <- newIORef IntMap.empty + pipeDrainThread <- newEmptyMVar + pipeNextReq <- newIORef 0 + let p = Pipe { .. } + + -- Fork thread to drain read end. + drainTid <- forkIO (drainPipe p) + putMVar pipeDrainThread drainTid + -- todo: proper thread clean up and no leaks -remoteCall :: Binary a => Pipe -> Message a -> IO a + pure p + +remoteCall :: (Show a, Binary a) => Pipe -> Message a -> IO a remoteCall pipe msg = do - writePipe pipe (putMessage msg) - readPipe pipe get -writePipe :: Pipe -> Put -> IO () -writePipe Pipe{..} put = putAll $ execPut put + -- Mark pending + uq <- freshReqId pipe + wv <- newEmptyMVar + atomicModifyIORef' (pipePending pipe) $ + \m -> (IntMap.insert uq wv m, ()) + + -- Write request + writePipe pipe (put (Request uq (Msg msg))) + + -- Block waiting for reply + runGet get <$> takeMVar wv remoteTHCall :: Binary a => Pipe -> THMessage a -> IO a remoteTHCall pipe msg = do writePipe pipe (putTHMessage msg) readPipe pipe get +drainPipe :: Pipe -> IO () +drainPipe p = forever $ do + Reply uq bs <- readPipe p get + wv <- atomicModifyIORef' (pipePending p) $ \m -> do + case IntMap.lookup uq m of + Nothing -> error "drainPipe: Received reply to no pending request" + Just wv -> (IntMap.delete uq m, wv) + putMVar wv bs -- signal matching request + +writePipe :: Pipe -> Put -> IO () +writePipe p@Pipe{..} put = withLock p $ putAll $ execPut put + readPipe :: Pipe -> Get a -> IO a -readPipe Pipe{..} get = do +readPipe p@Pipe{..} get = withLock p $ do leftovers <- readIORef pipeLeftovers m <- getBin getSome get leftovers case m of @@ -786,3 +845,9 @@ getBin getsome get leftover = go leftover (runGetIncremental get) else go Nothing (fun (Just b)) go _lft (Fail _rest _off str) = throwIO (ErrorCall ("getBin: " ++ str)) + +withLock :: Pipe -> IO c -> IO c +withLock Pipe{..} = bracket (takeMVar pipeLock) (putMVar pipeLock) . const + +freshReqId :: Pipe -> IO Int +freshReqId Pipe{pipeNextReq} = atomicModifyIORef' pipeNextReq $ \i -> (i+1, i) ===================================== libraries/ghci/GHCi/Run.hs ===================================== @@ -128,6 +128,15 @@ run m = case m of Seq ref -> doSeq ref ResumeSeq ref -> resumeSeq ref + WaitThreadResult tid -> localRef tid >>= atomically . readThreadEvalStatus + WaitAnyThreadBreak -> atomically $ readAnyThreadEvalBreak + WaitThreadResultOrAnyThreadBreak t -> do + tid <- localRef t + atomically $ + readThreadEvalStatus tid + `orElse` + (EvalPaused <$> readAnyThreadEvalBreak) + Shutdown -> unexpectedMessage m RunTH {} -> unexpectedMessage m RunModFinalizers {} -> unexpectedMessage m @@ -206,7 +215,7 @@ tryEval io = do -------------------------------------------------------------------------------- -evalStmt :: EvalOpts -> EvalExpr HValueRef -> IO (EvalStatus [HValueRef]) +evalStmt :: EvalOpts -> EvalExpr HValueRef -> IO (RemoteRef ThreadId) evalStmt opts expr = do io <- mkIO expr sandboxIO opts $ do @@ -248,16 +257,16 @@ evalStringToString r str = do -- The UI process has more and therefore also can show more -- information about the breakpoint than the current iserv -- process. -doSeq :: RemoteRef a -> IO (EvalStatus ()) -doSeq ref = clearEvalStatus <$> do +doSeq :: RemoteRef a -> IO (RemoteRef ThreadId) +doSeq ref = do sandboxIO evalOptsSeq $ do _ <- (void $ evaluate =<< localRef ref) return [] -- | Process a ResumeSeq message. Continue the :force processing #2950 -- after a breakpoint. -resumeSeq :: RemoteRef ThreadId -> IO (EvalStatus ()) -resumeSeq hvref = clearEvalStatus <$> resumeStmt evalOptsSeq hvref +resumeSeq :: RemoteRef ThreadId -> IO () +resumeSeq hvref = resumeStmt evalOptsSeq hvref evalOptsSeq :: EvalOpts evalOptsSeq = EvalOpts @@ -296,15 +305,16 @@ abandonStmt hvref = do -------------------------------------------------------------------------------- -resumeStmt :: EvalOpts -> RemoteRef ThreadId -> IO (EvalStatus [HValueRef]) +resumeStmt :: EvalOpts -> RemoteRef ThreadId -> IO () resumeStmt opts rtid = do resumeThreadId <- localRef rtid ResumeContext{..} <- getThreadResumeContext resumeThreadId - withBreakAction opts resumeThreadId $ \waitForResult -> - mask_ $ do - putMVar resumeBreakMVar () -- this awakens the stopped thread... - redirectInterrupts resumeThreadId $ - waitForResult + setBreakAction opts resumeThreadId -- See Note [TODO: all break options must be set every time] + -- withBreakAction opts resumeThreadId $ \waitForResult -> + -- mask_ $ do + putMVar resumeBreakMVar () -- this awakens the stopped thread... + -- redirectInterrupts resumeThreadId $ + -- waitForResult -- When running a computation, we redirect ^C exceptions to the running -- thread. ToDo: we might want a way to continue even if the target @@ -316,7 +326,7 @@ resumeStmt opts rtid = do -- only while we execute the user's code. We can't afford to lose the final -- putMVar, otherwise deadlock ensues. (#1583, #1922, #1946) -sandboxIO :: EvalOpts -> IO [HValueRef] -> IO (EvalStatus [HValueRef]) +sandboxIO :: EvalOpts -> IO [HValueRef] -> IO (RemoteRef ThreadId) sandboxIO opts io | useSandboxThread opts = do @@ -336,17 +346,31 @@ sandboxIO opts io tid <- forkIO $ do takeMVar wait -- don't run anything in this thread before - -- `withBreakAction` has a chance to set up. + -- `setBreakAction` has a chance to set up. tid <- myThreadId labelThread tid "GHCi sandbox" + + -- WHEN THE THREAD IS FINISHED, write the value itself. + -- + -- we only write the TID status map in two places: + -- + -- When the "main" thread finishes (here) + -- When any thread hits a breakpoint (global break action) unsafeUnmask runIt >>= writeThreadEvalStatus tid + -- TODO: This unsafe-unmask is very dubious. Consult with Andrea. I + -- don't think we should be uninterruptable masking interp commands + -- somewhere else. This should just be runIt? - -- We are running in uninterruptibleMask - withBreakAction opts tid $ \waitForResult -> do - putMVar wait () - redirectInterrupts tid $ unsafeUnmask $ - waitForResult + -- We are running in uninterruptibleMask -- TODO WHY?? + setBreakAction opts tid + putMVar wait () + + mkRemoteRef tid + -- withBreakAction opts tid $ \waitForResult -> do + -- putMVar wait () + -- redirectInterrupts tid $ unsafeUnmask $ + -- waitForResult | otherwise = do @@ -360,7 +384,17 @@ sandboxIO opts io -- tracing, etc.) need the expression to be running in a -- separate thread, so debugging is only enabled when -- using the sandbox. - runIt + tid <- forkIO (pure ()) + runIt >>= writeThreadEvalStatus tid + mkRemoteRef tid + -- We `runIt` on the main thread, but fork a new thread to store to the + -- result with `writeThreadEvalStatus` to fit the asynchronous API. We + -- can't use myThreadId because if the user happens to runIt twice + -- before awaiting, the second result will clobber the first. + + -- TODO We have to be careful evacuating this map bc there may be + -- uncollected results from finished threads. So the thread map can't + -- be weak refs ? Or maybe because of the RemoteRef it's fine. Ah, prob. where runIt = measureAlloc $ tryEval $ rethrow opts $ clearCCS io @@ -415,12 +449,16 @@ sandboxIO opts io -- whenever "main" or one of the threads spawned by "main" hit a breakpoint, -- rather than just "main" being observed. This should be used for the main -- debuggee execution of a debugger. -withBreakAction :: EvalOpts -> ThreadId -> (IO (EvalStatus [HValueRef]) -> IO a) -> IO a -withBreakAction opts tid@(ThreadId tid#) act = do +-- withBreakAction :: EvalOpts -> ThreadId -> (IO (EvalStatus [HValueRef]) -> IO a) -> IO a +-- withBreakAction opts tid@(ThreadId tid#) act = do + +setBreakAction :: EvalOpts -> ThreadId -> IO () +setBreakAction opts tid@(ThreadId tid#) = do ctx <- getThreadResumeContext tid - bracket (setBreakAction ctx) (resetBreakAction ctx) (\_ -> act (waitForResult ctx)) + setBreakAction' ctx + -- bracket _ (resetBreakAction ctx) (\_ -> act (waitForResult ctx)) where - setBreakAction ctx = do + setBreakAction' ctx = do -- TODO: TERRIBLY NOT MULTI-THREAD FRIENDLY. Each thread needs its own RTS flag exceptionFlag, like we have for step-in or step-out. runIf breakOnException $ \_ -> poke exceptionFlag 1 -- Breaking on exceptions is not enabled by default, since it @@ -433,7 +471,7 @@ withBreakAction opts tid@(ThreadId tid#) act = do runIfCtx ctx isolateThreadBreaks setIsolatedCtx resetBreakAction ctx () = do - poke exceptionFlag 0 + poke exceptionFlag 0 -- just wrong, expection flag should be per-thread rts_disableStopAfterReturn tid# rts_disableStopNextBreakpoint tid# runIfCtx ctx isolateThreadBreaks unsetIsolatedCtx @@ -485,9 +523,3 @@ redirectInterrupts target wait = do Nothing -> wait Just target -> do throwTo target (e :: SomeException); wait -clearEvalStatus :: EvalStatus a -> EvalStatus () -clearEvalStatus = \case - EvalComplete w (EvalException se) -> EvalComplete w (EvalException se) - EvalComplete w (EvalSuccess _) -> EvalComplete w (EvalSuccess ()) - EvalBreak ap mb rt rccs -> EvalBreak ap mb rt rccs - ===================================== libraries/ghci/GHCi/Server.hs ===================================== @@ -70,6 +70,14 @@ servWithCustom verbose hook pipe restore customHandler = loop when verbose $ trace ("msg: " ++ (show msg)) case msg of + -- The expressions which block waiting for a result musn't block the + -- server thread, so send them off and notify over the wire when done. + -- + -- The other end (see 'remoteCall') will be processing notifications, and + -- it will use the returned pointer as the index for the matching notification. + WaitThreadResult {} -> forkIO (run msg >>= (notification =<< myThreadId)) >>= reply . mkRemoteRef + WaitAnyThreadBreak {} -> forkIO (run msg >>= (notification =<< myThreadId)) >>= reply . mkRemoteRef + WaitThreadResultOrAnyThreadBreak {} -> forkIO (run msg >>= (notification =<< myThreadId)) >>= reply . mkRemoteRef CustomMessage tag payload -> handleCustom tag payload Shutdown -> return () RunTH st q ty loc -> wrapRunTH $ runTH pipe st q ty loc View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/a7150ad38a9ef92ff9a00e345b5b3f1... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/compare/a7150ad38a9ef92ff9a00e345b5b3f1... 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