packages feed

ki 1.0.1.1 → 1.0.1.2

raw patch · 13 files changed

+503/−457 lines, 13 filesdep +int-supplydep ~basedep ~containers

Dependencies added: int-supply

Dependency ranges changed: base, containers

Files

CHANGELOG.md view
@@ -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.
LICENSE view
@@ -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:
ki.cabal view
@@ -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 <mitchellwrosen@gmail.com>, Travis Staton <hello@travisstaton.com>+maintainer: Mitchell Dalvi Rosen <mitchellwrosen@gmail.com>, Travis Staton <hello@travisstaton.com> 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
src/Ki.hs view
@@ -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 --
− src/Ki/Internal/Counter.hs 
@@ -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 #-}
src/Ki/Internal/IO.hs view
@@ -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))
+ src/Ki/Internal/NonblockingSTM.hs view
@@ -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)
+ src/Ki/Internal/Propagating.hs view
@@ -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 _ = "<<internal ki exception: propagating>>"++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}
src/Ki/Internal/Scope.hs view
@@ -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 _ = "<<internal ki exception: scope closing>>"  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 ()
src/Ki/Internal/Thread.hs view
@@ -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
+ src/Ki/Internal/ThreadAffinity.hs view
@@ -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
+ src/Ki/Internal/ThreadOptions.hs view
@@ -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+    }
test/Tests.hs view
@@ -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 <<internal ki exception: scope closing>>"+            -- 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 <<internal ki exception: scope closing>>"+        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)