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..b98cad0 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 @@ -98,6 +99,7 @@ library test , tasty >=1.2 , tasty-hunit >=0.10 , tasty-quickcheck >=0.10 + , stm , quickcheck-instances >=0.3.33 , exceptions , containers @@ -118,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 @@ -129,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 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 a6124ab..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 @@ -16,6 +17,15 @@ module Python.Internal.Eval , runPy , runPyInMain , unsafeRunPy + -- ** Async + , PyAsync + , PyAsyncCancelled(..) + , waitPy + , waitPyCatch + , cancelPy + , uninterruptibleCancelPy + , runPyAsync + , withPyAsync -- * GC-related , newPyObject -- * C-API wrappers @@ -54,6 +64,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 @@ -98,32 +109,27 @@ 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 --- would 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. -- +-- There's support for running python code concurrently but it's very +-- experimental. See NOTE [Py Async] for details + + + +-- 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. @@ -274,6 +280,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 () ---------------------------------------------------------------- @@ -286,8 +299,8 @@ releaseLock tid = 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. @@ -326,8 +339,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 @@ -360,7 +373,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 @@ -369,7 +382,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 () @@ -388,8 +401,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 @@ -516,13 +529,139 @@ 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 +---------------------------------------------------------------- +-- 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) + +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 -- Thread ID + , asyncPyTID :: !(IO Word64) -- Thread ID used by python + , asyncAlive :: !(MVar Bool) -- Holds True while thread is alive + , 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) + +-- | Create new OS thread and execute python code on it. +runPyAsync :: Py a -> IO (PyAsync a) +runPyAsync py = do + atomically ensureInit + 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 + atomically $ putTMVar result a + ) `finally` uninterruptibleMask_ (modifyMVar_ alive (\_ -> pure False)) + pure PyAsync + { asyncTID = tid + , asyncPyTID = readMVar py_tid_mv + , asyncWait = takeTMVar result + , 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 +withPyAsync py = bracket (runPyAsync py) uninterruptibleCancelPy + ---------------------------------------------------------------- -- GC-related functions @@ -597,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 ---------------------------------------------------------------- @@ -623,7 +766,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. 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 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 diff --git a/test/TST/Run.hs b/test/TST/Run.hs index 6f5892f..67982c9 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 @@ -22,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 |] @@ -164,8 +168,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 = []