ki: A lightweight structured concurrency library

[ bsd3, concurrency, library ] [ Propose Tags ]

A lightweight structured concurrency library.

For a variant of this API generalized to MonadUnliftIO, see ki-unlifted.

Remember to link your program with -threaded to use the threaded runtime!


[Skip to Readme]

Modules

[Index] [Quick Jump]

Downloads

Maintainer's Corner

Package maintainers

For package maintainers and hackage trustees

Candidates

Versions [RSS] 0.1.0, 0.1.0.1, 0.2.0, 0.2.0.1, 1.0.0, 1.0.0.1, 1.0.0.2, 1.0.1.0, 1.0.1.1 (info)
Change log CHANGELOG.md
Dependencies base (>=4.12 && <4.20), containers (>=0.6 && <0.8) [details]
License BSD-3-Clause
Copyright Copyright (C) 2020-2023 Mitchell Rosen, Travis Staton
Author Mitchell Rosen
Maintainer Mitchell Rosen <mitchellwrosen@gmail.com>, Travis Staton <hello@travisstaton.com>
Category Concurrency
Home page https://github.com/awkward-squad/ki
Bug tracker https://github.com/awkward-squad/ki/issues
Source repo head: git clone https://github.com/awkward-squad/ki.git(ki)
Uploaded by mitchellwrosen at 2023-10-10T23:20:34Z
Distributions LTSHaskell:1.0.1.1, NixOS:1.0.1.1, Stackage:1.0.1.1
Reverse Dependencies 8 direct, 3 indirect [details]
Downloads 1065 total (32 in the last 30 days)
Rating 2.25 (votes: 2) [estimated by Bayesian average]
Your Rating
  • λ
  • λ
  • λ
Status Docs available [build log]
Last success reported on 2023-10-11 [all 1 reports]

Readme for ki-1.0.1.1

[back to package description]
ki ki-unlifted
GitHub CI
Hackage Hackage
Stackage LTS Stackage LTS
Stackage Nightly Stackage Nightly
Dependencies Dependencies

Overview

ki is a lightweight structured-concurrency library inspired by many other projects and blog posts:

A previous version of ki also included a mechanism for soft-cancellation/graceful shutdown, which took inspiration from:

However, this feature was removed (perhaps temporarily) because the design of the API was unsatisfactory.

Documentation

Hackage documentation

Example: Happy Eyeballs

The Happy Eyeballs algorithm is a particularly common example used to demonstrate the advantages of structured concurrency, because it is simple to describe, but can be surprisingly difficult to implement.

The problem can be abstractly described as follows: we have a small set of actions to run, each of which can take arbitrarily long, or fail. Each action is a different way of computing the same value, so we only need to wait for one action to return successfully. We don't want to run the actions one at a time (because that is likely to take too long), nor all at once (because that is an improper use of resources). Rather, we will begin executing the first action, then wait 250 milliseconds, then begin executing the second, and so on, until one returns successfully.

There are of course a number of ways to implement this algorithm. We'll do something non-optimal, but simple. Let's get the imports out of the way first.

import Control.Concurrent
import Control.Monad (when)
import Control.Monad.STM (atomically)
import Data.Function ((&))
import Data.Functor (void)
import Data.List qualified as List
import Data.Maybe (isJust)
import Ki qualified

Next, let's define a staggeredSpawner helper that implements the majority of the core algorithm: given a list of actions, spawn them all at 250 millisecond intervals. After all actions are spawned, we block until all of them have returned.

staggeredSpawner :: [IO ()] -> IO ()
staggeredSpawner actions = do
  Ki.scoped \scope -> do
    actions
      & map (\action -> void (Ki.fork scope action))
      & List.intersperse (threadDelay 250_000)
      & sequence_
    atomically (Ki.awaitAll scope)

And finally, we wrap this helper with happyEyeballs, which accepts a list of actions, and returns when one action returns successfully, or returns Nothing if all actions fail. Note that in a real implementation, we may want to consider what to do if an action throws an exception. Here, we trust each action to signal failure by returning Nothing.

happyEyeballs :: [IO (Maybe a)] -> IO (Maybe a)
happyEyeballs actions = do
  resultVar <- newEmptyMVar

  let worker action = do
        result <- action
        when (isJust result) do
          _ <- tryPutMVar resultVar result
          pure ()

  Ki.scoped \scope -> do
    _ <-
      Ki.fork scope do
        staggeredSpawner (map worker actions)
        tryPutMVar resultVar Nothing
    takeMVar resultVar