diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,31 @@ ## [Unreleased] +## [0.3.0.0] - 2021-02-23 + +### Added + +- Added `cast`, `call`, and `handle` functions, and `Envelope` message wrapper +type, which enforce you get the response you expect from certain messages +- Added convenient `Server` type alias for processes using `Envelope`s +- Added `examples/use.hs` to experiment with "use pattern" +- Moved things to separate modules (everything still re-exported from top-level +`Drama` module) +- Added `NoMsg`, to be used instead of `Void` + +### Changed + +- Renamed "actor" to "process" +- Renamed `Message` to `HasMsg` +- Renamed `exit` to `stop` +- Generalized `loop`, `continue`, and `stop` to any monad +- Updated examples to use `cast` and `call` where possible +- Improved error message for unreachable `msg ~ Void` state + +## Fixed + +- Fixed incorrect use of `-XCPP` with `deriving newtype MonadFail` + ## [0.2.0.0] - 2021-02-20 ### Added @@ -60,7 +85,8 @@ Initial release -[Unreleased]: https://github.com/evanrelf/drama/compare/v0.2.0.0...HEAD +[Unreleased]: https://github.com/evanrelf/drama/compare/v0.3.0.0...HEAD +[0.3.0.0]: https://github.com/evanrelf/drama/releases/tag/v0.3.0.0 [0.2.0.0]: https://github.com/evanrelf/drama/releases/tag/v0.2.0.0 [0.1.0.3]: https://github.com/evanrelf/drama/releases/tag/v0.1.0.3 [0.1.0.2]: https://github.com/evanrelf/drama/releases/tag/v0.1.0.2 diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -12,7 +12,7 @@ notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. -* Neither the name of the nor the names of its +* 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. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -3,4 +3,4 @@ [![Hackage](https://img.shields.io/hackage/v/drama.svg?logo=haskell&label=drama)](https://hackage.haskell.org/package/drama) [![CI](https://github.com/evanrelf/drama/actions/workflows/ci.yml/badge.svg)](https://github.com/evanrelf/drama/actions/workflows/ci.yml) -:woman_dancing: Simple actor library for Haskell +🎭 Actor library for Haskell diff --git a/bench/Main.hs b/bench/Main.hs deleted file mode 100644 --- a/bench/Main.hs +++ /dev/null @@ -1,41 +0,0 @@ -{-# LANGUAGE BlockArguments #-} -{-# LANGUAGE NumericUnderscores #-} - - -module Main (main) where - -import Criterion.Types -import Criterion.Main -import Drama -import Prelude hiding (log, max, sum) - - -main :: IO () -main = defaultMainWith defaultConfig{timeLimit = 30, verbosity = Verbose} - [ bgroup "benchCounterSummer" - [ bench "1_000_000" $ nfIO (benchCounterSummer 1_000_000) - ] - ] - - -benchCounterSummer :: Int -> IO () -benchCounterSummer max = run do - summerAddress <- spawn (summer max) - _ <- spawn (counter max summerAddress) - wait - - -summer :: Int -> Actor Int () -summer max = loop (max, 0) \(count, sum) -> do - n <- receive - if count > 0 - then continue (count - 1, sum + n) - else exit () - - -counter :: Int -> Address Int -> Actor msg () -counter max summerAddress = loop 0 \count -> do - send summerAddress count - if count < max - then continue (count + 1) - else exit () diff --git a/drama.cabal b/drama.cabal --- a/drama.cabal +++ b/drama.cabal @@ -1,16 +1,16 @@ -cabal-version: 3.0 +cabal-version: 2.4 name: drama -version: 0.2.0.0 -synopsis: Simple actor library for Haskell -description: Simple actor library for Haskell +version: 0.3.0.0 +synopsis: Actor library for Haskell +description: Actor library for Haskell category: Concurrency author: Evan Relf maintainer: Evan Relf homepage: https://github.com/evanrelf/drama license: BSD-3-Clause copyright: 2021 Evan Relf -tested-with: GHC == 8.6.5, GHC == 8.10.3 +tested-with: GHC == 8.6.5, GHC == 8.8.4, GHC == 8.10.3 license-file: LICENSE extra-source-files: @@ -37,13 +37,14 @@ common executable + build-depends: drama ghc-options: -threaded -rtsopts -with-rtsopts=-N -common stan +common hie if impl(ghc >= 8.8) ghc-options: -fwrite-ide-info @@ -51,35 +52,31 @@ library - import: common, stan + import: common, hie hs-source-dirs: src build-depends: - , ki ^>= 0.2.0.1 + ki >= 0.2.0.1 && < 0.3 , transformers >= 0.5.6.2 && < 1.0 - , unagi-chan ^>= 0.4.1.3 + , unagi-chan >= 0.4.1.3 && < 0.5 exposed-modules: - , Drama - , Drama.Internal - , Drama.Reexports - - -benchmark bench - import: common, stan - type: exitcode-stdio-1.0 - hs-source-dirs: bench - main-is: Main.hs - build-depends: - , criterion ^>= 1.5.9.0 - , drama + Drama + , Drama.Loop + , Drama.Process + , Drama.Process.Internal + , Drama.Server + , Drama.Server.Internal executable example-shared-resource - import: common, executable, stan + import: common, executable, hie main-is: examples/shared-resource.hs - build-depends: drama +executable example-use + import: common, executable, hie + main-is: examples/use.hs + + executable example-workers - import: common, executable, stan + import: common, executable, hie main-is: examples/workers.hs - build-depends: drama diff --git a/examples/shared-resource.hs b/examples/shared-resource.hs --- a/examples/shared-resource.hs +++ b/examples/shared-resource.hs @@ -11,32 +11,32 @@ main :: IO () main = run_ do - -- Spawn `logger` actor, which starts waiting for requests. - loggerAddress <- spawn logger + -- Spawn `logger` process, which starts waiting for requests. + loggerAddr <- spawn logger - -- Spawn two noisy actors, `fizzBuzz` and `navi`, which want to constantly + -- Spawn two noisy processes, `fizzBuzz` and `navi`, which want to constantly -- print to the console. Instead of running `putStrLn`, they will send -- messages to `logger`. - spawn_ (fizzBuzz loggerAddress) - spawn_ (navi loggerAddress) + spawn_ (fizzBuzz loggerAddr) + spawn_ (navi loggerAddr) -- Block `main` thread forever. wait --- | Actor which encapsulates access to the console (a shared resource). By +-- | Process which encapsulates access to the console (a shared resource). By -- sending log messages to `logger`, instead of running `putStrLn` directly, we --- can avoid interleaving logs from actors running in parallel. -logger :: Actor String () +-- can avoid interleaving logs from processes running in parallel. +logger :: Process String () logger = forever do string <- receive liftIO $ putStrLn string --- | Silly example actor which wants to print to the console -fizzBuzz :: Address String -> Actor Void () -fizzBuzz loggerAddress = do - let log = send loggerAddress +-- | Silly example process which wants to print to the console +fizzBuzz :: Address String -> Process NoMsg () +fizzBuzz loggerAddr = do + let log = send loggerAddr loop (0 :: Int) \n -> do if | n `mod` 15 == 0 -> log "FizzBuzz" @@ -50,10 +50,10 @@ continue (n + 1) --- | Silly example actor which wants to print to the console -navi :: Address String -> Actor Void () -navi loggerAddress = do - let log = send loggerAddress +-- | Silly example process which wants to print to the console +navi :: Address String -> Process NoMsg () +navi loggerAddr = do + let log = send loggerAddr forever do log "Hey, listen!" diff --git a/examples/use.hs b/examples/use.hs new file mode 100644 --- /dev/null +++ b/examples/use.hs @@ -0,0 +1,123 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE RankNTypes #-} + +module Main (main) where + +import Control.Monad (when) +import Data.IORef (modifyIORef, newIORef, readIORef, writeIORef) +import Drama +import System.Exit (exitSuccess) + + +main :: IO () +main = run_ do + bottles <- useCounter 99 + + forever do + n <- getCount bottles + + when (n <= 0) do + liftIO exitSuccess + + liftIO $ putStrLn (show n <> " bottle(s) of beer on the wall, " <> show n <> " bottle(s) of beer") + + bottles -= 1 + n' <- getCount bottles + + liftIO $ putStrLn ("Take one down and pass it around, " <> show n' <> " bottle(s) of beer") + + +-------------------------------------------------------------------------------- + + +data CounterMsg res where + Increment :: Int -> CounterMsg () + Decrement :: Int -> CounterMsg () + GetCount :: CounterMsg Int + + +counter :: Int -> Server CounterMsg () +counter count0 = do + UseState{get, modify} <- useState count0 + + forever $ receive >>= handle \case + Increment n -> modify (+ n) + Decrement n -> modify (+ negate n) + GetCount -> get + + +data UseCounter = UseCounter + { increment :: forall msg. Int -> Process msg () + , decrement :: forall msg. Int -> Process msg () + , (+=) :: forall msg. Int -> Process msg () + , (-=) :: forall msg. Int -> Process msg () + , getCount :: forall msg. Process msg Int + } + + +useCounter :: Int -> Process msg UseCounter +useCounter count0 = do + counterAddr <- spawn (counter count0) + + pure UseCounter + { increment = \n -> cast counterAddr (Increment n) + , decrement = \n -> cast counterAddr (Decrement n) + , (+=) = \n -> cast counterAddr (Increment n) + , (-=) = \n -> cast counterAddr (Decrement n) + , getCount = call counterAddr GetCount + } + + +-------------------------------------------------------------------------------- + + +data StateMsg s res where + GetState :: StateMsg s s + GetsState :: (s -> a) -> StateMsg s a + PutState :: s -> StateMsg s () + ModifyState :: (s -> s) -> StateMsg s () + + +state :: s -> Server (StateMsg s) () +state s0 = do + stateIORef <- liftIO $ newIORef s0 + + forever $ receive >>= handle \case + GetState -> + liftIO $ readIORef stateIORef + + GetsState f -> do + s <- liftIO $ readIORef stateIORef + pure (f s) + + PutState s -> + liftIO $ writeIORef stateIORef s + + ModifyState f -> + liftIO $ modifyIORef stateIORef f + + +data UseState s = UseState + { get :: forall msg. HasMsg s => Process msg s + , gets :: forall a msg. HasMsg a => (s -> a) -> Process msg a + , put :: forall msg. s -> Process msg () + , modify :: forall msg. (s -> s) -> Process msg () + } + + +useState :: s -> Process msg (UseState s) +useState s0 = do + stateAddr <- spawn (state s0) + + pure UseState + { get = call stateAddr GetState + , gets = \f -> call stateAddr (GetsState f) + , put = \s -> cast stateAddr (PutState s) + , modify = \f -> cast stateAddr (ModifyState f) + } + + +{- HLINT ignore "Avoid lambda" -} diff --git a/examples/workers.hs b/examples/workers.hs --- a/examples/workers.hs +++ b/examples/workers.hs @@ -1,47 +1,45 @@ {-# LANGUAGE BlockArguments #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE LambdaCase #-} module Main (main) where -import Control.Monad (replicateM, when) import Data.Function ((&)) import Drama main :: IO () -main = run do - -- Spawn `fib` actor, which starts waiting for requests. - fibAddress <- spawn fib +main = run_ do + -- Spawn `fib` process, which starts waiting for requests. + fibAddr <- spawn fib + let ns = [200, 400, 600] + -- Request fibonacci numbers from `fib`. `fib` will spawn three `fibWorker`s -- to do all the work in the background, and then send `main` their results -- once they finish. - myAddress <- here - [200, 400, 600] & mapM_ \n -> send fibAddress (myAddress, n) + fs <- mapM (call fibAddr . GetFibNumber) ns - -- Receive results sent back from the `fibWorker`s, and print them to the - -- console. - replicateM 3 receive >>= mapM_ \(n, f) -> + fs & mapM_ \(n, f) -> liftIO $ putStrLn ("Fibonacci number " <> show n <> " is " <> show f) - -- Tell `fib` to stop. - send fibAddress (myAddress, -1) - -- Wait for `fib` to stop before exiting. - wait +data FibMsg res where + GetFibNumber :: Int -> FibMsg (Int, Integer) --- | Actor which immediately spawns and delegates to "worker" actors. -fib :: Actor (Address (Int, Integer), Int) () -fib = do - (responseAddress, n) <- receive - when (n >= 0) do - spawn_ (fibWorker responseAddress n) - fib +-- | Process which immediately spawns and delegates to "worker" processes. +fib :: Server FibMsg () +fib = forever do + envelope <- receive + spawn_ (fibWorker envelope) --- | "Worker" actor responsible for doing the real, time-consuming work. -fibWorker :: Address (Int, Integer) -> Int -> Actor Void () -fibWorker responseAddress n = send responseAddress (n, fibs !! n) +-- | "Worker" process responsible for doing the real, time-consuming work. Dies +-- after sending its result to the return address. +fibWorker :: Envelope FibMsg -> Process NoMsg () +fibWorker = handle \case + GetFibNumber n -> pure (n, fibs !! n) -- | Infinite list of fibonacci numbers. diff --git a/src/Drama.hs b/src/Drama.hs --- a/src/Drama.hs +++ b/src/Drama.hs @@ -5,43 +5,20 @@ -- Copyright: © 2021 Evan Relf -- Maintainer: evan@evanrelf.com -- --- Simple actor library for Haskell --- +-- Actor library for Haskell module Drama - ( Actor - - -- * Spawning actors - , spawn - , spawn_ - , wait - - -- * Messages - - -- ** Addresses - , Address - , here - - -- ** Sending messages - , send - - -- ** Receiving messages - , receive - , tryReceive - - -- * Managing state - , loop - , continue - , exit + ( -- * Lower-level processes + module Drama.Process - -- * Running your program - , run - , run_ + -- * Higher-level processes + , module Drama.Server - -- * Re-exports - , module Drama.Reexports + -- * Helpful utilities + , module Drama.Loop ) where -import Drama.Internal -import Drama.Reexports +import Drama.Loop +import Drama.Process +import Drama.Server diff --git a/src/Drama/Internal.hs b/src/Drama/Internal.hs deleted file mode 100644 --- a/src/Drama/Internal.hs +++ /dev/null @@ -1,295 +0,0 @@ -{-# LANGUAGE BlockArguments #-} -{-# LANGUAGE CPP #-} -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE DerivingStrategies #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE LambdaCase #-} -{-# LANGUAGE NamedFieldPuns #-} -{-# LANGUAGE TypeFamilies #-} -{-# LANGUAGE UndecidableInstances #-} - --- For `Message msg` -{-# OPTIONS_GHC -Wno-redundant-constraints #-} - -{-# OPTIONS_HADDOCK not-home #-} - --- | --- Module: Drama.Internal --- Stability: experimental --- License: BSD-3-Clause --- Copyright: © 2021 Evan Relf --- Maintainer: evan@evanrelf.com - -module Drama.Internal where - -import Control.Applicative (Alternative) -import Control.Monad (MonadPlus) -import Control.Monad.Fix (MonadFix) -import Control.Monad.IO.Class (MonadIO (..)) -import Control.Monad.Trans.Reader (ReaderT (..), asks) -import Data.Kind (Constraint) -import Data.Void (Void) -import GHC.TypeLits (ErrorMessage (..), TypeError) - -import qualified Control.Concurrent.Chan.Unagi as Unagi -import qualified Ki - --- Support `MonadFail` on GHC 8.6.5 -#if MIN_VERSION_base(4,9,0) -import Control.Monad.Fail (MonadFail) -#endif -#if MIN_VERSION_base(4,13,0) -import Prelude hiding (MonadFail) -#endif - - --- | Forbid use of functions which use the underlying `Unagi.Chan` when an --- actor's message type is `()` or `Void`. --- --- Forces users to use the more efficient `spawn_` and `run_` functions, and --- prevents runtime exceptions. --- --- @since 0.2.0.0 -type family Message msg :: Constraint where - Message Void = TypeError ('Text "Actors with 'msg ~ Void' cannot receive messages") - Message () = TypeError ('Text "Use 'msg ~ Void' instead of 'msg ~ ()' for actors which do not receive messages") - Message msg = () - - --- | The `Actor` monad, where you can `spawn` other actors, and `send` and --- `receive` messages. --- --- @since 0.1.0.0 -newtype Actor msg a = Actor (ReaderT (ActorEnv msg) IO a) - deriving newtype - ( Functor - , Applicative - , Monad - , MonadIO - , Alternative -#if MIN_VERSION_base(4,9,0) - , MonadPlus -#endif - , MonadFail - , MonadFix - ) - - --- | @since 0.1.0.0 -runActor :: MonadIO m => ActorEnv msg -> Actor msg a -> m a -runActor actorEnv (Actor m) = liftIO $ runReaderT m actorEnv - - --- | Environment for the `Actor` monad. --- --- @since 0.1.0.0 -data ActorEnv msg = ActorEnv - { address :: Address msg - , mailbox :: Mailbox msg - , scope :: Scope - } - - --- | The address for an actor. Returned after `spawn`ing an actor or asking for --- the current actor's address with `here`. Used to `send` messages to specific --- actors. --- --- @since 0.1.0.0 -newtype Address msg = Address (Unagi.InChan msg) - - --- | Where messages are delivered. Implicitly provided to `receive` and --- `tryReceive` by the `Actor` monad. --- --- @since 0.1.0.0 -newtype Mailbox msg = Mailbox (Unagi.OutChan msg) - - --- | @since 0.1.0.0 -newtype Scope = Scope Ki.Scope - - --- | Spawn a new actor. Returns the spawned actor's address. --- --- Example: --- --- > printerAddress <- spawn printer --- --- @since 0.1.0.0 -spawn :: Message childMsg => Actor childMsg () -> Actor msg (Address childMsg) -spawn actor = do - (inChan, outChan) <- liftIO Unagi.newChan - let address = Address inChan - let mailbox = Mailbox outChan - - Scope kiScope <- Actor $ asks scope - liftIO $ Ki.fork_ kiScope $ Ki.scoped \childKiScope -> - let childScope = Scope childKiScope - childEnv = ActorEnv{address, mailbox, scope = childScope} - in runActor childEnv actor - - pure address - - --- | More efficient version of `spawn`, for actors which receive no messages --- (`msg ~ Void`). See docs for `spawn` for more information. --- --- @since 0.2.0.0 -spawn_ :: Actor Void () -> Actor msg () -spawn_ actor = do - let address = Address (error "unreachable") - let mailbox = Mailbox (error "unreachable") - - Scope kiScope <- Actor $ asks scope - liftIO $ Ki.fork_ kiScope $ Ki.scoped \childKiScope -> - let childScope = Scope childKiScope - childEnv = ActorEnv{address, mailbox, scope = childScope} - in runActor childEnv actor - - --- | Wait for all actors spawned by the current actor to terminate. --- --- Example: --- --- > fooAddress <- spawn foo --- > barAddress <- spawn bar --- > wait --- --- @since 0.1.0.0 -wait :: Actor msg () -wait = do - Scope kiScope <- Actor $ asks scope - liftIO $ Ki.wait kiScope - - --- | Return the current actor's own address. Useful for sending your address to --- other actors, or for sending yourself a message. --- --- @since 0.1.0.0 -here :: Message msg => Actor msg (Address msg) -here = Actor $ asks address - - --- | Given an actor's address, send it a message. --- --- Example: --- --- > send printerAddress "Hello, world!" --- --- @since 0.1.0.0 -send - :: Message recipientMsg - => Address recipientMsg - -> recipientMsg - -> Actor msg () -send (Address inChan) msg = liftIO $ Unagi.writeChan inChan msg - - --- | Receive a message sent to the actor's mailbox. This function blocks until --- a message is received. --- --- Example: --- --- > printer :: Actor String () --- > printer = forever do --- > string <- receive --- > liftIO $ putStrLn string --- --- @since 0.1.0.0 -receive :: Message msg => Actor msg msg -receive = do - Mailbox outChan <- Actor $ asks mailbox - liftIO $ Unagi.readChan outChan - - --- | Receive a message sent to the actor's mailbox. This function blocks until --- a message is received. --- --- Example: --- --- > printer :: Actor String () --- > printer = forever do --- > tryReceive >>= \case --- > Just string -> liftIO $ putStrLn string --- > Nothing -> ... --- --- @since 0.1.0.0 -tryReceive :: Message msg => Actor msg (Maybe msg) -tryReceive = do - Mailbox outChan <- Actor $ asks mailbox - (element, _) <- liftIO $ Unagi.tryReadChan outChan - liftIO $ Unagi.tryRead element - - --- | Run a top-level actor. Intended to be used at the entry point of your --- program. --- --- @since 0.1.0.0 -run :: (Message msg, MonadIO m) => Actor msg a -> m a -run actor = do - (inChan, outChan) <- liftIO Unagi.newChan - let address = Address inChan - let mailbox = Mailbox outChan - - liftIO $ Ki.scoped \kiScope -> do - let scope = Scope kiScope - runActor ActorEnv{address, mailbox, scope} actor - - --- | More efficient version of `run`, for actors which receive no messages --- (`msg ~ Void`). See docs for `run` for more information. --- --- @since 0.2.0.0 -run_ :: MonadIO m => Actor Void a -> m a -run_ actor = do - let address = Address (error "unreachable") - let mailbox = Mailbox (error "unreachable") - - liftIO $ Ki.scoped \kiScope -> do - let scope = Scope kiScope - runActor ActorEnv{address, mailbox, scope} actor - - --- | Loop indefinitely with state. Use `Control.Monad.forever` for stateless --- infinite loops. --- --- Example: --- --- > counter :: Actor () Int --- > counter = loop 10 \count -> do --- > liftIO $ print count --- > if count > 0 --- > then continue (count - 1) --- > else exit count --- --- @since 0.1.0.0 -loop - :: s - -- ^ Initial state - -> (s -> Actor msg (Either s a)) - -- ^ Action to perform, either returning a new state to continue looping, or - -- a final value to stop looping. - -> Actor msg a -loop s0 k = - k s0 >>= \case - Left s -> loop s k - Right x -> pure x - - --- | Continue looping with state. --- --- prop> continue s = pure (Left s) --- --- @since 0.1.0.0 -continue :: s -> Actor msg (Either s a) -continue s = pure (Left s) - - --- | Exit loop with value. --- --- prop> exit x = pure (Right x) --- --- @since 0.1.0.0 -exit :: a -> Actor msg (Either s a) -exit x = pure (Right x) diff --git a/src/Drama/Loop.hs b/src/Drama/Loop.hs new file mode 100644 --- /dev/null +++ b/src/Drama/Loop.hs @@ -0,0 +1,65 @@ +{-# LANGUAGE LambdaCase #-} + +-- | +-- Module: Drama.Loop +-- Stability: experimental +-- License: BSD-3-Clause +-- Copyright: © 2021 Evan Relf +-- Maintainer: evan@evanrelf.com + +module Drama.Loop + ( loop + , continue + , stop + + -- * Re-exports + , forever + ) +where + +import Control.Monad (forever) + + +-- | Loop indefinitely with state. Use `Control.Monad.forever` for stateless +-- infinite loops. +-- +-- ===== __ Example __ +-- +-- > counter :: Process NoMsg () +-- > counter = loop (10 :: Int) \count -> do +-- > liftIO $ print count +-- > if count > 0 +-- > then continue (count - 1) +-- > else exit () +-- +-- @since 0.3.0.0 +loop + :: Monad m + => s + -- ^ Initial state + -> (s -> m (Either s a)) + -- ^ Action to perform, returning either a new state to continue looping, or + -- a final value to stop looping. + -> m a +loop s0 k = + k s0 >>= \case + Left s -> loop s k + Right x -> pure x + + +-- | Continue looping with some new state. +-- +-- prop> continue s = pure (Left s) +-- +-- @since 0.3.0.0 +continue :: Monad m => s -> m (Either s a) +continue s = pure (Left s) + + +-- | Stop looping and return with a final value. +-- +-- prop> exit x = pure (Right x) +-- +-- @since 0.3.0.0 +stop :: Monad m => a -> m (Either s a) +stop x = pure (Right x) diff --git a/src/Drama/Process.hs b/src/Drama/Process.hs new file mode 100644 --- /dev/null +++ b/src/Drama/Process.hs @@ -0,0 +1,42 @@ +-- | +-- Module: Drama.Process +-- Stability: experimental +-- License: BSD-3-Clause +-- Copyright: © 2021 Evan Relf +-- Maintainer: evan@evanrelf.com +-- +-- Lower-level processes, supporting `spawn`, `send`, `receive` and other +-- related operations. Inspired by Elixir and Erlang's processes. + +module Drama.Process + ( Process + + -- * Spawning processes + , spawn + , wait + + -- * Sending messages + , Address + , send + , here + + -- * Receiving messages + , receive + , tryReceive + + -- * Running your program + , run + + -- * Not receiving messages + , NoMsg + , spawn_ + , run_ + , HasMsg + + -- * Re-exports + , liftIO + ) +where + +import Control.Monad.IO.Class (liftIO) +import Drama.Process.Internal diff --git a/src/Drama/Process/Internal.hs b/src/Drama/Process/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Drama/Process/Internal.hs @@ -0,0 +1,266 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE CPP #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE UndecidableInstances #-} + +-- For `HasMsg msg` +{-# OPTIONS_GHC -Wno-redundant-constraints #-} + +{-# OPTIONS_HADDOCK not-home #-} +{-# OPTIONS_HADDOCK prune #-} + +-- | +-- Module: Drama.Process.Internal +-- Stability: experimental +-- License: BSD-3-Clause +-- Copyright: © 2021 Evan Relf +-- Maintainer: evan@evanrelf.com + +module Drama.Process.Internal where + +import Control.Applicative (Alternative) +import Control.Monad (MonadPlus) +import Control.Monad.Fix (MonadFix) +import Control.Monad.IO.Class (MonadIO (..)) +import Control.Monad.Trans.Reader (ReaderT (..), asks) +import Data.Kind (Constraint) +import Data.Void (Void) +import GHC.TypeLits (ErrorMessage (..), TypeError) + +import qualified Control.Concurrent.Chan.Unagi as Unagi +import qualified Ki + +-- Support `MonadFail` on GHC 8.6.5 +#if MIN_VERSION_base(4,9,0) +import Control.Monad.Fail (MonadFail) +#endif +#if MIN_VERSION_base(4,13,0) +import Prelude hiding (MonadFail) +#endif + + +-- | Monad supporting actor operations. Inspired by Elixir and Erlang's +-- processes. +-- +-- @since 0.3.0.0 +newtype Process msg a = Process (ReaderT (ProcessEnv msg) IO a) + deriving newtype + ( Functor + , Applicative + , Monad + , MonadIO + , Alternative + , MonadPlus +#if MIN_VERSION_base(4,9,0) + , MonadFail +#endif + , MonadFix + ) + + +-- | Provided some `ProcessEnv`, convert a `Process` action into an `IO` +-- action. +-- +-- @since 0.3.0.0 +runProcess :: MonadIO m => ProcessEnv msg -> Process msg a -> m a +runProcess processEnv (Process reader) = liftIO $ runReaderT reader processEnv + + +-- | Ambient context provided by the `Process` monad. +-- +-- Values in `ProcessEnv` are scoped to the current process and cannot be safely +-- shared. Functions like `spawn`, `receive`, and `here` use these values as +-- implicit parameters to avoid leaking internals (and for convenience). +-- +-- @since 0.3.0.0 +data ProcessEnv msg = ProcessEnv + { address :: Address msg + -- ^ Current process' address. + , mailbox :: Mailbox msg + -- ^ Current process' mailbox. + , scope :: Scope + -- ^ Current process' token used for spawning threads. + } + + +-- | Address for sending messages to a process. Obtained by running `spawn`, +-- `here`, or `receive` (if another process sends you an address). +-- +-- @since 0.3.0.0 +newtype Address msg = Address (Unagi.InChan msg) + + +-- | Mailbox where a process receives messages. Cannot be shared with other +-- processes; used implicitly by `receive` and `tryReceive`. +-- +-- @since 0.3.0.0 +newtype Mailbox msg = Mailbox (Unagi.OutChan msg) + + +-- | Token delimiting the lifetime of child processes (threads) created by a +-- process. +-- +-- @since 0.3.0.0 +newtype Scope = Scope Ki.Scope + + +-- | Constraint which prevents setting `msg ~ Void`, and provides helpful type +-- errors. +-- +-- @since 0.3.0.0 +type family HasMsg msg :: Constraint where + HasMsg NoMsg = TypeError ('Text "Processes with 'msg ~ NoMsg' cannot receive messages") + HasMsg Void = TypeError ('Text "Use 'msg ~ NoMsg' instead of 'msg ~ Void' for processes which do not receive messages") + HasMsg () = TypeError ('Text "Use 'msg ~ NoMsg' instead of 'msg ~ ()' for processes which do not receive messages") + HasMsg msg = () + + +-- | Message type used by processes which do not receive messages. +-- +-- @since 0.3.0.0 +data NoMsg + + +-- | Spawn a child process and return its address. +-- +-- @since 0.3.0.0 +spawn + :: HasMsg msg + => Process msg () + -- ^ Process to spawn + -> Process _msg (Address msg) + -- ^ Spawned process' address +spawn process = do + (inChan, outChan) <- liftIO Unagi.newChan + let address = Address inChan + let mailbox = Mailbox outChan + spawnImpl address mailbox process + pure address + + +-- | More efficient version of `spawn`, for processes which receive no messages +-- (@msg ~ `NoMsg`@). See docs for `spawn` for more information. +-- +-- @since 0.3.0.0 +spawn_ :: Process NoMsg () -> Process msg () +spawn_ process = do + let address = Address (error voidMsgError) + let mailbox = Mailbox (error voidMsgError) + spawnImpl address mailbox process + + +spawnImpl + :: Address msg + -> Mailbox msg + -> Process msg () + -> Process _msg () +spawnImpl address mailbox process = do + Scope kiScope <- Process $ asks scope + liftIO $ Ki.fork_ kiScope $ runImpl address mailbox process + + +-- | Block until all child processes have terminated. +-- +-- @since 0.3.0.0 +wait :: Process msg () +wait = do + Scope kiScope <- Process $ asks scope + liftIO $ Ki.wait kiScope + + +-- | Return the current process' address. +-- +-- @since 0.3.0.0 +here :: HasMsg msg => Process msg (Address msg) +here = Process $ asks address + + +-- | Send a message to another process. +-- +-- @since 0.3.0.0 +send + :: HasMsg msg + => Address msg + -- ^ Other process' address + -> msg + -- ^ Message to send + -> Process _msg () +send (Address inChan) msg = liftIO $ Unagi.writeChan inChan msg + + +-- | Receive a message. When the mailbox is empty, blocks until a message +-- arrives. +-- +-- @since 0.3.0.0 +receive :: HasMsg msg => Process msg msg +receive = do + Mailbox outChan <- Process $ asks mailbox + liftIO $ Unagi.readChan outChan + + +-- | Try to receive a message. When the mailbox is empty, returns `Nothing`. +-- +-- @since 0.3.0.0 +tryReceive :: HasMsg msg => Process msg (Maybe msg) +tryReceive = do + Mailbox outChan <- Process $ asks mailbox + (element, _) <- liftIO $ Unagi.tryReadChan outChan + liftIO $ Unagi.tryRead element + + +-- | Run a top-level process. Intended to be used at the entry point of your +-- program. +-- +-- If your program is designed with processes in mind, you can use `Process` as +-- your program's base monad: +-- +-- > main :: IO () +-- > main = run do +-- > ... +-- +-- Otherwise, use `run` like you would with @run@ functions from libraries like +-- @transformers@ or @mtl@. +-- +-- @since 0.3.0.0 +run :: (HasMsg msg, MonadIO m) => Process msg a -> m a +run process = do + (inChan, outChan) <- liftIO Unagi.newChan + let address = Address inChan + let mailbox = Mailbox outChan + runImpl address mailbox process + + +-- | More efficient version of `run`, for processes which receive no messages +-- (@msg ~ `NoMsg`@). See docs for `run` for more information. +-- +-- @since 0.3.0.0 +run_ :: MonadIO m => Process NoMsg a -> m a +run_ process = do + let address = Address (error voidMsgError) + let mailbox = Mailbox (error voidMsgError) + runImpl address mailbox process + + +runImpl :: MonadIO m => Address msg -> Mailbox msg -> Process msg a -> m a +runImpl address mailbox process = do + liftIO $ Ki.scoped \kiScope -> do + let scope = Scope kiScope + runProcess ProcessEnv{address, mailbox, scope} process + + +voidMsgError :: String +voidMsgError = unlines . fmap unwords $ + [ ["[!] drama internal error"] + , [] + , [ "Attempted to use the address or mailbox of a process which cannot send" + , "or receive messages (msg ~ NoMsg)." + ] + , [ "This should be impossible using non-internal modules!" ] + , [] + , [ "Please report this issue at https://github.com/evanrelf/drama/issues" + ] + ] diff --git a/src/Drama/Reexports.hs b/src/Drama/Reexports.hs deleted file mode 100644 --- a/src/Drama/Reexports.hs +++ /dev/null @@ -1,5 +0,0 @@ -module Drama.Reexports (module X) where - -import Control.Monad as X (forever) -import Control.Monad.IO.Class as X (MonadIO (..)) -import Data.Void as X (Void) diff --git a/src/Drama/Server.hs b/src/Drama/Server.hs new file mode 100644 --- /dev/null +++ b/src/Drama/Server.hs @@ -0,0 +1,54 @@ +-- | +-- Module: Drama.Server +-- Stability: experimental +-- License: BSD-3-Clause +-- Copyright: © 2021 Evan Relf +-- Maintainer: evan@evanrelf.com +-- +-- Higher-level process operations, allowing responses to messages. Inspired by +-- Elixir and Erlang's generic servers (@GenServer@ / @gen_server@). +-- +-- ===== __Example__ +-- +-- A server which encapsulates a piece of mutable state. Its @StateMsg@ type +-- specifies which messages it accepts, which messages return a response, and +-- what type that response is. +-- +-- > data StateMsg s res where +-- > GetState :: StateMsg s s +-- > GetsState :: (s -> a) -> StateMsg s a +-- > PutState :: s -> StateMsg s () +-- > ModifyState :: (s -> s) -> StateMsg s () +-- > +-- > state :: s -> Server (StateMsg s) () +-- > state s0 = do +-- > stateIORef <- liftIO $ newIORef s0 +-- > +-- > forever $ receive >>= handle \case +-- > GetState -> +-- > liftIO $ readIORef stateIORef +-- > +-- > GetsState f -> do +-- > s <- liftIO $ readIORef stateIORef +-- > pure (f s) +-- > +-- > PutState s -> +-- > liftIO $ writeIORef stateIORef s +-- > +-- > ModifyState f -> +-- > liftIO $ modifyIORef stateIORef f + +module Drama.Server + ( Server + , Envelope + + -- * Sending messages + , cast + , call + + -- * Handling messages + , handle + ) +where + +import Drama.Server.Internal diff --git a/src/Drama/Server/Internal.hs b/src/Drama/Server/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Drama/Server/Internal.hs @@ -0,0 +1,88 @@ +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE GADTSyntax #-} +{-# LANGUAGE KindSignatures #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE RankNTypes #-} + +{-# OPTIONS_HADDOCK not-home #-} + +-- | +-- Module: Drama.Server.Internal +-- Stability: experimental +-- License: BSD-3-Clause +-- Copyright: © 2021 Evan Relf +-- Maintainer: evan@evanrelf.com + +module Drama.Server.Internal where + +import Control.Monad.IO.Class (MonadIO (..)) +import Data.Kind (Type) +import Drama.Process (HasMsg, Process, send) +import Drama.Process.Internal (Address (..)) + +import qualified Control.Concurrent.Chan.Unagi as Unagi + + +-- | @since 0.3.0.0 +type Server msg a = Process (Envelope msg) a + + +-- | Wrapper around higher-kinded message types, to make them compatible with +-- the lower-level `Process` machinery. +-- +-- Higher-kinded message types are defined as GADTs with a type parameter. This +-- allows specifying the response type for messages. +-- +-- @since 0.3.0.0 +data Envelope (msg :: Type -> Type) where + Cast :: msg () -> Envelope msg + Call :: HasMsg res => Address res -> msg res -> Envelope msg + + +-- | Send a message to another process, expecting no response. Returns +-- immediately without blocking. +-- +-- @since 0.3.0.0 +cast + :: Address (Envelope msg) + -- ^ Process' address + -> msg () + -- ^ Message to send + -> Process _msg () +cast addr msg = send addr (Cast msg) + + +-- | Send a message to another process, and wait for a response. +-- +-- @since 0.3.0.0 +call + :: HasMsg res + => Address (Envelope msg) + -- ^ Process' address + -> msg res + -- ^ Message to send + -> Process _msg res + -- ^ Response +call addr msg = do + (inChan, outChan) <- liftIO Unagi.newChan + let returnAddr = Address inChan + send addr (Call returnAddr msg) + liftIO $ Unagi.readChan outChan + + +-- | Handle messages which may require a response. This is the only way to +-- consume an `Envelope`. +-- +-- @since 0.3.0.0 +handle + :: (forall res. msg res -> Process _msg res) + -- ^ Callback function that responds to messages + -> Envelope msg + -- ^ Message to handle + -> Process _msg () +handle callback = \case + Cast msg -> + callback msg + Call returnAddr msg -> do + res <- callback msg + send returnAddr res