From 239da09e55953b99b82c0d7ac838e220aaf03341 Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Mon, 27 Jul 2026 23:54:00 +0300 Subject: [PATCH 01/15] Add runPyAsync for running python code asynchronously --- src/Python/Inline.hs | 2 ++ src/Python/Internal/Eval.hs | 28 +++++++++++++++++++++++++++- 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/src/Python/Inline.hs b/src/Python/Inline.hs index 82624bf..ce05467 100644 --- a/src/Python/Inline.hs +++ b/src/Python/Inline.hs @@ -45,6 +45,8 @@ module Python.Inline , Py , runPy , runPyInMain + , runPyAsync + , runPyAsyncEither , PyObject , PyError(..) , PyException(..) diff --git a/src/Python/Internal/Eval.hs b/src/Python/Internal/Eval.hs index a6124ab..a38db80 100644 --- a/src/Python/Internal/Eval.hs +++ b/src/Python/Internal/Eval.hs @@ -15,6 +15,8 @@ module Python.Internal.Eval -- * Evaluator , runPy , runPyInMain + , runPyAsync + , runPyAsyncEither , unsafeRunPy -- * GC-related , newPyObject @@ -274,6 +276,13 @@ releaseLock tid = readTVar globalPyLock >>= \case [] -> LockUnlocked t':ts -> Locked t' ts +ensureInit :: STM () +ensureInit = readTVar globalPyLock >>= \case + LockUninialized -> throwSTM PythonNotInitialized + LockFinalized -> throwSTM PythonIsFinalized + LockedByGC -> pure () + LockUnlocked -> pure () + Locked{} -> pure () ---------------------------------------------------------------- @@ -517,7 +526,24 @@ runPyInMain py either throwM pure r --- | Execute python action. This function is unsafe and should be only + +runPyAsyncEither :: Py a -> IO (STM (Either SomeException a)) +runPyAsyncEither py = do + atomically ensureInit + result <- newEmptyTMVarIO + -- FIXME: Should we rethrow only python expections? Sound sensible + _ <- forkOS $ do + a <- try $ unsafeRunPy $ ensureGIL py + atomically $ putTMVar result a + pure $ takeTMVar result + +runPyAsync :: Py a -> IO (STM a) +runPyAsync py = do + res <- runPyAsyncEither py + return $ either throwSTM pure =<< res + + + -- | Execute python action. This function is unsafe and should be only -- called in thread of interpreter. unsafeRunPy :: Py a -> IO a unsafeRunPy (Py io) = io From ab2f2040382b73edb124fc2071d4ae6ac7d71f08 Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Wed, 29 Jul 2026 10:52:32 +0300 Subject: [PATCH 02/15] Add API modelled after async library with support of cancelling As it turns out it's possible to cancel python threads asynchronously --- cbits/python.c | 11 ++++ include/inline-python.h | 9 +++ inline-python.cabal | 1 + src/Python/Inline.hs | 2 - src/Python/Inline/Async.hs | 23 ++++++++ src/Python/Internal/Eval.hs | 107 +++++++++++++++++++++++++++++------- 6 files changed, 132 insertions(+), 21 deletions(-) create mode 100644 src/Python/Inline/Async.hs diff --git a/cbits/python.c b/cbits/python.c index 8376d04..933a758 100644 --- a/cbits/python.c +++ b/cbits/python.c @@ -210,3 +210,14 @@ void inline_py_Integer_FromPy( PyLong_AsNativeBytes(p, buf, size, -1); #endif } + + + +static PyObject* AsyncError = 0; + +PyObject* inline_py_AsyncError() { + if( AsyncError == 0 ) { + AsyncError = PyErr_NewException("inline_py.AsyncError", PyExc_BaseException, 0); + } + return AsyncError; +} diff --git a/include/inline-python.h b/include/inline-python.h index 872de92..a6dcca4 100644 --- a/include/inline-python.h +++ b/include/inline-python.h @@ -81,3 +81,12 @@ void inline_py_Integer_FromPy( void* buf, size_t size ); + + + +// ================================================================ +// Async exceptions +// ================================================================ + +// Obtain class for async exception +PyObject* inline_py_AsyncError(); diff --git a/inline-python.cabal b/inline-python.cabal index 44bb8a1..9b0b46e 100644 --- a/inline-python.cabal +++ b/inline-python.cabal @@ -80,6 +80,7 @@ Library Python.Inline.QQ Python.Inline.Eval Python.Inline.Types + Python.Inline.Async Other-modules: Python.Internal.CAPI Python.Internal.Eval diff --git a/src/Python/Inline.hs b/src/Python/Inline.hs index ce05467..82624bf 100644 --- a/src/Python/Inline.hs +++ b/src/Python/Inline.hs @@ -45,8 +45,6 @@ module Python.Inline , Py , runPy , runPyInMain - , runPyAsync - , runPyAsyncEither , PyObject , PyError(..) , PyException(..) diff --git a/src/Python/Inline/Async.hs b/src/Python/Inline/Async.hs new file mode 100644 index 0000000..759c8b1 --- /dev/null +++ b/src/Python/Inline/Async.hs @@ -0,0 +1,23 @@ +-- | +-- Asynchronous computation using python. Normally library tries to +-- execute python code in the same thread. Moreover it use global lock +-- in addition to GIL in order to avoid blocking capability on GIL. +-- This module provide API for working with concurrent python. +-- Its API is heavily modelled after @async@ package. +-- +-- Note it's very experimental and not well tested. Also mixing +-- concurrency primitives from two languages makes difficult task of +-- concurrent programming even more complicated. +module Python.Inline.Async + ( PyAsync + , PyAsyncCancelled(..) + , runPyAsync + , withPyAsync + , waitPy + , waitPyCatch + , cancelPy + , uninterruptibleCancelPy + ) where + +import Python.Internal.Eval + diff --git a/src/Python/Internal/Eval.hs b/src/Python/Internal/Eval.hs index a38db80..99c619d 100644 --- a/src/Python/Internal/Eval.hs +++ b/src/Python/Internal/Eval.hs @@ -15,9 +15,16 @@ module Python.Internal.Eval -- * Evaluator , runPy , runPyInMain - , runPyAsync - , runPyAsyncEither , unsafeRunPy + -- ** Async + , PyAsync + , PyAsyncCancelled(..) + , waitPy + , waitPyCatch + , cancelPy + , uninterruptibleCancelPy + , runPyAsync + , withPyAsync -- * GC-related , newPyObject -- * C-API wrappers @@ -56,6 +63,7 @@ import Control.Monad.Trans.Cont import Data.Maybe import Data.Function import Data.ByteString.Unsafe qualified as BS +import Data.Word import Foreign.Concurrent qualified as GHC import Foreign.Ptr import Foreign.ForeignPtr @@ -525,29 +533,90 @@ runPyInMain py takeMVar resp `onException` throwTo tid_main InterruptMain either throwM pure r +-- | Execute python action. This function is unsafe and should be only +-- called in thread of interpreter. +unsafeRunPy :: Py a -> IO a +unsafeRunPy (Py io) = io -runPyAsyncEither :: Py a -> IO (STM (Either SomeException a)) -runPyAsyncEither py = do +---------------------------------------------------------------- +-- Async running +---------------------------------------------------------------- + +-- | Exception thrown to a thread doing async python computation. +data PyAsyncCancelled = PyAsyncCancelled + deriving (Show, Eq) + +instance Exception PyAsyncCancelled + +-- | Handle to asynchronous python computation spawned by +-- 'runPyAsync'. It's performed on separate OS thread. Use +-- 'wait'\/'waitCatch' to obtain computation result. +data PyAsync a = PyAsync + { asyncTID :: !ThreadId + , asyncPyTID :: !Word64 + , asyncWait :: STM (Either SomeException a) + } + +-- | Wait for result of asynchronous computation. If it threw an +-- exception it will be rethrown by @wait@. +waitPy :: PyAsync a -> STM a +waitPy a = either throwSTM pure =<< a.asyncWait + +-- | Wait for result of asynchronous computation. Exception thrown by +-- it will be returned as @Left@. +waitPyCatch :: PyAsync a -> STM (Either SomeException a) +waitPyCatch = (.asyncWait) + +-- | Cancel execution of asynchronous computation. Most likely thread +-- will be executing some python so first it attempts to raise async +-- exception in python code. Then it throws 'PyAsyncCancelled' in case +-- it executes haskell code. This means thread could be terminate +-- either with 'PyError' or 'PyAsyncCancelled'. +-- +-- Note that python code generally is not written under assumption +-- that it could be smitten with exception at an absolutely any +-- moment. +cancelPy :: PyAsync a -> IO () +cancelPy PyAsync{asyncTID=tid, asyncPyTID=py_tid} + | rtsSupportsBoundThreads = runInBoundThread go + | otherwise = go + where + go = do + [C.block| void { + int gil = PyGILState_Ensure(); + int n = PyThreadState_SetAsyncExc($(uint64_t py_tid), inline_py_AsyncError()); + printf("My job is done (%d) tid=(%ld)\n", n, $(uint64_t py_tid)); + PyGILState_Release(gil); + }|] + throwTo tid PyAsyncCancelled + +-- | Variant of 'cancel' which isn't interruptible. +uninterruptibleCancelPy :: PyAsync a -> IO () +uninterruptibleCancelPy = uninterruptibleMask_ . cancelPy + +-- | Create new OS thread and execute python code on it. +runPyAsync :: Py a -> IO (PyAsync a) +runPyAsync py = do atomically ensureInit - result <- newEmptyTMVarIO - -- FIXME: Should we rethrow only python expections? Sound sensible - _ <- forkOS $ do + result <- newEmptyTMVarIO + py_tid_mv <- newEmptyMVar + tid <- forkOS $ do + -- Obtain python thread ID + putMVar py_tid_mv =<< [CU.exp| uint64_t { PyThread_get_thread_ident() } |] a <- try $ unsafeRunPy $ ensureGIL py atomically $ putTMVar result a - pure $ takeTMVar result - -runPyAsync :: Py a -> IO (STM a) -runPyAsync py = do - res <- runPyAsyncEither py - return $ either throwSTM pure =<< res - - - -- | Execute python action. This function is unsafe and should be only --- called in thread of interpreter. -unsafeRunPy :: Py a -> IO a -unsafeRunPy (Py io) = io + py_tid <- takeMVar py_tid_mv + pure PyAsync + { asyncTID = tid + , asyncPyTID = py_tid + , asyncWait = takeTMVar result + } +-- | Create new OS thread and execute python code on it. Will use +-- 'uninterruptibleCancel' after callback finishes execution. +withPyAsync :: Py a -> (PyAsync a -> IO b) -> IO b +withPyAsync py = bracket (runPyAsync py) uninterruptibleCancelPy ---------------------------------------------------------------- From 0ead691562986f8810a7182bcb8b566f6dd3bba1 Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Wed, 29 Jul 2026 11:37:52 +0300 Subject: [PATCH 03/15] Add tests for cancelling python --- inline-python.cabal | 1 + test/TST/Run.hs | 40 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 41 insertions(+) diff --git a/inline-python.cabal b/inline-python.cabal index 9b0b46e..73c9a82 100644 --- a/inline-python.cabal +++ b/inline-python.cabal @@ -99,6 +99,7 @@ library test , tasty >=1.2 , tasty-hunit >=0.10 , tasty-quickcheck >=0.10 + , stm , quickcheck-instances >=0.3.33 , exceptions , containers diff --git a/test/TST/Run.hs b/test/TST/Run.hs index 6f5892f..0dd1094 100644 --- a/test/TST/Run.hs +++ b/test/TST/Run.hs @@ -3,6 +3,7 @@ module TST.Run(tests) where import Control.Concurrent +import Control.Concurrent.STM import Control.Exception import Control.Monad import Control.Monad.IO.Class @@ -12,6 +13,7 @@ import Test.Tasty.HUnit import Python.Inline import Python.Inline.QQ import Python.Inline.Eval +import Python.Inline.Async import TST.Util tests :: TestTree @@ -164,8 +166,46 @@ tests = testGroup "Run python" assert m_hs.a == 12 assert m_hs.b == 'asd' |] + , testGroup "async" $ guardThreaded + [ -- We can run async computation at all + testCase "runPyAsync" $ do + runPy [pymain| dct = {} |] + a <- runPyAsync $ [py_| dct[1] = 100 |] + _ <- atomically $ waitPy a + n <- runPy $ fromPy =<< [pye| dct[1] |] + assertEqual "x" (Just (100::Int)) n + runPy [pymain| del dct |] + , -- Cancellation of python code + testCase "cancelPy [python]" $ do + a <- runPyAsync $ forever $ [py_| + import time + while True: + time.sleep(1e-3) + |] + d <- registerDelay 100_000 + cancelPy a + _ <- atomically $ waitPyCatch a `orElse` do readTVar d >>= \case + True -> error "Timeout" + False -> retry + return () + -- , -- Cancellation of haskell code + -- testCase "cancelPy [haskell]" $ do + -- a <- runPyAsync $ do + -- liftIO $ forever $ threadDelay 1_000_000 + -- d <- registerDelay 100_000 + -- cancelPy a + -- _ <- atomically $ waitPyCatch a `orElse` do readTVar d >>= \case + -- True -> error "Timeout" + -- False -> retry + -- return () + ] ] data Stop = Stop deriving stock Show deriving anyclass Exception + +guardThreaded :: [TestTree] -> [TestTree] +guardThreaded ts + | rtsSupportsBoundThreads = ts + | otherwise = [] From aac39be3aec484665f0cab6d5ce128613608c132 Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Sun, 9 Aug 2026 20:03:41 +0300 Subject: [PATCH 04/15] Amend NOTE on concurrency --- src/Python/Internal/Eval.hs | 25 +++++++++++-------------- 1 file changed, 11 insertions(+), 14 deletions(-) diff --git a/src/Python/Internal/Eval.hs b/src/Python/Internal/Eval.hs index 99c619d..b333d55 100644 --- a/src/Python/Internal/Eval.hs +++ b/src/Python/Internal/Eval.hs @@ -110,30 +110,27 @@ C.include "" -- -- One could think that running python code in bound threads and -- making sure that GIL is held would suffice. It doesn't. Doing so --- would quickly results in deadlock. Exact reason for that is not +-- quickly results in deadlock. Exact reason for that is not -- understood. -- -- Another problem is GHC may schedule two threads each running python --- code on same capability. They won't have any problems taking GIL --- and will run concurrently stepping on each other's toes. --- --- Only way to solve this problem is to introduce another lock on --- haskell side. It's visible to haskell RTS so we won't get deadlocks --- and it makes sure that only one haskell thread interacts with --- python at a time. --- +-- code on same capability. It seems very likely that they'll step on +-- each others' toes. -- +-- Current solution is to protect execution of python code with global +-- lock. Since it's visible to haskell RTS we don't get deadlocks. +-- This also means we can't execute python code concurrently. + + + +-- NOTE: [Main thread] +-- ~~~~~~~~~~~~~~~~~~~ -- -- Also python designate thread in which python interpreter was -- initialized as a main thread. It has special status for example -- some libraries may run only in main thread (e.g. tkinter). But if -- we don't take special precautions we won't know which thread it -- is. --- --- --- --- There's of course question how well python threading interacts with --- haskell. No one knows, probably it won't work well. From 1a7175703d73cc6f4ff04cf4593aee6aeec0f857 Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Wed, 19 Aug 2026 20:55:16 +0300 Subject: [PATCH 05/15] Implement cancelPy correctly We must to try interrupt thread concurrently by throwing haskell and python thread --- src/Python/Internal/Eval.hs | 44 ++++++++++++++++++++++++------------- 1 file changed, 29 insertions(+), 15 deletions(-) diff --git a/src/Python/Internal/Eval.hs b/src/Python/Internal/Eval.hs index b333d55..62cc5ae 100644 --- a/src/Python/Internal/Eval.hs +++ b/src/Python/Internal/Eval.hs @@ -551,7 +551,7 @@ instance Exception PyAsyncCancelled -- 'wait'\/'waitCatch' to obtain computation result. data PyAsync a = PyAsync { asyncTID :: !ThreadId - , asyncPyTID :: !Word64 + , asyncPyTID :: !(IO Word64) , asyncWait :: STM (Either SomeException a) } @@ -575,18 +575,33 @@ waitPyCatch = (.asyncWait) -- that it could be smitten with exception at an absolutely any -- moment. cancelPy :: PyAsync a -> IO () -cancelPy PyAsync{asyncTID=tid, asyncPyTID=py_tid} - | rtsSupportsBoundThreads = runInBoundThread go - | otherwise = go +cancelPy PyAsync{asyncTID=tid, asyncPyTID} + = runInBoundThread go where go = do - [C.block| void { - int gil = PyGILState_Ensure(); - int n = PyThreadState_SetAsyncExc($(uint64_t py_tid), inline_py_AsyncError()); - printf("My job is done (%d) tid=(%ld)\n", n, $(uint64_t py_tid)); - PyGILState_Release(gil); - }|] + -- As long as thread is running python code it couldn't be + -- interrupted by haskell exception so we spawn separate thread + -- which attempts to repeatedly interrupt python evaluation. + -- + -- PyThreadState_SetAsyncExc won't do anything if thread isn't + -- running python at the moment so we attempt to cancel python + -- repeatedly until we get + py_tid <- asyncPyTID + mv_done <- newEmptyMVar + -- Interrupting python + _ <- forkIO $ runInBoundThread $ fix $ \loop -> do + [C.block| void { + int gil = PyGILState_Ensure(); + int n = PyThreadState_SetAsyncExc($(uint64_t py_tid), inline_py_AsyncError()); + printf("My job is done (%d) tid=(%ld)\n", n, $(uint64_t py_tid)); + PyGILState_Release(gil); + }|] + tryReadMVar mv_done >>= \case + Just () -> return () + Nothing -> threadDelay 50 >> loop -- Avoid hammering interrupt too hard throwTo tid PyAsyncCancelled + putMVar mv_done () + -- | Variant of 'cancel' which isn't interruptible. uninterruptibleCancelPy :: PyAsync a -> IO () @@ -598,15 +613,14 @@ runPyAsync py = do atomically ensureInit result <- newEmptyTMVarIO py_tid_mv <- newEmptyMVar - tid <- forkOS $ do - -- Obtain python thread ID - putMVar py_tid_mv =<< [CU.exp| uint64_t { PyThread_get_thread_ident() } |] + tid <- forkOS $ mask_ $ do + -- Obtain python thread ID. + putMVar py_tid_mv =<< [C.exp| uint64_t { PyThread_get_thread_ident() } |] a <- try $ unsafeRunPy $ ensureGIL py atomically $ putTMVar result a - py_tid <- takeMVar py_tid_mv pure PyAsync { asyncTID = tid - , asyncPyTID = py_tid + , asyncPyTID = readMVar py_tid_mv , asyncWait = takeTMVar result } From dc3aec2036b2b23189951e5915d4ac8eed99f451 Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Fri, 21 Aug 2026 16:50:52 +0300 Subject: [PATCH 06/15] Better error message --- test/TST/Run.hs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/test/TST/Run.hs b/test/TST/Run.hs index 0dd1094..339b539 100644 --- a/test/TST/Run.hs +++ b/test/TST/Run.hs @@ -24,7 +24,9 @@ tests = testGroup "Run python" , testCase "Nested runPyInMain" $ runPyInMain $ liftIO $ runPyInMain $ pure () , testCase "runPyInMain" $ runPyInMain $ [py_| import threading - assert threading.main_thread() == threading.current_thread() + tid_main = threading.main_thread() + tid_our = threading.current_thread() + assert tid_main == tid_our, f"TID[main]={tid_main}, TID[our]={tid_our}" |] , testCase "Python exceptions are converted (py)" $ runPy $ throwsPy [py_| 1 / 0 |] , testCase "Python exceptions are converted (std)" $ throwsPyIO $ runPy [py_| 1 / 0 |] From 794633ff3affe69dfb24aeae7ec6e76a20da78a8 Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Fri, 21 Aug 2026 16:53:28 +0300 Subject: [PATCH 07/15] Python<=3.11 have no exception object when interrupted --- src/Python/Internal/Eval.hs | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/src/Python/Internal/Eval.hs b/src/Python/Internal/Eval.hs index 62cc5ae..1bc8e33 100644 --- a/src/Python/Internal/Eval.hs +++ b/src/Python/Internal/Eval.hs @@ -729,7 +729,18 @@ convertPy2Haskell = runProgram $ do PyErr_Fetch(p, p+1, p+2); }|] p_type <- peekElemOff p_errors 0 - p_value <- peekElemOff p_errors 1 + -- NOTE: When we set exception using PyThreadState_SetAsyncExc + -- this field remains NULL on python<=3.11. In this case we + -- assume it's our AsyncError: + p_value <- peekElemOff p_errors 1 >>= \case + NULL -> [CU.block| PyObject* { + PyObject *err_class = inline_py_AsyncError(); + PyObject *tuple = PyTuple_New(0); + PyObject *err = PyObject_Call(err_class, tuple, NULL); + Py_DECREF(tuple); + return err; + } |] + p -> pure p -- Traceback is not used ATM pure (p_type,p_value) -- Convert exception type and value to strings. From 6745d8201d867fe85a2e6e9895e0d7ecf18940e9 Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Fri, 21 Aug 2026 17:23:48 +0300 Subject: [PATCH 08/15] Typo in name --- src/Python/Internal/Eval.hs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/src/Python/Internal/Eval.hs b/src/Python/Internal/Eval.hs index 1bc8e33..88d2a2c 100644 --- a/src/Python/Internal/Eval.hs +++ b/src/Python/Internal/Eval.hs @@ -300,8 +300,8 @@ ensureInit = readTVar globalPyLock >>= \case initializePython :: IO () -- See NOTE: [Python and threading] initializePython = [CU.exp| int { Py_IsInitialized() } |] >>= \case - 0 | rtsSupportsBoundThreads -> runInBoundThread $ doInializePython - | otherwise -> doInializePython + 0 | rtsSupportsBoundThreads -> runInBoundThread $ doInitializePython + | otherwise -> doInitializePython _ -> pure () -- | Destroy python interpreter. @@ -340,8 +340,8 @@ withPython :: IO a -> IO a withPython = bracket_ initializePython finalizePython -doInializePython :: IO () -doInializePython = do +doInitializePython :: IO () +doInitializePython = do -- First we need to grab global python lock on haskell side join $ atomically $ do readTVar globalPyState >>= \case @@ -374,7 +374,7 @@ doInializePython = do fini $ RunningN gc_chan lock_eval tid_main tid_gc -- Nothing special is needed on single threaded RTS | otherwise -> do - doInializePythonIO >>= \case + doInitializePythonIO >>= \case True -> pure () False -> throwM PyInitializationFailed fini Running1 @@ -383,7 +383,7 @@ doInializePython = do -- This action is executed on python's main thread mainThread :: MVar Bool -> MVar EvalReq -> IO () mainThread lock_init lock_eval = do - r_init <- doInializePythonIO + r_init <- doInitializePythonIO putMVar lock_init r_init case r_init of False -> pure () @@ -402,8 +402,8 @@ mainThread lock_init lock_eval = do HereWeGoAgain -> loop -doInializePythonIO :: IO Bool -doInializePythonIO = do +doInitializePythonIO :: IO Bool +doInitializePythonIO = do -- FIXME: I'd like more direct access to argv argv0 <- getProgName argv <- getArgs From 362e982102df16f0d9b8357d1d4f18928171f440 Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Sat, 22 Aug 2026 21:36:01 +0300 Subject: [PATCH 09/15] Hopefully proper fix for cancelPy This prevents it from interrupting next forkOS which happens to be scheduled on same OS thread --- src/Python/Internal/Eval.hs | 40 ++++++++++++++++++++----------------- 1 file changed, 22 insertions(+), 18 deletions(-) diff --git a/src/Python/Internal/Eval.hs b/src/Python/Internal/Eval.hs index 88d2a2c..15f4112 100644 --- a/src/Python/Internal/Eval.hs +++ b/src/Python/Internal/Eval.hs @@ -552,6 +552,7 @@ instance Exception PyAsyncCancelled data PyAsync a = PyAsync { asyncTID :: !ThreadId , asyncPyTID :: !(IO Word64) + , asyncAlive :: !(MVar Bool) , asyncWait :: STM (Either SomeException a) } @@ -575,8 +576,8 @@ waitPyCatch = (.asyncWait) -- that it could be smitten with exception at an absolutely any -- moment. cancelPy :: PyAsync a -> IO () -cancelPy PyAsync{asyncTID=tid, asyncPyTID} - = runInBoundThread go +cancelPy PyAsync{asyncTID=tid, asyncPyTID, asyncAlive} + = go where go = do -- As long as thread is running python code it couldn't be @@ -587,20 +588,21 @@ cancelPy PyAsync{asyncTID=tid, asyncPyTID} -- running python at the moment so we attempt to cancel python -- repeatedly until we get py_tid <- asyncPyTID - mv_done <- newEmptyMVar -- Interrupting python _ <- forkIO $ runInBoundThread $ fix $ \loop -> do - [C.block| void { - int gil = PyGILState_Ensure(); - int n = PyThreadState_SetAsyncExc($(uint64_t py_tid), inline_py_AsyncError()); - printf("My job is done (%d) tid=(%ld)\n", n, $(uint64_t py_tid)); - PyGILState_Release(gil); - }|] - tryReadMVar mv_done >>= \case - Just () -> return () - Nothing -> threadDelay 50 >> loop -- Avoid hammering interrupt too hard + join $ withMVar asyncAlive $ \case + False -> return $ return () + True -> do + [C.block| void { + int gil = PyGILState_Ensure(); + int n = PyThreadState_SetAsyncExc($(uint64_t py_tid), inline_py_AsyncError()); + printf("My job is done (%d) tid=(%ld)\n", n, $(uint64_t py_tid)); + PyGILState_Release(gil); + }|] + return $ do + threadDelay 50 -- Avoid hammering interrupt too hard + loop throwTo tid PyAsyncCancelled - putMVar mv_done () -- | Variant of 'cancel' which isn't interruptible. @@ -613,15 +615,17 @@ runPyAsync py = do atomically ensureInit result <- newEmptyTMVarIO py_tid_mv <- newEmptyMVar - tid <- forkOS $ mask_ $ do - -- Obtain python thread ID. - putMVar py_tid_mv =<< [C.exp| uint64_t { PyThread_get_thread_ident() } |] - a <- try $ unsafeRunPy $ ensureGIL py - atomically $ putTMVar result a + alive <- newMVar True + tid <- forkOS $ mask_ $ + (do putMVar py_tid_mv =<< [C.exp| uint64_t { PyThread_get_thread_ident() } |] + a <- try $ unsafeRunPy $ ensureGIL py + atomically $ putTMVar result a + ) `finally` uninterruptibleMask_ (modifyMVar_ alive (\_ -> pure False)) pure PyAsync { asyncTID = tid , asyncPyTID = readMVar py_tid_mv , asyncWait = takeTMVar result + , asyncAlive = alive } -- | Create new OS thread and execute python code on it. Will use From fdc209fdeb36c756301f10e33b133ff87d2f6e9a Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Sun, 23 Aug 2026 10:13:12 +0300 Subject: [PATCH 10/15] Add -rtsopts to test suites It's needed to be able pass interesting RTS options --- inline-python.cabal | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/inline-python.cabal b/inline-python.cabal index 73c9a82..b98cad0 100644 --- a/inline-python.cabal +++ b/inline-python.cabal @@ -120,7 +120,7 @@ library test test-suite inline-python-tests import: language type: exitcode-stdio-1.0 - Ghc-options: -threaded -with-rtsopts=-N2 + Ghc-options: -threaded -rtsopts -with-rtsopts=-N2 hs-source-dirs: test/exe main-is: main.hs build-depends: base @@ -131,6 +131,7 @@ test-suite inline-python-tests test-suite inline-python-tests1 import: language type: exitcode-stdio-1.0 + Ghc-options: -rtsopts hs-source-dirs: test/exe main-is: main.hs build-depends: base From 31819a32d125d1a19a02b3d874fb08d1e1e3b7ca Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Sun, 23 Aug 2026 10:14:17 +0300 Subject: [PATCH 11/15] We don't need runInBoundThread here We're calling single C function. We don't care from where --- src/Python/Internal/Eval.hs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Python/Internal/Eval.hs b/src/Python/Internal/Eval.hs index 15f4112..1ce505f 100644 --- a/src/Python/Internal/Eval.hs +++ b/src/Python/Internal/Eval.hs @@ -589,7 +589,7 @@ cancelPy PyAsync{asyncTID=tid, asyncPyTID, asyncAlive} -- repeatedly until we get py_tid <- asyncPyTID -- Interrupting python - _ <- forkIO $ runInBoundThread $ fix $ \loop -> do + _ <- forkIO $ fix $ \loop -> do join $ withMVar asyncAlive $ \case False -> return $ return () True -> do From 997a0d40e4d213cb5a8c3f765e56d591e2781d8b Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Sun, 23 Aug 2026 10:50:55 +0300 Subject: [PATCH 12/15] Document and clean up async code --- src/Python/Internal/Eval.hs | 130 ++++++++++++++++++++++-------------- 1 file changed, 79 insertions(+), 51 deletions(-) diff --git a/src/Python/Internal/Eval.hs b/src/Python/Internal/Eval.hs index 1ce505f..e5d90e3 100644 --- a/src/Python/Internal/Eval.hs +++ b/src/Python/Internal/Eval.hs @@ -108,11 +108,6 @@ C.include "" -- implement N-M threading and schedules N green thread on M OS -- threads as it see fit. -- --- One could think that running python code in bound threads and --- making sure that GIL is held would suffice. It doesn't. Doing so --- quickly results in deadlock. Exact reason for that is not --- understood. --- -- Another problem is GHC may schedule two threads each running python -- code on same capability. It seems very likely that they'll step on -- each others' toes. @@ -120,6 +115,9 @@ C.include "" -- Current solution is to protect execution of python code with global -- lock. Since it's visible to haskell RTS we don't get deadlocks. -- This also means we can't execute python code concurrently. +-- +-- There's support for running python code concurrently but it's very +-- experimental. See NOTE [Py Async] for details @@ -540,6 +538,38 @@ unsafeRunPy (Py io) = io -- Async running ---------------------------------------------------------------- +-- NOTE: [Py Async] +-- ~~~~~~~~~~~~~~~~ +-- +-- Interaction with concurrent python in multithreaded environments +-- stays on rather shaky foundations. I'm not sure that RTS won't +-- schedule regular threads on forkOS'd thread and they won't cause +-- problems there. +-- +-- General idea of python asyncs is: we start new thread using forkOS +-- and run python code there and hope that it won't interfere with +-- anything. +-- +-- Separate problem is interrupting such threads. There're several +-- constraints which severly limit possible implementations: +-- +-- 1. Haskell exception cannot be delivered while thread is running +-- python. We're in the middle of foreign call. We need to +-- interrupt python as well. +-- +-- 2. PyThreadState_SetAsyncExc doesn't queue exception. If python +-- thread isn't running (e.g. released GIL by calling liftIO) it's +-- a noop. +-- +-- 3. PyThreadState_SetAsyncExc uses OS thread id as key for thread +-- interruption. And haskell runtime can schedule another thread +-- on same OS thread. So we must not to attempt to interrupt +-- thread after it finished. +-- +-- So we try to throw both haskell and python exceptions concurrently +-- and add MVar lock to check liveliness of worker thread, + + -- | Exception thrown to a thread doing async python computation. data PyAsyncCancelled = PyAsyncCancelled deriving (Show, Eq) @@ -550,9 +580,9 @@ instance Exception PyAsyncCancelled -- 'runPyAsync'. It's performed on separate OS thread. Use -- 'wait'\/'waitCatch' to obtain computation result. data PyAsync a = PyAsync - { asyncTID :: !ThreadId - , asyncPyTID :: !(IO Word64) - , asyncAlive :: !(MVar Bool) + { asyncTID :: !ThreadId -- Thread ID + , asyncPyTID :: !(IO Word64) -- Thread ID used by python + , asyncAlive :: !(MVar Bool) -- Holds True while thread is alive , asyncWait :: STM (Either SomeException a) } @@ -566,49 +596,6 @@ waitPy a = either throwSTM pure =<< a.asyncWait waitPyCatch :: PyAsync a -> STM (Either SomeException a) waitPyCatch = (.asyncWait) --- | Cancel execution of asynchronous computation. Most likely thread --- will be executing some python so first it attempts to raise async --- exception in python code. Then it throws 'PyAsyncCancelled' in case --- it executes haskell code. This means thread could be terminate --- either with 'PyError' or 'PyAsyncCancelled'. --- --- Note that python code generally is not written under assumption --- that it could be smitten with exception at an absolutely any --- moment. -cancelPy :: PyAsync a -> IO () -cancelPy PyAsync{asyncTID=tid, asyncPyTID, asyncAlive} - = go - where - go = do - -- As long as thread is running python code it couldn't be - -- interrupted by haskell exception so we spawn separate thread - -- which attempts to repeatedly interrupt python evaluation. - -- - -- PyThreadState_SetAsyncExc won't do anything if thread isn't - -- running python at the moment so we attempt to cancel python - -- repeatedly until we get - py_tid <- asyncPyTID - -- Interrupting python - _ <- forkIO $ fix $ \loop -> do - join $ withMVar asyncAlive $ \case - False -> return $ return () - True -> do - [C.block| void { - int gil = PyGILState_Ensure(); - int n = PyThreadState_SetAsyncExc($(uint64_t py_tid), inline_py_AsyncError()); - printf("My job is done (%d) tid=(%ld)\n", n, $(uint64_t py_tid)); - PyGILState_Release(gil); - }|] - return $ do - threadDelay 50 -- Avoid hammering interrupt too hard - loop - throwTo tid PyAsyncCancelled - - --- | Variant of 'cancel' which isn't interruptible. -uninterruptibleCancelPy :: PyAsync a -> IO () -uninterruptibleCancelPy = uninterruptibleMask_ . cancelPy - -- | Create new OS thread and execute python code on it. runPyAsync :: Py a -> IO (PyAsync a) runPyAsync py = do @@ -616,6 +603,9 @@ runPyAsync py = do result <- newEmptyTMVarIO py_tid_mv <- newEmptyMVar alive <- newMVar True + -- Worker thread. We must modify liveliness MVar under + -- uninterruptibleMask otherwise it could be interrupted and + -- cancelPy will consider thread alive forever tid <- forkOS $ mask_ $ (do putMVar py_tid_mv =<< [C.exp| uint64_t { PyThread_get_thread_ident() } |] a <- try $ unsafeRunPy $ ensureGIL py @@ -628,6 +618,44 @@ runPyAsync py = do , asyncAlive = alive } + +-- | Cancel execution of asynchronous computation. Most likely thread +-- will be executing some python so first it attempts to raise async +-- exception in python code. Then it throws 'PyAsyncCancelled' in case +-- it executes haskell code. This means thread could be terminate +-- either with 'PyError' or 'PyAsyncCancelled'. +-- +-- Note that python code generally is not written under assumption +-- that it could be smitten with exception at an absolutely any +-- moment. +cancelPy :: PyAsync a -> IO () +cancelPy PyAsync{asyncTID=tid, asyncPyTID, asyncAlive} = do + -- See NOTE: [Py Async] + py_tid <- asyncPyTID + -- Interrupting python + _ <- forkIO $ fix $ \loop -> do + -- Attempt to interrupt python. Only if thread is still alive + n <- withMVar asyncAlive $ \case + False -> return 1 + True -> [C.block| int { + int gil = PyGILState_Ensure(); + int n = PyThreadState_SetAsyncExc($(uint64_t py_tid), inline_py_AsyncError()); + PyGILState_Release(gil); + return n; + }|] + case n of + 0 -> do + threadDelay 50 -- Avoid hammering interrupt too hard + loop + _ -> return () + -- Interrupt haskell + throwTo tid PyAsyncCancelled + + +-- | Variant of 'cancel' which isn't interruptible. +uninterruptibleCancelPy :: PyAsync a -> IO () +uninterruptibleCancelPy = uninterruptibleMask_ . cancelPy + -- | Create new OS thread and execute python code on it. Will use -- 'uninterruptibleCancel' after callback finishes execution. withPyAsync :: Py a -> (PyAsync a -> IO b) -> IO b From 0e952411bef2e5b3c7012f741dcaacbb1a4c995c Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Sun, 23 Aug 2026 10:56:16 +0300 Subject: [PATCH 13/15] Warnings --- src/Python/Internal/EvalQQ.hs | 1 - src/Python/Internal/Program.hs | 1 - 2 files changed, 2 deletions(-) diff --git a/src/Python/Internal/EvalQQ.hs b/src/Python/Internal/EvalQQ.hs index 3a1948b..ce3609f 100644 --- a/src/Python/Internal/EvalQQ.hs +++ b/src/Python/Internal/EvalQQ.hs @@ -11,7 +11,6 @@ module Python.Internal.EvalQQ import Control.Monad.IO.Class import Control.Monad.Catch -import Control.Monad.Trans.Cont (ContT(..)) import Data.Bits import Data.Char import Data.List (intercalate) diff --git a/src/Python/Internal/Program.hs b/src/Python/Internal/Program.hs index 50075ab..7c713a0 100644 --- a/src/Python/Internal/Program.hs +++ b/src/Python/Internal/Program.hs @@ -38,7 +38,6 @@ import Foreign.C.Types import Foreign.Storable import Language.C.Inline qualified as C -import Language.C.Inline.Unsafe qualified as CU import Python.Internal.Types import Python.Internal.Util From a0542aca64ba3f70192fc3c6f0735c61f68315a7 Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Sun, 23 Aug 2026 11:03:15 +0300 Subject: [PATCH 14/15] Make MonadIO release GIL --- src/Python/Internal/Eval.hs | 5 +++++ src/Python/Internal/Types.hs | 4 ---- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/src/Python/Internal/Eval.hs b/src/Python/Internal/Eval.hs index e5d90e3..4eeb7cb 100644 --- a/src/Python/Internal/Eval.hs +++ b/src/Python/Internal/Eval.hs @@ -2,6 +2,7 @@ {-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE TemplateHaskell #-} +{-# OPTIONS_GHC -Wno-orphans #-} -- | -- Evaluation of python expressions. module Python.Internal.Eval @@ -735,6 +736,10 @@ dropGIL action = do `finally` [C.exp| void { PyEval_RestoreThread($(PyThreadState *st)) } |] +-- | Removes exception masking and releases GIL temporarily +instance MonadIO Py where + liftIO = dropGIL . interruptible + ---------------------------------------------------------------- -- Conversion of exceptions ---------------------------------------------------------------- diff --git a/src/Python/Internal/Types.hs b/src/Python/Internal/Types.hs index f617602..d5be642 100644 --- a/src/Python/Internal/Types.hs +++ b/src/Python/Internal/Types.hs @@ -120,10 +120,6 @@ newtype Py a = Py (IO a) pyIO :: IO a -> Py a pyIO = Py --- | Removes exception masking -instance MonadIO Py where - liftIO = Py . interruptible - instance PrimMonad Py where type PrimState Py = RealWorld primitive = Py . primitive From 8fcc8f4b2627fc67c248c61dc09bd2cb3fe93cdd Mon Sep 17 00:00:00 2001 From: Alexey Khudyakov Date: Sun, 23 Aug 2026 11:27:15 +0300 Subject: [PATCH 15/15] Reenable test for interrupting haskell --- test/TST/Run.hs | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/test/TST/Run.hs b/test/TST/Run.hs index 339b539..67982c9 100644 --- a/test/TST/Run.hs +++ b/test/TST/Run.hs @@ -190,16 +190,16 @@ tests = testGroup "Run python" True -> error "Timeout" False -> retry return () - -- , -- Cancellation of haskell code - -- testCase "cancelPy [haskell]" $ do - -- a <- runPyAsync $ do - -- liftIO $ forever $ threadDelay 1_000_000 - -- d <- registerDelay 100_000 - -- cancelPy a - -- _ <- atomically $ waitPyCatch a `orElse` do readTVar d >>= \case - -- True -> error "Timeout" - -- False -> retry - -- return () + , -- Cancellation of haskell code + testCase "cancelPy [haskell]" $ do + a <- runPyAsync $ do + liftIO $ forever $ threadDelay 1_000_000 + d <- registerDelay 100_000 + cancelPy a + _ <- atomically $ waitPyCatch a `orElse` do readTVar d >>= \case + True -> error "Timeout" + False -> retry + return () ] ]