diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,11 @@ +## [1.0.1.0] - 2023-04-03 + +- Change [#25](https://github.com/awkward-squad/ki/pull/25): Attempting to fork a thread in a closing scope now acts as + if it were a child being terminated due to the scope closing. Previously, attempting to fork a thread in a closing + scope would throw a runtime exception like `error "ki: scope closed"`. +- Change [#27](https://github.com/awkward-squad/ki/pull/27): Calling `awaitAll` on a closed scope now returns `()` + instead of blocking forever. + ## [1.0.0.2] - 2023-01-25 - Bugfix [#20](https://github.com/awkward-squad/ki/pull/20): previously, a child thread could deadlock when attempting diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -32,3 +32,68 @@ # Documentation [Hackage documentation](https://hackage.haskell.org/package/ki/docs/Ki.html) + +# Example: Happy Eyeballs + +The [Happy Eyeballs](https://en.wikipedia.org/wiki/Happy_Eyeballs) algorithm is a particularly common example used to +demonstrate the advantages of structured concurrency, because it is simple to describe, but can be surprisingly +difficult to implement. + +The problem can be abstractly described as follows: we have a small set of actions to run, each of which can take +arbitrarily long, or fail. Each action is a different way of computing the same value, so we only need to wait for +one action to return successfully. We don't want to run the actions one at a time (because that is likely to take too +long), nor all at once (because that is an improper use of resources). Rather, we will begin executing the first action, +then wait 250 milliseconds, then begin executing the second, and so on, until one returns successfully. + +There are of course a number of ways to implement this algorithm. We'll do something non-optimal, but simple. Let's get +the imports out of the way first. + +```haskell +import Control.Concurrent +import Control.Monad (when) +import Control.Monad.STM (atomically) +import Data.Function ((&)) +import Data.Functor (void) +import Data.List qualified as List +import Data.Maybe (isJust) +import Ki qualified +``` + +Next, let's define a `staggeredSpawner` helper that implements the majority of the core algorithm: given a list of +actions, spawn them all at 250 millisecond intervals. After all actions are spawned, we block until all of them have +returned. + +```haskell +staggeredSpawner :: [IO ()] -> IO () +staggeredSpawner actions = do + Ki.scoped \scope -> do + actions + & map (\action -> void (Ki.fork scope action)) + & List.intersperse (threadDelay 250_000) + & sequence_ + atomically (Ki.awaitAll scope) +``` + +And finally, we wrap this helper with `happyEyeballs`, which accepts a list of actions, and returns when one action +returns successfully, or returns `Nothing` if all actions fail. Note that in a real implementation, we may want to +consider what to do if an action throws an exception. Here, we trust each action to signal failure by returning +`Nothing`. + +```haskell +happyEyeballs :: [IO (Maybe a)] -> IO (Maybe a) +happyEyeballs actions = do + resultVar <- newEmptyMVar + + let worker action = do + result <- action + when (isJust result) do + _ <- tryPutMVar resultVar result + pure () + + Ki.scoped \scope -> do + _ <- + Ki.fork scope do + staggeredSpawner (map worker actions) + tryPutMVar resultVar Nothing + takeMVar resultVar +``` diff --git a/ki.cabal b/ki.cabal --- a/ki.cabal +++ b/ki.cabal @@ -3,7 +3,7 @@ author: Mitchell Rosen bug-reports: https://github.com/awkward-squad/ki/issues category: Concurrency -copyright: Copyright (C) 2020-2022 Mitchell Rosen, Travis Staton +copyright: Copyright (C) 2020-2023 Mitchell Rosen, Travis Staton homepage: https://github.com/awkward-squad/ki license: BSD-3-Clause license-file: LICENSE @@ -12,7 +12,7 @@ stability: experimental synopsis: A lightweight structured concurrency library tested-with: GHC == 9.0.2, GHC == 9.2.5, GHC == 9.4.4 -version: 1.0.0.2 +version: 1.0.1.0 description: A lightweight structured concurrency library. @@ -30,10 +30,11 @@ source-repository head type: git location: https://github.com/awkward-squad/ki.git + subdir: ki common component build-depends: - base ^>= 4.12 || ^>= 4.13 || ^>= 4.14 || ^>= 4.15 || ^>= 4.16 || ^>= 4.17, + base ^>= 4.12 || ^>= 4.13 || ^>= 4.14 || ^>= 4.15 || ^>= 4.16 || ^>= 4.17 || ^>= 4.18, default-extensions: AllowAmbiguousTypes BangPatterns @@ -50,7 +51,6 @@ InstanceSigs LambdaCase NamedFieldPuns - NoImplicitPrelude NumericUnderscores PartialTypeSignatures PatternSynonyms @@ -86,7 +86,7 @@ other-modules: Ki.Internal.ByteCount Ki.Internal.Counter - Ki.Internal.Prelude + Ki.Internal.IO Ki.Internal.Scope Ki.Internal.Thread diff --git a/src/Ki/Internal/ByteCount.hs b/src/Ki/Internal/ByteCount.hs --- a/src/Ki/Internal/ByteCount.hs +++ b/src/Ki/Internal/ByteCount.hs @@ -6,7 +6,9 @@ ) where -import Ki.Internal.Prelude +import Data.Coerce (coerce) +import Data.Int (Int64) +import Numeric.Natural (Natural) -- | A number of bytes. newtype ByteCount = ByteCount Int64 diff --git a/src/Ki/Internal/Counter.hs b/src/Ki/Internal/Counter.hs --- a/src/Ki/Internal/Counter.hs +++ b/src/Ki/Internal/Counter.hs @@ -29,7 +29,6 @@ import Data.Bits import GHC.Base -import Ki.Internal.Prelude -- | A thread-safe counter implemented with atomic fetch-and-add. data Counter diff --git a/src/Ki/Internal/IO.hs b/src/Ki/Internal/IO.hs new file mode 100644 --- /dev/null +++ b/src/Ki/Internal/IO.hs @@ -0,0 +1,98 @@ +{-# LANGUAGE MagicHash #-} +{-# LANGUAGE UnboxedTuples #-} + +-- | Miscellaneous IO utilities +module Ki.Internal.IO + ( -- * Unexceptional IO + UnexceptionalIO (..), + IOResult (..), + unexceptionalTry, + unexceptionalTryEither, + + -- * Exception utils + isAsyncException, + interruptiblyMasked, + uninterruptiblyMasked, + tryEitherSTM, + + -- * Fork utils + forkIO, + forkOn, + ) +where + +import Control.Exception +import Control.Monad (join) +import Data.Coerce (coerce) +import GHC.Base (maskAsyncExceptions#, maskUninterruptible#) +import GHC.Conc (STM, ThreadId (ThreadId), catchSTM) +import GHC.Exts (Int (I#), fork#, forkOn#) +import GHC.IO (IO (IO)) +import Prelude + +-- A little promise that this IO action cannot throw an exception. +-- +-- Yeah it's verbose, and maybe not that necessary, but the code that bothers to use it really does require +-- un-exceptiony IO actions for correctness, so here we are. +newtype UnexceptionalIO a = UnexceptionalIO + {runUnexceptionalIO :: IO a} + deriving newtype (Applicative, Functor, Monad) + +data IOResult a + = Failure !SomeException -- sync or async exception + | Success a + +unexceptionalTry :: forall a. IO a -> UnexceptionalIO (IOResult a) +unexceptionalTry action = + UnexceptionalIO do + (Success <$> action) `catch` \exception -> + pure (Failure exception) + +-- Like try, but with continuations. Also, catches all exceptions, because that's the only flavor we need. +unexceptionalTryEither :: + forall a b. + (SomeException -> UnexceptionalIO b) -> + (a -> UnexceptionalIO b) -> + IO a -> + UnexceptionalIO b +unexceptionalTryEither onFailure onSuccess action = + UnexceptionalIO do + join do + catch + (coerce @_ @(a -> IO b) onSuccess <$> action) + (pure . coerce @_ @(SomeException -> IO b) onFailure) + +isAsyncException :: SomeException -> Bool +isAsyncException exception = + case fromException @SomeAsyncException exception of + Nothing -> False + Just _ -> True + +-- | Call an action with asynchronous exceptions interruptibly masked. +interruptiblyMasked :: IO a -> IO a +interruptiblyMasked (IO io) = + IO (maskAsyncExceptions# io) + +-- | Call an action with asynchronous exceptions uninterruptibly masked. +uninterruptiblyMasked :: IO a -> IO a +uninterruptiblyMasked (IO io) = + IO (maskUninterruptible# io) + +-- Like try, but with continuations +tryEitherSTM :: Exception e => (e -> STM b) -> (a -> STM b) -> STM a -> STM b +tryEitherSTM onFailure onSuccess action = + join (catchSTM (onSuccess <$> action) (pure . onFailure)) + +-- Control.Concurrent.forkIO without the exception handler +forkIO :: IO () -> IO ThreadId +forkIO (IO action) = + IO \s0 -> + case fork# action s0 of + (# s1, tid #) -> (# s1, ThreadId tid #) + +-- Control.Concurrent.forkOn without the exception handler +forkOn :: Int -> IO () -> IO ThreadId +forkOn (I# cap) (IO action) = + IO \s0 -> + case forkOn# cap action s0 of + (# s1, tid #) -> (# s1, ThreadId tid #) diff --git a/src/Ki/Internal/Prelude.hs b/src/Ki/Internal/Prelude.hs deleted file mode 100644 --- a/src/Ki/Internal/Prelude.hs +++ /dev/null @@ -1,62 +0,0 @@ -{-# LANGUAGE MagicHash #-} -{-# LANGUAGE UnboxedTuples #-} - -module Ki.Internal.Prelude - ( forkIO, - forkOn, - interruptiblyMasked, - uninterruptiblyMasked, - module X, - ) -where - -import Control.Applicative as X (optional, (<|>)) -import Control.Concurrent hiding (forkIO, forkOn) -import Control.Concurrent as X (ThreadId, myThreadId, threadDelay, throwTo) -import Control.Concurrent.MVar as X -import Control.Exception -import Control.Exception as X (Exception, SomeException, mask_, throwIO, try, uninterruptibleMask, uninterruptibleMask_) -import Control.Monad as X (join, when) -import Data.Coerce as X (coerce) -import Data.Data as X (Data) -import Data.Foldable as X (for_, traverse_) -import Data.Function as X (fix) -import Data.Functor as X (void, ($>), (<&>)) -import Data.Int as X -import Data.IntMap.Strict as X (IntMap) -import Data.Map.Strict as X (Map) -import Data.Maybe as X (fromMaybe) -import Data.Sequence as X (Seq) -import Data.Set as X (Set) -import Data.Word as X (Word32) -import GHC.Base (maskAsyncExceptions#, maskUninterruptible#) -import GHC.Conc (ThreadId (ThreadId)) -import GHC.Exts (Int (I#), fork#, forkOn#) -import GHC.Generics as X (Generic) -import GHC.IO (IO (IO)) -import Numeric.Natural as X (Natural) -import Prelude as X - --- | Call an action with asynchronous exceptions interruptibly masked. -interruptiblyMasked :: IO a -> IO a -interruptiblyMasked (IO io) = - IO (maskAsyncExceptions# io) - --- | Call an action with asynchronous exceptions uninterruptibly masked. -uninterruptiblyMasked :: IO a -> IO a -uninterruptiblyMasked (IO io) = - IO (maskUninterruptible# io) - --- Control.Concurrent.forkIO without the dumb exception handler -forkIO :: IO () -> IO ThreadId -forkIO (IO action) = - IO \s0 -> - case fork# action s0 of - (# s1, tid #) -> (# s1, ThreadId tid #) - --- Control.Concurrent.forkOn without the dumb exception handler -forkOn :: Int -> IO () -> IO ThreadId -forkOn (I# cap) (IO action) = - IO \s0 -> - case forkOn# cap action s0 of - (# s1, tid #) -> (# s1, ThreadId tid #) diff --git a/src/Ki/Internal/Scope.hs b/src/Ki/Internal/Scope.hs --- a/src/Ki/Internal/Scope.hs +++ b/src/Ki/Internal/Scope.hs @@ -11,17 +11,25 @@ ) where -import qualified Control.Concurrent +import Control.Concurrent (ThreadId, myThreadId, throwTo) +import Control.Concurrent.MVar (MVar, newEmptyMVar, tryPutMVar, tryTakeMVar) import Control.Exception ( Exception (fromException, toException), MaskingState (..), - SomeAsyncException, + SomeException, + assert, asyncExceptionFromException, asyncExceptionToException, - catch, + throwIO, + try, + uninterruptibleMask, pattern ErrorCall, ) -import qualified Data.IntMap.Lazy as IntMap +import Control.Monad (when, guard) +import Data.Foldable (for_) +import Data.Functor (void) +import Data.IntMap (IntMap) +import qualified Data.IntMap.Lazy as IntMap.Lazy import Data.Void (Void, absurd) import GHC.Conc ( STM, @@ -36,12 +44,21 @@ throwSTM, writeTVar, ) +import GHC.Conc.Sync (readTVarIO) import GHC.IO (unsafeUnmask) import Ki.Internal.ByteCount import Ki.Internal.Counter -import Ki.Internal.Prelude +import Ki.Internal.IO + ( IOResult (..), + UnexceptionalIO (..), + interruptiblyMasked, + isAsyncException, + unexceptionalTry, + unexceptionalTryEither, + uninterruptiblyMasked, + ) import Ki.Internal.Thread -import GHC.Conc.Sync (readTVarIO) +import Data.Maybe (isJust) -- | A scope. -- @@ -71,14 +88,29 @@ nextChildIdCounter :: {-# UNPACK #-} !Counter, -- The id of the thread that created the scope, which is considered the parent of all threads created within it. parentThreadId :: {-# UNPACK #-} !ThreadId, - -- The number of child threads that are guaranteed to be about to start, in the sense that only the GHC scheduler - -- can continue to delay; there's no opportunity for an async exception to strike and prevent one of these threads - -- from starting. - -- - -- Sentinel value: -1 means the scope is closed. - startingVar :: {-# UNPACK #-} !(TVar Int) + statusVar :: {-# UNPACK #-} !(TVar ScopeStatus) } +-- The scope status: either open (allowing new threads to be created), closing (disallowing new threads to be +-- created, and in the process of killing living children), or closed (at the very end of `scoped`) +type ScopeStatus = Int + +-- The number of child threads that are guaranteed to be about to start, in the sense that only the GHC scheduler +-- can continue to delay; there's no opportunity for an async exception to strike and prevent one of these threads +-- from starting. +pattern Open :: Int +pattern Open <- ((>= 0) -> True) + +-- The scope is closing. +pattern Closing :: Int +pattern Closing = -1 + +-- The scope is closed. +pattern Closed :: Int +pattern Closed = -2 + +{-# COMPLETE Open, Closing, Closed #-} + -- Internal async exception thrown by a parent thread to its children when the scope is closing. data ScopeClosing = ScopeClosing @@ -95,9 +127,7 @@ -- throw it to some other thread)... but who would do that? isScopeClosingException :: SomeException -> Bool isScopeClosingException exception = - case fromException exception of - Just ScopeClosing -> True - _ -> False + isJust (fromException @ScopeClosing exception) pattern IsScopeClosingException :: SomeException pattern IsScopeClosingException <- (isScopeClosingException -> True) @@ -116,7 +146,7 @@ -- * The parent thread blocks until those threads terminate. scoped :: (Scope -> IO a) -> IO a scoped action = do - scope@Scope {childExceptionVar, childrenVar, startingVar} <- allocateScope + scope@Scope {childExceptionVar, childrenVar, statusVar} <- allocateScope uninterruptibleMask \restore -> do result <- try (restore (action scope)) @@ -126,30 +156,35 @@ atomically do -- Block until we haven't committed to starting any threads. Without this, we may create a thread concurrently -- with closing its scope, and not grab its thread id to throw an exception to. - blockUntil0 startingVar - -- Write the sentinel value indicating that this scope is closed, and it is an error to try to create a thread - -- within it. - writeTVar startingVar (-1) + n <- readTVar statusVar + assert (n >= 0) (guard (n == 0)) + -- Indicate that this scope is closing, so attempts to create a new thread within it will throw ScopeClosing + -- (as if the calling thread was a parent of this scope, which it should be, and we threw it a ScopeClosing + -- ourselves). + writeTVar statusVar Closing -- Return the list of currently-running children to kill. Some of them may have *just* started (e.g. if we - -- initially retried in `blockUntil0` above). That's fine - kill them all! + -- initially retried in `guard (n == 0)` above). That's fine - kill them all! readTVar childrenVar -- If one of our children propagated an exception to us, then we know it's about to terminate, so we don't bother -- throwing an exception to it. pure case result of - Left (fromException -> Just ThreadFailed {childId}) -> IntMap.delete childId livingChildren0 + Left (fromException -> Just ThreadFailed {childId}) -> IntMap.Lazy.delete childId livingChildren0 _ -> livingChildren0 -- Deliver a ScopeClosing exception to every living child. -- -- This happens to throw in the order the children were created... but I think we decided this feature isn't very -- useful in practice, so maybe we should simplify the internals and just keep a set of children? - for_ (IntMap.elems livingChildren) \livingChild -> throwTo livingChild ScopeClosing + for_ (IntMap.Lazy.elems livingChildren) \livingChild -> throwTo livingChild ScopeClosing - -- Block until all children have terminated; this relies on children respecting the async exception, which they - -- must, for correctness. Otherwise, a thread could indeed outlive the scope in which it's created, which is - -- definitely not structured concurrency! - atomically (blockUntilEmpty childrenVar) + atomically do + -- Block until all children have terminated; this relies on children respecting the async exception, which they + -- must, for correctness. Otherwise, a thread could indeed outlive the scope in which it's created, which is + -- definitely not structured concurrency! + blockUntilEmpty childrenVar + -- Record the scope as closed (from closing), so subsequent attempts to use it will throw a runtime exception + writeTVar statusVar Closed -- By now there are three sources of exception: -- @@ -174,17 +209,17 @@ allocateScope :: IO Scope allocateScope = do childExceptionVar <- newEmptyMVar - childrenVar <- newTVarIO IntMap.empty + childrenVar <- newTVarIO IntMap.Lazy.empty nextChildIdCounter <- newCounter parentThreadId <- myThreadId - startingVar <- newTVarIO 0 - pure Scope {childExceptionVar, childrenVar, nextChildIdCounter, parentThreadId, startingVar} + statusVar <- newTVarIO 0 + pure Scope {childExceptionVar, childrenVar, nextChildIdCounter, parentThreadId, statusVar} -- Spawn a thread in a scope, providing it its child id and a function that sets the masking state to the requested -- masking state. The given action is called with async exceptions interruptibly masked. -spawn :: Scope -> ThreadOptions -> (Int -> (forall x. IO x -> IO x) -> UnexceptionalIO ()) -> IO ThreadId +spawn :: Scope -> ThreadOptions -> (Tid -> (forall x. IO x -> IO x) -> UnexceptionalIO ()) -> IO ThreadId spawn - Scope {childrenVar, nextChildIdCounter, startingVar} + Scope {childrenVar, nextChildIdCounter, statusVar} ThreadOptions {affinity, allocationLimit, label, maskingState = requestedChildMaskingState} action = do -- Interruptible mask is enough so long as none of the STM operations below block. @@ -194,10 +229,12 @@ interruptiblyMasked do -- Record the thread as being about to start. Not allowed to retry. atomically do - n <- readTVar startingVar - if n < 0 - then throwSTM (ErrorCall "ki: scope closed") - else writeTVar startingVar $! n + 1 + n <- readTVar statusVar + assert (n >= -2) do + case n of + Open -> writeTVar statusVar $! n + 1 + Closing -> throwSTM ScopeClosing + Closed -> throwSTM (ErrorCall "ki: scope closed") childId <- incrCounter nextChildIdCounter @@ -227,8 +264,8 @@ -- Record the child as having started. Not allowed to retry. atomically do - n <- readTVar startingVar - writeTVar startingVar $! n - 1 -- it's actually ok to go from e.g. -1 to -2 here (very unlikely) + n <- readTVar statusVar + writeTVar statusVar $! n - 1 recordChild childrenVar childId childThreadId pure childThreadId @@ -239,10 +276,10 @@ -- * Flipping `Just _` to `Nothing` (uncommon case: we observe that a child already unrecorded itself) -- -- Never retries. -recordChild :: TVar (IntMap ThreadId) -> Int -> ThreadId -> STM () +recordChild :: TVar (IntMap ThreadId) -> Tid -> ThreadId -> STM () recordChild childrenVar childId childThreadId = do children <- readTVar childrenVar - writeTVar childrenVar $! IntMap.alter (maybe (Just childThreadId) (const Nothing)) childId children + writeTVar childrenVar $! IntMap.Lazy.alter (maybe (Just childThreadId) (const Nothing)) childId children -- Unrecord a child (ourselves) by either: -- @@ -250,35 +287,26 @@ -- * Flipping `Nothing` to `Just undefined` (uncommon case: we terminate and unrecord before parent can record us). -- -- Never retries. -unrecordChild :: TVar (IntMap ThreadId) -> Int -> STM () +unrecordChild :: TVar (IntMap ThreadId) -> Tid -> STM () unrecordChild childrenVar childId = do children <- readTVar childrenVar - writeTVar childrenVar $! IntMap.alter (maybe (Just undefined) (const Nothing)) childId children - --- forkIO/forkOn/forkOS, switching on affinity -forkWithAffinity :: ThreadAffinity -> IO () -> IO ThreadId -forkWithAffinity = \case - Unbound -> forkIO - Capability n -> forkOn n - OsThread -> Control.Concurrent.forkOS + writeTVar childrenVar $! IntMap.Lazy.alter (maybe (Just undefined) (const Nothing)) childId children -- | Wait until all threads created within a scope terminate. awaitAll :: Scope -> STM () -awaitAll Scope {childrenVar, startingVar} = do +awaitAll Scope {childrenVar, statusVar} = do blockUntilEmpty childrenVar - blockUntil0 startingVar + n <- readTVar statusVar + case n of + Open -> guard (n == 0) + Closing -> retry -- block until closed + Closed -> pure () -- Block until an IntMap becomes empty. blockUntilEmpty :: TVar (IntMap a) -> STM () blockUntilEmpty var = do x <- readTVar var - if IntMap.null x then pure () else retry - --- Block until a TVar becomes 0. -blockUntil0 :: TVar Int -> STM () -blockUntil0 var = do - x <- readTVar var - if x == 0 then pure () else retry + guard (IntMap.Lazy.null x) -- | Create a child thread to execute an action within a scope. -- @@ -296,24 +324,25 @@ -- | Variant of 'Ki.fork' that takes an additional options argument. forkWith :: Scope -> ThreadOptions -> IO a -> IO (Thread a) forkWith scope opts action = do - resultVar <- newTVarIO Nothing + resultVar <- newTVarIO NoResultYet + let done result = UnexceptionalIO (atomically (writeTVar resultVar result)) ident <- spawn scope opts \childId masking -> do result <- unexceptionalTry (masking action) case result of - Left exception -> + Failure exception -> do when (not (isScopeClosingException exception)) (propagateException scope childId exception) - Right _ -> pure () - -- even put async exceptions that we propagated. this isn't totally ideal because a caller awaiting this thread - -- would not be able to distinguish between async exceptions delivered to this thread, or itself - UnexceptionalIO (atomically (writeTVar resultVar (Just result))) + -- even put async exceptions that we propagated. this isn't totally ideal because a caller awaiting this + -- thread would not be able to distinguish between async exceptions delivered to this thread, or itself + done (BadResult exception) + Success value -> done (GoodResult value) let doAwait = readTVar resultVar >>= \case - Nothing -> retry - Just (Left exception) -> throwSTM exception - Just (Right value) -> pure value + NoResultYet -> retry + BadResult exception -> throwSTM exception + GoodResult value -> pure value pure (makeThread ident doAwait) -- | Variant of 'Ki.forkWith' for threads that never return. @@ -335,15 +364,21 @@ forkTry scope = forkTryWith scope defaultThreadOptions +data Result a + = NoResultYet + | BadResult !SomeException -- sync or async + | GoodResult a + -- | Variant of 'Ki.forkTry' that takes an additional options argument. forkTryWith :: forall e a. Exception e => Scope -> ThreadOptions -> IO a -> IO (Thread (Either e a)) forkTryWith scope opts action = do - resultVar <- newTVarIO Nothing + resultVar <- newTVarIO NoResultYet + let done result = UnexceptionalIO (atomically (writeTVar resultVar result)) childThreadId <- spawn scope opts \childId masking -> do result <- unexceptionalTry (masking action) case result of - Left exception -> do + Failure exception -> do let shouldPropagate = if isScopeClosingException exception then False @@ -352,42 +387,36 @@ -- if the user calls `forkTry @MyAsyncException`, we still want to propagate the async exception Just _ -> isAsyncException exception when shouldPropagate (propagateException scope childId exception) - Right _value -> pure () - UnexceptionalIO (atomically (writeTVar resultVar (Just result))) + done (BadResult exception) + Success value -> done (GoodResult value) let doAwait = readTVar resultVar >>= \case - Nothing -> retry - Just (Left exception) -> + NoResultYet -> retry + BadResult exception -> case fromException @e exception of Nothing -> throwSTM exception Just expectedException -> pure (Left expectedException) - Just (Right value) -> pure (Right value) + GoodResult value -> pure (Right value) pure (makeThread childThreadId doAwait) - where - isAsyncException :: SomeException -> Bool - isAsyncException exception = - case fromException @SomeAsyncException exception of - Nothing -> False - Just _ -> True -- We have a non-`ScopeClosing` exception to propagate to our parent. -- --- If our scope has already begun closing (`startingVar` is -1), then either... +-- If our scope has already begun closing (`statusVar` is Closing), then either... -- -- (A) We already received a `ScopeClosing`, but then ended up trying to propagate an exception anyway, because we -- threw a synchronous exception (or were hit by a different asynchronous exception) during our teardown procedure. -- -- or -- --- (B) We will receive a `ScopeClosing` imminently, because our parent has *just* finished setting `startingVar` to --- -1, and will proceed to throw ScopeClosing to all of its children. +-- (B) We will receive a `ScopeClosing` imminently, because our parent has *just* finished setting `statusVar` to +-- Closing, and will proceed to throw ScopeClosing to all of its children. -- -- If (A), our parent has asynchronous exceptions masked, so we must inform it of our exception via `childExceptionVar` -- rather than throwTo. If (B), either mechanism would work. And because we don't if we're in case (A) or (B), we just -- `childExceptionVar`. -- --- And if our scope has not already begun closing (`startingVar` is not -1), then we ought to throw our exception to it. --- But that might fail due to either... +-- And if our scope has not already begun closing (`statusVar` is not Closing), then we ought to throw our exception to +-- it. But that might fail due to either... -- -- (C) Our parent concurrently closing the scope and sending us a `ScopeClosing`; because it has asynchronous -- exceptions uninterruptibly masked and we only have asynchronous exception *synchronously* masked, its `throwTo` @@ -400,46 +429,19 @@ -- propagate, we just scoot these freaky exceptions under the rug. -- -- Precondition: interruptibly masked -propagateException :: Scope -> Int -> SomeException -> UnexceptionalIO () -propagateException Scope {childExceptionVar, parentThreadId, startingVar} childId exception = - UnexceptionalIO (readTVarIO startingVar) >>= \case - -1 -> tryPutChildExceptionVar -- (A) / (B) - _ -> loop +propagateException :: Scope -> Tid -> SomeException -> UnexceptionalIO () +propagateException Scope {childExceptionVar, parentThreadId, statusVar} childId exception = + UnexceptionalIO (readTVarIO statusVar) >>= \case + Closing -> tryPutChildExceptionVar -- (A) / (B) + _ -> loop -- we know status is Open here where loop :: UnexceptionalIO () loop = unexceptionalTry (throwTo parentThreadId ThreadFailed {childId, exception}) >>= \case - Left IsScopeClosingException -> tryPutChildExceptionVar -- (C) - Left _ -> loop -- (D) - Right _ -> pure () + Failure IsScopeClosingException -> tryPutChildExceptionVar -- (C) + Failure _ -> loop -- (D) + Success _ -> pure () tryPutChildExceptionVar :: UnexceptionalIO () tryPutChildExceptionVar = UnexceptionalIO (void (tryPutMVar childExceptionVar exception)) - - --- A little promise that this IO action cannot throw an exception. --- --- Yeah it's verbose, and maybe not that necessary, but the code that bothers to use it really does require --- un-exceptiony IO actions for correctness, so here we are. -newtype UnexceptionalIO a = UnexceptionalIO - {runUnexceptionalIO :: IO a} - deriving newtype (Applicative, Functor, Monad) - -unexceptionalTry :: forall a. IO a -> UnexceptionalIO (Either SomeException a) -unexceptionalTry = - coerce @(IO a -> IO (Either SomeException a)) try - --- Like try, but with continuations. Also, catches all exceptions, because that's the only flavor we need. -unexceptionalTryEither :: - forall a b. - (SomeException -> UnexceptionalIO b) -> - (a -> UnexceptionalIO b) -> - IO a -> - UnexceptionalIO b -unexceptionalTryEither onFailure onSuccess action = - UnexceptionalIO do - join do - catch - (coerce @_ @(a -> IO b) onSuccess <$> action) - (pure . coerce @_ @(SomeException -> IO b) onFailure) diff --git a/src/Ki/Internal/Thread.hs b/src/Ki/Internal/Thread.hs --- a/src/Ki/Internal/Thread.hs +++ b/src/Ki/Internal/Thread.hs @@ -2,7 +2,9 @@ ( Thread, makeThread, await, + Tid, ThreadAffinity (..), + forkWithAffinity, ThreadOptions (..), defaultThreadOptions, ThreadFailed (..), @@ -10,16 +12,18 @@ ) where +import Control.Concurrent (ThreadId, forkOS) import Control.Exception ( BlockedIndefinitelyOnSTM (..), Exception (fromException, toException), MaskingState (..), + SomeException, asyncExceptionFromException, asyncExceptionToException, ) -import GHC.Conc (STM, catchSTM) +import GHC.Conc (STM) import Ki.Internal.ByteCount -import Ki.Internal.Prelude +import Ki.Internal.IO (forkIO, forkOn, tryEitherSTM) -- | A thread. -- @@ -57,6 +61,10 @@ await_ = tryEitherSTM (\BlockedIndefinitelyOnSTM -> action) pure action } +-- A unique identifier for a thread within a scope. (Internal type alias) +type Tid = + Int + -- | What, if anything, a thread is bound to. data ThreadAffinity = -- | Unbound. @@ -67,6 +75,13 @@ OsThread deriving stock (Eq, Show) +-- forkIO/forkOn/forkOS, switching on affinity +forkWithAffinity :: ThreadAffinity -> IO () -> IO ThreadId +forkWithAffinity = \case + Unbound -> forkIO + Capability n -> forkOn n + OsThread -> Control.Concurrent.forkOS + -- | -- -- [@affinity@]: @@ -125,7 +140,7 @@ -- Internal exception type thrown by a child thread to its parent, if it fails unexpectedly. data ThreadFailed = ThreadFailed - { childId :: {-# UNPACK #-} !Int, + { childId :: {-# UNPACK #-} !Tid, exception :: !SomeException } deriving stock (Show) @@ -134,6 +149,7 @@ toException = asyncExceptionToException fromException = asyncExceptionFromException +-- Peel an outer ThreadFailed layer off of some exception, if there is one. unwrapThreadFailed :: SomeException -> SomeException unwrapThreadFailed e0 = case fromException e0 of @@ -144,8 +160,3 @@ await :: Thread a -> STM a await = await_ - --- Like try, but with continuations -tryEitherSTM :: Exception e => (e -> STM b) -> (a -> STM b) -> STM a -> STM b -tryEitherSTM onFailure onSuccess action = - join (catchSTM (onSuccess <$> action) (pure . onFailure)) diff --git a/test/Tests.hs b/test/Tests.hs --- a/test/Tests.hs +++ b/test/Tests.hs @@ -19,6 +19,17 @@ scope <- Ki.scoped pure (atomically . Ki.await =<< Ki.fork scope (pure ())) `shouldThrow` ErrorCall "ki: scope closed" pure (), + testCase "`fork` throws ScopeClosing when the scope is closing" do + Ki.scoped \scope -> do + _ <- + Ki.forkWith scope Ki.defaultThreadOptions {Ki.maskingState = MaskedInterruptible} do + -- Naughty: catch and ignore the ScopeClosing delivered to us + result1 <- try @SomeException (threadDelay maxBound) + show result1 `shouldBe` "Left ScopeClosing" + -- Try forking a new thread in the closing scope, and assert that (synchronously) throws ScopeClosing + result2 <- try @SomeException (Ki.fork_ scope undefined) + show result2 `shouldBe` "Left ScopeClosing" + pure (), testCase "`awaitAll` succeeds when no threads are alive" do Ki.scoped (atomically . Ki.awaitAll), testCase "`fork` propagates exceptions" do @@ -125,10 +136,14 @@ toException = asyncExceptionToException fromException = asyncExceptionFromException +shouldBe :: (Eq a, Show a) => a -> a -> IO () +shouldBe actual expected = do + unless (actual == expected) (fail ("expected " ++ show expected ++ ", got " ++ show actual)) + shouldReturn :: (Eq a, Show a) => IO a -> a -> IO () shouldReturn action expected = do actual <- action - unless (actual == expected) (fail ("expected " ++ show expected ++ ", got " ++ show actual)) + actual `shouldBe` expected shouldThrow :: (Show a, Eq e, Exception e) => IO a -> e -> IO () shouldThrow action expected =