ki 0.2.0.1 → 1.0.0

raw patch · 27 files changed

+1027/−1880 lines, 27 filesdep +tastydep +tasty-hunitdep −concurrencydep −dejafudep ~basedep ~containersdep ~stmPVP ok

version bump matches the API change (PVP)

Dependencies added: tasty, tasty-hunit

Dependencies removed: concurrency, dejafu

Dependency ranges changed: base, containers, stm

API changes (from Hackage documentation)

Files

• CHANGELOG.md +22/−17
• LICENSE +1/−1
• README.md +26/−145
• ki.cabal +34/−91
• src/Ki.hs +92/−64
• src/Ki/CancelToken.hs +0/−18
• src/Ki/Concurrency.hs +0/−187
• src/Ki/Context.hs +0/−49
• src/Ki/Ctx.hs +0/−92
• src/Ki/Debug.hs +0/−22
• src/Ki/Duration.hs +0/−36
• src/Ki/Implicit.hs +0/−197
• src/Ki/Internal.hs +0/−18
• src/Ki/Internal/ByteCount.hs +37/−0
• src/Ki/Internal/Counter.hs +57/−0
• src/Ki/Internal/Prelude.hs +62/−0
• src/Ki/Internal/Scope.hs +416/−0
• src/Ki/Internal/Thread.hs +151/−0
• src/Ki/Prelude.hs +0/−47
• src/Ki/Scope.hs +0/−198
• src/Ki/Thread.hs +0/−160
• src/Ki/Timeout.hs +0/−15
• test/Tests.hs +129/−0
• test/dejafu-tests/DejaFuTestUtils.hs +0/−149
• test/dejafu-tests/DejaFuTests.hs +0/−65
• test/unit-tests/TestUtils.hs +0/−77
• test/unit-tests/Tests.hs +0/−232

@@ -1,30 +1,35 @@-# Changelog+## [1.0.0] - 2022-06-30 -All notable changes to this project will be documented in this file.+- 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`.+- Breaking: Remove `Ki.Internal` module.+- Breaking: Generalize `async` to `forkTry`.+- Breaking: Generalize `forkWithUnmask` to `forkWith`.+- Breaking: Make `fork_` take an `IO Void` rather than an `IO ()`.+- Breaking: Make `fork` create an unmasked thread, rather than inherit the parent's masking state.+- Breaking: Rename `waitSTM` to `awaitAll` (replacing the old `wait` in `IO`). -The format is based on [Keep a Changelog](http://keepachangelog.com/)-and this project adheres to the [Haskell Package Versioning Policy](https://pvp.haskell.org/).+- Change: Make `scoped` kill threads in the order they were created. -## [0.2.0.1] - 2020-12-20+- Bugfix: Fix small memory leak related to closing a scope.+- Bugfix: Fix subtle bug related to GHC's treatment of deadlocked threads.+- Bugfix: make `async` (now `forkTry`) propagate async exceptions.+- Bugfix: make `scoped` safe to run with asynchronous exceptions masked.+- Bugfix: propagate exceptions to creator of scope, not creator of thread -### Changed-- Marked dejafu test suite as "not buildable" by default+- Performance: Use atomic fetch-and-add rather than a `TVar` to track internal child thread ids.  ## [0.2.0] - 2020-12-17 -### Changed-- Rename `cancelScope` to `cancel`.--### Removed-- Remove `ThreadFailed` exception wrapper.+- Breaking: Remove `ThreadFailed` exception wrapper.+- Breaking: Rename `cancelScope` to `cancel`.  ## [0.1.0.1] - 2020-11-30 -### Changed-- Lower `cabal-version` from 3.0 to 2.2 because `stack` cannot parse 3.0-- Replace `AtomicCounter` with `Int` (to drop the `atomic-primops` dependency)+- 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 -### Added-- Initial release+- Initial release.

@@ -1,4 +1,4 @@-Copyright 2020 Mitchell Rosen+Copyright 2020-2022 Mitchell Rosen, Travis Staton  Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: 

@@ -1,155 +1,36 @@-# ki--[![GitHub CI](https://github.com/mitchellwrosen/ki/workflows/CI/badge.svg)](https://github.com/mitchellwrosen/ki/actions)-[![Hackage](https://img.shields.io/hackage/v/ki.svg?label=ki&logo=haskell)](https://hackage.haskell.org/package/ki-0/candidate)-[![Stackage LTS](https://stackage.org/package/ki/badge/lts)](https://www.stackage.org/lts/package/ki)-[![Stackage Nightly](https://stackage.org/package/ki/badge/nightly)](https://www.stackage.org/nightly/package/ki)-[![Dependencies](https://img.shields.io/hackage-deps/v/ki)](https://packdeps.haskellers.com/reverse/ki)-+# Overview -`ki` is a lightweight structured-concurrency library inspired by many other projects:+`ki` is a lightweight structured-concurrency library inspired by many other projects and blog posts: -* [`libdill`](http://libdill.org/)-* [`trio`](https://github.com/python-trio/trio)+* [libdill](http://libdill.org/)+* [trio](https://github.com/python-trio/trio) * [Kotlin coroutines](https://kotlinlang.org/docs/reference/coroutines-overview.html)-* [Go Concurrency Patterns: Context](https://blog.golang.org/context)-* [.NET 4 Cancellation Framework](https://devblogs.microsoft.com/pfxteam/net-4-cancellation-framework/)--## Overview--### Structured concurrency--Structured concurrency aims to make concurrent programs easier to understand by delimiting the lifetime of all-concurrently threads to a syntactic block, akin to structured programming.--This library defines five primary functions; please read the Haddocks for more comprehensive usage information.--```haskell--- Perform an IO action within a new scope-scoped :: (Scope -> IO a) -> IO a---- Create a background thread (propagates exceptions to its parent)-fork :: Scope -> IO a -> IO (Thread a)---- Create a background thread (does not propagate exceptions to its parent)-async :: Scope -> IO a -> IO (Either ThreadFailed a)---- Wait for a thread to finish-await :: Thread a -> IO a---- Wait for all threads created within a scope to finish-wait :: Scope -> IO ()-```--A `Scope` is an explicit data structure from which threads can be created, with the property that by the time the-`Scope` itself "goes out of scope", all threads created within it will have finished.--When viewing a concurrent program as a "call tree" (analogous to a call stack), this approach, in contrast to to-directly creating green threads in the style of Haskell's `forkIO` or Golang's `go`, respects the basic function-abstraction, in that each function has a single ingress and a single egress.--Please read [Notes on structured concurrency](https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/)-for a more detailed overview on structured concurrency.--### Error propagation--When a parent thread throws or is thrown an exception, it first throws exceptions to all of its children and waits for-them to finish. This makes threads hierarchical: a thread cannot outlive the thread that created it.--When a child thread throws or is thrown an exception, depending on how it was created (see `fork` and `async` above), it-_may_ propagate the exception to its parent. This is intended to cover both of the following cases:--  * It is is _unexpected_ for a thread to fail; if it does, the program should crash loudly.-  * It is _conceivable_ for a thread to fail; if it does, this is not an exceptional circumstance, so should not require-    installing an exception handler.--### Soft-cancellation--Sometimes it is desirable to inform threads that they should endeavor to complete their work and then gracefully-terminate. This is a "cooperative" or "soft" cancellation, in contrast to throwing a thread an exception so that it-terminates immediately.--In `ki`, soft-cancellation is exposed as an alternative superset of the core API, because it involves additional-plumbing of an opaque `Context` type.--```haskell-withGlobalContext :: (Context => IO a) -> IO a--scoped :: Context => (Context => Scope -> IO a) -> IO a--fork :: Scope -> (Context => IO a) -> IO (Thread a)-```--Creating a new scope _requires_ a context, whereas the callbacks provided to `scoped` and `fork` are _provided_-a context. (Above, the context is passed around as an implicit parameter, but could instead be passed around in a-reader monad or similar).--The core API is extended with two functions to soft-cancel a scope, and to observe whether one's own scope has been-canceled.--```haskell-cancel :: Scope -> IO ()--cancelled :: Context => IO (Maybe CancelToken)-```--Canceling a scope is observable by all threads created within it, all threads created within _those_ threads, and so on.--#### A small soft-cancellation example--A worker thread may be written to perform a task in a loop, and cooperatively check for cancellation before doing work.--```haskell-worker :: Ki.Context => IO ()-worker =-  forever do-    checkCancellation-    doWork-  where-    checkCancellation :: IO ()-    checkCancellation = do-      maybeCancelToken <- Ki.cancelled-      case maybeCancelToken of-        Nothing -> pure ()-        Just cancelToken -> do-          putStrLn "I'm cancelled! Time to clean up."-          doCleanup-          throwIO cancelToken-```--The parent of such worker threads may (via some signaling mechanism) determine that it should cancel them, do so, and-then defensively fall back to _hard_-cancelling in case some worker is not respecting the soft-cancel signal, for-whatever reason.--```haskell-Ki.scoped \scope -> do-  worker--  -- Some time later, we decide to soft-cancel-  Ki.cancel scope--  -- Give the workers up to 10 seconds to finish-  Ki.waitFor scope (10 * Ki.seconds)+* [Notes on structured concurrency, or: Go statement considered harmful](https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful)+* [Structured Concurrency in High-level Languages](https://250bpm.com/blog:124)+* [Update on Structured Concurrency](https://250bpm.com/blog:137)+* [Two Approaches to Structured Concurrency](https://250bpm.com/blog:139)+* [libdill: Structured Concurrency for C](https://libdill.org/structured-concurrency.html) -  -- Fall through the bottom of `scoped`, which throws hard-cancels all-  -- remaining threads by throwing each one an asynchronous exceptions-```+A previous version of `ki` also included a mechanism for soft-cancellation/graceful shutdown, which took inspiration+from: -### Testing+* [Go Concurrency Patterns: Context](https://blog.golang.org/context)+* [.NET 4 Cancellation Framework](https://devblogs.microsoft.com/pfxteam/net-4-cancellation-framework)+* [Timeouts and cancellation for humans](https://vorpus.org/blog/timeouts-and-cancellation-for-humans)+* [Graceful Shutdown](https://250bpm.com/blog:146) -(Some of) the implementation is tested for deadlocks, race conditions, and other concurrency anomalies by-[`dejafu`](http://hackage.haskell.org/package/dejafu), a fantastic unit-testing library for concurrent programs.+However, this feature was removed (perhaps temporarily) because the design of the API was unsatisfactory. -Nonetheless this library should not considered production-ready!+# Documentation -## Recommended reading+[Hackage documentation](https://hackage.haskell.org/package/ki/docs/Ki.html) -In chronological order of publication,+# Badges / meta -  * https://vorpus.org/blog/timeouts-and-cancellation-for-humans/-  * https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/-  * http://250bpm.com/blog:124-  * http://250bpm.com/blog:137-  * http://250bpm.com/blog:139-  * http://250bpm.com/blog:146-  * http://libdill.org/structured-concurrency.html+| `ki` | `ki-unlifted` |+| --- | --- |+| [![GitHub CI](https://github.com/awkward-squad/ki/workflows/CI/badge.svg)](https://github.com/awkward-squad/ki/actions) | |+| [![Hackage](https://img.shields.io/hackage/v/ki.svg?label=ki&logo=haskell)](https://hackage.haskell.org/package/ki) | [![Hackage](https://img.shields.io/hackage/v/ki-unlifted.svg?label=ki-unlifted&logo=haskell)](https://hackage.haskell.org/package/ki-unlifted) |+| [![Stackage LTS](https://stackage.org/package/ki/badge/lts)](https://www.stackage.org/lts/package/ki) | [![Stackage LTS](https://stackage.org/package/ki-unlifted/badge/lts)](https://www.stackage.org/lts/package/ki-unlifted) |+| [![Stackage Nightly](https://stackage.org/package/ki/badge/nightly)](https://www.stackage.org/nightly/package/ki) | [![Stackage Nightly](https://stackage.org/package/ki-unlifted/badge/nightly)](https://www.stackage.org/nightly/package/ki-unlifted) |+| [![Dependencies](https://img.shields.io/hackage-deps/v/ki)](https://packdeps.haskellers.com/reverse/ki) | [![Dependencies](https://img.shields.io/hackage-deps/v/ki-unlifted)](https://packdeps.haskellers.com/reverse/ki-unlifted) |

