oath (empty) → 0.0

raw patch · 6 files changed

+279/−0 lines, 6 filesdep +asyncdep +basedep +futures

Dependencies added: async, base, futures, oath, promise, stm, stm-delay, streamly, tasty-bench, unsafe-promises

Files

CHANGELOG.md +5/−0
README.md +87/−0
benchmarks/bench.hs +15/−0
oath.cabal +60/−0
src/Oath.hs +54/−0
tests/test.hs +58/−0

+ CHANGELOG.md view

@@ -0,0 +1,5 @@+# Revision history for oath++## 0.1.0.0 -- YYYY-mm-dd++* First version. Released on an unsuspecting world.

+ README.md view

@@ -0,0 +1,87 @@+Oath: composable concurrent computation done right+----++Oath is an Applicative structures that makes concurrent actions composable.++```haskell+newtype Oath a = Oath { runOath :: forall r. (STM a -> IO r) -> IO r }+```++`Oath` is a continuation-passing IO action which takes a transaction to obtain the final result (`STM a`).+The continuation-passing style makes it easier to release resources in time.+The easiest way to construct `Oath` is `oath`. It run the supplied action in a separate thread as long as the continuation is running.++```haskell+oath :: IO a -> Oath a+oath act = Oath $ \cont -> do+  v <- newEmptyTMVarIO+  tid <- forkFinally act (atomically . putTMVar v)+  let await = takeTMVar v >>= either throwSTM pure+  cont await `finally` killThread tid++evalOath :: Oath a -> IO a+evalOath m = runOath m atomically+```++`Oath` is an `Applicative`, so you can combine multiple `Oath`s. It starts computations without waiting for the results. Run `evalOath` to get the final result. The following code run concurrently `foo :: IO a` and `bar :: IO b`, then applies `f` to these results.++```haskell+main = evalOath $ f <$> oath foo <*> oath bar+```++It _does not_ provide a Monad instance because it is logically impossible to define one consistent with the Applicative instance.++Usage+----++`Oath` abstracts a triple of sending a request, waiting for response, and cancelling a request. If you want to send requests in a deterministic order, you can construct `Oath` directly instead of calling `oath`.++```haskell+Oath $ \cont -> bracket sendRequest cancelRequest (cont . waitForResponse)+```++Timeout behaviour can be easily added using the `Alternative` instance and `delay :: Int -> Oath ()`. Or more in general, `<|>` runs both computations until one of them finishes.++```haskell+-- | An 'Oath' that finishes once the given number of microseconds elapses+delay :: Int -> Oath ()++oath action <|> delay 100000+```++Comparison to other packages+----++[future](https://hackage.haskell.org/package/future-2.0.0/docs/Control-Concurrent-Future.html), [caf](https://hackage.haskell.org/package/caf-0.0.3/docs/Control-Concurrent-Futures.html) and [async](https://hackage.haskell.org/package/async-2.2.4/docs/Control-Concurrent-Async.html) seem solve the same problem. They define abstractions to asynchronous computations. `async` has an applicative `Concurrently` wrapper.++[spawn](https://hackage.haskell.org/package/spawn-0.3/docs/Control-Concurrent-Spawn.html) does not define any datatype. Instead it provides an utility function for `IO` (`spawn :: IO a -> IO (IO a)`). It does not offer a way to cancel a computation.++[promises](https://hackage.haskell.org/package/promises-0.3/docs/Data-Promise.html) provides a monadic interface for pure demand-driven computation. It has nothing to do with concurrency.++[unsafe-promises](https://hackage.haskell.org/package/unsafe-promises-0.0.1.3/docs/Control-Concurrent-Promise-Unsafe.html) creates an IO action that waits for the result on-demand using `unsafeInterleaveIO`.++[futures](https://hackage.haskell.org/package/futures-0.1/docs/Futures.html) provides a wrapper of `forkIO`. There is no way to terminate an action and it does not propagate exceptions.++[promise](https://hackage.haskell.org/package/promise-0.1.0.0/docs/Control-Concurrent-Promise.html) has illegal Applicative and Monad instances; `(<*>)` is not associative and has a bind that's not consistent with `(<*>).`++Performance+----++```haskell+bench "oath 10" $ nfIO $ O.evalOath $ traverse (O.oath . pure) [0 :: Int ..9]+bench "async 10" $ nfIO $ A.runConcurrently $ traverse (A.Concurrently . pure) [0 :: Int ..9]+```++`Oath`'s overhead of `(<*>)` is less than `Concurrently`. Unlike `Concurrently`, `<*>` itself does not fork threads.++```+All+  oath 10:   OK (1.63s)+    5.78 μs ± 265 ns+  async 10:  OK (0.21s)+    12.3 μs ± 767 ns+  oath 100:  OK (0.22s)+    52.6 μs ± 4.4 μs+  async 100: OK (0.23s)+    109  μs ± 8.4 μs+```

+ benchmarks/bench.hs view

@@ -0,0 +1,15 @@+{-# LANGUAGE RankNTypes #-}+import qualified Oath as O+import qualified Control.Concurrent.Async as A+import Test.Tasty.Bench+import qualified Streamly.Prelude as S++main :: IO ()+main = defaultMain+  [ bench "oath 10" $ nfIO $ O.evalOath $ traverse (O.oath . pure) [0 :: Int ..9]+  , bench "async 10" $ nfIO $ A.runConcurrently $ traverse (A.Concurrently . pure) [0 :: Int ..9]+  , bench "streamly 10" $ nfIO $ S.drain $ S.fromZipAsync $ traverse (S.fromEffect . pure) [0 :: Int ..9]+  , bench "oath 100" $ nfIO $ O.evalOath $ traverse (O.oath . pure) [0 :: Int ..99]+  , bench "async 100" $ nfIO $ A.runConcurrently $ traverse (A.Concurrently . pure) [0 :: Int ..99]+  , bench "streamly 100" $ nfIO $ S.toList $ S.fromZipAsync $ traverse (S.fromEffect . pure) [0 :: Int ..99]+  ]

+ oath.cabal view

@@ -0,0 +1,60 @@+cabal-version:      2.4+name:               oath+version:            0.0++-- A short (one-line) description of the package.+synopsis:           Composable concurrent computation done right++-- A longer description of the package.+description:        See README.md for details++-- A URL where users can report bugs.+bug-reports:        https://github.com/fumieval/oath++-- The license under which the package is released.+license:            BSD-3-Clause+author:             Fumiaki Kinoshita+maintainer:         fumiexcel@gmail.com++-- A copyright notice.+copyright:          Copyright (c) 2021 Fumiaki Kinoshita+category:           Concurrency+extra-source-files: CHANGELOG.md, README.md++source-repository head+  type: git+  location: https://github.com/fumieval/oath.git++library+    exposed-modules: Oath+    -- Modules included in this executable, other than Main.+    -- other-modules:++    -- LANGUAGE extensions used by modules in this package.+    -- other-extensions:+    build-depends:    base >=4.14.1.0 && <4.17, stm, stm-delay+    hs-source-dirs:   src+    default-language: Haskell2010+    ghc-options: -Wall -Wcompat++test-suite test+    build-depends: base >=4.14.1.0+        , futures+        , unsafe-promises+        , promise+        , oath+        , async+        , streamly+    type: exitcode-stdio-1.0+    hs-source-dirs: tests+    default-language: Haskell2010+    main-is: test.hs+    ghc-options: -Wall -Wcompat++benchmark bench+    type:                exitcode-stdio-1.0+    main-is:             bench.hs+    hs-source-dirs:      benchmarks+    build-depends:       base, tasty-bench, oath, async, streamly+    ghc-options:         -Wall -threaded -O2+    default-language:    Haskell2010

+ src/Oath.hs view

@@ -0,0 +1,54 @@+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE DerivingVia #-}+module Oath+  ( Oath(..)+  , hoistOath+  , evalOath+  , oath+  , delay+  , timeout) where++import Control.Applicative+import Control.Concurrent+import Control.Concurrent.STM+import Control.Concurrent.STM.Delay+import Control.Exception+import Data.Monoid++-- 'Oath' is an 'Applicative' structure that collects results of one or more computations.+newtype Oath a = Oath { runOath :: forall r. (STM a -> IO r) -> IO r }+  deriving Functor+  deriving (Semigroup, Monoid) via Ap Oath a++hoistOath :: (forall x. STM x -> STM x) -> Oath a -> Oath a+hoistOath t (Oath m) = Oath $ \cont -> m $ cont . t++evalOath :: Oath a -> IO a+evalOath m = runOath m atomically++instance Applicative Oath where+  pure a = Oath $ \cont -> cont (pure a)+  Oath m <*> Oath n = Oath $ \cont -> m $ \f -> n $ \x -> cont (f <*> x)++instance Alternative Oath where+  empty = Oath $ \cont -> cont empty+  Oath m <|> Oath n = Oath $ \cont -> m $ \a -> n $ \b -> cont (a <|> b)++-- | Lift an IO action into an 'Oath', forking a thread.+-- When the continuation terminates, it kills the thread.+-- Exception thrown in the thread will be propagated to the result.+oath :: IO a -> Oath a+oath act = Oath $ \cont -> do+  v <- newEmptyTMVarIO+  tid <- forkFinally act (atomically . putTMVar v)+  let await = readTMVar v >>= either throwSTM pure+  cont await `finally` killThread tid++-- | An 'Oath' that finishes once the given number of microseconds elapses+delay :: Int -> Oath ()+delay dur = Oath $ \cont -> bracket (newDelay dur) cancelDelay (cont . waitDelay)++-- | Wrap an 'Oath'+timeout :: Int -> Oath a -> Oath (Maybe a)+timeout dur m = Just <$> m <|> Nothing <$ delay dur

+ tests/test.hs view

@@ -0,0 +1,58 @@+import qualified Oath+import qualified Futures+import qualified Control.Concurrent.Promise.Unsafe as UP+import qualified Control.Concurrent.Promise as Promise+import qualified Control.Concurrent.Async as Async+import Control.Applicative+import Control.Concurrent+import Control.Exception+import Data.Functor.Compose+import qualified Streamly.Prelude as S++action :: Int -> String -> Double -> IO ()+action order name dur = do+  -- delay a bit so that "Begin" gets printed in a consistent order+  threadDelay $ 1000 * 10 * order+  putStrLn $ "    Begin " <> name+  threadDelay (floor $ 1000000 * dur)+  putStrLn $ "    End " <> name+  `onException` putStrLn ("    Killed " <> name)++tester :: Applicative m => (m () -> IO ()) -> (IO () -> m ()) -> IO ()+tester run lift = do+  putStrLn "  Left:"+  run $ (lift (action 0 "foo" 0.1) *> lift (action 1 "bar" 0.3)) *> lift (action 2 "baz" 0.2)+  putStrLn "  Right:"+  run $ lift (action 0 "foo" 0.1) *> (lift (action 1 "bar" 0.3) *> lift (action 2 "baz" 0.2))++  putStrLn "  Left Error:"+  try (run $ lift (fail "Fail") *> lift (action 0 "success" 0.1)) >>= print'+  putStrLn "  Right Error:"+  try (run $ lift (action 0 "success" 0.1) *> lift (fail "Fail")) >>= print'+  where+    print' :: Either SomeException () -> IO ()+    print' = putStrLn . ("    "++) . show++main :: IO ()+main = do+  putStrLn "oath timeout:"+  Oath.evalOath $ Oath.oath (action 0 "delay" 1.0) <|> Oath.delay 200000+  threadDelay 10000++  putStrLn "oath: "+  tester Oath.evalOath Oath.oath+  putStrLn "async: "+  tester Async.runConcurrently Async.Concurrently+  putStrLn "futures:"+  tester ((>>=Futures.block) . getCompose) (Compose . Futures.fork)++  putStrLn "streamly:"+  tester (S.drain . S.fromZipAsync . S.fromAhead) S.fromEffect++  putStrLn "promise:"+  tester Promise.runPromise Promise.liftIO++  -- random+  putStrLn "unsafe-promises:"+  try (tester id UP.promise) >>= (print :: Either SomeException () -> IO ())+