diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,8 +5,25 @@ The format is based on [Keep a Changelog][changelog], and this project adheres to the [Haskell Package Versioning Policy][pvp]. -## [Unreleased] +## [0.4.0.0] - Unreleased +- Remove low-level `Process` API in favor of high-level `Server` API +- Rename "process" to "actor" everywhere +- Consolidate everything under `Drama` and `Drama.Internal` modules +- Remove `Drama.Loop` module +- Rename `run{,_}` to `runActor{,_}` +- Rename `here` to `getSelf` +- Make `tryReceive` indicate whether a message was received by returning a + `Bool` +- Add `Actor_` convenience type synonym +- Allow using `call` with response type of `()` + + So that you can send synchronous/blocking messages that don't have a + meaningful response. Useful if you need to wait until a message is handled + before proceeding. +- Send `MVar` instead of `Unagi.Chan` in `Envelope` +- Loosened dependency version bounds + ## [0.3.0.0] - 2021-02-23 ### Added @@ -85,7 +102,7 @@ Initial release -[Unreleased]: https://github.com/evanrelf/drama/compare/v0.3.0.0...HEAD +[0.4.0.0]: 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 diff --git a/drama.cabal b/drama.cabal --- a/drama.cabal +++ b/drama.cabal @@ -1,7 +1,7 @@ cabal-version: 2.4 name: drama -version: 0.3.0.0 +version: 0.4.0.0 synopsis: Actor library for Haskell description: Actor library for Haskell category: Concurrency @@ -55,16 +55,12 @@ import: common, hie hs-source-dirs: src build-depends: - ki >= 0.2.0.1 && < 0.3 - , transformers >= 0.5.6.2 && < 1.0 - , unagi-chan >= 0.4.1.3 && < 0.5 + ki >= 0.2 && < 0.3 + , transformers >= 0.5 && < 1.0 + , unagi-chan >= 0.4 && < 0.5 exposed-modules: Drama - , Drama.Loop - , Drama.Process - , Drama.Process.Internal - , Drama.Server - , Drama.Server.Internal + , Drama.Internal executable example-shared-resource @@ -75,8 +71,3 @@ executable example-use import: common, executable, hie main-is: examples/use.hs - - -executable example-workers - import: common, executable, hie - main-is: examples/workers.hs diff --git a/examples/shared-resource.hs b/examples/shared-resource.hs --- a/examples/shared-resource.hs +++ b/examples/shared-resource.hs @@ -1,20 +1,24 @@ {-# LANGUAGE BlockArguments #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiWayIf #-} {-# LANGUAGE NumericUnderscores #-} module Main (main) where import Control.Concurrent (threadDelay) +import Control.Monad (forever) +import Control.Monad.IO.Class (liftIO) import Drama import Prelude hiding (log) main :: IO () -main = run_ do - -- Spawn `logger` process, which starts waiting for requests. +main = runActor_ do + -- Spawn `logger` actor, which starts waiting for requests. loggerAddr <- spawn logger - -- Spawn two noisy processes, `fizzBuzz` and `navi`, which want to constantly + -- Spawn two noisy actors, `fizzBuzz` and `navi`, which want to constantly -- print to the console. Instead of running `putStrLn`, they will send -- messages to `logger`. spawn_ (fizzBuzz loggerAddr) @@ -24,36 +28,43 @@ wait --- | Process which encapsulates access to the console (a shared resource). By +-- | Message type for `logger` +data LogMsg res where + LogMsg :: String -> LogMsg () + + +-- | Actor 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 processes running in parallel. -logger :: Process String () -logger = forever do - string <- receive - liftIO $ putStrLn string +-- can avoid interleaving logs from actors running in parallel. +logger :: Actor LogMsg () +logger = forever $ receive \case + LogMsg string -> liftIO $ putStrLn string --- | Silly example process which wants to print to the console -fizzBuzz :: Address String -> Process NoMsg () -fizzBuzz loggerAddr = do - let log = send loggerAddr +-- | Silly example actor which wants to print to the console +fizzBuzz :: Address LogMsg -> Actor_ () +fizzBuzz loggerAddr = go 0 + where + log :: String -> Actor_ () + log string = cast loggerAddr (LogMsg string) - loop (0 :: Int) \n -> do - if | n `mod` 15 == 0 -> log "FizzBuzz" - | n `mod` 3 == 0 -> log "Fizz" - | n `mod` 5 == 0 -> log "Buzz" - | otherwise -> log (show n) + go :: Int -> Actor_ () + go n = do + if | n `mod` 15 == 0 -> log "FizzBuzz" + | n `mod` 3 == 0 -> log "Fizz" + | n `mod` 5 == 0 -> log "Buzz" + | otherwise -> log (show n) - -- Delay included for nicer (slow) output - liftIO $ threadDelay 500_000 + -- Delay included for nicer (slow) output + liftIO $ threadDelay 500_000 - continue (n + 1) + go (n + 1) --- | Silly example process which wants to print to the console -navi :: Address String -> Process NoMsg () +-- | Silly example actor which wants to print to the console +navi :: Address LogMsg -> Actor_ () navi loggerAddr = do - let log = send loggerAddr + let log string = cast loggerAddr (LogMsg string) forever do log "Hey, listen!" diff --git a/examples/use.hs b/examples/use.hs --- a/examples/use.hs +++ b/examples/use.hs @@ -6,14 +6,15 @@ module Main (main) where -import Control.Monad (when) +import Control.Monad (forever, when) +import Control.Monad.IO.Class (liftIO) import Data.IORef (modifyIORef, newIORef, readIORef, writeIORef) import Drama import System.Exit (exitSuccess) main :: IO () -main = run_ do +main = runActor_ do bottles <- useCounter 99 forever do @@ -39,26 +40,26 @@ GetCount :: CounterMsg Int -counter :: Int -> Server CounterMsg () +counter :: Int -> Actor CounterMsg () counter count0 = do UseState{get, modify} <- useState count0 - forever $ receive >>= handle \case + forever $ receive \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 + { increment :: forall msg. Int -> Actor msg () + , decrement :: forall msg. Int -> Actor msg () + , (+=) :: forall msg. Int -> Actor msg () + , (-=) :: forall msg. Int -> Actor msg () + , getCount :: forall msg. Actor msg Int } -useCounter :: Int -> Process msg UseCounter +useCounter :: Int -> Actor msg UseCounter useCounter count0 = do counterAddr <- spawn (counter count0) @@ -81,11 +82,11 @@ ModifyState :: (s -> s) -> StateMsg s () -state :: s -> Server (StateMsg s) () +state :: s -> Actor (StateMsg s) () state s0 = do stateIORef <- liftIO $ newIORef s0 - forever $ receive >>= handle \case + forever $ receive \case GetState -> liftIO $ readIORef stateIORef @@ -101,14 +102,14 @@ 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 () + { get :: forall msg. Actor msg s + , gets :: forall a msg. (s -> a) -> Actor msg a + , put :: forall msg. s -> Actor msg () + , modify :: forall msg. (s -> s) -> Actor msg () } -useState :: s -> Process msg (UseState s) +useState :: s -> Actor msg (UseState s) useState s0 = do stateAddr <- spawn (state s0) diff --git a/examples/workers.hs b/examples/workers.hs deleted file mode 100644 --- a/examples/workers.hs +++ /dev/null @@ -1,47 +0,0 @@ -{-# LANGUAGE BlockArguments #-} -{-# LANGUAGE GADTs #-} -{-# LANGUAGE LambdaCase #-} - -module Main (main) where - -import Data.Function ((&)) -import Drama - - -main :: IO () -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. - fs <- mapM (call fibAddr . GetFibNumber) ns - - fs & mapM_ \(n, f) -> - liftIO $ putStrLn ("Fibonacci number " <> show n <> " is " <> show f) - - -data FibMsg res where - GetFibNumber :: Int -> FibMsg (Int, Integer) - - --- | Process which immediately spawns and delegates to "worker" processes. -fib :: Server FibMsg () -fib = forever do - envelope <- receive - spawn_ (fibWorker envelope) - - --- | "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. -fibs :: [Integer] -fibs = 0 : 1 : zipWith (+) fibs (tail fibs) diff --git a/src/Drama.hs b/src/Drama.hs --- a/src/Drama.hs +++ b/src/Drama.hs @@ -6,19 +6,63 @@ -- Maintainer: evan@evanrelf.com -- -- Actor library for Haskell +-- +-- ===== __Example__ +-- +-- An actor 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 -> Actor (StateMsg s) () +-- > state s0 = do +-- > stateIORef <- liftIO $ newIORef s0 +-- > +-- > forever $ receive \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 - ( -- * Lower-level processes - module Drama.Process + ( Actor - -- * Higher-level processes - , module Drama.Server + -- * Spawning actors + , spawn + , wait - -- * Helpful utilities - , module Drama.Loop + -- * Sending messages + , Address + , cast + , call + , getSelf + + -- * Receiving messages + , receive + , tryReceive + + -- * Running your program + , runActor + + -- * Not receiving messages + , Actor_ + , NoMsg + , spawn_ + , runActor_ ) where -import Drama.Loop -import Drama.Process -import Drama.Server +import Drama.Internal diff --git a/src/Drama/Internal.hs b/src/Drama/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Drama/Internal.hs @@ -0,0 +1,290 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE CPP #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE KindSignatures #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE RankNTypes #-} + +{-# OPTIONS_HADDOCK not-home #-} +{-# OPTIONS_HADDOCK prune #-} + +-- | +-- 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.Concurrent (MVar, newEmptyMVar, putMVar, takeMVar) +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 (Type) + +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. +-- +-- @since 0.4.0.0 +newtype Actor (msg :: Type -> Type) a = Actor (ReaderT (ActorEnv msg) IO a) + deriving newtype + ( Functor + , Applicative + , Monad + , MonadIO + , Alternative + , MonadPlus +#if MIN_VERSION_base(4,9,0) + , MonadFail +#endif + , MonadFix + ) + + +-- | Ambient context provided by the `Actor` monad. +-- +-- Values in `ActorEnv` are scoped to the current actor and cannot be safely +-- shared. Functions like `spawn`, `receive`, and `getSelf` use these values as +-- implicit parameters to avoid leaking internals (and for convenience). +-- +-- @since 0.4.0.0 +data ActorEnv msg = ActorEnv + { address :: Address msg + -- ^ Current actor's address. + , mailbox :: Mailbox msg + -- ^ Current actor's mailbox. + , scope :: Ki.Scope + -- ^ Current actor's token used for spawning threads. Delimits the lifetime + -- of child actors (threads). + } + + +-- | Address for sending messages to an actor. Obtained by running `spawn`, +-- `getSelf`, or `receive` (if another actor sends you an address). +-- +-- @since 0.4.0.0 +newtype Address msg = Address (Unagi.InChan (Envelope msg)) + + +-- | Mailbox where an actor receives messages. Cannot be shared with other +-- actors; used implicitly by `receive` and `tryReceive`. +-- +-- @since 0.4.0.0 +newtype Mailbox msg = Mailbox (Unagi.OutChan (Envelope msg)) + + +-- | Wrapper around higher-kinded message types. +-- +-- Higher-kinded message types are defined as GADTs with a type parameter. This +-- allows specifying the response type for messages. +-- +-- @since 0.4.0.0 +data Envelope (msg :: Type -> Type) where + Cast :: msg () -> Envelope msg + Call :: MVar res -> msg res -> Envelope msg + + +-- | Message type used by actors which do not receive messages. +-- +-- @since 0.4.0.0 +data NoMsg res + + +-- | @since 0.4.0.0 +type Actor_ = Actor NoMsg + + +-- | Spawn a child actor and return its address. +-- +-- @since 0.4.0.0 +spawn + :: Actor msg () + -- ^ Actor to spawn + -> Actor _msg (Address msg) + -- ^ Spawned actor's address +spawn actor = do + (inChan, outChan) <- liftIO Unagi.newChan + let address = Address inChan + let mailbox = Mailbox outChan + spawnImpl address mailbox actor + pure address + + +-- | More efficient version of `spawn`, for actors which receive no messages +-- (@msg ~ `NoMsg`@). See docs for `spawn` for more information. +-- +-- @since 0.4.0.0 +spawn_ :: Actor_ () -> Actor msg () +spawn_ actor = do + let address = Address (error noMsgError) + let mailbox = Mailbox (error noMsgError) + spawnImpl address mailbox actor + + +spawnImpl + :: Address msg + -> Mailbox msg + -> Actor msg () + -> Actor _msg () +spawnImpl address mailbox actor = do + scope <- Actor $ asks scope + liftIO $ Ki.fork_ scope $ runActorImpl address mailbox actor + + +-- | Block until all child actors have terminated. +-- +-- @since 0.4.0.0 +wait :: Actor msg () +wait = do + scope <- Actor $ asks scope + liftIO $ Ki.wait scope + + +-- | Return the current actor's address. +-- +-- @since 0.4.0.0 +getSelf :: Actor msg (Address msg) +getSelf = Actor $ asks address + + +-- | Send a message to another actor, expecting no response. Returns immediately +-- without blocking. +-- +-- @since 0.4.0.0 +cast + :: Address msg + -- ^ Actor's address + -> msg () + -- ^ Message to send + -> Actor _msg () +cast (Address inChan) msg = liftIO $ Unagi.writeChan inChan (Cast msg) + + +-- | Send a message to another actor, and wait for a response. +-- +-- @since 0.4.0.0 +call + :: Address msg + -- ^ Actor's address + -> msg res + -- ^ Message to send + -> Actor _msg res + -- ^ Response +call (Address inChan) msg = liftIO do + resMVar <- newEmptyMVar + Unagi.writeChan inChan (Call resMVar msg) + takeMVar resMVar + + +-- | Receive a message. When the mailbox is empty, blocks until a message +-- arrives. +-- +-- @since 0.4.0.0 +receive + :: (forall res. msg res -> Actor msg res) + -- ^ Callback function that responds to messages + -> Actor msg () +receive callback = do + Mailbox outChan <- Actor $ asks mailbox + envelope <- liftIO $ Unagi.readChan outChan + case envelope of + Cast msg -> + callback msg + Call resMVar msg -> do + res <- callback msg + liftIO $ putMVar resMVar res + + +-- | Try to receive a message. When the mailbox is empty, returns immediately. +-- +-- @since 0.4.0.0 +tryReceive + :: (forall res. msg res -> Actor msg res) + -- ^ Callback function that responds to messages + -> Actor msg Bool +tryReceive callback = do + Mailbox outChan <- Actor $ asks mailbox + (element, _) <- liftIO $ Unagi.tryReadChan outChan + envelope <- liftIO $ Unagi.tryRead element + case envelope of + Nothing -> + pure False + Just (Cast msg) -> do + callback msg + pure True + Just (Call resMVar msg) -> do + res <- callback msg + liftIO $ putMVar resMVar res + pure True + + +-- | Run a top-level actor. Intended to be used at the entry point of your +-- program. +-- +-- If your program is designed with actors in mind, you can use `Actor` as +-- your program's base monad: +-- +-- > main :: IO () +-- > main = runActor root +-- > +-- > root :: Actor RootMsg () +-- > root = do +-- > ... +-- +-- Otherwise, use `runActor` like you would with @run@ functions from libraries +-- like @transformers@ or @mtl@. +-- +-- @since 0.4.0.0 +runActor :: MonadIO m => Actor msg a -> m a +runActor actor = do + (inChan, outChan) <- liftIO Unagi.newChan + let address = Address inChan + let mailbox = Mailbox outChan + runActorImpl address mailbox actor + + +-- | More efficient version of `runActor`, for actors which receive no messages +-- (@msg ~ `NoMsg`@). See docs for `runActor` for more information. +-- +-- @since 0.4.0.0 +runActor_ :: MonadIO m => Actor_ a -> m a +runActor_ actor = do + let address = Address (error noMsgError) + let mailbox = Mailbox (error noMsgError) + runActorImpl address mailbox actor + + +runActorImpl :: MonadIO m => Address msg -> Mailbox msg -> Actor msg a -> m a +runActorImpl address mailbox (Actor reader) = + liftIO $ Ki.scoped \scope -> + runReaderT reader ActorEnv{address, mailbox, scope} + + +noMsgError :: String +noMsgError = unlines . fmap unwords $ + [ ["[!] drama internal error"] + , [] + , [ "Attempted to use the address or mailbox of a actor 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/Loop.hs b/src/Drama/Loop.hs deleted file mode 100644 --- a/src/Drama/Loop.hs +++ /dev/null @@ -1,65 +0,0 @@ -{-# 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 deleted file mode 100644 --- a/src/Drama/Process.hs +++ /dev/null @@ -1,42 +0,0 @@ --- | --- 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 deleted file mode 100644 --- a/src/Drama/Process/Internal.hs +++ /dev/null @@ -1,266 +0,0 @@ -{-# 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/Server.hs b/src/Drama/Server.hs deleted file mode 100644 --- a/src/Drama/Server.hs +++ /dev/null @@ -1,54 +0,0 @@ --- | --- 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 deleted file mode 100644 --- a/src/Drama/Server/Internal.hs +++ /dev/null @@ -1,88 +0,0 @@ -{-# 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