diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -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: diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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) | diff --git a/ki.cabal b/ki.cabal --- a/ki.cabal +++ b/ki.cabal @@ -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 +maintainer: Mitchell Rosen , Travis Staton 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 - - 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 + @@, + see @@. . - @ - 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 - diff --git a/src/Ki.hs b/src/Ki.hs --- a/src/Ki.hs +++ b/src/Ki.hs @@ -1,82 +1,110 @@ +-- | `ki` is a lightweight structured concurrency library. +-- +-- For a variant of this API generalized to +-- @@, +-- see @@. +-- +-- 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, +-- . +-- +-- ==== __👉 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 +-- @ diff --git a/src/Ki/CancelToken.hs b/src/Ki/CancelToken.hs deleted file mode 100644 --- a/src/Ki/CancelToken.hs +++ /dev/null @@ -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 diff --git a/src/Ki/Concurrency.hs b/src/Ki/Concurrency.hs deleted file mode 100644 --- a/src/Ki/Concurrency.hs +++ /dev/null @@ -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 diff --git a/src/Ki/Context.hs b/src/Ki/Context.hs deleted file mode 100644 --- a/src/Ki/Context.hs +++ /dev/null @@ -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 - } diff --git a/src/Ki/Ctx.hs b/src/Ki/Ctx.hs deleted file mode 100644 --- a/src/Ki/Ctx.hs +++ /dev/null @@ -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 diff --git a/src/Ki/Debug.hs b/src/Ki/Debug.hs deleted file mode 100644 --- a/src/Ki/Debug.hs +++ /dev/null @@ -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 #-} diff --git a/src/Ki/Duration.hs b/src/Ki/Duration.hs deleted file mode 100644 --- a/src/Ki/Duration.hs +++ /dev/null @@ -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) diff --git a/src/Ki/Implicit.hs b/src/Ki/Implicit.hs deleted file mode 100644 --- a/src/Ki/Implicit.hs +++ /dev/null @@ -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 diff --git a/src/Ki/Internal.hs b/src/Ki/Internal.hs deleted file mode 100644 --- a/src/Ki/Internal.hs +++ /dev/null @@ -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 diff --git a/src/Ki/Internal/ByteCount.hs b/src/Ki/Internal/ByteCount.hs new file mode 100644 --- /dev/null +++ b/src/Ki/Internal/ByteCount.hs @@ -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) diff --git a/src/Ki/Internal/Counter.hs b/src/Ki/Internal/Counter.hs new file mode 100644 --- /dev/null +++ b/src/Ki/Internal/Counter.hs @@ -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 #-} diff --git a/src/Ki/Internal/Prelude.hs b/src/Ki/Internal/Prelude.hs new file mode 100644 --- /dev/null +++ b/src/Ki/Internal/Prelude.hs @@ -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 #) diff --git a/src/Ki/Internal/Scope.hs b/src/Ki/Internal/Scope.hs new file mode 100644 --- /dev/null +++ b/src/Ki/Internal/Scope.hs @@ -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)) diff --git a/src/Ki/Internal/Thread.hs b/src/Ki/Internal/Thread.hs new file mode 100644 --- /dev/null +++ b/src/Ki/Internal/Thread.hs @@ -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)) diff --git a/src/Ki/Prelude.hs b/src/Ki/Prelude.hs deleted file mode 100644 --- a/src/Ki/Prelude.hs +++ /dev/null @@ -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 diff --git a/src/Ki/Scope.hs b/src/Ki/Scope.hs deleted file mode 100644 --- a/src/Ki/Scope.hs +++ /dev/null @@ -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 diff --git a/src/Ki/Thread.hs b/src/Ki/Thread.hs deleted file mode 100644 --- a/src/Ki/Thread.hs +++ /dev/null @@ -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 diff --git a/src/Ki/Timeout.hs b/src/Ki/Timeout.hs deleted file mode 100644 --- a/src/Ki/Timeout.hs +++ /dev/null @@ -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) diff --git a/test/Tests.hs b/test/Tests.hs new file mode 100644 --- /dev/null +++ b/test/Tests.hs @@ -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) diff --git a/test/dejafu-tests/DejaFuTestUtils.hs b/test/dejafu-tests/DejaFuTestUtils.hs deleted file mode 100644 --- a/test/dejafu-tests/DejaFuTestUtils.hs +++ /dev/null @@ -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 diff --git a/test/dejafu-tests/DejaFuTests.hs b/test/dejafu-tests/DejaFuTests.hs deleted file mode 100644 --- a/test/dejafu-tests/DejaFuTests.hs +++ /dev/null @@ -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 diff --git a/test/unit-tests/TestUtils.hs b/test/unit-tests/TestUtils.hs deleted file mode 100644 --- a/test/unit-tests/TestUtils.hs +++ /dev/null @@ -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 () diff --git a/test/unit-tests/Tests.hs b/test/unit-tests/Tests.hs deleted file mode 100644 --- a/test/unit-tests/Tests.hs +++ /dev/null @@ -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