diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,8 +1,14 @@ -## [1.0.1.1] - 2023-10-10 +## [1.0.1.2] - July 15, 2024 +- Bugfix [#33](https://github.com/awkward-squad/ki/issues/33): A scope could erroneously fail to propagate an exception + to one of its children. +- Refactor: depend on (rather than inline) `int-supply` package. + +## [1.0.1.1] - October 10, 2023 + - Compat: support GHC 9.8.1 -## [1.0.1.0] - 2023-04-03 +## [1.0.1.0] - April 3, 2023 - 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 @@ -10,16 +16,16 @@ - 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 +## [1.0.0.2] - January 25, 2023 - Bugfix [#20](https://github.com/awkward-squad/ki/pull/20): previously, a child thread could deadlock when attempting - to propagate an exception to its parent + to propagate an exception to its parent. -## [1.0.0.1] - 2022-08-14 +## [1.0.0.1] - August 14, 2022 - Compat: support GHC 9.4.1 -## [1.0.0] - 2022-06-30 +## [1.0.0] - June 30, 2022 - Breaking: Remove `Context` type, `Ki.Implicit` module, and the ability to soft-cancel a `Scope`. - Breaking: Remove `Duration` type and its associated API, including `waitFor` and `awaitFor`. @@ -40,17 +46,17 @@ - Performance: Use atomic fetch-and-add rather than a `TVar` to track internal child thread ids. -## [0.2.0] - 2020-12-17 +## [0.2.0] - December 17, 2020 - Breaking: Remove `ThreadFailed` exception wrapper. - Breaking: Rename `cancelScope` to `cancel`. -## [0.1.0.1] - 2020-11-30 +## [0.1.0.1] - November 30, 2020 - Misc: Replace `AtomicCounter` with `Int` to drop the `atomic-primops` dependency. - Bounds: Lower `cabal-version` from 3.0 to 2.2 because `stack` cannot parse 3.0. -## [0.1.0] - 2020-11-11 +## [0.1.0] - November 11, 2020 - Initial release. diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright 2020-2022 Mitchell Rosen, Travis Staton +Copyright 2020-2024 Mitchell Dalvi Rosen, Travis Staton Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: diff --git a/ki.cabal b/ki.cabal --- a/ki.cabal +++ b/ki.cabal @@ -1,18 +1,18 @@ cabal-version: 2.2 -author: Mitchell Rosen +author: Mitchell Dalvi Rosen, Travis Staton bug-reports: https://github.com/awkward-squad/ki/issues category: Concurrency -copyright: Copyright (C) 2020-2023 Mitchell Rosen, Travis Staton +copyright: Copyright (C) 2020-2024 Mitchell Dalvi Rosen, Travis Staton homepage: https://github.com/awkward-squad/ki license: BSD-3-Clause license-file: LICENSE -maintainer: Mitchell Rosen , Travis Staton +maintainer: Mitchell Dalvi Rosen , Travis Staton name: ki stability: stable synopsis: A lightweight structured concurrency library -tested-with: GHC == 9.4.7, GHC == 9.6.3, GHC == 9.8.1 -version: 1.0.1.1 +tested-with: GHC == 9.6.5, GHC == 9.8.2, GHC == 9.10.1 +version: 1.0.1.2 description: A lightweight structured concurrency library. @@ -23,7 +23,7 @@ . Remember to link your program with @-threaded@ to use the threaded runtime! -extra-source-files: +extra-doc-files: CHANGELOG.md README.md @@ -34,7 +34,7 @@ common component build-depends: - base ^>= 4.12 || ^>= 4.13 || ^>= 4.14 || ^>= 4.15 || ^>= 4.16 || ^>= 4.17 || ^>= 4.18 || ^>= 4.19, + base ^>= 4.12 || ^>= 4.13 || ^>= 4.14 || ^>= 4.15 || ^>= 4.16 || ^>= 4.17 || ^>= 4.18 || ^>= 4.19 || ^>= 4.20, default-extensions: AllowAmbiguousTypes BangPatterns @@ -83,15 +83,19 @@ import: component build-depends: containers ^>= 0.6 || ^>= 0.7, + int-supply ^>= 1.0.0, exposed-modules: Ki hs-source-dirs: src other-modules: Ki.Internal.ByteCount - Ki.Internal.Counter Ki.Internal.IO + Ki.Internal.NonblockingSTM + Ki.Internal.Propagating Ki.Internal.Scope Ki.Internal.Thread + Ki.Internal.ThreadAffinity + Ki.Internal.ThreadOptions test-suite tests import: component diff --git a/src/Ki.hs b/src/Ki.hs --- a/src/Ki.hs +++ b/src/Ki.hs @@ -48,13 +48,9 @@ fork_, scoped, ) -import Ki.Internal.Thread - ( Thread, - ThreadAffinity (..), - ThreadOptions (..), - await, - defaultThreadOptions, - ) +import Ki.Internal.Thread (Thread, await) +import Ki.Internal.ThreadAffinity (ThreadAffinity (..)) +import Ki.Internal.ThreadOptions (ThreadOptions (..), defaultThreadOptions) -- $introduction -- diff --git a/src/Ki/Internal/Counter.hs b/src/Ki/Internal/Counter.hs deleted file mode 100644 --- a/src/Ki/Internal/Counter.hs +++ /dev/null @@ -1,56 +0,0 @@ --- Some code modified from the atomic-primops library; license included below. --- Copyright (c)2012-2013, Ryan R. Newton --- --- All rights reserved. --- --- Redistribution and use in source and binary forms, with or without --- modification, are permitted provided that the following conditions are met: --- --- * Redistributions of source code must retain the above copyright --- notice, this list of conditions and the following disclaimer. --- --- * Redistributions in binary form must reproduce the above --- copyright notice, this list of conditions and the following --- disclaimer in the documentation and/or other materials provided --- with the distribution. --- --- * Neither the name of Ryan R. Newton nor the names of other --- contributors may be used to endorse or promote products derived --- from this software without specific prior written permission. -{-# LANGUAGE MagicHash #-} -{-# LANGUAGE UnboxedTuples #-} - -module Ki.Internal.Counter - ( Counter, - newCounter, - incrCounter, - ) -where - -import Data.Bits -import GHC.Base - --- | A thread-safe counter implemented with atomic fetch-and-add. -data Counter - = Counter (MutableByteArray# RealWorld) - --- | Create a new counter initialized to 0. -newCounter :: IO Counter -newCounter = - IO \s0# -> - case newByteArray# size s0# of - (# s1#, arr# #) -> - case writeIntArray# arr# 0# 0# s1# of - s2# -> (# s2#, Counter arr# #) - where - !(I# size) = - finiteBitSize (undefined :: Int) `div` 8 -{-# INLINE newCounter #-} - --- | Increment a counter and return the value prior to incrementing. -incrCounter :: Counter -> IO Int -incrCounter (Counter arr#) = - IO \s0# -> - case fetchAddIntArray# arr# 0# 1# s0# of - (# s1#, n# #) -> (# s1#, I# n# #) -{-# INLINE incrCounter #-} diff --git a/src/Ki/Internal/IO.hs b/src/Ki/Internal/IO.hs --- a/src/Ki/Internal/IO.hs +++ b/src/Ki/Internal/IO.hs @@ -10,7 +10,9 @@ unexceptionalTryEither, -- * Exception utils - isAsyncException, + assertIO, + assertM, + exceptionIs, interruptiblyMasked, uninterruptiblyMasked, tryEitherSTM, @@ -24,13 +26,16 @@ import Control.Exception import Control.Monad (join) import Data.Coerce (coerce) +import Data.Maybe (isJust) 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 System.IO.Unsafe (unsafePerformIO) import Prelude --- A little promise that this IO action cannot throw an exception. +-- A little promise that this IO action cannot throw an exception (*including* async exceptions, which you normally +-- think of as being able to strike at any time). -- -- 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. @@ -42,13 +47,17 @@ = Failure !SomeException -- sync or async exception | Success a +-- Try an action, catching any exception it throws. +-- +-- The caller is responsible for ensuring that async exceptions are masked (at whatever masking level is appropriate), +-- as (again) `UnexceptionalIO` implies async exceptions won't be thrown either. 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. +-- Like try, but with continuations. unexceptionalTryEither :: forall a b. (SomeException -> UnexceptionalIO b) -> @@ -62,24 +71,35 @@ (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 +-- | Make an assertion in a IO that requires IO. +assertIO :: IO Bool -> IO () +assertIO b = + assert (unsafePerformIO b) (pure ()) +{-# INLINE assertIO #-} +-- | Make an assertion in a monad. +assertM :: (Applicative m) => Bool -> m () +assertM b = + assert b (pure ()) +{-# INLINE assertM #-} + +-- | @exceptionIs \@e exception@ returns whether @exception@ is an instance of @e@. +exceptionIs :: forall e. (Exception e) => SomeException -> Bool +exceptionIs = + isJust . fromException @e + -- | Call an action with asynchronous exceptions interruptibly masked. -interruptiblyMasked :: IO a -> IO a -interruptiblyMasked (IO io) = - IO (maskAsyncExceptions# io) +interruptiblyMasked :: forall a. IO a -> IO a +interruptiblyMasked = + coerce (maskAsyncExceptions# @a) -- | Call an action with asynchronous exceptions uninterruptibly masked. -uninterruptiblyMasked :: IO a -> IO a -uninterruptiblyMasked (IO io) = - IO (maskUninterruptible# io) +uninterruptiblyMasked :: forall a. IO a -> IO a +uninterruptiblyMasked = + coerce (maskUninterruptible# @a) -- Like try, but with continuations -tryEitherSTM :: Exception e => (e -> STM b) -> (a -> STM b) -> STM a -> STM b +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/src/Ki/Internal/NonblockingSTM.hs b/src/Ki/Internal/NonblockingSTM.hs new file mode 100644 --- /dev/null +++ b/src/Ki/Internal/NonblockingSTM.hs @@ -0,0 +1,36 @@ +-- | STM minus retry. These STM actions are guaranteed not to block, and thus guaranteed not to be interrupted by an +-- async exception. +module Ki.Internal.NonblockingSTM + ( NonblockingSTM, + nonblockingAtomically, + nonblockingThrowSTM, + + -- * TVar + nonblockingReadTVar, + nonblockingWriteTVar', + ) +where + +import Control.Exception (Exception) +import Data.Coerce (coerce) +import GHC.Conc (STM, TVar, atomically, readTVar, throwSTM, writeTVar) + +newtype NonblockingSTM a + = NonblockingSTM (STM a) + deriving newtype (Applicative, Functor, Monad) + +nonblockingAtomically :: forall a. NonblockingSTM a -> IO a +nonblockingAtomically = + coerce @(STM a -> IO a) atomically + +nonblockingThrowSTM :: forall e x. (Exception e) => e -> NonblockingSTM x +nonblockingThrowSTM = + coerce @(e -> STM x) throwSTM + +nonblockingReadTVar :: forall a. TVar a -> NonblockingSTM a +nonblockingReadTVar = + coerce @(TVar a -> STM a) readTVar + +nonblockingWriteTVar' :: forall a. TVar a -> a -> NonblockingSTM () +nonblockingWriteTVar' var !x = + NonblockingSTM (writeTVar var x) diff --git a/src/Ki/Internal/Propagating.hs b/src/Ki/Internal/Propagating.hs new file mode 100644 --- /dev/null +++ b/src/Ki/Internal/Propagating.hs @@ -0,0 +1,40 @@ +module Ki.Internal.Propagating + ( Tid, + peelOffPropagating, + propagate, + ) +where + +import Control.Concurrent (ThreadId) +import Control.Exception (Exception (..), SomeException, asyncExceptionFromException, asyncExceptionToException, throwTo) + +-- Internal exception type thrown by a child thread to its parent, if the child fails unexpectedly. +data Propagating = Propagating + { childId :: {-# UNPACK #-} !Tid, + exception :: !SomeException + } + +instance Exception Propagating where + toException = asyncExceptionToException + fromException = asyncExceptionFromException + +instance Show Propagating where + show _ = "<>" + +pattern PropagatingThe :: SomeException -> SomeException +pattern PropagatingThe exception <- (fromException -> Just Propagating {exception}) + +-- A unique identifier for a thread within a scope. (Internal type alias) +type Tid = + Int + +-- Peel an outer Propagating layer off of some exception, if there is one. +peelOffPropagating :: SomeException -> SomeException +peelOffPropagating = \case + PropagatingThe exception -> exception + exception -> exception + +-- @propagate exception child parent@ propagates @exception@ from @child@ to @parent@. +propagate :: SomeException -> Tid -> ThreadId -> IO () +propagate exception childId parentThreadId = + throwTo parentThreadId Propagating {childId, exception} 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 @@ -16,8 +16,8 @@ import Control.Exception ( Exception (fromException, toException), MaskingState (..), + SomeAsyncException, SomeException, - assert, asyncExceptionFromException, asyncExceptionToException, throwIO, @@ -25,7 +25,7 @@ uninterruptibleMask, pattern ErrorCall, ) -import Control.Monad (when, guard) +import Control.Monad (guard, when) import Data.Foldable (for_) import Data.Functor (void) import Data.IntMap (IntMap) @@ -46,19 +46,24 @@ ) import GHC.Conc.Sync (readTVarIO) import GHC.IO (unsafeUnmask) -import Ki.Internal.ByteCount -import Ki.Internal.Counter +import IntSupply (IntSupply) +import qualified IntSupply +import Ki.Internal.ByteCount (byteCountToInt64) import Ki.Internal.IO ( IOResult (..), UnexceptionalIO (..), + assertM, + exceptionIs, interruptiblyMasked, - isAsyncException, unexceptionalTry, unexceptionalTryEither, uninterruptiblyMasked, ) -import Ki.Internal.Thread -import Data.Maybe (isJust) +import Ki.Internal.NonblockingSTM +import Ki.Internal.Propagating (Tid, peelOffPropagating, propagate) +import Ki.Internal.Thread (Thread, makeThread) +import Ki.Internal.ThreadAffinity (forkWithAffinity) +import Ki.Internal.ThreadOptions (ThreadOptions (..), defaultThreadOptions) -- | A scope. -- @@ -84,15 +89,15 @@ childExceptionVar :: {-# UNPACK #-} !(MVar SomeException), -- The set of child threads that are currently running, each keyed by a monotonically increasing int. childrenVar :: {-# UNPACK #-} !(TVar (IntMap ThreadId)), - -- The counter that holds the (int) key to use for the next child thread. - nextChildIdCounter :: {-# UNPACK #-} !Counter, + -- The supply that holds the (int) key to use for the next child thread. + nextChildIdSupply :: {-# UNPACK #-} !IntSupply, -- The id of the thread that created the scope, which is considered the parent of all threads created within it. parentThreadId :: {-# UNPACK #-} !ThreadId, 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`) +-- created, and in the process of killing running 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 @@ -112,26 +117,20 @@ {-# COMPLETE Open, Closing, Closed #-} -- Internal async exception thrown by a parent thread to its children when the scope is closing. +-- +-- In various places we trust without verifying that any 'ScopeClosing' exception, which is not exported by this module, +-- was indeed thrown to a thread by its parent. It is possible to write a program that violates this (just catch the +-- async exception and throw it to some other thread)... but who would do that? data ScopeClosing = ScopeClosing instance Show ScopeClosing where - show _ = "ScopeClosing" + show _ = "<>" instance Exception ScopeClosing where toException = asyncExceptionToException fromException = asyncExceptionFromException --- Trust without verifying that any 'ScopeClosed' exception, which is not exported by this module, was indeed thrown to --- a thread by its parent. It is possible to write a program that violates this (just catch the async exception and --- throw it to some other thread)... but who would do that? -isScopeClosingException :: SomeException -> Bool -isScopeClosingException exception = - isJust (fromException @ScopeClosing exception) - -pattern IsScopeClosingException :: SomeException -pattern IsScopeClosingException <- (isScopeClosingException -> True) - -- | Open a scope, perform an IO action with it, then close the scope. -- -- ==== __👉 Details__ @@ -151,45 +150,42 @@ uninterruptibleMask \restore -> do result <- try (restore (action scope)) - !livingChildren <- do - livingChildren0 <- - 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. - 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 `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.Lazy.delete childId livingChildren0 - _ -> livingChildren0 + !runningChildren <- do + 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. + starting <- readTVar statusVar + assertM (starting >= 0) + guard (starting == 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 `guard (n == 0)` above). That's fine - kill them all! + readTVar childrenVar - -- Deliver a ScopeClosing exception to every living child. + -- Deliver a ScopeClosing exception to every running 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.Lazy.elems livingChildren) \livingChild -> throwTo livingChild ScopeClosing + -- This happens to throw in the order the children were created, but that isn't an important/useful enough feature + -- to be worth documenting, so users shouldn't rely on it. It's definitely not the case that child 1 will completely + -- terminate before child 2 is delivered an exception: each child may delay arbitrarily while cleaning up. + for_ runningChildren \child -> throwTo child ScopeClosing 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 + children <- readTVar childrenVar + guard (IntMap.Lazy.null children) + -- 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: -- - -- 1) A sync or async exception thrown during the callback, captured in `result`. If applicable, we want to unwrap - -- the `ThreadFailed` off of this, which was only used to indicate it came from one of our children. + -- 1) A sync or async exception thrown during the callback, captured in `result`. If applicable, we want to peel + -- the `Propagating` off of this, which was only used to indicate it came from one of our children. -- -- 2) A sync or async exception left for us in `childExceptionVar` by a child that tried to propagate it to us -- directly, but failed (because we killed it concurrently). @@ -199,7 +195,7 @@ -- -- We cannot throw more than one, so throw them in that priority order. case result of - Left exception -> throwIO (unwrapThreadFailed exception) + Left exception -> throwIO (peelOffPropagating exception) Right value -> tryTakeMVar childExceptionVar >>= \case Nothing -> pure value @@ -210,104 +206,105 @@ allocateScope = do childExceptionVar <- newEmptyMVar childrenVar <- newTVarIO IntMap.Lazy.empty - nextChildIdCounter <- newCounter + nextChildIdSupply <- IntSupply.new parentThreadId <- myThreadId statusVar <- newTVarIO 0 - pure Scope {childExceptionVar, childrenVar, nextChildIdCounter, parentThreadId, statusVar} + pure Scope {childExceptionVar, childrenVar, nextChildIdSupply, 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 -> (Tid -> (forall x. IO x -> IO x) -> UnexceptionalIO ()) -> IO ThreadId -spawn - 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. - -- - -- Unconditionally set masking state to MaskedInterruptible, even though we might already be at MaskedInterruptible - -- or MaskedUninterruptible, to avoid a branch on parentMaskingState. - interruptiblyMasked do - -- Record the thread as being about to start. Not allowed to retry. - atomically do - n <- readTVar statusVar - assert (n >= -2) do - case n of - Open -> writeTVar statusVar $! n + 1 - Closing -> throwSTM ScopeClosing - Closed -> throwSTM (ErrorCall "ki: scope closed") +spawn :: Scope -> ThreadOptions -> (Tid -> (forall x. IO x -> IO x) -> UnexceptionalIO ()) -> IO ChildIds +spawn scope@Scope {childrenVar, statusVar} options action = do + -- Interruptible mask is enough so long as none of the STM operations below block. + -- + -- Unconditionally set masking state to MaskedInterruptible, even though we might already be at MaskedInterruptible + -- or MaskedUninterruptible, to avoid a branch on parentMaskingState. + interruptiblyMasked do + -- Record the thread as being about to start. Not allowed to retry. + nonblockingAtomically do + status <- nonblockingReadTVar statusVar + assertM (status >= -2) + case status of + Open -> nonblockingWriteTVar' statusVar (status + 1) + Closing -> nonblockingThrowSTM ScopeClosing + Closed -> nonblockingThrowSTM (ErrorCall "ki: scope closed") - childId <- incrCounter nextChildIdCounter + childIds <- spawnChild scope options action - childThreadId <- - forkWithAffinity affinity do - when (not (null label)) do - childThreadId <- myThreadId - labelThread childThreadId label + -- Record the child as having started. Not allowed to retry. + nonblockingAtomically do + starting <- nonblockingReadTVar statusVar + assertM (starting >= 1) + nonblockingWriteTVar' statusVar (starting - 1) + recordChild childrenVar childIds - case allocationLimit of - Nothing -> pure () - Just bytes -> do - setAllocationCounter (byteCountToInt64 bytes) - enableAllocationLimit + pure childIds - let -- Action that sets the masking state from the current (MaskedInterruptible) to the requested one. - atRequestedMaskingState :: IO a -> IO a - atRequestedMaskingState = - case requestedChildMaskingState of - Unmasked -> unsafeUnmask - MaskedInterruptible -> id - MaskedUninterruptible -> uninterruptiblyMasked +data ChildIds + = ChildIds + {-# UNPACK #-} !Tid + {-# UNPACK #-} !ThreadId - runUnexceptionalIO (action childId atRequestedMaskingState) +spawnChild :: Scope -> ThreadOptions -> (Tid -> (forall x. IO x -> IO x) -> UnexceptionalIO ()) -> IO ChildIds +spawnChild scope options action = do + childId <- IntSupply.next nextChildIdSupply + childThreadId <- + forkWithAffinity affinity do + when (not (null label)) do + childThreadId <- myThreadId + labelThread childThreadId label - atomically (unrecordChild childrenVar childId) + for_ allocationLimit \bytes -> do + setAllocationCounter (byteCountToInt64 bytes) + enableAllocationLimit - -- Record the child as having started. Not allowed to retry. - atomically do - n <- readTVar statusVar - writeTVar statusVar $! n - 1 - recordChild childrenVar childId childThreadId + let -- Action that sets the masking state from the current (MaskedInterruptible) to the requested one. + atRequestedMaskingState :: IO a -> IO a + atRequestedMaskingState = + case requestedChildMaskingState of + Unmasked -> unsafeUnmask + MaskedInterruptible -> id + MaskedUninterruptible -> uninterruptiblyMasked - pure childThreadId + runUnexceptionalIO (action childId atRequestedMaskingState) + nonblockingAtomically (unrecordChild childrenVar childId) + pure (ChildIds childId childThreadId) + where + Scope {childrenVar, nextChildIdSupply} = scope + ThreadOptions {affinity, allocationLimit, label, maskingState = requestedChildMaskingState} = options +{-# INLINE spawnChild #-} + -- Record our child by either: -- -- * Flipping `Nothing` to `Just childThreadId` (common case: we record child before it unrecords itself) -- * Flipping `Just _` to `Nothing` (uncommon case: we observe that a child already unrecorded itself) --- --- Never retries. -recordChild :: TVar (IntMap ThreadId) -> Tid -> ThreadId -> STM () -recordChild childrenVar childId childThreadId = do - children <- readTVar childrenVar - writeTVar childrenVar $! IntMap.Lazy.alter (maybe (Just childThreadId) (const Nothing)) childId children +recordChild :: TVar (IntMap ThreadId) -> ChildIds -> NonblockingSTM () +recordChild childrenVar (ChildIds childId childThreadId) = do + children <- nonblockingReadTVar childrenVar + nonblockingWriteTVar' childrenVar (IntMap.Lazy.alter (maybe (Just childThreadId) (const Nothing)) childId children) -- Unrecord a child (ourselves) by either: -- -- * Flipping `Just childThreadId` to `Nothing` (common case: parent recorded us first) -- * Flipping `Nothing` to `Just undefined` (uncommon case: we terminate and unrecord before parent can record us). --- --- Never retries. -unrecordChild :: TVar (IntMap ThreadId) -> Tid -> STM () +unrecordChild :: TVar (IntMap ThreadId) -> Tid -> NonblockingSTM () unrecordChild childrenVar childId = do - children <- readTVar childrenVar - writeTVar childrenVar $! IntMap.Lazy.alter (maybe (Just undefined) (const Nothing)) childId children + children <- nonblockingReadTVar childrenVar + nonblockingWriteTVar' 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, statusVar} = do - blockUntilEmpty childrenVar - n <- readTVar statusVar - case n of - Open -> guard (n == 0) + children <- readTVar childrenVar + guard (IntMap.Lazy.null children) + status <- readTVar statusVar + assertM (status >= -2) + case status of + Open -> guard (status == 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 - guard (IntMap.Lazy.null x) - -- | Create a child thread to execute an action within a scope. -- -- /Note/: The child thread does not mask asynchronous exceptions, regardless of the parent thread's masking state. To @@ -326,14 +323,12 @@ forkWith scope opts action = do resultVar <- newTVarIO NoResultYet let done result = UnexceptionalIO (atomically (writeTVar resultVar result)) - ident <- + ChildIds _ childThreadId <- spawn scope opts \childId masking -> do - result <- unexceptionalTry (masking action) - case result of + unexceptionalTry (masking action) >>= \case Failure exception -> do - when - (not (isScopeClosingException exception)) - (propagateException scope childId exception) + when (not (exceptionIs @ScopeClosing exception)) do + propagateException scope childId exception -- 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) @@ -343,7 +338,7 @@ NoResultYet -> retry BadResult exception -> throwSTM exception GoodResult value -> pure value - pure (makeThread ident doAwait) + pure (makeThread childThreadId doAwait) -- | Variant of 'Ki.forkWith' for threads that never return. forkWith_ :: Scope -> ThreadOptions -> IO Void -> IO () @@ -351,7 +346,10 @@ _childThreadId <- spawn scope opts \childId masking -> unexceptionalTryEither - (\exception -> when (not (isScopeClosingException exception)) (propagateException scope childId exception)) + ( \exception -> + when (not (exceptionIs @ScopeClosing exception)) do + propagateException scope childId exception + ) absurd (masking action) pure () @@ -360,7 +358,7 @@ -- -- * Synchronous (/i.e./ not an instance of 'SomeAsyncException'). -- * An instance of @e@. -forkTry :: forall e a. Exception e => Scope -> IO a -> IO (Thread (Either e a)) +forkTry :: forall e a. (Exception e) => Scope -> IO a -> IO (Thread (Either e a)) forkTry scope = forkTryWith scope defaultThreadOptions @@ -370,22 +368,21 @@ | 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 :: forall e a. (Exception e) => Scope -> ThreadOptions -> IO a -> IO (Thread (Either e a)) forkTryWith scope opts action = do resultVar <- newTVarIO NoResultYet let done result = UnexceptionalIO (atomically (writeTVar resultVar result)) - childThreadId <- + ChildIds _ childThreadId <- spawn scope opts \childId masking -> do result <- unexceptionalTry (masking action) case result of Failure exception -> do + -- then-branch explanation: if the user calls `forkTry @MyAsyncException` for some reason, we want to ignore + -- this request and propagate the async exception. `forkTry` can only be used to catch synchronous exceptions. let shouldPropagate = - if isScopeClosingException exception - then False - else case fromException @e exception of - Nothing -> True - -- if the user calls `forkTry @MyAsyncException`, we still want to propagate the async exception - Just _ -> isAsyncException exception + if exceptionIs @e exception + then exceptionIs @SomeAsyncException exception + else not (exceptionIs @ScopeClosing exception) when shouldPropagate (propagateException scope childId exception) done (BadResult exception) Success value -> done (GoodResult value) @@ -432,14 +429,17 @@ 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 + Closing -> tryPutChildExceptionVar -- (A) or (B), we don't care which + status -> do + assertM (status >= 0) -- we know status is Open (0+) here; can't be Closed (-2) + loop where loop :: UnexceptionalIO () loop = - unexceptionalTry (throwTo parentThreadId ThreadFailed {childId, exception}) >>= \case - Failure IsScopeClosingException -> tryPutChildExceptionVar -- (C) - Failure _ -> loop -- (D) + unexceptionalTry (propagate exception childId parentThreadId) >>= \case + Failure secondException + | exceptionIs @ScopeClosing secondException -> tryPutChildExceptionVar -- (C) + | otherwise -> loop -- (D) Success _ -> pure () tryPutChildExceptionVar :: UnexceptionalIO () 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,28 +2,13 @@ ( Thread, makeThread, await, - Tid, - ThreadAffinity (..), - forkWithAffinity, - ThreadOptions (..), - defaultThreadOptions, - ThreadFailed (..), - unwrapThreadFailed, ) where -import Control.Concurrent (ThreadId, forkOS) -import Control.Exception - ( BlockedIndefinitelyOnSTM (..), - Exception (fromException, toException), - MaskingState (..), - SomeException, - asyncExceptionFromException, - asyncExceptionToException, - ) +import Control.Concurrent (ThreadId) +import Control.Exception (BlockedIndefinitelyOnSTM (..)) import GHC.Conc (STM) -import Ki.Internal.ByteCount -import Ki.Internal.IO (forkIO, forkOn, tryEitherSTM) +import Ki.Internal.IO (tryEitherSTM) -- | A thread. -- @@ -60,101 +45,6 @@ -- cover this use case and prevent any accidental infinite loops. 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. - Unbound - | -- | Bound to a capability. - Capability Int - | -- | Bound to an OS thread. - 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@]: --- --- The affinity of a thread. A thread can be unbound, bound to a specific capability, or bound to a specific OS --- thread. --- --- Default: 'Unbound' --- --- [@allocationLimit@]: --- --- The maximum number of bytes a thread may allocate before it is delivered an --- 'Control.Exception.AllocationLimitExceeded' exception. If caught, the thread is allowed to allocate an additional --- 100kb (tunable with @+RTS -xq@) to perform any necessary cleanup actions; if exceeded, the thread is delivered --- another. --- --- Default: @Nothing@ (no limit) --- --- [@label@]: --- --- The label of a thread, visible in the [event log](https://downloads.haskell.org/ghc/latest/docs/html/users_guide/runtime_control.html#rts-eventlog) (@+RTS -l@). --- --- Default: @""@ (no label) --- --- [@maskingState@]: --- --- The masking state a thread is created in. To unmask, use 'GHC.IO.unsafeUnmask'. --- --- Default: @Unmasked@ -data ThreadOptions = ThreadOptions - { affinity :: ThreadAffinity, - allocationLimit :: Maybe ByteCount, - label :: String, - maskingState :: MaskingState - } - deriving stock (Eq, Show) - --- | Default thread options. --- --- @ --- 'Ki.ThreadOptions' --- { 'Ki.affinity' = 'Ki.Unbound' --- , 'Ki.allocationLimit' = Nothing --- , 'Ki.label' = "" --- , 'Ki.maskingState' = 'Unmasked' --- } --- @ -defaultThreadOptions :: ThreadOptions -defaultThreadOptions = - ThreadOptions - { affinity = Unbound, - allocationLimit = Nothing, - label = "", - maskingState = Unmasked - } - --- Internal exception type thrown by a child thread to its parent, if it fails unexpectedly. -data ThreadFailed = ThreadFailed - { childId :: {-# UNPACK #-} !Tid, - exception :: !SomeException - } - deriving stock (Show) - -instance Exception ThreadFailed where - 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 - Just (ThreadFailed _ e1) -> e1 - Nothing -> e0 -- | Wait for a thread to terminate. await :: Thread a -> STM a diff --git a/src/Ki/Internal/ThreadAffinity.hs b/src/Ki/Internal/ThreadAffinity.hs new file mode 100644 --- /dev/null +++ b/src/Ki/Internal/ThreadAffinity.hs @@ -0,0 +1,25 @@ +module Ki.Internal.ThreadAffinity + ( ThreadAffinity (..), + forkWithAffinity, + ) +where + +import Control.Concurrent (ThreadId, forkOS) +import Ki.Internal.IO (forkIO, forkOn) + +-- | What, if anything, a thread is bound to. +data ThreadAffinity + = -- | Unbound. + Unbound + | -- | Bound to a capability. + Capability Int + | -- | Bound to an OS thread. + 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 -> forkOS diff --git a/src/Ki/Internal/ThreadOptions.hs b/src/Ki/Internal/ThreadOptions.hs new file mode 100644 --- /dev/null +++ b/src/Ki/Internal/ThreadOptions.hs @@ -0,0 +1,65 @@ +module Ki.Internal.ThreadOptions + ( ThreadOptions (..), + defaultThreadOptions, + ) +where + +import Control.Exception (MaskingState (..)) +import Ki.Internal.ByteCount (ByteCount) +import Ki.Internal.ThreadAffinity (ThreadAffinity (..)) + +-- | +-- +-- [@affinity@]: +-- +-- The affinity of a thread. A thread can be unbound, bound to a specific capability, or bound to a specific OS +-- thread. +-- +-- Default: 'Unbound' +-- +-- [@allocationLimit@]: +-- +-- The maximum number of bytes a thread may allocate before it is delivered an +-- 'Control.Exception.AllocationLimitExceeded' exception. If caught, the thread is allowed to allocate an additional +-- 100kb (tunable with @+RTS -xq@) to perform any necessary cleanup actions; if exceeded, the thread is delivered +-- another. +-- +-- Default: @Nothing@ (no limit) +-- +-- [@label@]: +-- +-- The label of a thread, visible in the [event log](https://downloads.haskell.org/ghc/latest/docs/html/users_guide/runtime_control.html#rts-eventlog) (@+RTS -l@). +-- +-- Default: @""@ (no label) +-- +-- [@maskingState@]: +-- +-- The masking state a thread is created in. To unmask, use 'GHC.IO.unsafeUnmask'. +-- +-- Default: @Unmasked@ +data ThreadOptions = ThreadOptions + { affinity :: ThreadAffinity, + allocationLimit :: Maybe ByteCount, + label :: String, + maskingState :: MaskingState + } + deriving stock (Eq, Show) + +-- | Default thread options. +-- +-- @ +-- 'Ki.ThreadOptions' +-- { 'Ki.affinity' = 'Ki.Unbound' +-- , 'Ki.allocationLimit' = Nothing +-- , 'Ki.label' = "" +-- , 'Ki.maskingState' = 'Unmasked' +-- } +-- @ +defaultThreadOptions :: ThreadOptions +defaultThreadOptions = + ThreadOptions + { affinity = Unbound, + allocationLimit = Nothing, + label = "", + maskingState = Unmasked + } diff --git a/test/Tests.hs b/test/Tests.hs --- a/test/Tests.hs +++ b/test/Tests.hs @@ -1,125 +1,145 @@ module Main (main) where -import Control.Concurrent (newEmptyMVar, putMVar, takeMVar, threadDelay) +import Control.Concurrent (newEmptyMVar, putMVar, readMVar, takeMVar, threadDelay) import Control.Concurrent.STM (atomically) import Control.Exception import Control.Monad +import Data.IORef import GHC.IO (unsafeUnmask) import qualified Ki -import Test.Tasty -import Test.Tasty.HUnit +import Test.Tasty (TestTree, defaultMain, testGroup) +import Test.Tasty.HUnit (testCase) import Prelude main :: IO () main = - defaultMain do - testGroup - "Unit tests" - [ testCase "`fork` throws ErrorCall when the scope is closed" do - 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 - (`shouldThrow` A) do - Ki.scoped \scope -> do - Ki.fork_ scope (throwIO A) - atomically (Ki.awaitAll scope), - testCase "`fork` puts exceptions after propagating" do - (`shouldThrow` A) do - Ki.scoped \scope -> do - mask \restore -> do - thread :: Ki.Thread () <- Ki.fork scope (throwIO A) - restore (atomically (Ki.awaitAll scope)) `catch` \(e :: SomeException) -> print e - atomically (Ki.await thread), - testCase "`fork` forks in unmasked state regardless of parent's masking state" do - Ki.scoped \scope -> do - _ <- Ki.fork scope (getMaskingState `shouldReturn` Unmasked) - _ <- mask_ (Ki.fork scope (getMaskingState `shouldReturn` Unmasked)) - _ <- uninterruptibleMask_ (Ki.fork scope (getMaskingState `shouldReturn` Unmasked)) - atomically (Ki.awaitAll scope), - testCase "`forkWith` can fork in interruptibly masked state regardless of paren't masking state" do - Ki.scoped \scope -> do - _ <- - Ki.forkWith - scope - Ki.defaultThreadOptions {Ki.maskingState = MaskedInterruptible} - (getMaskingState `shouldReturn` MaskedInterruptible) - _ <- - mask_ do - Ki.forkWith - scope - Ki.defaultThreadOptions {Ki.maskingState = MaskedInterruptible} - (getMaskingState `shouldReturn` MaskedInterruptible) - _ <- - uninterruptibleMask_ do - Ki.forkWith - scope - Ki.defaultThreadOptions {Ki.maskingState = MaskedInterruptible} - (getMaskingState `shouldReturn` MaskedInterruptible) - atomically (Ki.awaitAll scope), - testCase "`forkWith` can fork in uninterruptibly masked state regardless of paren't masking state" do - Ki.scoped \scope -> do - _ <- - Ki.forkWith - scope - Ki.defaultThreadOptions {Ki.maskingState = MaskedUninterruptible} - (getMaskingState `shouldReturn` MaskedUninterruptible) - _ <- - mask_ do - Ki.forkWith - scope - Ki.defaultThreadOptions {Ki.maskingState = MaskedUninterruptible} - (getMaskingState `shouldReturn` MaskedUninterruptible) + defaultMain (testGroup "Unit tests" tests) + +tests :: [TestTree] +tests = + [ testCase "`fork` throws ErrorCall when the scope is closed" do + scope <- Ki.scoped pure + (atomically . Ki.await =<< Ki.fork scope (pure ())) `shouldThrow` ErrorCall "ki: scope closed" + pure (), + testCase "`fork` throws ScopeClosing to children 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 <>" + -- 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 <>" + pure (), + testCase "`awaitAll` succeeds when no threads are alive" do + Ki.scoped (atomically . Ki.awaitAll), + testCase "`fork` propagates exceptions" do + (`shouldThrow` A) do + Ki.scoped \scope -> do + Ki.fork_ scope (throwIO A) + atomically (Ki.awaitAll scope), + testCase "`fork` puts exceptions after propagating" do + (`shouldThrow` A) do + Ki.scoped \scope -> do + mask \restore -> do + thread :: Ki.Thread () <- Ki.fork scope (throwIO A) + restore (atomically (Ki.awaitAll scope)) `catch` \(_ :: SomeException) -> pure () + atomically (Ki.await thread), + testCase "`fork` forks in unmasked state regardless of parent's masking state" do + Ki.scoped \scope -> do + _ <- Ki.fork scope (getMaskingState `shouldReturn` Unmasked) + _ <- mask_ (Ki.fork scope (getMaskingState `shouldReturn` Unmasked)) + _ <- uninterruptibleMask_ (Ki.fork scope (getMaskingState `shouldReturn` Unmasked)) + atomically (Ki.awaitAll scope), + testCase "`forkWith` can fork in interruptibly masked state regardless of paren't masking state" do + Ki.scoped \scope -> do + _ <- + Ki.forkWith + scope + Ki.defaultThreadOptions {Ki.maskingState = MaskedInterruptible} + (getMaskingState `shouldReturn` MaskedInterruptible) + _ <- + mask_ do + Ki.forkWith + scope + Ki.defaultThreadOptions {Ki.maskingState = MaskedInterruptible} + (getMaskingState `shouldReturn` MaskedInterruptible) + _ <- + uninterruptibleMask_ do + Ki.forkWith + scope + Ki.defaultThreadOptions {Ki.maskingState = MaskedInterruptible} + (getMaskingState `shouldReturn` MaskedInterruptible) + atomically (Ki.awaitAll scope), + testCase "`forkWith` can fork in uninterruptibly masked state regardless of paren't masking state" do + Ki.scoped \scope -> do + _ <- + Ki.forkWith + scope + Ki.defaultThreadOptions {Ki.maskingState = MaskedUninterruptible} + (getMaskingState `shouldReturn` MaskedUninterruptible) + _ <- + mask_ do + Ki.forkWith + scope + Ki.defaultThreadOptions {Ki.maskingState = MaskedUninterruptible} + (getMaskingState `shouldReturn` MaskedUninterruptible) + _ <- + uninterruptibleMask_ do + Ki.forkWith + scope + Ki.defaultThreadOptions {Ki.maskingState = MaskedUninterruptible} + (getMaskingState `shouldReturn` MaskedUninterruptible) + atomically (Ki.awaitAll scope), + testCase "`forkTry` can catch sync exceptions" do + Ki.scoped \scope -> do + result :: Ki.Thread (Either A ()) <- Ki.forkTry scope (throw A) + atomically (Ki.await result) `shouldReturn` Left A, + testCase "`forkTry` can propagate sync exceptions" do + (`shouldThrow` A) do + Ki.scoped \scope -> do + thread :: Ki.Thread (Either A2 ()) <- Ki.forkTry scope (throw A) + atomically (Ki.await thread), + testCase "`forkTry` propagates async exceptions" do + (`shouldThrow` B) do + Ki.scoped \scope -> do + thread :: Ki.Thread (Either B ()) <- Ki.forkTry scope (throw B) + atomically (Ki.await thread), + testCase "`forkTry` puts exceptions after propagating" do + (`shouldThrow` A2) do + Ki.scoped \scope -> do + mask \restore -> do + thread :: Ki.Thread (Either A ()) <- Ki.forkTry scope (throwIO A2) + restore (atomically (Ki.awaitAll scope)) `catch` \(_ :: SomeException) -> pure () + atomically (Ki.await thread), + testCase "child propagates exceptions thrown during cleanup" do + (`shouldThrow` A) do + Ki.scoped \scope -> do + ready <- newEmptyMVar + Ki.forkWith_ scope Ki.defaultThreadOptions {Ki.maskingState = MaskedInterruptible} do + putMVar ready () + unsafeUnmask (forever (threadDelay maxBound)) `finally` throwIO A + takeMVar ready, + testCase "regression test https://github.com/awkward-squad/ki/issues/33" do + ref <- newIORef False + ready <- newEmptyMVar + + handle (\A -> pure ()) do + Ki.scoped \scope1 -> do + _ <- + Ki.fork scope1 do + readMVar ready + throwIO A + Ki.scoped \scope2 -> do _ <- - uninterruptibleMask_ do - Ki.forkWith - scope - Ki.defaultThreadOptions {Ki.maskingState = MaskedUninterruptible} - (getMaskingState `shouldReturn` MaskedUninterruptible) - atomically (Ki.awaitAll scope), - testCase "`forkTry` can catch sync exceptions" do - Ki.scoped \scope -> do - result :: Ki.Thread (Either A ()) <- Ki.forkTry scope (throw A) - atomically (Ki.await result) `shouldReturn` Left A, - testCase "`forkTry` can propagate sync exceptions" do - (`shouldThrow` A) do - Ki.scoped \scope -> do - thread :: Ki.Thread (Either A2 ()) <- Ki.forkTry scope (throw A) - atomically (Ki.await thread), - testCase "`forkTry` propagates async exceptions" do - (`shouldThrow` B) do - Ki.scoped \scope -> do - thread :: Ki.Thread (Either B ()) <- Ki.forkTry scope (throw B) - atomically (Ki.await thread), - testCase "`forkTry` puts exceptions after propagating" do - (`shouldThrow` A2) do - Ki.scoped \scope -> do - mask \restore -> do - thread :: Ki.Thread (Either A ()) <- Ki.forkTry scope (throwIO A2) - restore (atomically (Ki.awaitAll scope)) `catch` \(_ :: SomeException) -> pure () - atomically (Ki.await thread), - testCase "child propagates exceptions thrown during cleanup" do - (`shouldThrow` A) do - Ki.scoped \scope -> do - ready <- newEmptyMVar - Ki.forkWith_ scope Ki.defaultThreadOptions {Ki.maskingState = MaskedInterruptible} do - putMVar ready () - unsafeUnmask (forever (threadDelay maxBound)) `finally` throwIO A - takeMVar ready - ] + Ki.fork scope2 do + (putMVar ready () >> threadDelay 1_000_000) `catch` \(_ :: SomeException) -> + writeIORef ref True + atomically (Ki.awaitAll scope2) + + readIORef ref `shouldReturn` True + ] data A = A deriving stock (Eq, Show)