@@ -1,47 +1,26 @@ cabal-version: 2.2  author: Mitchell Rosen-bug-reports: https://github.com/mitchellwrosen/ki/issues+bug-reports: https://github.com/awkward-squad/ki/issues category: Concurrency-copyright: Copyright (C) 2020 Mitchell Rosen-homepage: https://github.com/mitchellwrosen/ki+copyright: Copyright (C) 2020-2022 Mitchell Rosen, Travis Staton+homepage: https://github.com/awkward-squad/ki license: BSD-3-Clause license-file: LICENSE-maintainer: Mitchell Rosen <mitchellwrosen@gmail.com>+maintainer: Mitchell Rosen <mitchellwrosen@gmail.com>, Travis Staton <hello@travisstaton.com> name: ki stability: experimental-synopsis: A lightweight, structured-concurrency library-version: 0.2.0.1+synopsis: A lightweight structured concurrency library+version: 1.0.0  description:-  A lightweight, structured-concurrency library.-  .-  This package comes in two variants:-  .-  * "Ki" exposes the most stripped-down variant; start here.-  .-  * "Ki.Implicit" extends "Ki" with an implicit context that's used to-    propagate soft cancellation signals.-  .-      Using this variant comes at a cost:-  .-        * You must manually add constraints to propagate the implicit context to-          where it's needed.-  .-        * To remain warning-free, you must delete the implicit context constraints-          where they are no longer needed.-  .-    If you don't need soft-cancellation, there is no benefit to using this-    variant, and you should stick with "Ki".+  A lightweight structured concurrency library.   .-  Because you'll only ever need one variant at a time, I recommend using a-  <https://cabal.readthedocs.io/en/latest/cabal-package.html#pkg-field-mixins mixin stanza>-  to rename one module to @Ki@ while hiding the others. This also simplifies the-  process of upgrading from "Ki.Implicit" to "Ki" if necessary.+  For a variant of this API generalized to+  @<https://hackage.haskell.org/package/unliftio-core/docs/Control-Monad-IO-Unlift.html#t:MonadUnliftIO MonadUnliftIO>@,+  see @<https://hackage.haskell.org/package/ki-unlifted ki-unlifted>@.   .-  @-  mixins: ki (Ki.Implicit as Ki)-  @+  Remember to link your program with @-threaded@ to use the threaded runtime!  extra-source-files:   CHANGELOG.md@@ -49,11 +28,14 @@  source-repository head   type: git-  location: https://github.com/mitchellwrosen/ki.git+  location: https://github.com/awkward-squad/ki.git  common component+  build-depends:+    base ^>= 4.12 || ^>= 4.13 || ^>= 4.14 || ^>= 4.15 || ^>= 4.16,   default-extensions:     AllowAmbiguousTypes+    BangPatterns     BlockArguments     ConstraintKinds     DeriveAnyClass@@ -64,14 +46,17 @@     DuplicateRecordFields     ExistentialQuantification     GeneralizedNewtypeDeriving-    ImplicitParams+    InstanceSigs     LambdaCase     NamedFieldPuns     NoImplicitPrelude     NumericUnderscores+    PartialTypeSignatures+    PatternSynonyms     RankNTypes     RoleAnnotations     ScopedTypeVariables+    TypeApplications     ViewPatterns   default-language: Haskell2010   ghc-options:@@ -86,74 +71,32 @@     ghc-options:       -Wno-missing-safe-haskell-mode       -Wno-prepositive-qualified-module--flag dejafu-tests-  description: Internal flag used by DejaFu test suite-  default: False-  manual: True+  if impl(ghc >= 9.2)+    ghc-options:+      -Wno-missing-kind-signatures  library   import: component   build-depends:-    base >= 4.12.0.0 && < 4.15,-    containers-  if !flag(dejafu-tests)-    build-depends:-      stm+    containers ^>= 0.6,   exposed-modules:-    Ki,-    Ki.Implicit,-    Ki.Internal+    Ki   hs-source-dirs: src   other-modules:-    Ki.CancelToken-    Ki.Concurrency-    Ki.Context-    Ki.Ctx-    Ki.Debug-    Ki.Duration-    Ki.Prelude-    Ki.Scope-    Ki.Thread-    Ki.Timeout-  if flag(dejafu-tests)-    build-depends:-      concurrency ^>= 1.11.0.0,-      dejafu ^>= 2.4.0.0,-    cpp-options: -DTEST+    Ki.Internal.ByteCount+    Ki.Internal.Counter+    Ki.Internal.Prelude+    Ki.Internal.Scope+    Ki.Internal.Thread -test-suite dejafu-tests+test-suite tests   import: component-  if flag(dejafu-tests)-    buildable: True-  else-    buildable: False   build-depends:-    base,-    concurrency,-    dejafu,     ki,-  ghc-options: -rtsopts -threaded-  hs-source-dirs: test/dejafu-tests-  main-is: DejaFuTests.hs-  other-modules:-    DejaFuTestUtils-  type: exitcode-stdio-1.0--test-suite unit-tests-  import: component-  if flag(dejafu-tests)-    buildable: False-  else-    buildable: True-  build-depends:-    base,-    ki,-    stm,+    stm ^>= 2.5,+    tasty ^>= 1.4.2,+    tasty-hunit ^>= 0.10,   ghc-options: -rtsopts -threaded-  hs-source-dirs: test/unit-tests+  hs-source-dirs: test   main-is: Tests.hs-  other-modules:-    TestUtils   type: exitcode-stdio-1.0-

@@ -1,82 +1,110 @@+-- | `ki` is a lightweight structured concurrency library.+--+-- For a variant of this API generalized to+-- @<https://hackage.haskell.org/package/unliftio-core/docs/Control-Monad-IO-Unlift.html#t:MonadUnliftIO MonadUnliftIO>@,+-- see @<https://hackage.haskell.org/package/ki-unlifted ki-unlifted>@.+--+-- Remember to link your program with @-threaded@ to use the threaded runtime! module Ki-  ( -- * Scope-    Scope,-    scoped,-    Scope.wait,-    Scope.waitSTM,-    Scope.waitFor,+  ( -- * Introduction+    -- $introduction -    -- * Spawning threads-    -- $spawning-threads+    -- * Core API+    Scope,     Thread,--    -- ** Fork-    Thread.fork,-    Thread.fork_,-    Thread.forkWithUnmask,-    Thread.forkWithUnmask_,+    scoped,+    fork,+    forkTry,+    await,+    awaitAll, -    -- ** Async-    Thread.async,-    Thread.asyncWithUnmask,+    -- * Extended API+    fork_,+    forkWith,+    forkWith_,+    forkTryWith, -    -- ** Await-    Thread.await,-    Thread.awaitSTM,-    Thread.awaitFor,+    -- ** Thread options+    ThreadOptions (..),+    defaultThreadOptions,+    ThreadAffinity (..), -    -- * Miscellaneous-    Duration,-    Duration.microseconds,-    Duration.milliseconds,-    Duration.seconds,-    Timeout.timeoutSTM,-    sleep,+    -- ** Byte count+    ByteCount,+    kilobytes,+    megabytes,   ) where -import qualified Ki.Context as Context-import Ki.Duration (Duration)-import qualified Ki.Duration as Duration-import Ki.Prelude-import Ki.Scope (Scope)-import qualified Ki.Scope as Scope-import Ki.Thread (Thread)-import qualified Ki.Thread as Thread-import qualified Ki.Timeout as Timeout+import Ki.Internal.ByteCount (ByteCount, kilobytes, megabytes)+import Ki.Internal.Scope+  ( Scope,+    awaitAll,+    fork,+    forkTry,+    forkTryWith,+    forkWith,+    forkWith_,+    fork_,+    scoped,+  )+import Ki.Internal.Thread+  ( Thread,+    ThreadAffinity (..),+    ThreadOptions (..),+    await,+    defaultThreadOptions,+  ) --- $spawning-threads+-- $introduction ----- There are two variants of __thread__-creating functions with different exception-propagation semantics.+-- Structured concurrency is a paradigm of concurrent programming in which a lexical scope delimits the lifetime of each+-- thread. Threads therefore form a "call tree" hierarchy in which no child can outlive its parent. ----- * If a __thread__ created with 'Ki.fork' throws an exception, it is immediately propagated up the call tree to the--- __thread__ that created its __scope__.+-- Exceptions are propagated promptly from child to parent and vice-versa: ----- * If a __thread__ created with 'Ki.async' throws an exception, it is not propagated up the call tree, but can be--- observed by 'Ki.await'.---- | Open a __scope__, perform an @IO@ action with it, then close the __scope__.+--     * If an exception is raised in a child thread, the child raises the same exception in its parent, then+--       terminates. ----- When the __scope__ is closed, all remaining __threads__ created within it are killed.+--     * If an exception is raised in a parent thread, the parent first raises an exception in all of its living+--       children, waits for them to terminate, then re-raises the original exception. ----- /Throws/:+-- All together, this library: -----   * The exception thrown by the callback to 'scoped' itself, if any.---   * The first exception thrown by or to a __thread__ created with 'Ki.fork', if any.+--     * Guarantees the absence of "ghost threads" (/i.e./ threads that accidentally continue to run alongside the main+--       thread after the function that spawned them returns). ----- ==== __Examples__+--     * Performs prompt, bidirectional exception propagation when an exception is raised anywhere in the call tree. ----- @--- 'scoped' \\scope -> do---   'Ki.fork_' scope worker1---   'Ki.fork_' scope worker2---   'Ki.wait' scope--- @-scoped :: (Scope -> IO a) -> IO a-scoped =-  Scope.scoped Context.dummyContext---- | Duration-based @threadDelay@.-sleep :: Duration -> IO ()-sleep duration =-  threadDelay (Duration.toMicroseconds duration)+--     * Provides a safe and flexible API that can be used directly, or with which higher-level concurrency patterns can+--       be built on top, such as worker queues, pub-sub pipelines, and supervision trees.+--+-- For a longer introduction to structured concurrency, including an educative analogy to structured programming, please+-- read Nathaniel J. Smith's blog post,+-- <https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/ "Notes on structured concurrency, or: Go statement considered harmful">.+--+-- ==== __👉 Quick start examples__+--+-- * Perform two actions concurrently, and wait for both of them to complete.+--+--     @+--     concurrently :: IO a -> IO b -> IO (a, b)+--     concurrently action1 action2 =+--       Ki.'Ki.scoped' \\scope -> do+--         thread1 <- Ki.'Ki.fork' scope action1+--         result2 <- action2+--         result1 <- atomically (Ki.'Ki.await' thread1)+--         pure (result1, result2)+--     @+--+-- * Perform two actions concurrently, and when the first action terminates, stop executing the other.+--+--     @+--     race :: IO a -> IO a -> IO a+--     race action1 action2 =+--       Ki.'Ki.scoped' \\scope -> do+--         resultVar \<- newEmptyMVar+--         _ \<- Ki.'Ki.fork' scope (action1 \>>= tryPutMVar resultVar)+--         _ \<- Ki.'Ki.fork' scope (action2 \>>= tryPutMVar resultVar)+--         takeMVar resultVar+--     @

@@ -1,18 +0,0 @@-module Ki.CancelToken-  ( CancelToken (..),-    newCancelToken,-  )-where--import Ki.Prelude---- | A cancel token represents a request for /cancellation/; this request can be fulfilled by throwing the token as an--- exception.-newtype CancelToken-  = CancelToken Int-  deriving stock (Eq, Show)-  deriving anyclass (Exception)--newCancelToken :: IO CancelToken-newCancelToken =-  coerce uniqueInt

