ki 1.0.0.2 → 1.0.1.0
raw patch · 10 files changed
+336/−198 lines, 10 filesdep ~containersdep ~tastyPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: containers, tasty
API changes (from Hackage documentation)
Files
- CHANGELOG.md +8/−0
- README.md +65/−0
- ki.cabal +5/−5
- src/Ki/Internal/ByteCount.hs +3/−1
- src/Ki/Internal/Counter.hs +0/−1
- src/Ki/Internal/IO.hs +98/−0
- src/Ki/Internal/Prelude.hs +0/−62
- src/Ki/Internal/Scope.hs +122/−120
- src/Ki/Internal/Thread.hs +19/−8
- test/Tests.hs +16/−1
CHANGELOG.md view
@@ -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
README.md view
@@ -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+```
ki.cabal view
@@ -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
src/Ki/Internal/ByteCount.hs view
@@ -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
src/Ki/Internal/Counter.hs view
@@ -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
+ src/Ki/Internal/IO.hs view
@@ -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 #)
− src/Ki/Internal/Prelude.hs
@@ -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 #)
src/Ki/Internal/Scope.hs view
@@ -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)
src/Ki/Internal/Thread.hs view
@@ -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))
test/Tests.hs view
@@ -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 =