packages feed

ki (empty) → 0.1.0

raw patch · 23 files changed

+2084/−0 lines, 23 filesdep +atomic-primopsdep +basedep +concurrency

Dependencies added: atomic-primops, base, concurrency, containers, dejafu, ki, stm

Files

+ CHANGELOG.md view
@@ -0,0 +1,11 @@+# Changelog++All notable changes to this project will be documented in this file.++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/).++## [0.1.0] - 2020-11-11++### Added+- Initial release
+ LICENSE view
@@ -0,0 +1,11 @@+Copyright 2020 Mitchell Rosen++Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:++1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.++2. 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.++3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,155 @@+# 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)+++`ki` is a lightweight structured-concurrency library inspired by many other projects:++* [`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+cancelScope :: 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)++  -- Fall through the bottom of `scoped`, which throws hard-cancels all+  -- remaining threads by throwing each one an asynchronous exceptions+```++### Testing++(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.++Nonetheless this library should not considered production-ready!++## Recommended reading++In chronological order of publication,++  * 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.cabal view
@@ -0,0 +1,153 @@+cabal-version: 3.0++author: Mitchell Rosen+bug-reports: https://github.com/mitchellwrosen/ki/issues+category: Concurrency+copyright: Copyright (C) 2020 Mitchell Rosen+homepage: https://github.com/mitchellwrosen/ki+license: BSD-3-Clause+license-file: LICENSE+maintainer: Mitchell Rosen <mitchellwrosen@gmail.com>+name: ki+stability: experimental+synopsis: A lightweight, structured concurrency library+version: 0.1.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".++  Because you'll only ever need one variant at a time, I recommend using a+  <https://www.haskell.org/cabal/users-guide/developing-packages.html#pkg-field-mixins mixin stanza>+  to rename one module to @Ki@ while hiding the others. This also simplifies the+  process of upgrading from "Ki.Implicit" to "Ki" if necessary.++  @+  mixins: ki (Ki.Implicit as Ki)+  @++extra-source-files:+  CHANGELOG.md+  README.md++source-repository head+  type: git+  location: https://github.com/mitchellwrosen/ki.git++common component+  default-extensions:+    AllowAmbiguousTypes+    BlockArguments+    ConstraintKinds+    DeriveAnyClass+    DeriveDataTypeable+    DeriveFunctor+    DeriveGeneric+    DerivingStrategies+    DuplicateRecordFields+    ExistentialQuantification+    GeneralizedNewtypeDeriving+    ImplicitParams+    LambdaCase+    NamedFieldPuns+    NoImplicitPrelude+    RankNTypes+    RoleAnnotations+    ScopedTypeVariables+    ViewPatterns+  default-language: Haskell2010+  ghc-options:+    -Weverything+    -Wno-all-missed-specialisations+    -Wno-implicit-prelude+    -Wno-missed-specialisations+    -Wno-missing-import-lists+    -Wno-safe+    -Wno-unsafe+  if impl(ghc >= 8.10)+    ghc-options:+      -Wno-missing-safe-haskell-mode+      -Wno-prepositive-qualified-module++flag test+  description: Internal flag used by DejaFu test suite+  default: False+  manual: True++library+  import: component+  build-depends:+    base >= 4.12.0.0 && < 4.15,+    containers,+    stm,+  exposed-modules:+    Ki,+    Ki.Implicit,+    Ki.Internal,+  hs-source-dirs: src+  other-modules:+    Ki.CancelToken+    Ki.Concurrency+    Ki.Context+    Ki.Ctx+    Ki.Debug+    Ki.Duration+    Ki.Prelude+    Ki.Scope+    Ki.ScopeClosing+    Ki.Thread+    Ki.ThreadFailed+    Ki.Timeout+  if flag(test)+    build-depends:+      concurrency ^>= 1.11.0.0,+      dejafu ^>= 2.4.0.0,+    cpp-options: -DTEST+  else+    build-depends:+      atomic-primops,++test-suite dejafu-tests+  import: component+  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+  build-depends:+    base,+    ki,+    stm,+  ghc-options: -rtsopts -threaded+  hs-source-dirs: test/unit-tests+  main-is: Tests.hs+  other-modules:+    TestUtils+  type: exitcode-stdio-1.0+
+ src/Ki.hs view
@@ -0,0 +1,113 @@+module Ki+  ( -- * Scope+    Scope,+    scoped,+    Scope.wait,+    Scope.waitSTM,+    Scope.waitFor,++    -- * Spawning threads+    -- $spawning-threads+    Thread,++    -- ** Fork+    Thread.fork,+    Thread.fork_,+    Thread.forkWithUnmask,+    Thread.forkWithUnmask_,++    -- ** Async+    Thread.async,+    Thread.asyncWithUnmask,++    -- ** Await+    await,+    awaitSTM,+    awaitFor,++    -- * Miscellaneous+    Duration,+    Duration.microseconds,+    Duration.milliseconds,+    Duration.seconds,+    Timeout.timeoutSTM,+    sleep,++    -- * Exceptions+    ThreadFailed (..),+  )+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 Ki.ThreadFailed (ThreadFailed (..))+import qualified Ki.Timeout as Timeout++-- | Wait for a __thread__ to finish.+--+-- /Throws/:+--+--   * 'ThreadFailed' if the __thread__ threw an exception and was created with 'Ki.fork'.+await :: Thread a -> IO a+await =+  Thread.await++-- | @STM@ variant of 'await'.+--+-- /Throws/:+--+--   * 'ThreadFailed' if the __thread__ threw an exception and was created with 'Ki.fork'.+awaitSTM :: Thread a -> STM a+awaitSTM =+  Thread.awaitSTM++-- | Variant of 'await' that waits for up to the given duration.+--+-- /Throws/:+--+--   * 'ThreadFailed' if the __thread__ threw an exception and was created with 'Ki.fork'.+awaitFor :: Thread a -> Duration -> IO (Maybe a)+awaitFor =+  Thread.awaitFor++-- $spawning-threads+--+-- There are two variants of __thread__-creating functions with different exception-propagation semantics.+--+-- * 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__.+--+-- * 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__.+--+-- 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.+--   * 'ThreadFailed' containing the first exception a __thread__ created with 'Ki.fork' throws, if any.+--+-- ==== __Examples__+--+-- @+-- '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)
+ src/Ki/CancelToken.hs view
@@ -0,0 +1,18 @@+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
+ src/Ki/Concurrency.hs view
@@ -0,0 +1,187 @@+{-# 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.Atomics.Counter+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 =+  incrCounter 1 counter++counter :: AtomicCounter+counter =+  unsafePerformIO (newCounter 0)+{-# NOINLINE counter #-}++#endif
+ src/Ki/Context.hs view
@@ -0,0 +1,49 @@+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+    }
+ src/Ki/Ctx.hs view
@@ -0,0 +1,92 @@+{-# 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
+ src/Ki/Debug.hs view
@@ -0,0 +1,22 @@+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 #-}
+ src/Ki/Duration.hs view
@@ -0,0 +1,36 @@+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)
+ src/Ki/Implicit.hs view
@@ -0,0 +1,229 @@+{-# 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+    await,+    awaitSTM,+    awaitFor,++    -- * Soft-cancellation+    Scope.cancelScope,+    cancelled,+    cancelledSTM,+    CancelToken,++    -- * Miscellaneous+    Duration,+    Duration.microseconds,+    Duration.milliseconds,+    Duration.seconds,+    timeoutSTM,+    sleep,++    -- * Exceptions+    ThreadFailed (..),+  )+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.ThreadFailed (ThreadFailed (..))+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__ to compute a value concurrently.+--+-- /Throws/:+--+--   * Calls 'error' if the __scope__ is /closed/.+async :: Scope -> (Context => IO a) -> IO (Thread (Either ThreadFailed 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 ThreadFailed a))+asyncWithUnmask scope action =+  Thread.asyncWithUnmask scope (let ?context = Scope.context scope in action)++-- | Wait for a __thread__ to finish.+--+-- /Throws/:+--+--   * 'ThreadFailed' if the __thread__ threw an exception and was created with 'fork'.+await :: Thread a -> IO a+await =+  Thread.await++-- | @STM@ variant of 'await'.+--+-- /Throws/:+--+--   * 'ThreadFailed' if the __thread__ threw an exception and was created with 'fork'.+awaitSTM :: Thread a -> STM a+awaitSTM =+  Thread.awaitSTM++-- | Variant of 'await' that waits for up to the given duration.+--+-- /Throws/:+--+--   * 'ThreadFailed' if the __thread__ threw an exception and was created with 'fork'.+awaitFor :: Thread a -> Duration -> IO (Maybe a)+awaitFor =+  Thread.awaitFor++-- | 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__ to compute a value concurrently.+--+-- If the __thread__ throws an exception, the exception is wrapped in 'ThreadFailed' and 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.+--   * 'ThreadFailed' containing the first exception a __thread__ created with 'fork' throws, 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 /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
+ src/Ki/Internal.hs view
@@ -0,0 +1,22 @@+module Ki.Internal+  ( module Ki.CancelToken,+    module Ki.Context,+    module Ki.Ctx,+    module Ki.Duration,+    module Ki.Scope,+    module Ki.ScopeClosing,+    module Ki.Thread,+    module Ki.ThreadFailed,+    module Ki.Timeout,+  )+where++import Ki.CancelToken+import Ki.Context+import Ki.Ctx+import Ki.Duration+import Ki.Scope+import Ki.ScopeClosing+import Ki.Thread+import Ki.ThreadFailed+import Ki.Timeout
+ src/Ki/Prelude.hs view
@@ -0,0 +1,47 @@+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
+ src/Ki/Scope.hs view
@@ -0,0 +1,172 @@+{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE TypeApplications #-}++module Ki.Scope+  ( Scope (..),+    cancelScope,+    scopeCancelledSTM,+    scopeFork,+    scoped,+    wait,+    waitFor,+    waitSTM,+  )+where++import Control.Exception (fromException, 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.ScopeClosing (ScopeClosing (..))+import Ki.ThreadFailed (ThreadFailedAsync (..))+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__.+cancelScope :: Scope -> IO ()+cancelScope 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) -> (ThreadId -> 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 childThreadId 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 'AsyncThreadFailed' (assumed to have come from one of our children).+    throw :: SomeException -> IO a+    throw exception =+      case fromException exception of+        Just (ThreadFailedAsync 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)++--------------------------------------------------------------------------------+-- 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
+ src/Ki/ScopeClosing.hs view
@@ -0,0 +1,16 @@+module Ki.ScopeClosing+  ( ScopeClosing (..),+  )+where++import Control.Exception (Exception (..), asyncExceptionFromException, asyncExceptionToException)+import Ki.Prelude++-- | 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
+ src/Ki/Thread.hs view
@@ -0,0 +1,166 @@+module Ki.Thread+  ( Thread,+    async,+    asyncWithUnmask,+    await,+    awaitSTM,+    awaitFor,+    fork,+    fork_,+    forkWithUnmask,+    forkWithUnmask_,+  )+where++import Control.Exception (Exception (fromException))+import Data.Bifunctor (first)+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.ScopeClosing (ScopeClosing (ScopeClosing))+import Ki.ThreadFailed (ThreadFailed (ThreadFailed), ThreadFailedAsync (ThreadFailedAsync))+import Ki.Timeout (timeoutSTM)++-- | A running __thread__.+data Thread a = Thread+  { threadId :: !ThreadId,+    action :: !(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__ to compute a value concurrently.+--+-- /Throws/:+--+--   * Calls 'error' if the __scope__ is /closed/.+async :: Scope -> IO a -> IO (Thread (Either ThreadFailed 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 ThreadFailed 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 ThreadFailed a))+asyncWithRestore scope action = do+  resultVar <- newEmptyTMVarIO+  childThreadId <-+    Scope.scopeFork scope action \childThreadId result ->+      putTMVarIO resultVar (first (ThreadFailed childThreadId) result)+  pure (Thread childThreadId (readTMVar resultVar))++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 duration =+--   'timeout' duration (pure . Just \<$\> 'awaitSTM' thread) (pure Nothing)+-- @+awaitFor :: Thread a -> Duration -> IO (Maybe a)+awaitFor thread duration =+  timeoutSTM duration (pure . Just <$> awaitSTM thread) (pure Nothing)++-- | Create a __thread__ within a __scope__ to compute a value concurrently.+--+-- If the __thread__ throws an exception, the exception is wrapped in 'ThreadFailed' and 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 \childThreadId -> \case+      Left exception -> do+        whenM+          (shouldPropagateException scope exception)+          (throwTo parentThreadId (ThreadFailedAsync threadFailedException))+        putTMVarIO resultVar (Left threadFailedException)+        where+          threadFailedException :: ThreadFailed+          threadFailedException =+            ThreadFailed childThreadId exception+      Right result -> putTMVarIO resultVar (Right result)+  pure (Thread childThreadId (readTMVar resultVar >>= either throwSTM pure))++forkWithRestore_ :: Scope -> ((forall x. IO x -> IO x) -> IO ()) -> IO ()+forkWithRestore_ scope action = do+  parentThreadId <- myThreadId+  _childThreadId <-+    Scope.scopeFork scope action \childThreadId ->+      onLeft \exception -> do+        whenM+          (shouldPropagateException scope exception)+          (throwTo parentThreadId (ThreadFailedAsync (ThreadFailed childThreadId exception)))+  pure ()++shouldPropagateException :: Scope -> SomeException -> IO Bool+shouldPropagateException Scope {closedVar, context} exception =+  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 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
+ src/Ki/ThreadFailed.hs view
@@ -0,0 +1,27 @@+module Ki.ThreadFailed+  ( ThreadFailed (..),+    ThreadFailedAsync (..),+  )+where++import Control.Exception (Exception (..), asyncExceptionFromException, asyncExceptionToException)+import Ki.Prelude++-- | A __thread__ failed, either by throwing or being thrown an exception.+data ThreadFailed = ThreadFailed+  { threadId :: ThreadId,+    exception :: SomeException+  }+  deriving stock (Show)+  deriving anyclass (Exception)++-- | An async wrapper around 'ThreadFailed', used when a child __thread__ communicates its failure to its parent. This+-- is preferred to throwing 'ThreadFailed' directly, so that client code (outside of this library) can follow+-- best-practices when encountering a mysterious async exception: clean up resources and re-throw it.+newtype ThreadFailedAsync+  = ThreadFailedAsync ThreadFailed+  deriving stock (Show)++instance Exception ThreadFailedAsync where+  toException = asyncExceptionToException+  fromException = asyncExceptionFromException
+ src/Ki/Timeout.hs view
@@ -0,0 +1,15 @@+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)
+ test/dejafu-tests/DejaFuTestUtils.hs view
@@ -0,0 +1,149 @@+{-# 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
+ test/dejafu-tests/DejaFuTests.hs view
@@ -0,0 +1,111 @@+{-# LANGUAGE PatternSynonyms #-}+{-# LANGUAGE TypeApplications #-}++module Main+  ( main,+  )+where++import Control.Concurrent.Classy hiding (fork, forkWithUnmask, wait)+import Control.Exception+  ( Exception (..),+    SomeException (..),+    asyncExceptionFromException,+    asyncExceptionToException,+  )+import Control.Monad+import Data.Maybe+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+  -}++  -- TODO move to unit test+  test "`scoped` kills threads when it throws" (returns True) do+    ref <- newIORef False+    ignoring @A do+      scoped \scope -> do+        var <- newEmptyMVar+        uninterruptibleMask_ do+          forkWithUnmask_ scope \unmask -> do+            putMVar var ()+            unmask block `onException` writeIORef ref True+        takeMVar var+        void (throw A)+    readIORef ref++  -- TODO move to unit test+  test "`scoped` kills threads when `fork` throws" (returns True) do+    ref <- newIORef False+    catch+      ( scoped \scope -> do+          uninterruptibleMask_ (forkWithUnmask_ scope \unmask -> unmask block `onException` writeIORef ref True)+          fork_ scope (throw A)+          wait scope+          pure False+      )+      ( \exception ->+          case fromException exception of+            Just (ThreadFailed _threadId (fromException -> Just A)) -> readIORef ref+            _ -> pure False+      )++  test "`scoped` closing while `fork` propagating never deadlocks" (nondeterministic [Right False, Right True]) do+    ref <- newIORef False+    catch+      (scoped \scope -> fork_ scope (throw A))+      (\(ThreadFailed _threadId _exception) -> 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++-- finally :: P a -> P b -> P a+-- finally action after =+--   mask \restore -> do+--     result <- restore action `onException` after+--     _ <- after+--     pure result++onException :: P a -> P b -> P a+onException action cleanup =+  catch @_ @SomeException action \ex -> do+    _ <- cleanup+    throw ex
+ test/unit-tests/TestUtils.hs view
@@ -0,0 +1,72 @@+{-# LANGUAGE TypeApplications #-}++module TestUtils+  ( 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++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 ()
+ test/unit-tests/Tests.hs view
@@ -0,0 +1,211 @@+{-# 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 "`cancelScope` observable by scope's `cancelled`" do+    Ki.scoped \scope -> do+      Ki.cancelScope scope+      (isJust <$> Ki.cancelled) `shouldReturn` True++  test "`cancelScope` observable by inner scope's `cancelled`" do+    Ki.scoped \scope ->+      Ki.scoped \_ -> do+        Ki.cancelScope scope+        (isJust <$> Ki.cancelled) `shouldReturn` True++  childtest "`cancelScope` observable by child's `cancelled`" \fork -> do+    ref <- newIORef Nothing+    Ki.scoped \scope -> do+      fork scope do+        Ki.cancelScope scope+        Ki.cancelled >>= writeIORef ref+      Ki.wait scope+    (isJust <$> readIORef ref) `shouldReturn` True++  childtest "`cancelScope` 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.cancelScope 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.cancelScope scope1+      Ki.scoped \_ -> (isJust <$> Ki.cancelled) `shouldReturn` True++  childtest "child thread inherits cancellation" \fork -> do+    ref <- newIORef Nothing+    Ki.scoped \scope -> do+      Ki.cancelScope 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 -> do+    shouldThrowSuchThat+      ( Ki.scoped \scope -> do+          fork scope (throwIO A)+          Ki.wait scope+      )+      (\(Ki.ThreadFailed _threadId exception) -> fromException exception == Just A)++  forktest "propagates async exceptions" \fork -> do+    shouldThrowSuchThat+      ( Ki.scoped \scope -> do+          fork scope (throwIO B)+          Ki.wait scope+      )+      (\(Ki.ThreadFailed _threadId exception) -> fromException exception == Just B)++  forktest "doesn't propagate own cancel token exceptions" \fork ->+    Ki.scoped \scope -> do+      Ki.cancelScope scope+      fork scope (atomically Ki.cancelledSTM >>= throwIO)+      Ki.wait scope++  forktest "propagates ScopeClosing if it isn't ours" \fork ->+    shouldThrowSuchThat+      ( Ki.scoped \scope -> do+          fork scope (throwIO Ki.Internal.ScopeClosing)+          Ki.wait scope+      )+      (\(Ki.ThreadFailed _threadId exception) -> fromException exception == Just Ki.Internal.ScopeClosing)++  forktest "propagates others' cancel token exceptions" \fork ->+    shouldThrowSuchThat+      ( Ki.scoped \scope -> do+          Ki.cancelScope scope+          fork scope (throwIO (Ki.Internal.CancelToken 0))+          Ki.wait scope+      )+      (\(Ki.ThreadFailed _threadId exception) -> fromException exception == Just (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 (Ki.ThreadFailed _threadId exception) -> fromException exception == Just A+        _ -> False++  test "`async` returns async exceptions" do+    Ki.scoped \scope -> do+      result <- Ki.async @() scope (throw B)+      Ki.await result `shouldReturnSuchThat` \case+        Left (Ki.ThreadFailed _threadId exception) -> fromException exception == Just B+        _ -> False++  test "awaiting a failed `fork`ed thread throws the sync exception it failed with" do+    Ki.scoped \scope -> do+      mask \unmask -> do+        thread <- Ki.fork @() scope (throw A)+        unmask (Ki.wait scope) `catch` \(Ki.Internal.ThreadFailedAsync _) -> pure ()+        Ki.await thread+          `shouldThrowSuchThat` \(Ki.ThreadFailed _threadId exception) -> fromException exception == Just A++  test "awaiting a failed `fork`ed thread throws the async exception it failed with" do+    Ki.scoped \scope -> do+      mask \unmask -> do+        thread <- Ki.fork @() scope (throw B)+        unmask (Ki.wait scope) `catch` \(Ki.Internal.ThreadFailedAsync _) -> pure ()+        Ki.await thread+          `shouldThrowSuchThat` \(Ki.ThreadFailed _threadId exception) -> fromException exception == Just B++  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` ()++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