@@ -1,187 +0,0 @@-{-# LANGUAGE CPP #-}-{-# LANGUAGE MagicHash #-}-{-# LANGUAGE TypeApplications #-}-{-# LANGUAGE UnboxedTuples #-}--module Ki.Concurrency-  ( IO,-    MVar,-    STM,-    TBQueue,-    TMVar,-    TQueue,-    TVar,-    ThreadId,-    atomically,-    catch,-    check,-    forkIO,-    modifyTVar',-    myThreadId,-    newEmptyTMVarIO,-    newMVar,-    newTBQueueIO,-    newTQueueIO,-    newTVar,-    newTVarIO,-    onException,-    putTMVar,-    putTMVarIO,-    readTBQueue,-    readTMVar,-    readTQueue,-    readTVar,-    readTVarIO,-    registerDelay,-    retry,-    threadDelay,-    throwIO,-    throwSTM,-    throwTo,-    try,-    uninterruptibleMask,-    uniqueInt,-    unsafeUnmask,-    withMVar,-    writeTBQueue,-    writeTQueue,-    writeTVar,-  )-where--#ifdef TEST--import Control.Concurrent.Classy hiding (MVar, STM, TBQueue, TMVar, TQueue, TVar, ThreadId, registerDelay)-import qualified Control.Concurrent.Classy-import Control.Exception (Exception, SomeException)-import Numeric.Natural (Natural)-import qualified Test.DejaFu-import qualified Test.DejaFu.Conc.Internal.Common-import qualified Test.DejaFu.Conc.Internal.STM-import Test.DejaFu.Types (ThreadId)-import qualified Prelude-import Prelude hiding (IO)--type IO =-  Test.DejaFu.ConcIO--type MVar =-  Test.DejaFu.Conc.Internal.Common.ModelMVar Prelude.IO--type STM =-  Test.DejaFu.Conc.Internal.STM.ModelSTM Prelude.IO--type TBQueue =-  Control.Concurrent.Classy.TBQueue STM--type TQueue =-  Control.Concurrent.Classy.TQueue STM--type TMVar =-  Control.Concurrent.Classy.TMVar STM--type TVar =-  Test.DejaFu.Conc.Internal.STM.ModelTVar Prelude.IO--forkIO :: IO () -> IO ThreadId-forkIO =-  fork--newTBQueueIO :: Natural -> IO (TBQueue a)-newTBQueueIO =-  atomically . newTBQueue--newTQueueIO :: IO (TQueue a)-newTQueueIO =-  atomically newTQueue--newEmptyTMVarIO :: IO (TMVar a)-newEmptyTMVarIO =-  atomically newEmptyTMVar--newTVarIO :: a -> IO (TVar a)-newTVarIO =-  atomically . newTVar--onException :: IO a -> IO b -> IO a-onException action cleanup =-  catch @_ @SomeException action \exception -> do-    _ <- cleanup-    throwIO exception--putTMVarIO :: TMVar a -> a -> IO ()-putTMVarIO var x =-  atomically (putTMVar var x)--readTVarIO :: TVar a -> IO a-readTVarIO =-  atomically . readTVar--registerDelay :: Int -> IO (STM (), IO ())-registerDelay micros = do-  var <- Control.Concurrent.Classy.registerDelay micros-  pure (readTVar var >>= check, pure ())--throwIO :: Exception e => e -> IO a-throwIO =-  throw--try :: Exception e => IO a -> IO (Either e a)-try action =-  catch (Right <$> action) (pure . Left)--uniqueInt :: IO Int-uniqueInt =-  pure 0--#else--import Control.Concurrent hiding (forkIO)-import Control.Concurrent.STM hiding (registerDelay)-import Control.Exception-import Control.Monad (unless)-import Data.IORef (IORef, atomicModifyIORef', newIORef)-import GHC.Conc (ThreadId (ThreadId))-#if defined(mingw32_HOST_OS)-import GHC.Conc.Windows-#else-import GHC.Event-#endif-import GHC.Exts (fork#)-import GHC.IO (IO (IO), unsafePerformIO, unsafeUnmask)-import Prelude--forkIO :: IO () -> IO ThreadId-forkIO action =-  IO \s ->-    case fork# action s of-      (# s1, tid #) -> (# s1, ThreadId tid #)--putTMVarIO :: TMVar a -> a -> IO ()-putTMVarIO var x =-  atomically (putTMVar var x)--#if defined(mingw32_HOST_OS)-registerDelay :: Int -> IO (STM (), IO ())-registerDelay micros = do-  var <- GHC.Conc.Windows.registerDelay micros-  pure (readTVar var >>= \b -> unless b retry, pure ()) -- no unregister on Windows =P-#else-registerDelay :: Int -> IO (STM (), IO ())-registerDelay micros = do-  var <- newTVarIO False-  manager <- getSystemTimerManager-  key <- registerTimeout manager micros (atomically (writeTVar var True))-  pure (readTVar var >>= \b -> unless b retry, unregisterTimeout manager key)-#endif--uniqueInt :: IO Int-uniqueInt =-  atomicModifyIORef' counter \n -> let n' = n + 1 in (n', n')--counter :: IORef Int-counter =-  unsafePerformIO (newIORef 0)-{-# NOINLINE counter #-}--#endif

@@ -1,49 +0,0 @@-module Ki.Context-  ( Context (..),-    dummyContext,-    globalContext,-  )-where--import Ki.CancelToken (CancelToken)-import Ki.Ctx (Ctx)-import qualified Ki.Ctx as Ctx-import Ki.Prelude--data Context = Context-  { cancelContext :: IO (),-    contextCancelTokenSTM :: STM CancelToken,-    -- | Derive a child context from a parent context.-    ---    --   * If the parent is already cancelled, so is the child.-    --   * If the parent isn't already canceled, the child registers itself with the-    --     parent such that:-    --       * When the parent is cancelled, so is the child-    --       * When the child is cancelled, it removes the parent's reference to it-    deriveContext :: STM Context-  }--newContext :: Ctx -> Context-newContext ctx =-  Context-    { cancelContext = Ctx.cancelCtx ctx,-      contextCancelTokenSTM = Ctx.ctxCancelToken ctx,-      deriveContext = newContext <$> Ctx.deriveCtx ctx-    }--dummyContext :: Context-dummyContext =-  Context-    { cancelContext = pure (),-      contextCancelTokenSTM = retry,-      deriveContext = pure dummyContext-    }---- | The global context. It cannot be cancelled.-globalContext :: Context-globalContext =-  Context-    { cancelContext = pure (),-      contextCancelTokenSTM = retry,-      deriveContext = newContext <$> Ctx.newCtxSTM-    }

@@ -1,92 +0,0 @@-{-# LANGUAGE PatternSynonyms #-}-{-# LANGUAGE TypeApplications #-}--module Ki.Ctx-  ( Ctx (..),-    newCtxSTM,-    deriveCtx,-    cancelCtx,-    cancelCtxSTM,-    ctxCancelToken,-  )-where--import qualified Data.IntMap.Strict as IntMap-import Ki.CancelToken-import Ki.Prelude--data Ctx = Ctx-  { cancelTokenVar :: TVar (Maybe CancelToken),-    childrenVar :: TVar (IntMap Ctx),-    -- | The next id to assign to a child context. The child needs a unique identifier so it can delete itself from its-    -- parent's children map if it's cancelled independently. Wrap-around seems ok; that's a *lot* of children for one-    -- parent to have.-    nextIdVar :: TVar Int,-    -- | When I'm cancelled, this action removes myself from my parent's context. This isn't simply a pointer to the-    -- parent 'Ctx' for three reasons:-    ---    --   * "Root" contexts don't have a parent, so it'd have to be a Maybe (one more pointer indirection)-    --   * We don't really need a reference to the parent, because we only want to be able to remove ourselves from its-    --     children map, so just storing the STM action that does exactly seems a bit safer, even if conceptually it's-    --     a bit indirect.-    --   * If we stored a reference to the parent, we'd also have to store our own id, rather than just currying it into-    --     this action.-    onCancel :: STM ()-  }--newCtxSTM :: STM Ctx-newCtxSTM =-  newCtxSTM_ (pure ())--newCtxSTM_ :: STM () -> STM Ctx-newCtxSTM_ onCancel = do-  cancelTokenVar <- newTVar Nothing-  childrenVar <- newTVar IntMap.empty-  nextIdVar <- newTVar 0-  pure Ctx {cancelTokenVar, childrenVar, nextIdVar, onCancel}--deriveCtx :: Ctx -> STM Ctx-deriveCtx context@Ctx {cancelTokenVar, childrenVar, nextIdVar} =-  readTVar cancelTokenVar >>= \case-    Nothing -> do-      childId <- readTVar nextIdVar-      writeTVar nextIdVar $! childId + 1-      child <- newCtxSTM_ (modifyTVar' childrenVar (IntMap.delete childId))-      children <- readTVar childrenVar-      writeTVar childrenVar $! IntMap.insert childId child children-      pure child-    Just (CancelToken _) -> pure context--cancelCtx :: Ctx -> IO ()-cancelCtx context = do-  token <- newCancelToken-  atomically (cancelCtxSTM context token)--cancelCtxSTM :: Ctx -> CancelToken -> STM ()-cancelCtxSTM ctx@Ctx {onCancel} token =-  whenCanceling ctx token do-    cancelChildren ctx token-    onCancel--ctxCancelSTM_ :: CancelToken -> Ctx -> STM ()-ctxCancelSTM_ token ctx =-  whenCanceling ctx token (cancelChildren ctx token)--whenCanceling :: Ctx -> CancelToken -> STM () -> STM ()-whenCanceling Ctx {cancelTokenVar} token action =-  readTVar cancelTokenVar >>= \case-    Nothing -> do-      writeTVar cancelTokenVar $! Just token-      action-    Just (CancelToken _) -> pure ()--cancelChildren :: Ctx -> CancelToken -> STM ()-cancelChildren Ctx {childrenVar} token = do-  children <- readTVar childrenVar-  for_ (IntMap.elems children) (ctxCancelSTM_ token)--ctxCancelToken :: Ctx -> STM CancelToken-ctxCancelToken Ctx {cancelTokenVar} =-  readTVar cancelTokenVar >>= \case-    Nothing -> retry-    Just token -> pure token

@@ -1,22 +0,0 @@-module Ki.Debug-  ( debug,-  )-where--import Control.Concurrent-import System.IO.Unsafe (unsafePerformIO)-import Prelude--debug :: Monad m => String -> m ()-debug message =-  unsafePerformIO output `seq` pure ()-  where-    output :: IO ()-    output = do-      threadId <- myThreadId-      withMVar lock \_ -> putStrLn ("[" ++ show threadId ++ "] " ++ message)--lock :: MVar ()-lock =-  unsafePerformIO (newMVar ())-{-# NOINLINE lock #-}

@@ -1,36 +0,0 @@-module Ki.Duration-  ( Duration (..),-    toMicroseconds,-    microseconds,-    milliseconds,-    seconds,-  )-where--import Data.Data (Data)-import Data.Fixed-import Ki.Prelude---- | A length of time with microsecond precision. Numeric literals are treated as seconds.-newtype Duration = Duration (Fixed E6)-  deriving stock (Data, Generic)-  deriving newtype (Enum, Eq, Fractional, Num, Ord, Read, Real, RealFrac, Show)--toMicroseconds :: Duration -> Int-toMicroseconds (Duration (MkFixed us)) =-  fromIntegral us---- | One microsecond.-microseconds :: Duration-microseconds =-  Duration (MkFixed 1)---- | One millisecond.-milliseconds :: Duration-milliseconds =-  Duration (MkFixed 1000)---- | One second.-seconds :: Duration-seconds =-  Duration (MkFixed 1000000)

@@ -1,197 +0,0 @@-{-# LANGUAGE PatternSynonyms #-}--module Ki.Implicit-  ( -- * Context-    Context,-    withGlobalContext,--    -- * Scope-    Scope,-    scoped,-    Scope.wait,-    Scope.waitSTM,-    waitFor,--    -- * Spawning threads-    -- $spawning-threads-    Thread,--    -- ** Fork-    fork,-    fork_,-    forkWithUnmask,-    forkWithUnmask_,--    -- ** Async-    async,-    asyncWithUnmask,--    -- ** Await-    Thread.await,-    Thread.awaitSTM,-    Thread.awaitFor,--    -- * Soft-cancellation-    CancelToken,-    Scope.cancel,-    cancelled,-    cancelledSTM,--    -- * Miscellaneous-    Duration,-    Duration.microseconds,-    Duration.milliseconds,-    Duration.seconds,-    timeoutSTM,-    sleep,-  )-where--import Ki.CancelToken (CancelToken)-import qualified Ki.Context as Context-import Ki.Duration (Duration)-import qualified Ki.Duration as Duration-import Ki.Prelude-import Ki.Scope (Scope)-import qualified Ki.Scope as Scope-import Ki.Thread (Thread)-import qualified Ki.Thread as Thread-import Ki.Timeout (timeoutSTM)---- $spawning-threads------ There are two variants of __thread__-creating functions with different exception-propagation semantics.------ * If a __thread__ created with 'fork' throws an exception, it is immediately propagated up the call tree to the--- __thread__ that created its __scope__.------ * If a __thread__ created with 'async' throws an exception, it is not propagated up the call tree, but can be--- observed by 'Ki.Implicit.await'.---- | A __context__ models a program's call tree, and is used as a mechanism to propagate /cancellation/ requests to--- every __thread__ created within a __scope__.------ Every __thread__ is provided its own __context__, which is derived from its __scope__.------ A __thread__ can query whether its __context__ has been /cancelled/, which is a suggestion to perform a graceful--- termination.-type Context =-  ?context :: Context.Context---- | Create a __thread__ within a __scope__.------ /Throws/:------   * Calls 'error' if the __scope__ is /closed/.-async :: Scope -> (Context => IO a) -> IO (Thread (Either SomeException a))-async scope action =-  Thread.async scope (with scope action)---- | Variant of 'async' that provides the __thread__ a function that unmasks asynchronous exceptions.------ /Throws/:------   * Calls 'error' if the __scope__ is /closed/.-asyncWithUnmask :: Scope -> (Context => (forall x. IO x -> IO x) -> IO a) -> IO (Thread (Either SomeException a))-asyncWithUnmask scope action =-  Thread.asyncWithUnmask scope (let ?context = Scope.context scope in action)---- | Return whether the current __context__ is /cancelled/.------ __Threads__ running in a /cancelled/ __context__ should terminate as soon as possible. The cancel token may be thrown--- to fulfill the /cancellation/ request in case the __thread__ is unable or unwilling to terminate normally with a--- value.-cancelled :: Context => IO (Maybe CancelToken)-cancelled =-  atomically (optional cancelledSTM)---- | @STM@ variant of 'cancelled'; blocks until the current __context__ is /cancelled/.-cancelledSTM :: Context => STM CancelToken-cancelledSTM =-  Context.contextCancelTokenSTM ?context---- | Create a __thread__ within a __scope__.------ If the __thread__ throws an exception, the exception is immediately propagated up the call tree to the __thread__--- that opened its __scope__, unless that exception is a 'CancelToken' that fulfills a /cancellation/ request.------ /Throws/:------   * Calls 'error' if the __scope__ is /closed/.-fork :: Scope -> (Context => IO a) -> IO (Thread a)-fork scope action =-  Thread.fork scope (with scope action)---- | Variant of 'fork' that does not return a handle to the created __thread__.------ /Throws/:------   * Calls 'error' if the __scope__ is /closed/.-fork_ :: Scope -> (Context => IO ()) -> IO ()-fork_ scope action =-  Thread.fork_ scope (with scope action)---- | Variant of 'fork' that provides the __thread__ a function that unmasks asynchronous exceptions.------ /Throws/:------   * Calls 'error' if the __scope__ is /closed/.-forkWithUnmask :: Scope -> (Context => (forall x. IO x -> IO x) -> IO a) -> IO (Thread a)-forkWithUnmask scope action =-  Thread.forkWithUnmask scope (let ?context = Scope.context scope in action)---- | Variant of 'forkWithUnmask' that does not return a handle to the created __thread__.------ /Throws/:------   * Calls 'error' if the __scope__ is /closed/.-forkWithUnmask_ :: Scope -> (Context => (forall x. IO x -> IO x) -> IO ()) -> IO ()-forkWithUnmask_ scope action =-  Thread.forkWithUnmask_ scope (let ?context = Scope.context scope in action)---- | Perform an @IO@ action in the global __context__. The global __context__ cannot be /cancelled/.-withGlobalContext :: (Context => IO a) -> IO a-withGlobalContext action =-  let ?context = Context.globalContext in action---- | Open a __scope__, perform an @IO@ action with it, then close the __scope__.------ When the __scope__ is closed, all remaining __threads__ created within it are killed.------ /Throws/:------   * The exception thrown by the callback to 'scoped' itself, if any.---   * The first exception thrown by or to a __thread__ created with 'fork', if any.------ ==== __Examples__------ @--- 'scoped' \\scope -> do---   'fork_' scope worker1---   'fork_' scope worker2---   'Ki.Implicit.wait' scope--- @-scoped :: Context => (Context => Scope -> IO a) -> IO a-scoped action =-  Scope.scoped ?context \scope -> with scope (action scope)---- | __Context__-aware, duration-based @threadDelay@.------ /Throws/:------   * Throws 'CancelToken' if the current __context__ is (or becomes) /cancelled/.-sleep :: Context => Duration -> IO ()-sleep duration =-  timeoutSTM duration (cancelledSTM >>= throwSTM) (pure ())---- | Variant of 'Ki.Implicit.wait' that waits for up to the given duration. This is useful for giving __threads__ some--- time to fulfill a /cancellation/ request before killing them.-waitFor :: Scope -> Duration -> IO ()-waitFor =-  Scope.waitFor------with :: Scope -> (Context => a) -> a-with scope action =-  let ?context = Scope.context scope in action

@@ -1,18 +0,0 @@-module Ki.Internal-  ( module Ki.CancelToken,-    module Ki.Context,-    module Ki.Ctx,-    module Ki.Duration,-    module Ki.Scope,-    module Ki.Thread,-    module Ki.Timeout,-  )-where--import Ki.CancelToken-import Ki.Context-import Ki.Ctx-import Ki.Duration-import Ki.Scope-import Ki.Thread-import Ki.Timeout

@@ -0,0 +1,37 @@+module Ki.Internal.ByteCount+  ( ByteCount,+    kilobytes,+    megabytes,+    byteCountToInt64,+  )+where++import Ki.Internal.Prelude++-- | A number of bytes.+newtype ByteCount = ByteCount Int64+  deriving newtype (Eq, Ord)++instance Show ByteCount where+  show (ByteCount b)+    | (mb, 0) <- quotRem b 1048576, mb > 0 = "megabytes " ++ show mb+    | (kb, 0) <- quotRem b 1024 = "kilobytes " ++ show kb+    | otherwise = undefined++-- | A number of kilobytes.+kilobytes :: Natural -> ByteCount+kilobytes n =+  ByteCount (snip (n * 1024))++-- | A number of megabytes.+megabytes :: Natural -> ByteCount+megabytes n =+  ByteCount (snip (n * 1048576))++byteCountToInt64 :: ByteCount -> Int64+byteCountToInt64 =+  coerce++snip :: Natural -> Int64+snip n =+  fromIntegral (min (fromIntegral (maxBound :: Int64)) n)

@@ -0,0 +1,57 @@+-- 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+import Ki.Internal.Prelude++-- | 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 #-}

@@ -0,0 +1,62 @@+{-# LANGUAGE MagicHash #-}+{-# LANGUAGE UnboxedTuples #-}++module Ki.Internal.Prelude+  ( forkIO,+    forkOn,+    interruptiblyMasked,+    uninterruptiblyMasked,+    module X,+  )+where++import Control.Applicative as X (optional, (<|>))+import Control.Concurrent hiding (forkIO, forkOn)+import Control.Concurrent as X (ThreadId, myThreadId, threadDelay, throwTo)+import Control.Concurrent.MVar as X+import Control.Exception+import Control.Exception as X (Exception, SomeException, mask_, throwIO, try, uninterruptibleMask, uninterruptibleMask_)+import Control.Monad as X (join, when)+import Data.Coerce as X (coerce)+import Data.Data as X (Data)+import Data.Foldable as X (for_, traverse_)+import Data.Function as X (fix)+import Data.Functor as X (void, ($>), (<&>))+import Data.Int as X+import Data.IntMap.Strict as X (IntMap)+import Data.Map.Strict as X (Map)+import Data.Maybe as X (fromMaybe)+import Data.Sequence as X (Seq)+import Data.Set as X (Set)+import Data.Word as X (Word32)+import GHC.Base (maskAsyncExceptions#, maskUninterruptible#)+import GHC.Conc (ThreadId (ThreadId))+import GHC.Exts (Int (I#), fork#, forkOn#)+import GHC.Generics as X (Generic)+import GHC.IO (IO (IO))+import Numeric.Natural as X (Natural)+import Prelude as X++-- | Call an action with asynchronous exceptions interruptibly masked.+interruptiblyMasked :: IO a -> IO a+interruptiblyMasked (IO io) =+  IO (maskAsyncExceptions# io)++-- | Call an action with asynchronous exceptions uninterruptibly masked.+uninterruptiblyMasked :: IO a -> IO a+uninterruptiblyMasked (IO io) =+  IO (maskUninterruptible# io)++-- Control.Concurrent.forkIO without the dumb exception handler+forkIO :: IO () -> IO ThreadId+forkIO action =+  IO \s0 ->+    case fork# action s0 of+      (# s1, tid #) -> (# s1, ThreadId tid #)++-- Control.Concurrent.forkOn without the dumb exception handler+forkOn :: Int -> IO () -> IO ThreadId+forkOn (I# cap) action =+  IO \s0 ->+    case forkOn# cap action s0 of+      (# s1, tid #) -> (# s1, ThreadId tid #)

@@ -0,0 +1,416 @@+module Ki.Internal.Scope+  ( Scope,+    scoped,+    awaitAll,+    fork,+    forkWith,+    forkWith_,+    fork_,+    forkTry,+    forkTryWith,+  )+where++import qualified Control.Concurrent+import Control.Exception+  ( Exception (fromException, toException),+    MaskingState (..),+    SomeAsyncException,+    asyncExceptionFromException,+    asyncExceptionToException,+    catch,+    pattern ErrorCall,+  )+import qualified Data.IntMap.Lazy as IntMap+import Data.Void (Void, absurd)+import GHC.Conc+  ( STM,+    TVar,+    atomically,+    enableAllocationLimit,+    labelThread,+    newTVarIO,+    readTVar,+    retry,+    setAllocationCounter,+    throwSTM,+    writeTVar,+  )+import GHC.IO (unsafeUnmask)+import Ki.Internal.ByteCount+import Ki.Internal.Counter+import Ki.Internal.Prelude+import Ki.Internal.Thread++-- | A scope.+--+-- ==== __👉 Details__+--+-- * A scope delimits the lifetime of all threads created within it.+--+-- * A scope is only valid during the callback provided to 'Ki.scoped'.+--+-- * The thread that creates a scope is considered the parent of all threads created within it.+--+-- * All threads created within a scope can be awaited together (see 'Ki.awaitAll').+--+-- * All threads created within a scope are terminated when the scope closes.+data Scope = Scope+  { -- The MVar that a child tries to put to, in the case that it tries to propagate an exception to its parent, but+    -- gets delivered an exception from its parent concurrently (which interrupts the throw). The parent must raise+    -- exceptions in its children with asynchronous exceptions uninterruptibly masked for correctness, yet we don't want+    -- a parent in the process of tearing down to miss/ignore this exception that we're trying to propagate?+    --+    -- Why a single-celled MVar? What if two siblings are fighting to inform their parent of their death? Well, only+    -- one exception can be propagated by the parent anyway, so we wouldn't need or want both.+    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 id of the thread that created the scope, which is considered the parent of all threads created within it.+    parentThreadId :: {-# UNPACK #-} !ThreadId,+    -- The number of child threads that are guaranteed to be about to start, in the sense that only the GHC scheduler+    -- can continue to delay; there's no opportunity for an async exception to strike and prevent one of these threads+    -- from starting.+    --+    -- Sentinel value: -1 means the scope is closed.+    startingVar :: {-# UNPACK #-} !(TVar Int)+  }++-- Internal async exception thrown by a parent thread to its children when the scope is closing.+data ScopeClosing+  = ScopeClosing++instance Show ScopeClosing where+  show _ = "ScopeClosing"++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 =+  case fromException exception of+    Just ScopeClosing -> True+    _ -> False++pattern IsScopeClosingException :: SomeException+pattern IsScopeClosingException <- (isScopeClosingException -> True)++-- | Open a scope, perform an IO action with it, then close the scope.+--+-- ==== __👉 Details__+--+-- * The thread that creates a scope is considered the parent of all threads created within it.+--+-- * A scope is only valid during the callback provided to 'Ki.scoped'.+--+-- * When a scope closes (/i.e./ just before 'Ki.scoped' returns):+--+--     * The parent thread raises an exception in all of its living children.+--     * The parent thread blocks until those threads terminate.+scoped :: (Scope -> IO a) -> IO a+scoped action = do+  scope@Scope {childExceptionVar, childrenVar, startingVar} <- allocateScope++  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.+          blockUntil0 startingVar+          -- Write the sentinel value indicating that this scope is closed, and it is an error to try to create a thread+          -- within it.+          writeTVar startingVar (-1)+          -- Return the list of currently-running children to kill. Some of them may have *just* started (e.g. if we+          -- initially retried in `blockUntil0` above). That's fine - kill them all!+          readTVar childrenVar++      -- If one of our children propagated an exception to us, then we know it's about to terminate, so we don't bother+      -- throwing an exception to it.+      pure case result of+        Left (fromException -> Just ThreadFailed {childId}) -> IntMap.delete childId livingChildren0+        _ -> livingChildren0++    -- Deliver a ScopeClosing exception to every living child.+    --+    -- This happens to throw in the order the children were created... but I think we decided this feature isn't very+    -- useful in practice, so maybe we should simplify the internals and just keep a set of children?+    for_ (IntMap.elems livingChildren) \livingChild -> throwTo livingChild ScopeClosing++    -- Block until all children have terminated; this relies on children respecting the async exception, which they+    -- must, for correctness. Otherwise, a thread could indeed outlive the scope in which it's created, which is+    -- definitely not structured concurrency!+    atomically (blockUntilEmpty childrenVar)++    -- 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.+    --+    --   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).+    --+    --   3) An async exception waiting in our exception queue, because we still have async exceptions uninterruptibly+    --      masked.+    --+    -- We cannot throw more than one, so throw them in that priority order.+    case result of+      Left exception -> throwIO (unwrapThreadFailed exception)+      Right value ->+        tryTakeMVar childExceptionVar >>= \case+          Nothing -> pure value+          Just exception -> throwIO exception++-- Allocate a new scope.+allocateScope :: IO Scope+allocateScope = do+  childExceptionVar <- newEmptyMVar+  childrenVar <- newTVarIO IntMap.empty+  nextChildIdCounter <- newCounter+  parentThreadId <- myThreadId+  startingVar <- newTVarIO 0+  pure Scope {childExceptionVar, childrenVar, nextChildIdCounter, parentThreadId, startingVar}++-- Spawn a thread in a scope, providing it its child id and a function that sets the masking state to the requested+-- masking state. The given action is called with async exceptions interruptibly masked.+spawn :: Scope -> ThreadOptions -> (Int -> (forall x. IO x -> IO x) -> UnexceptionalIO ()) -> IO ThreadId+spawn+  Scope {childrenVar, nextChildIdCounter, startingVar}+  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 startingVar+        if n < 0+          then throwSTM (ErrorCall "ki: scope closed")+          else writeTVar startingVar $! n + 1++      childId <- incrCounter nextChildIdCounter++      childThreadId <-+        forkWithAffinity affinity do+          when (not (null label)) do+            childThreadId <- myThreadId+            labelThread childThreadId label++          case allocationLimit of+            Nothing -> pure ()+            Just bytes -> do+              setAllocationCounter (byteCountToInt64 bytes)+              enableAllocationLimit++          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++          runUnexceptionalIO (action childId atRequestedMaskingState)++          atomically (unrecordChild childrenVar childId)++      -- Record the child as having started. Not allowed to retry.+      atomically do+        n <- readTVar startingVar+        writeTVar startingVar $! n - 1 -- it's actually ok to go from e.g. -1 to -2 here (very unlikely)+        recordChild childrenVar childId childThreadId++      pure childThreadId++-- 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) -> Int -> ThreadId -> STM ()+recordChild childrenVar childId childThreadId = do+  children <- readTVar childrenVar+  writeTVar childrenVar $! IntMap.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) -> Int -> STM ()+unrecordChild childrenVar childId = do+  children <- readTVar childrenVar+  writeTVar childrenVar $! IntMap.alter (maybe (Just undefined) (const Nothing)) childId children++-- forkIO/forkOn/forkOS, switching on affinity+forkWithAffinity :: ThreadAffinity -> IO () -> IO ThreadId+forkWithAffinity = \case+  Unbound -> forkIO+  Capability n -> forkOn n+  OsThread -> Control.Concurrent.forkOS++-- | Wait until all threads created within a scope terminate.+awaitAll :: Scope -> STM ()+awaitAll Scope {childrenVar, startingVar} = do+  blockUntilEmpty childrenVar+  blockUntil0 startingVar++-- Block until an IntMap becomes empty.+blockUntilEmpty :: TVar (IntMap a) -> STM ()+blockUntilEmpty var = do+  x <- readTVar var+  if IntMap.null x then pure () else retry++-- Block until a TVar becomes 0.+blockUntil0 :: TVar Int -> STM ()+blockUntil0 var = do+  x <- readTVar var+  if x == 0 then pure () else retry++-- | 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+-- create a child thread with a different initial masking state, use 'Ki.forkWith'.+fork :: Scope -> IO a -> IO (Thread a)+fork scope =+  forkWith scope defaultThreadOptions++-- | Variant of 'Ki.fork' for threads that never return.+fork_ :: Scope -> IO Void -> IO ()+fork_ scope =+  forkWith_ scope defaultThreadOptions++-- | Variant of 'Ki.fork' that takes an additional options argument.+forkWith :: Scope -> ThreadOptions -> IO a -> IO (Thread a)+forkWith scope opts action = do+  resultVar <- newTVarIO Nothing+  ident <-+    spawn scope opts \childId masking -> do+      result <- unexceptionalTry (masking action)+      case result of+        Left exception ->+          when+            (not (isScopeClosingException exception))+            (propagateException scope childId exception)+        Right _ -> pure ()+      -- even put async exceptions that we propagated. this isn't totally ideal because a caller awaiting this thread+      -- would not be able to distinguish between async exceptions delivered to this thread, or itself+      UnexceptionalIO (atomically (writeTVar resultVar (Just result)))+  let doAwait =+        readTVar resultVar >>= \case+          Nothing -> retry+          Just (Left exception) -> throwSTM exception+          Just (Right value) -> pure value+  pure (makeThread ident doAwait)++-- | Variant of 'Ki.forkWith' for threads that never return.+forkWith_ :: Scope -> ThreadOptions -> IO Void -> IO ()+forkWith_ scope opts action = do+  _childThreadId <-+    spawn scope opts \childId masking ->+      unexceptionalTryEither+        (\exception -> when (not (isScopeClosingException exception)) (propagateException scope childId exception))+        absurd+        (masking action)+  pure ()++-- | Like 'Ki.fork', but the child thread does not propagate exceptions that are both:+--+-- * 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 scope =+  forkTryWith scope defaultThreadOptions++-- | Variant of 'Ki.forkTry' that takes an additional options argument.+forkTryWith :: forall e a. Exception e => Scope -> ThreadOptions -> IO a -> IO (Thread (Either e a))+forkTryWith scope opts action = do+  resultVar <- newTVarIO Nothing+  childThreadId <-+    spawn scope opts \childId masking -> do+      result <- unexceptionalTry (masking action)+      case result of+        Left exception -> do+          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+          when shouldPropagate (propagateException scope childId exception)+        Right _value -> pure ()+      UnexceptionalIO (atomically (writeTVar resultVar (Just result)))+  let doAwait =+        readTVar resultVar >>= \case+          Nothing -> retry+          Just (Left exception) ->+            case fromException @e exception of+              Nothing -> throwSTM exception+              Just expectedException -> pure (Left expectedException)+          Just (Right value) -> pure (Right value)+  pure (makeThread childThreadId doAwait)+  where+    isAsyncException :: SomeException -> Bool+    isAsyncException exception =+      case fromException @SomeAsyncException exception of+        Nothing -> False+        Just _ -> True++-- TODO document this+-- Precondition: interruptibly masked+propagateException :: Scope -> Int -> SomeException -> UnexceptionalIO ()+propagateException Scope {childExceptionVar, parentThreadId} childId exception =+  loop+  where+    loop :: UnexceptionalIO ()+    loop =+      unexceptionalTry (throwTo parentThreadId ThreadFailed {childId, exception}) >>= \case+        Left IsScopeClosingException -> unexceptionalTryPutMVar_ childExceptionVar exception+        -- while blocking on notifying the parent of this exception, we got hit by a random async exception from+        -- elsewhere. that's weird and unexpected, but we already have an exception to deliver, so it just gets tossed+        -- to the void...+        Left _ -> loop+        Right _ -> pure ()++-- A little promise that this IO action cannot throw an exception.+--+-- Yeah it's verbose, and maybe not that necessary, but the code that bothers to use it really does require+-- un-exceptiony IO actions for correctness, so here we are.+newtype UnexceptionalIO a = UnexceptionalIO+  {runUnexceptionalIO :: IO a}+  deriving newtype (Applicative, Functor, Monad)++unexceptionalTry :: forall a. IO a -> UnexceptionalIO (Either SomeException a)+unexceptionalTry =+  coerce @(IO a -> IO (Either SomeException a)) try++-- Like try, but with continuations. Also, catches all exceptions, because that's the only flavor we need.+unexceptionalTryEither ::+  forall a b.+  (SomeException -> UnexceptionalIO b) ->+  (a -> UnexceptionalIO b) ->+  IO a ->+  UnexceptionalIO b+unexceptionalTryEither onFailure onSuccess action =+  UnexceptionalIO do+    join do+      catch+        (coerce @_ @(a -> IO b) onSuccess <$> action)+        (pure . coerce @_ @(SomeException -> IO b) onFailure)++unexceptionalTryPutMVar_ :: MVar a -> a -> UnexceptionalIO ()+unexceptionalTryPutMVar_ var x =+  coerce (void (tryPutMVar var x))

@@ -0,0 +1,151 @@+module Ki.Internal.Thread+  ( Thread,+    makeThread,+    await,+    ThreadAffinity (..),+    ThreadOptions (..),+    defaultThreadOptions,+    ThreadFailed (..),+    unwrapThreadFailed,+  )+where++import Control.Exception+  ( BlockedIndefinitelyOnSTM (..),+    Exception (fromException, toException),+    MaskingState (..),+    asyncExceptionFromException,+    asyncExceptionToException,+  )+import GHC.Conc (STM, catchSTM)+import Ki.Internal.ByteCount+import Ki.Internal.Prelude++-- | A thread.+--+-- ==== __👉 Details__+--+-- * A thread's lifetime is delimited by the scope in which it was created.+--+-- * The thread that creates a scope is considered the parent of all threads created within it.+--+-- * If an exception is raised in a child thread, the child either propagates the exception to its parent (see+--   'Ki.fork'), or returns the exception as a value (see 'Ki.forkTry').+--+-- * All threads created within a scope are terminated when the scope closes.+data Thread a = Thread+  { threadId :: {-# UNPACK #-} !ThreadId,+    await_ :: !(STM a)+  }+  deriving stock (Functor)++instance Eq (Thread a) where+  Thread ix _ == Thread iy _ =+    ix == iy++instance Ord (Thread a) where+  compare (Thread ix _) (Thread iy _) =+    compare ix iy++makeThread :: ThreadId -> STM a -> Thread a+makeThread threadId action =+  Thread+    { threadId,+      -- If *they* are deadlocked, we will *both* will be delivered a wakeup from the RTS. We want to shrug this+      -- exception off, because afterwards they'll have put to the result var. But don't shield indefinitely, once will+      -- cover this use case and prevent any accidental infinite loops.+      await_ = tryEitherSTM (\BlockedIndefinitelyOnSTM -> action) pure action+    }++-- | 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)++-- |+--+-- [@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' = Nothing+--   , '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 #-} !Int,+    exception :: !SomeException+  }+  deriving stock (Show)++instance Exception ThreadFailed where+  toException = asyncExceptionToException+  fromException = asyncExceptionFromException++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+await =+  await_++-- Like try, but with continuations+tryEitherSTM :: Exception e => (e -> STM b) -> (a -> STM b) -> STM a -> STM b+tryEitherSTM onFailure onSuccess action =+  join (catchSTM (onSuccess <$> action) (pure . onFailure))

@@ -1,47 +0,0 @@-module Ki.Prelude-  ( atomicallyIO,-    onLeft,-    whenJust,-    whenLeft,-    whenM,-    module X,-  )-where--import Control.Applicative as X (optional, (<|>))-import Control.Exception as X (Exception, SomeException)-import Control.Monad as X (join, unless)-import Data.Coerce as X (coerce)-import Data.Foldable as X (for_)-import Data.Function as X (fix)-import Data.Functor as X (void, ($>), (<&>))-import Data.IntMap.Strict as X (IntMap)-import Data.Map.Strict as X (Map)-import Data.Maybe as X (fromMaybe)-import Data.Set as X (Set)-import Data.Word as X (Word32)-import GHC.Generics as X (Generic)-import Ki.Concurrency as X-import Prelude as X hiding (IO)--atomicallyIO :: STM (IO a) -> IO a-atomicallyIO =-  join . atomically--onLeft :: (a -> IO b) -> Either a b -> IO b-onLeft f =-  either f pure--whenJust :: Maybe a -> (a -> IO ()) -> IO ()-whenJust x f =-  maybe (pure ()) f x--whenLeft :: Either a b -> (a -> IO b) -> IO b-whenLeft x f =-  either f pure x--whenM :: IO Bool -> IO () -> IO ()-whenM x y =-  x >>= \case-    False -> pure ()-    True -> y

@@ -1,198 +0,0 @@-{-# LANGUAGE PatternSynonyms #-}-{-# LANGUAGE TypeApplications #-}--module Ki.Scope-  ( Scope (..),-    cancel,-    scopeCancelledSTM,-    scopeFork,-    scoped,-    wait,-    waitFor,-    waitSTM,-    ScopeClosing (..),-    ThreadFailed (..),-  )-where--import Control.Exception-  ( Exception (fromException, toException),-    asyncExceptionFromException,-    asyncExceptionToException,-    pattern ErrorCall,-  )-import qualified Data.Monoid as Monoid-import qualified Data.Set as Set-import Ki.Context (Context)-import qualified Ki.Context as Context-import Ki.Duration (Duration)-import Ki.Prelude-import Ki.Timeout (timeoutSTM)---- | A __scope__ delimits the lifetime of all __threads__ created within it.-data Scope = Scope-  { context :: Context,-    -- | Whether this scope is closed.-    -- Invariant: if closed, no threads are starting.-    closedVar :: TVar Bool,-    -- | The set of threads that are currently running.-    runningVar :: TVar (Set ThreadId),-    -- | The number of threads that are *guaranteed* to be about to start, in the sense that only the GHC scheduler can-    -- continue to delay; no async exception can strike here and prevent one of these threads from starting.-    ---    -- If this number is non-zero, and that's problematic (e.g. because we're trying to cancel this scope), we always-    -- respect it and wait for it to drop to zero before proceeding.-    startingVar :: TVar Int-  }--newScope :: Context -> IO Scope-newScope parentContext = do-  context <- atomically (Context.deriveContext parentContext)-  closedVar <- newTVarIO False-  runningVar <- newTVarIO Set.empty-  startingVar <- newTVarIO 0-  pure Scope {context, closedVar, runningVar, startingVar}---- | /Cancel/ all __contexts__ derived from a __scope__.-cancel :: Scope -> IO ()-cancel Scope {context} =-  Context.cancelContext context--scopeCancelledSTM :: Scope -> STM (IO a)-scopeCancelledSTM Scope {context} =-  throwIO <$> Context.contextCancelTokenSTM context---- | Close a scope, kill all of the running threads, and return the first async exception delivered to us while doing--- so, if any.------ Preconditions:---   * The set of threads doesn't include us---   * We're uninterruptibly masked-closeScope :: Scope -> IO (Maybe SomeException)-closeScope scope@Scope {closedVar, runningVar} = do-  threads <--    atomically do-      blockUntilNoneStarting scope-      writeTVar closedVar True-      readTVar runningVar-  exception <- killThreads (Set.toList threads)-  atomically (blockUntilNoneRunning scope)-  pure exception--scopeFork :: Scope -> ((forall x. IO x -> IO x) -> IO a) -> (Either SomeException a -> IO ()) -> IO ThreadId-scopeFork Scope {closedVar, runningVar, startingVar} action k =-  uninterruptibleMask \restore -> do-    -- Record the thread as being about to start-    atomically do-      readTVar closedVar >>= \case-        False -> modifyTVar' startingVar (+ 1)-        True -> throwSTM (ErrorCall "ki: scope closed")--    -- Fork the thread-    childThreadId <--      forkIO do-        childThreadId <- myThreadId-        result <- try (action restore)-        k result-        atomically do-          running <- readTVar runningVar-          case Set.splitMember childThreadId running of-            (xs, True, ys) -> writeTVar runningVar $! Set.union xs ys-            _ -> retry--    -- Record the thread as having started-    atomically do-      modifyTVar' startingVar \n -> n -1-      modifyTVar' runningVar (Set.insert childThreadId)--    pure childThreadId--scoped :: Context -> (Scope -> IO a) -> IO a-scoped context f = do-  scope <- newScope context-  uninterruptibleMask \restore -> do-    result <- try (restore (f scope))-    closeScopeException <- closeScope scope-    -- If the callback failed, we don't care if we were thrown an async exception while closing the scope.-    -- Otherwise, throw that exception (if it exists).-    case result of-      Left exception -> throw exception-      Right value -> do-        whenJust closeScopeException throw-        pure value-  where-    -- If applicable, unwrap the 'ThreadFailed' (assumed to have come from one of our children).-    throw :: SomeException -> IO a-    throw exception =-      case fromException exception of-        Just (ThreadFailed threadFailedException) -> throwIO threadFailedException-        Nothing -> throwIO exception---- | Wait until all __threads__ created within a __scope__ finish.-wait :: Scope -> IO ()-wait =-  atomically . waitSTM---- | Variant of 'wait' that waits for up to the given duration.-waitFor :: Scope -> Duration -> IO ()-waitFor scope duration =-  timeoutSTM duration (pure <$> waitSTM scope) (pure ())---- | @STM@ variant of 'wait'.-waitSTM :: Scope -> STM ()-waitSTM scope = do-  blockUntilNoneRunning scope-  blockUntilNoneStarting scope------------------------------------------------------------------------------------- Scope helpers--blockUntilNoneRunning :: Scope -> STM ()-blockUntilNoneRunning Scope {runningVar} =-  blockUntilTVar runningVar Set.null--blockUntilNoneStarting :: Scope -> STM ()-blockUntilNoneStarting Scope {startingVar} =-  blockUntilTVar startingVar (== 0)------------------------------------------------------------------------------------- Internal exception types---- | Exception thrown by a parent __thread__ to its children when the __scope__ is closing.-data ScopeClosing-  = ScopeClosing-  deriving stock (Eq, Show)--instance Exception ScopeClosing where-  toException = asyncExceptionToException-  fromException = asyncExceptionFromException---- | Exception thrown by a child __thread__ to its parent, if it fails unexpectedly.-newtype ThreadFailed-  = ThreadFailed SomeException-  deriving stock (Show)--instance Exception ThreadFailed where-  toException = asyncExceptionToException-  fromException = asyncExceptionFromException------------------------------------------------------------------------------------- Misc. utils--killThreads :: [ThreadId] -> IO (Maybe SomeException)-killThreads =-  (`fix` mempty) \loop acc -> \case-    [] -> pure (Monoid.getFirst acc)-    threadId : threadIds ->-      -- We unmask because we don't want to deadlock with a thread-      -- that is concurrently trying to throw an exception to us with-      -- exceptions masked.-      try (unsafeUnmask (throwTo threadId ScopeClosing)) >>= \case-        -- don't drop thread we didn't (necessarily) deliver the exception to-        Left exception -> loop (acc <> Monoid.First (Just exception)) (threadId : threadIds)-        Right () -> loop acc threadIds--blockUntilTVar :: TVar a -> (a -> Bool) -> STM ()-blockUntilTVar var f = do-  value <- readTVar var-  unless (f value) retry

@@ -1,160 +0,0 @@-module Ki.Thread-  ( Thread (..),-    async,-    asyncWithUnmask,-    await,-    awaitSTM,-    awaitFor,-    fork,-    fork_,-    forkWithUnmask,-    forkWithUnmask_,-  )-where--import Control.Exception (Exception (fromException))-import qualified Ki.Context as Context-import Ki.Duration (Duration)-import Ki.Prelude-import Ki.Scope (Scope (Scope))-import qualified Ki.Scope as Scope-import Ki.Timeout (timeoutSTM)---- | A running __thread__.-data Thread a-  = Thread !ThreadId !(STM a)-  deriving stock (Functor, Generic)--instance Eq (Thread a) where-  Thread id1 _ == Thread id2 _ =-    id1 == id2--instance Ord (Thread a) where-  compare (Thread id1 _) (Thread id2 _) =-    compare id1 id2---- | Create a __thread__ within a __scope__.------ /Throws/:------   * Calls 'error' if the __scope__ is /closed/.-async :: Scope -> IO a -> IO (Thread (Either SomeException a))-async scope action =-  asyncWithRestore scope \restore -> restore action---- | Variant of 'async' that provides the __thread__ a function that unmasks asynchronous exceptions.------ /Throws/:------   * Calls 'error' if the __scope__ is /closed/.-asyncWithUnmask :: Scope -> ((forall x. IO x -> IO x) -> IO a) -> IO (Thread (Either SomeException a))-asyncWithUnmask scope action =-  asyncWithRestore scope \restore -> restore (action unsafeUnmask)--asyncWithRestore :: forall a. Scope -> ((forall x. IO x -> IO x) -> IO a) -> IO (Thread (Either SomeException a))-asyncWithRestore scope action = do-  resultVar <- newEmptyTMVarIO-  childThreadId <- Scope.scopeFork scope action (putTMVarIO resultVar)-  pure (Thread childThreadId (readTMVar resultVar))---- | Wait for a __thread__ to finish.-await :: Thread a -> IO a-await =-  atomically . awaitSTM---- | @STM@ variant of 'await'.-awaitSTM :: Thread a -> STM a-awaitSTM (Thread _ action) =-  action---- | Variant of 'await' that gives up after the given duration.-awaitFor :: Thread a -> Duration -> IO (Maybe a)-awaitFor thread duration =-  timeoutSTM duration (pure . Just <$> awaitSTM thread) (pure Nothing)---- | Create a __thread__ within a __scope__.------ If the __thread__ throws an exception, the exception is immediately propagated up the call tree to the __thread__--- that opened its __scope__.------ /Throws/:------   * Calls 'error' if the __scope__ is /closed/.-fork :: Scope -> IO a -> IO (Thread a)-fork scope action =-  forkWithRestore scope \restore -> restore action---- | Variant of 'fork' that does not return a handle to the created __thread__.------ /Throws/:------   * Calls 'error' if the __scope__ is /closed/.-fork_ :: Scope -> IO () -> IO ()-fork_ scope action =-  forkWithRestore_ scope \restore -> restore action---- | Variant of 'fork' that provides the __thread__ a function that unmasks asynchronous exceptions.------ /Throws/:------   * Calls 'error' if the __scope__ is /closed/.-forkWithUnmask :: Scope -> ((forall x. IO x -> IO x) -> IO a) -> IO (Thread a)-forkWithUnmask scope action =-  forkWithRestore scope \restore -> restore (action unsafeUnmask)---- | Variant of 'forkWithUnmask' that does not return a handle to the created __thread__.------ /Throws/:------   * Calls 'error' if the __scope__ is /closed/.-forkWithUnmask_ :: Scope -> ((forall x. IO x -> IO x) -> IO ()) -> IO ()-forkWithUnmask_ scope action =-  forkWithRestore_ scope \restore -> restore (action unsafeUnmask)--forkWithRestore :: Scope -> ((forall x. IO x -> IO x) -> IO a) -> IO (Thread a)-forkWithRestore scope action = do-  parentThreadId <- myThreadId-  resultVar <- newEmptyTMVarIO-  childThreadId <--    Scope.scopeFork scope action \case-      Left exception ->-        -- Intentionally don't fill the result var.-        ---        -- Prior to 0.2.0, we did put a 'Left exception' in the result var, so that if another thread awaited it, we'd-        -- promptly deliver them the exception that brought this thread down. However, that exception was *wrapped* in-        -- a 'ThreadFailed' exception, so the caller could distinguish between async exceptions *delivered to them* and-        -- async exceptions coming *synchronously* out of the call to 'await'.-        ---        -- At some point I reasoned that if one is following some basic structured concurrency guidelines, and not doing-        -- weird/complicated things like passing threads around, then it is likely that a failed forked thread is just-        -- about to propagate its exception to all callers of 'await' (presumably, its direct parent).-        ---        -- Might GHC deliver a BlockedIndefinitelyOnSTM in the meantime, though?-        maybePropagateException scope parentThreadId exception-      Right result -> putTMVarIO resultVar result-  pure (Thread childThreadId (readTMVar resultVar))--forkWithRestore_ :: Scope -> ((forall x. IO x -> IO x) -> IO ()) -> IO ()-forkWithRestore_ scope action = do-  parentThreadId <- myThreadId-  _childThreadId <- Scope.scopeFork scope action (onLeft (maybePropagateException scope parentThreadId))-  pure ()--maybePropagateException :: Scope -> ThreadId -> SomeException -> IO ()-maybePropagateException Scope {closedVar, context} parentThreadId exception =-  whenM shouldPropagateException (throwTo parentThreadId (Scope.ThreadFailed exception))-  where-    shouldPropagateException :: IO Bool-    shouldPropagateException =-      case fromException exception of-        -- Our scope is (presumably) closing, so don't propagate this exception that presumably just came from our-        -- parent. But if our scope's closedVar isn't True, that means this 'ScopeClosing' definitely came from-        -- somewhere else...-        Just Scope.ScopeClosing -> not <$> readTVarIO closedVar-        Nothing ->-          case fromException exception of-            -- We (presumably) are honoring our own cancellation request, so don't propagate that either.-            -- It's a bit complicated looking because we *do* want to throw this token if we (somehow) threw it-            -- "inappropriately" in the sense that it wasn't ours to throw - it was smuggled from elsewhere.-            Just token -> atomically ((/= token) <$> Context.contextCancelTokenSTM context <|> pure True)-            Nothing -> pure True

@@ -1,15 +0,0 @@-module Ki.Timeout-  ( timeoutSTM,-  )-where--import Ki.Duration (Duration)-import qualified Ki.Duration as Duration-import Ki.Prelude---- | Wait for an @STM@ action to return an @IO@ action, or if the given duration elapses, return the given @IO@ action--- instead.-timeoutSTM :: Duration -> STM (IO a) -> IO a -> IO a-timeoutSTM duration action fallback = do-  (delay, unregister) <- registerDelay (Duration.toMicroseconds duration)-  atomicallyIO (delay $> fallback <|> (unregister >>) <$> action)

@@ -0,0 +1,129 @@+module Main (main) where++import Control.Concurrent.STM (atomically)+import Control.Exception+import Control.Monad+import qualified Ki+import Test.Tasty+import Test.Tasty.HUnit+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 "`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)+            _ <-+              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)+      ]++data A = A+  deriving stock (Eq, Show)+  deriving anyclass (Exception)++data A2 = A2+  deriving stock (Eq, Show)+  deriving anyclass (Exception)++data B = B+  deriving stock (Eq, Show)++instance Exception B where+  toException = asyncExceptionToException+  fromException = asyncExceptionFromException++shouldReturn :: (Eq a, Show a) => IO a -> a -> IO ()+shouldReturn action expected = do+  actual <- action+  unless (actual == expected) (fail ("expected " ++ show expected ++ ", got " ++ show actual))++shouldThrow :: (Show a, Eq e, Exception e) => IO a -> e -> IO ()+shouldThrow action expected =+  try @SomeException action >>= \case+    Left exception | fromException exception == Just expected -> pure ()+    Left exception ->+      fail ("expected exception " ++ displayException expected ++ ", got exception " ++ displayException exception)+    Right value -> fail ("expected exception " ++ displayException expected ++ ", got " ++ show value)

@@ -1,149 +0,0 @@-{-# LANGUAGE TypeApplications #-}--module DejaFuTestUtils-  ( P,-    _failingTest,-    block,-    deadlocks,-    ignoring,-    nondeterministic,-    returns,-    test,-    throws,-    todo,-  )-where--import Control.Concurrent.Classy hiding (fork, forkWithUnmask, wait)-import Control.Exception (Exception (fromException))-import Control.Monad-import Data.Foldable-import Data.Function-import Data.List (intercalate)-import Data.Maybe-import qualified Ki.Implicit as Ki-import System.Exit-import qualified Test.DejaFu as DejaFu-import qualified Test.DejaFu.Types as DejaFu-import Text.Printf (printf)-import Prelude--type P =-  DejaFu.Program DejaFu.Basic IO--test :: (Eq a, Show a) => String -> DejaFu.Predicate a -> (Ki.Context => P a) -> IO ()-test name predicate t = do-  result <- DejaFu.runTestWithSettings settings (DejaFu.representative predicate) (Ki.withGlobalContext t)-  printf "[%s] %s\n" (if DejaFu._pass result then "x" else " ") name-  for_ (DejaFu._failures result) \(value, trace) -> prettyPrintTrace value trace-  unless (DejaFu._pass result) exitFailure-  where-    settings :: DejaFu.Settings IO a-    settings =-      DejaFu.fromWayAndMemType (DejaFu.systematically bounds) DejaFu.defaultMemType-      where-        bounds =-          DejaFu.Bounds-            { DejaFu.boundPreemp = Just 2,-              DejaFu.boundFair = Just 5-            }--_failingTest :: String -> IO (DejaFu.Result a) -> IO ()-_failingTest name action = do-  result <- action-  printf "[%s] %s\n" (if DejaFu._pass result then " " else "x") name-  when (DejaFu._pass result) exitFailure--todo :: String -> IO ()-todo =-  printf "[-] %s\n"--deadlocks :: DejaFu.Predicate a-deadlocks =-  DejaFu.alwaysTrue \case-    Left DejaFu.Deadlock -> True-    _ -> False--nondeterministic :: (Eq a, Show a) => [Either DejaFu.Condition a] -> DejaFu.Predicate a-nondeterministic =-  DejaFu.gives--returns :: Eq a => a -> DejaFu.Predicate a-returns expected =-  DejaFu.alwaysTrue \case-    Left _ -> False-    Right actual -> actual == expected--throws :: (Eq e, Exception e) => e -> DejaFu.Predicate a-throws expected =-  DejaFu.alwaysTrue \case-    Left (DejaFu.UncaughtException actual) -> fromException actual == Just expected-    _ -> False------block :: P ()-block =-  newEmptyMVar >>= takeMVar--ignoring :: forall e. Exception e => P () -> P ()-ignoring action =-  catch @_ @e action \_ -> pure ()------prettyPrintTrace :: Show a => Either DejaFu.Condition a -> DejaFu.Trace -> IO ()-prettyPrintTrace value trace = do-  print value-  flip fix trace \loop -> \case-    [] -> pure ()-    (decision, _, action) : xs -> do-      case decision of-        DejaFu.Start n -> putStrLn ("  [" ++ prettyThreadId n ++ "]")-        DejaFu.SwitchTo n -> putStrLn ("  [" ++ prettyThreadId n ++ "]")-        DejaFu.Continue -> pure ()-      putStrLn ("    " ++ prettyThreadAction action)-      loop xs--prettyThreadAction :: DejaFu.ThreadAction -> String-prettyThreadAction = \case-  DejaFu.BlockedSTM actions -> "atomically " ++ show actions ++ " (blocked)"-  DejaFu.BlockedTakeMVar n -> "takeMVar " ++ prettyMVarId n ++ " (blocked)"-  DejaFu.BlockedThrowTo n -> "throwTo " ++ prettyThreadId n ++ " (blocked)"-  DejaFu.Fork n -> "fork " ++ prettyThreadId n-  DejaFu.MyThreadId -> "myThreadId"-  DejaFu.NewIORef n -> prettyIORefId n ++ " <- newIORef"-  DejaFu.NewMVar n -> prettyMVarId n ++ " <- newMVar"-  DejaFu.PutMVar n [] -> "putMVar " ++ prettyMVarId n-  DejaFu.PutMVar n ts ->-    "putMVar " ++ prettyMVarId n ++ " (waking "-      ++ intercalate ", " (map prettyThreadId ts)-      ++ ")"-  DejaFu.ReadIORef n -> "readIORef " ++ prettyIORefId n-  DejaFu.ResetMasking _ state -> "setMaskingState " ++ show state-  DejaFu.Return -> "pure"-  DejaFu.STM actions _ -> "atomically " ++ show actions-  DejaFu.SetMasking _ state -> "setMaskingState " ++ show state-  DejaFu.Stop -> "stop"-  DejaFu.TakeMVar n [] -> "takeMVar " ++ prettyMVarId n-  DejaFu.TakeMVar n ts ->-    "takeMVar " ++ prettyMVarId n ++ " (waking "-      ++ intercalate ", " (map prettyThreadId ts)-      ++ ")"-  DejaFu.Throw Nothing -> "throw (thread died)"-  DejaFu.Throw (Just _) -> "throw (thread still alive)"-  DejaFu.ThrowTo n Nothing -> "throwTo " ++ prettyThreadId n ++ " (killed it)"-  DejaFu.ThrowTo n (Just _) -> "throwTo " ++ prettyThreadId n ++ " (didn't kill it)"-  action -> show action--prettyIORefId :: DejaFu.IORefId -> String-prettyIORefId n =-  "ioref#" ++ show n--prettyMVarId :: DejaFu.MVarId -> String-prettyMVarId n =-  "mvar#" ++ show n--prettyThreadId :: DejaFu.ThreadId -> String-prettyThreadId n =-  "thread#" ++ show n

@@ -1,65 +0,0 @@-{-# LANGUAGE PatternSynonyms #-}-{-# LANGUAGE TypeApplications #-}--module Main-  ( main,-  )-where--import Control.Concurrent.Classy hiding (fork, forkWithUnmask, wait)-import Control.Exception-  ( Exception (..),-    asyncExceptionFromException,-    asyncExceptionToException,-  )-import DejaFuTestUtils-import Ki.Implicit-import Prelude--main :: IO ()-main = do-  test "`waitFor` waits for a duration" (nondeterministic [Right False, Right True]) do-    ref <- newIORef False-    scoped \scope -> do-      fork_ scope (writeIORef ref True)-      waitFor scope 1-    readIORef ref--  {- seems like a dejafu bug-  test "`fork` propagates async exceptions to parent" (throws ThreadKilled) do-    scoped \scope -> do-      var <- newEmptyMVar-      fork_ scope do-        myThreadId >>= putMVar var-        block-      takeMVar var >>= killThread-      block-  -}--  test "`scoped` closing while `fork` propagating never deadlocks" (nondeterministic [Right False, Right True]) do-    ref <- newIORef False-    catch-      (scoped \scope -> fork_ scope (throw A))-      (\A -> writeIORef ref True)-    readIORef ref--  test "thread waiting on its own scope deadlocks" deadlocks do-    scoped \scope -> do-      fork_ scope (wait scope)-      wait scope--  test "thread waiting on its own scope allows async exceptions" (returns ()) do-    scoped \scope -> fork_ scope (wait scope)--data A-  = A-  deriving stock (Eq, Show)-  deriving anyclass (Exception)--data B-  = B-  deriving stock (Eq, Show)--instance Exception B where-  toException = asyncExceptionToException-  fromException = asyncExceptionFromException

@@ -1,77 +0,0 @@-{-# LANGUAGE TypeApplications #-}--module TestUtils-  ( ignoring,-    fail,-    shouldReturn,-    shouldReturnSuchThat,-    shouldThrow,-    shouldThrowSuchThat,-    test,-  )-where--import Control.Exception-import Control.Monad (unless)-import Data.Either (isRight)-import qualified Ki.Implicit as Ki-import System.Exit (exitFailure)-import Text.Printf (printf)-import Prelude hiding (fail)--newtype TestFailure-  = TestFailure String-  deriving stock (Show)--instance Exception TestFailure where-  displayException (TestFailure message) =-    message--ignoring :: forall e. Exception e => IO () -> IO ()-ignoring action =-  catch action \(_ :: e) -> pure ()--fail :: String -> IO ()-fail =-  throwIO . TestFailure--shouldReturn :: (Eq a, Show a) => IO a -> a -> IO ()-shouldReturn action expected = do-  actual <- action-  unless (actual == expected) (fail ("expected " ++ show expected ++ ", got " ++ show actual))--shouldReturnSuchThat :: Show a => IO a -> (a -> Bool) -> IO ()-shouldReturnSuchThat action predicate = do-  result <- action-  unless (predicate result) (fail ("result " ++ show result ++ " did not pass predicate"))--shouldThrow :: (Show a, Eq e, Exception e) => IO a -> e -> IO ()-shouldThrow action expected =-  try @SomeException action >>= \case-    Left exception | fromException exception == Just expected -> pure ()-    Left exception ->-      fail ("expected exception " ++ displayException expected ++ ", got exception " ++ displayException exception)-    Right value -> fail ("expected exception " ++ displayException expected ++ ", got " ++ show value)--shouldThrowSuchThat :: (Show a, Exception e) => IO a -> (e -> Bool) -> IO ()-shouldThrowSuchThat action predicate =-  try @SomeException action >>= \case-    Left exception ->-      case fromException exception of-        Nothing ->-          fail ("expected exception, got exception " ++ displayException exception)-        Just exception' ->-          unless-            (predicate exception')-            (fail ("exception " ++ displayException exception' ++ " did not pass predicate"))-    Right value -> fail ("expected exception, got " ++ show value)--test :: String -> (Ki.Context => IO ()) -> IO ()-test name action = do-  result <- try @SomeException (Ki.withGlobalContext action)-  printf "[%s] %s\n" (if isRight result then "x" else " ") name-  case result of-    Left exception -> do-      putStrLn (displayException exception)-      exitFailure-    Right () -> pure ()

@@ -1,232 +0,0 @@-{-# LANGUAGE PatternSynonyms #-}-{-# LANGUAGE TypeApplications #-}--module Main (main) where--import Control.Concurrent-import Control.Concurrent.STM-import Control.Exception-import Control.Monad (when)-import Data.Functor-import Data.IORef-import Data.Maybe-import qualified Ki.Implicit as Ki-import qualified Ki.Internal-import TestUtils-import Prelude hiding (fail)--main :: IO ()-main = do-  test "background context isn't cancelled" do-    (isJust <$> Ki.cancelled) `shouldReturn` False--  test "new scope doesn't start out cancelled" do-    Ki.scoped \_ -> (isJust <$> Ki.cancelled) `shouldReturn` False--  test "`cancel` observable by scope's `cancelled`" do-    Ki.scoped \scope -> do-      Ki.cancel scope-      (isJust <$> Ki.cancelled) `shouldReturn` True--  test "`cancel` observable by inner scope's `cancelled`" do-    Ki.scoped \scope ->-      Ki.scoped \_ -> do-        Ki.cancel scope-        (isJust <$> Ki.cancelled) `shouldReturn` True--  childtest "`cancel` observable by child's `cancelled`" \fork -> do-    ref <- newIORef Nothing-    Ki.scoped \scope -> do-      fork scope do-        Ki.cancel scope-        Ki.cancelled >>= writeIORef ref-      Ki.wait scope-    (isJust <$> readIORef ref) `shouldReturn` True--  childtest "`cancel` observable by grandchild's `cancelled`" \fork -> do-    ref <- newIORef Nothing-    Ki.scoped \scope1 -> do-      fork scope1 do-        Ki.scoped \scope2 -> do-          fork scope2 do-            Ki.cancel scope1-            Ki.cancelled >>= writeIORef ref-          Ki.wait scope2-      Ki.wait scope1-    (isJust <$> readIORef ref) `shouldReturn` True--  test "inner scope inherits cancellation" do-    Ki.scoped \scope1 -> do-      Ki.cancel scope1-      Ki.scoped \_ -> (isJust <$> Ki.cancelled) `shouldReturn` True--  childtest "child thread inherits cancellation" \fork -> do-    ref <- newIORef Nothing-    Ki.scoped \scope -> do-      Ki.cancel scope-      fork scope (Ki.cancelled >>= writeIORef ref)-      Ki.wait scope-    (isJust <$> readIORef ref) `shouldReturn` True--  childtest "creating a child thread throws ErrorCall when the scope is closed" \fork -> do-    scope <- Ki.scoped pure-    fork scope (pure ()) `shouldThrow` ErrorCall "ki: scope closed"--  test "cancelled child context removes parent's ref to it" do-    ctx0 <- atomically Ki.Internal.newCtxSTM-    ctx1 <- atomically (Ki.Internal.deriveCtx ctx0)-    (length <$> readTVarIO (Ki.Internal.childrenVar ctx0)) `shouldReturn` 1-    Ki.Internal.cancelCtx ctx1-    (length <$> readTVarIO (Ki.Internal.childrenVar ctx0)) `shouldReturn` 0--  test "`wait` succeeds when no threads are alive" do-    Ki.scoped Ki.wait--  childtest "creates a thread" \fork -> do-    parentThreadId <- myThreadId-    Ki.scoped \scope -> do-      fork scope do-        childThreadId <- myThreadId-        when (parentThreadId == childThreadId) (fail "didn't create a thread")-      Ki.wait scope--  forktest "propagates sync exceptions" \fork ->-    ( Ki.scoped \scope -> do-        fork scope (throwIO A)-        Ki.wait scope-    )-      `shouldThrow` A--  forktest "propagates async exceptions" \fork ->-    ( Ki.scoped \scope -> do-        fork scope (throwIO B)-        Ki.wait scope-    )-      `shouldThrow` B--  forktest "doesn't propagate own cancel token exceptions" \fork ->-    Ki.scoped \scope -> do-      Ki.cancel scope-      fork scope (atomically Ki.cancelledSTM >>= throwIO)-      Ki.wait scope--  forktest "propagates ScopeClosing if it isn't ours" \fork ->-    ( Ki.scoped \scope -> do-        fork scope (throwIO Ki.Internal.ScopeClosing)-        Ki.wait scope-    )-      `shouldThrow` Ki.Internal.ScopeClosing--  forktest "propagates others' cancel token exceptions" \fork ->-    ( Ki.scoped \scope -> do-        Ki.cancel scope-        fork scope (throwIO (Ki.Internal.CancelToken 0))-        Ki.wait scope-    )-      `shouldThrow` Ki.Internal.CancelToken 0--  test "`async` returns sync exceptions" do-    Ki.scoped \scope -> do-      result <- Ki.async @() scope (throw A)-      Ki.await result `shouldReturnSuchThat` \case-        Left (fromException -> Just A) -> True-        _ -> False--  test "`async` returns async exceptions" do-    Ki.scoped \scope -> do-      result <- Ki.async @() scope (throw B)-      Ki.await result `shouldReturnSuchThat` \case-        Left (fromException -> Just B) -> True-        _ -> False--  test "awaiting a failed `fork`ed thread blocks" do-    Ki.scoped \scope -> do-      mask \unmask -> do-        thread <- Ki.fork @() scope (throw A)-        unmask (Ki.wait scope) `catch` \(Ki.Internal.ThreadFailed _) -> pure ()-        Ki.await thread `shouldThrowSuchThat` \case-          (fromException -> Just BlockedIndefinitelyOnSTM) -> True-          _ -> False--  test "awaiting a failed `fork`ed thread blocks" do-    Ki.scoped \scope -> do-      mask \unmask -> do-        thread <- Ki.fork @() scope (throw B)-        unmask (Ki.wait scope) `catch` \(Ki.Internal.ThreadFailed _) -> pure ()-        Ki.await thread `shouldThrowSuchThat` \case-          (fromException -> Just BlockedIndefinitelyOnSTM) -> True-          _ -> False--  childtest "inherits masking state" \fork -> do-    Ki.scoped \scope -> do-      fork scope (getMaskingState `shouldReturn` Unmasked)-      mask_ (fork scope (getMaskingState `shouldReturn` MaskedInterruptible))-      uninterruptibleMask_ (fork scope (getMaskingState `shouldReturn` MaskedUninterruptible))-      Ki.wait scope--  test "provides an unmasking function (`forkWithUnmask`)" do-    Ki.scoped \scope -> do-      _thread <- mask_ (Ki.forkWithUnmask scope \unmask -> unmask getMaskingState `shouldReturn` Unmasked)-      Ki.wait scope--  test "provides an unmasking function (`forkWithUnmask_`)" do-    Ki.scoped \scope -> do-      mask_ (Ki.forkWithUnmask_ scope \unmask -> unmask getMaskingState `shouldReturn` Unmasked)-      Ki.wait scope--  test "provides an unmasking function (`asyncWithUnmask`)" do-    Ki.scoped \scope -> do-      _thread <- (Ki.asyncWithUnmask scope \unmask -> unmask getMaskingState `shouldReturn` Unmasked)-      Ki.wait scope--  test "thread can be awaited after its scope closes" do-    thread <--      Ki.scoped \scope -> do-        thread <- Ki.fork scope (pure ())-        Ki.wait scope-        pure thread-    Ki.await thread `shouldReturn` ()--  test "parent kills children when it throws" do-    ref <- newIORef False-    ignoring @A do-      Ki.scoped \scope -> do-        var <- newEmptyMVar-        uninterruptibleMask_ do-          Ki.forkWithUnmask_ scope \unmask -> do-            putMVar var ()-            unmask (threadDelay 1_000_000) `onException` writeIORef ref True-        takeMVar var-        void (throw A)-    readIORef ref `shouldReturn` True--  test "parent kills children when child throws" do-    ref <- newIORef False-    ignoring @A do-      Ki.scoped \scope -> do-        uninterruptibleMask_ do-          (Ki.forkWithUnmask_ scope \unmask -> unmask (threadDelay 1_000_000) `onException` writeIORef ref True)-        Ki.fork_ scope (throw A)-        Ki.wait scope-    readIORef ref `shouldReturn` True--forktest :: String -> (Ki.Context => (Ki.Scope -> (Ki.Context => IO ()) -> IO ()) -> IO ()) -> IO ()-forktest name theTest = do-  test (name ++ " (`fork`)") (theTest \scope action -> void (Ki.fork scope action))-  test (name ++ " (`fork_`)") (theTest Ki.fork_)--childtest :: String -> (Ki.Context => (Ki.Scope -> (Ki.Context => IO ()) -> IO ()) -> IO ()) -> IO ()-childtest name theTest = do-  forktest name theTest-  test (name ++ " (`async`)") (theTest \scope action -> void (Ki.async scope action))--data A = A-  deriving stock (Eq, Show)-  deriving anyclass (Exception)--data B = B-  deriving stock (Eq, Show)--instance Exception B where-  toException = asyncExceptionToException-  fromException = asyncExceptionFromException