diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,4 @@ +# CHANGELOG + +## v0.1.0.0 +* Initial release diff --git a/LICENSE.md b/LICENSE.md new file mode 100644 --- /dev/null +++ b/LICENSE.md @@ -0,0 +1,20 @@ +Copyright (c) 2022 Tristan de Cacqueray + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the +"Software"), to deal in the Software without restriction, including +without limitation the rights to use, copy, modify, merge, publish, +distribute, sublicense, and/or sell copies of the Software, and to +permit persons to whom the Software is furnished to do so, subject to +the following conditions: + +The above copyright notice and this permission notice shall be included +in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. +IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, +TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,41 @@ +# ki-effectful + +[![Build Status](https://github.com/TristanCacqueray/ki-effectful/workflows/Haskell-CI/badge.svg?branch=main)](https://github.com/TristanCacqueray/ki-effectful/actions?query=branch%3Amain) +[![Hackage](https://img.shields.io/hackage/v/ki-effectful.svg?logo=haskell)](https://hackage.haskell.org/package/ki-effectful) +[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE.md) + +## Description + +A `StructuredConcurrency` effect for the [`effectful`][effectful] ecosystem. + +## How to use + +This library exposes the following elements: + +* `StructuredConcurrency` — The type-level effect that you can declare in your type signatures. + +example: +```haskell +runStructuredConcurrency :: IOE :> es => Eff (StructuredConcurrency : es) a -> Eff es a +``` + +* The [`ki`][ki] api lifted to Eff using the effect local rep to store the scope: + +example: +```haskell +scoped :: StructuredConcurrency :> es => Eff es a -> Eff es a +fork :: StructuredConcurrency :> es => Eff es a -> Eff es (Thread a) +``` + +* Helper function to access the current scope: + +```haskell +testScopeLifting :: StructuredConcurrency :> es => Eff es Int +testScopeLifting = withCurrentScope $ \runInScope -> do + child <- scoped $ do + runInScope $ fork $ pure 42 + atomically $ await child +``` + +[effectful]: https://github.com/haskell-effectful/effectful +[ki]: https://github.com/awkward-squad/ki diff --git a/ki-effectful.cabal b/ki-effectful.cabal new file mode 100644 --- /dev/null +++ b/ki-effectful.cabal @@ -0,0 +1,59 @@ +cabal-version: 3.0 +name: ki-effectful +version: 0.1.0.0 +category: Concurrency +stability: experimental +synopsis: Adaptation of the ki library for the effectful ecosystem. +description: + Adaptation of the @@ library for the @@ ecosystem. + +bug-reports: https://github.com/TristanCacqueray/ki-effectful/issues +homepage: https://github.com/TristanCacqueray/ki-effectful#readme +author: Tristan de Cacqueray +maintainer: tdecacqu@redhat.com +copyright: 2022 Red Hat +license: MIT +license-file: LICENSE.md +build-type: Simple +extra-source-files: + CHANGELOG.md + README.md + +source-repository head + type: git + location: https://github.com/TristanCacqueray/ki-effectful + +common common + ghc-options: + -Weverything -Wno-redundant-constraints -Wno-implicit-prelude + -Wno-missing-import-lists -Wno-safe -Wno-unsafe + -Wno-missing-safe-haskell-mode -Wno-missing-local-signatures + + if impl(ghc >=9.2) + ghc-options: -Wno-missing-kind-signatures + + default-language: Haskell2010 + +library + import: common + hs-source-dirs: src + exposed-modules: Effectful.Ki + build-depends: + , base <5 + , effectful-core >=1.0 && <3.0 + , ki ^>=1.0 + , stm ^>=2.5 + +test-suite ki-effectful-test + import: common + ghc-options: -rtsopts -threaded -with-rtsopts=-N + type: exitcode-stdio-1.0 + main-is: Main.hs + hs-source-dirs: test + build-depends: + , base <5 + , effectful-core >=1.0 && <3.0 + , ki-effectful + , stm ^>=2.5 + , tasty ^>=1.4.2 + , tasty-hunit ^>=0.10 diff --git a/src/Effectful/Ki.hs b/src/Effectful/Ki.hs new file mode 100644 --- /dev/null +++ b/src/Effectful/Ki.hs @@ -0,0 +1,168 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE ImportQualifiedPost #-} +{-# LANGUAGE KindSignatures #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE TypeOperators #-} + +module Effectful.Ki ( + -- * Effect + StructuredConcurrency, + + -- * Handlers + runStructuredConcurrency, + withCurrentScope, + + -- * Core API + Scope, + Thread, + scoped, + fork, + forkTry, + await, + awaitAll, + withAwaitAll, + + -- * Extended API + fork_, + forkWith, + forkWith_, + forkTryWith, + + -- ** Thread options + ThreadOptions (..), + defaultThreadOptions, + ThreadAffinity (..), + + -- ** Byte count + ByteCount, + kilobytes, + megabytes, + + -- * STM re-export + Effectful.Ki.atomically, + Effectful.Ki.newTVarIO, + Effectful.Ki.newTMVarIO, + Effectful.Ki.newEmptyTMVarIO, +) where + +import Control.Concurrent.STM hiding (atomically) +import Control.Concurrent.STM qualified as STM +import Control.Exception (Exception) +import Data.Void (Void) +import Effectful +import Effectful.Dispatch.Static +import Effectful.Dispatch.Static.Primitive (cloneEnv) +import Effectful.Dispatch.Static.Unsafe (reallyUnsafeUnliftIO) + +import Ki hiding (fork, forkTry, forkTryWith, forkWith, forkWith_, fork_, scoped) +import Ki qualified + +data StructuredConcurrency :: Effect + +type instance DispatchOf StructuredConcurrency = 'Static 'WithSideEffects +data instance StaticRep StructuredConcurrency = StructuredConcurrency Scope + +-- | Run the 'StructuredConcurrency' effect. +runStructuredConcurrency :: IOE :> es => Eff (StructuredConcurrency : es) a -> Eff es a +runStructuredConcurrency k = withEffToIO $ \runInIO -> + Ki.scoped $ \scope -> + runInIO $ evalStaticRep (StructuredConcurrency scope) k + +scoped :: StructuredConcurrency :> es => Eff es a -> Eff es a +scoped k = reallyUnsafeUnliftIO $ \runInIO -> + Ki.scoped $ \scope -> + runInIO $ localStaticRep (const (StructuredConcurrency scope)) k + +-- | Provide a callback function to run an action within the current `Scope`. +withCurrentScope :: + StructuredConcurrency :> es => + ((forall es' a. StructuredConcurrency :> es' => Eff es' a -> Eff es' a) -> Eff es b) -> + Eff es b +withCurrentScope f = do + rep <- getStaticRep @StructuredConcurrency + f (localStaticRep (const rep)) + +fork :: + StructuredConcurrency :> es => + Eff es a -> + Eff es (Thread a) +fork action = do + StructuredConcurrency scope <- getStaticRep + unsafeEff $ \es -> do + es' <- cloneEnv es + Ki.fork scope (unEff action es') + +forkTry :: + Exception e => + StructuredConcurrency :> es => + Eff es a -> + Eff es (Thread (Either e a)) +forkTry action = do + StructuredConcurrency scope <- getStaticRep + unsafeEff $ \es -> do + es' <- cloneEnv es + Ki.forkTry scope (unEff action es') + +fork_ :: + StructuredConcurrency :> es => + Eff es Void -> + Eff es () +fork_ action = do + StructuredConcurrency scope <- getStaticRep + unsafeEff $ \es -> do + es' <- cloneEnv es + Ki.fork_ scope (unEff action es') + +forkWith :: + StructuredConcurrency :> es => + ThreadOptions -> + Eff es a -> + Eff es (Thread a) +forkWith threadOptions action = do + StructuredConcurrency scope <- getStaticRep + unsafeEff $ \es -> do + es' <- cloneEnv es + Ki.forkWith scope threadOptions (unEff action es') + +forkWith_ :: + StructuredConcurrency :> es => + ThreadOptions -> + Eff es Void -> + Eff es () +forkWith_ threadOptions action = do + StructuredConcurrency scope <- getStaticRep + unsafeEff $ \es -> do + es' <- cloneEnv es + Ki.forkWith_ scope threadOptions (unEff action es') + +forkTryWith :: + Exception e => + StructuredConcurrency :> es => + ThreadOptions -> + Eff es a -> + Eff es (Thread (Either e a)) +forkTryWith threadOptions action = do + StructuredConcurrency scope <- getStaticRep + unsafeEff $ \es -> do + es' <- cloneEnv es + Ki.forkTryWith scope threadOptions (unEff action es') + +withAwaitAll :: StructuredConcurrency :> es => (STM () -> STM a) -> Eff es a +withAwaitAll f = do + StructuredConcurrency scope <- getStaticRep + unsafeEff_ $ STM.atomically $ f $ Ki.awaitAll scope + +atomically :: StructuredConcurrency :> es => STM a -> Eff es a +atomically = unsafeEff_ . STM.atomically + +newTVarIO :: StructuredConcurrency :> es => a -> Eff es (TVar a) +newTVarIO = unsafeEff_ . STM.newTVarIO + +newTMVarIO :: StructuredConcurrency :> es => a -> Eff es (TMVar a) +newTMVarIO = unsafeEff_ . STM.newTMVarIO + +newEmptyTMVarIO :: StructuredConcurrency :> es => Eff es (TMVar a) +newEmptyTMVarIO = unsafeEff_ STM.newEmptyTMVarIO diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,97 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeOperators #-} + +module Main (main) where + +import Control.Applicative ((<|>)) +import Control.Concurrent (threadDelay) +import Control.Concurrent.STM (TMVar, putTMVar, readTMVar) +import Control.Exception +import Effectful +import Effectful.Ki +import Test.Tasty +import Test.Tasty.HUnit + +main :: IO () +main = + defaultMain $ + testGroup + "ki-effectful" + [ testCase "`fork` works" $ do + res <- runStructuredEff testFork + assertEqual "match" res 42 + , testCase "`fork` propagates exceptions" $ do + assertThrow $ runStructuredEff testThrow + , testCase "run a `fork` in a parent scope" $ do + res <- runStructuredEff testScopeLifting + assertEqual "match" res 42 + , testCase "`client` works" $ do + res <- runStructuredEff testClient + assertEqual "match" res (Just 42) + , testCase "`client` cancel" $ do + res <- runStructuredEff testClientCancel + assertEqual "match" res Nothing + ] + +runStructuredEff :: Eff '[StructuredConcurrency, IOE] a -> IO a +runStructuredEff = runEff . runStructuredConcurrency + +testFork :: StructuredConcurrency :> es => Eff es Int +testFork = do + child <- fork $ pure 42 + atomically $ await child + +testThrow :: StructuredConcurrency :> es => Eff es Int +testThrow = do + _ <- fork $ error "oops" + child <- fork $ pure 42 + withAwaitAll $ \waitAll -> do + waitAll + await child + +testScopeLifting :: StructuredConcurrency :> es => Eff es Int +testScopeLifting = withCurrentScope $ \runInScope -> do + child <- scoped $ do + runInScope $ fork $ pure 42 + atomically $ await child + +testClient :: StructuredConcurrency :> es => Eff es (Maybe Int) +testClient = do + hitman <- newEmptyTMVarIO + child <- fork $ client hitman (pure 42) + atomically $ await child + +testClientCancel :: (IOE :> es, StructuredConcurrency :> es) => Eff es (Maybe Int) +testClientCancel = do + hitman <- newEmptyTMVarIO + child <- fork $ client hitman $ do + -- liftIO $ putStrLn "running" + liftIO (threadDelay 500000) + pure 42 + liftIO (threadDelay 100000) + -- liftIO $ putStrLn "stopping" + atomically $ putTMVar hitman () + atomically $ await child + +-- | cancellable client implementation proposed in https://github.com/awkward-squad/ki/issues/11#issuecomment-1214159154 +client :: StructuredConcurrency :> es => TMVar () -> Eff es a -> Eff es (Maybe a) +client doneVar action = do + thread <- fork action + let waitDone = do + () <- readTMVar doneVar + pure Nothing + + waitThread = do + result <- await thread + pure (Just result) + + atomically $ waitDone <|> waitThread + +assertThrow :: IO a -> Assertion +assertThrow action = do + res <- try @SomeException action + case res of + Left _ -> pure () + Right _ -> assertFailure "Action did not throw"