diff --git a/bench/Main.hs b/bench/Main.hs new file mode 100644 --- /dev/null +++ b/bench/Main.hs @@ -0,0 +1,41 @@ +{-# 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 new file mode 100644 --- /dev/null +++ b/drama.cabal @@ -0,0 +1,52 @@ +cabal-version: 3.0 + +name: drama +version: 0.1.0.0 +synopsis: Simple actor library for Haskell +description: Simple 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 + + +common common + build-depends: base >= 4.12 && < 5.0 + default-language: Haskell2010 + ghc-options: + -Wall + -Wcompat + -Werror=incomplete-record-updates + -Werror=incomplete-uni-patterns + -Werror=missing-fields + -Werror=partial-fields + -Widentities + -Wmissing-home-modules + -Wredundant-constraints + -fshow-warning-groups + + +library + import: common + hs-source-dirs: src + build-depends: + , ki ^>= 0.2.0.1 + , transformers ^>= 0.5.6.2 + , unagi-chan ^>= 0.4.1.3 + exposed-modules: + , Drama + , Drama.Internal + other-modules: + , Drama.Demo + + +benchmark bench + import: common + type: exitcode-stdio-1.0 + hs-source-dirs: bench + main-is: Main.hs + build-depends: + , criterion ^>= 1.5.9.0 + , drama diff --git a/src/Drama.hs b/src/Drama.hs new file mode 100644 --- /dev/null +++ b/src/Drama.hs @@ -0,0 +1,101 @@ +-- | +-- Module: Drama +-- Stability: experimental +-- License: BSD-3-Clause +-- Copyright: © 2021 Evan Relf +-- Maintainer: evan@evanrelf.com +-- +-- Simple actor library for Haskell +-- +-- ==== __Example__ +-- +-- Fizz buzz, using three actors: @main@, @logger@, and @fizzBuzz@: +-- +-- > {-# LANGUAGE BlockArguments #-} +-- > {-# LANGUAGE MultiWayIf #-} +-- > {-# LANGUAGE NumericUnderscores #-} +-- > +-- > module Main (main) where +-- > +-- > import Control.Concurrent (threadDelay) +-- > import Control.Monad (forever) +-- > import Control.Monad.IO.Class (MonadIO (..)) +-- > import Drama +-- > import Prelude hiding (log) +-- > +-- > main :: IO () +-- > main = run do +-- > loggerAddress <- spawn logger +-- > _ <- spawn (fizzBuzz loggerAddress) +-- > wait +-- > +-- > logger :: Actor String () +-- > logger = forever do +-- > string <- receive +-- > liftIO $ putStrLn string +-- > +-- > fizzBuzz :: Address String -> Actor () () +-- > fizzBuzz loggerAddress = do +-- > let log = send loggerAddress +-- > +-- > 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) +-- > +-- > liftIO $ threadDelay 500_000 +-- > +-- > continue (n + 1) +-- +-- Output: +-- +-- > λ> main +-- > 1 +-- > 2 +-- > Fizz +-- > 4 +-- > Buzz +-- > Fizz +-- > 7 +-- > 8 +-- > Fizz +-- > Buzz +-- > 11 +-- > Fizz +-- > 13 +-- > 14 +-- > FizzBuzz +-- > ... + +module Drama + ( Actor + + -- * Spawning actors + , spawn + , wait + + -- * Messages + + -- ** Addresses + , Address + , here + + -- ** Sending messages + , send + + -- ** Receiving messages + , receive + , tryReceive + + -- * Managing state + , loop + , continue + , exit + + -- * Running your program + , run + ) +where + +import Drama.Internal diff --git a/src/Drama/Demo.hs b/src/Drama/Demo.hs new file mode 100644 --- /dev/null +++ b/src/Drama/Demo.hs @@ -0,0 +1,102 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE MultiWayIf #-} +{-# LANGUAGE NumericUnderscores #-} +{-# LANGUAGE RankNTypes #-} + +module Drama.Demo where + +import Control.Concurrent (threadDelay) +import Control.Monad (forever) +import Control.Monad.IO.Class (MonadIO (..)) +import Drama +import Prelude hiding (log) + + +main2 :: IO () +main2 = run do + loggerAddress <- spawn logger + _ <- spawn (fizzBuzz loggerAddress) + wait + + +logger :: Actor String () +logger = forever do + string <- receive + liftIO $ putStrLn string + + +fizzBuzz :: Address String -> Actor () () +fizzBuzz loggerAddress = do + let log = send loggerAddress + + loop (1 :: 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) + + liftIO $ threadDelay 500_000 + + continue (n + 1) + + +counter :: Actor msg Int +counter = loop 10 \count -> do + liftIO $ print count + if count > 0 + then continue (count - 1) + else exit count + + +echo :: (forall msg. String -> Actor msg ()) -> Actor () () +echo log = forever do + line <- liftIO getLine + if line == "ping" then + log "pong" + else + log line + + +add1 :: (forall msg. String -> Actor msg ()) -> Actor (Address Int, Int) () +add1 log = forever do + log "[add1] Waiting for request" + (returnAddress, number) <- receive + log "[add1] Received request" + log "[add1] Spawning worker" + spawn (add1Worker log number returnAddress) + + +add1Worker :: (forall msg. String -> Actor msg ()) -> Int -> Address Int -> Actor () () +add1Worker log number returnAddress = do + log "[add1Worker] Started" + send returnAddress (number + 1) + log "[add1Worker] Replied" + + +program :: Actor Int () +program = do + liftIO $ putStrLn "[main] START" + + liftIO $ putStrLn "[main] Spawning logger" + loggerAddress <- spawn logger + + let log = send loggerAddress + + liftIO $ putStrLn "[main] Spawning echo" + _ <- spawn (echo log) + + log "[main] Spawning add1" + add1Address <- spawn (add1 log) + + log "[main] Sending number to add1" + address <- here + send add1Address (address, 1) + two <- receive + log ("[main] Received response from add1: " <> show two) + + log "[main] FINISH" + wait + + +main :: IO () +main = run program 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,227 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NamedFieldPuns #-} + +{-# 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 qualified Control.Concurrent.Chan.Unagi as Unagi +import qualified Ki + + +-- | 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 + , MonadPlus + , 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 :: 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 + + +-- | 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 :: 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 :: 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 :: 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 :: 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 :: 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 + + +-- | 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)