diff --git a/Control/Concurrent/Actor.hs b/Control/Concurrent/Actor.hs --- a/Control/Concurrent/Actor.hs +++ b/Control/Concurrent/Actor.hs @@ -1,379 +1,191 @@ -- | -- Module : Control.Concurrent.Actor --- Copyright : (c) 2014 Alex Constandache & Forkk --- License : BSD3 +-- Copyright : (c) 2014 Forkk +-- License : MIT -- Maintainer : forkk@forkk.net -- Stability : experimental -- Portability : GHC only (requires throwTo) -- --- This module implements Erlang-style actors --- (what Erlang calls processes). It does not implement --- network distribution (yet?). Here is an example: +-- This module implements Erlang-style actors (what Erlang calls processes). -- -- @ ---act1 :: Actor ---act1 = do --- me <- self --- liftIO $ print "act1 started" --- forever $ receive --- [ Case $ \((n, a) :: (Int, Address)) -> --- if n > 10000 --- then do --- liftIO . throwIO $ NonTermination --- else do --- liftIO . putStrLn $ "act1 got " ++ (show n) ++ " from " ++ (show a) --- send a (n+1, me) --- , Case $ \(e :: RemoteException) -> --- liftIO . print $ "act1 received a remote exception" --- , Default $ liftIO . print $ "act1: received a malformed message" --- ] --- ---act2 :: Address -> Actor ---act2 addr = do --- monitor addr --- -- setFlag TrapRemoteExceptions --- me <- self --- send addr (0 :: Int, me) --- forever $ receive --- [ Case $ \((n, a) :: (Int, Address)) -> do --- liftIO . putStrLn $ "act2 got " ++ (show n) ++ " from " ++ (show a) --- send a (n+1, me) --- , Case $ \(e :: RemoteException) -> --- liftIO . print $ "act2 received a remote exception: " ++ (show e) --- ] --- ---act3 :: Address -> Actor ---act3 addr = do --- monitor addr --- setFlag TrapRemoteExceptions --- forever $ receive --- [ Case $ \(e :: RemoteException) -> --- liftIO . print $ "act3 received a remote exception: " ++ (show e) --- ] --- ---main = do --- addr1 <- spawn act1 --- addr2 <- spawn (act2 addr1) --- spawn (act3 addr2) --- threadDelay 20000000 --- @ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE ExistentialQuantification #-} -module Control.Concurrent.Actor ( - -- * Types - Address - , Handler(..) - , ActorM - , Actor - , ActorException - , ActorExit - , Flag(..) - -- * Actor actions - , send - , self - , receive - , receiveWithTimeout - , spawn - , runActor - , monitor - , link - , kill - , status - , setFlag - , clearFlag - , toggleFlag - , testFlag - ) where +module Control.Concurrent.Actor + -- * Types + ( ActorHandle + , ActorMessage + , ActorM + -- * Sending Messages + , send + , sendIO + -- * Receiving Messages + , receive + , receiveMaybe + -- * Spawning Actors + , spawnActor + , runActor + -- * Getting Information + , self + , actorThread + -- , monitor + -- , link + -- , kill + -- , status + -- , setFlag + -- , clearFlag + -- , toggleFlag + -- , testFlag + ) where import Control.Concurrent - ( forkIO - , killThread - , myThreadId - , ThreadId - ) -import Control.Exception - ( Exception(..) - , SomeException - , catches - , throwTo - , throwIO - , PatternMatchFail(..) - ) -import qualified Control.Exception as E (Handler(..)) -import Control.Concurrent.Chan - ( Chan - , newChan - , readChan - , writeChan - ) -import Control.Concurrent.MVar - ( MVar - , newMVar - , modifyMVar_ - , withMVar - , readMVar - ) -import GHC.Conc (ThreadStatus, threadStatus) +import Control.Concurrent.STM import Control.Monad.Reader - ( ReaderT - , runReaderT - , asks - , ask - , liftIO - ) -import System.Timeout (timeout) -import Data.Dynamic -import Data.Set - ( Set - , empty - , insert - , delete - , elems - ) -import Data.Word (Word64) -import Data.Bits ( - setBit - , clearBit - , complementBit - , testBit - ) --- | Exception raised by an actor on exit -data ActorExit = ActorExit deriving (Typeable, Show) -instance Exception ActorExit - -data ActorException = ActorException Address SomeException - deriving (Typeable, Show) - -instance Exception ActorException - -type Flags = Word64 - -data Flag = TrapActorExceptions - deriving (Eq, Enum) - -defaultFlags :: [Flag] -defaultFlags = [] - -setF :: Flag -> Flags -> Flags -setF = flip setBit . fromEnum +-- {{{ Types -clearF :: Flag -> Flags -> Flags -clearF = flip clearBit . fromEnum +-- {{{ Message -toggleF :: Flag -> Flags -> Flags -toggleF = flip complementBit . fromEnum +-- | The `ActorMessage` class must be implemented by any type that will be sent +-- as a message to actors. +-- Any given type of actor will have one `ActorMessage` type that is sent to +-- that actor. This ensures type safety. +-- Currently this is simply a dummy class with nothing in it, but things may be +-- added in the future. +class ActorMessage msg -isSetF :: Flag -> Flags -> Bool -isSetF = flip testBit . fromEnum +-- Allow actors that don't take messages. +instance ActorMessage () -data Context = Context - { ctxMonitors :: MVar (Set Address) - , ctxChan :: Chan Message - , ctxFlags :: MVar Flags - } deriving (Typeable) +-- }}} -newtype Message = Msg { unMsg :: Dynamic } - deriving (Typeable) +-- {{{ Handle and context -instance Show Message where - show = show . unMsg +-- | An `ActorHandle` acts as a reference to a specific actor. +data ActorMessage msg => ActorHandle msg = ActorHandle + { ahContext :: ActorContext msg -- Context for this handle's actor. + , ahThread :: ThreadId -- The actor's thread ID. + } -toMsg :: Typeable a => a -> Message -toMsg = Msg . toDyn +-- | The `ActorContext` holds shared information about a given actor. +-- This is information such as the actor's mail box, the list of actors it's +-- linked to, etc. +data ActorMessage msg => ActorContext msg = ActorContext + { acMailBox :: MailBox msg -- Channel for the actor's messages. + } -fromMsg :: Typeable a => Message -> Maybe a -fromMsg = fromDynamic . unMsg +-- | The type for the actor's mail box. +type MailBox msg = TChan msg --- | The address of an actor, used to send messages -data Address = Address - { addrThread :: ThreadId - , addrContext :: Context - } deriving (Typeable) +-- }}} -instance Show Address where - show (Address ti _) = "Address(" ++ (show ti) ++ ")" +-- }}} -instance Eq Address where - addr1 == addr2 = (addrThread addr1) == (addrThread addr2) +-- | The base actor monad. +type ActorM msg = ReaderT (ActorContext msg) IO -instance Ord Address where - addr1 `compare` addr2 = (addrThread addr1) `compare` (addrThread addr2) +-- {{{ Get info --- | The actor monad, just a reader monad on top of 'IO'. -type ActorM = ReaderT Context IO +-- | Gets a handle to the current actor. +self :: ActorMessage msg => ActorM msg (ActorHandle msg) +self = do + context <- ask + thread <- liftIO $ myThreadId + return $ ActorHandle context thread --- | The type of an actor. It is just a monadic action --- in the 'ActorM' monad, returning () -type Actor = ActorM () +-- | Retrieves the mail box for the current actor. +myMailBox :: ActorMessage msg => ActorM msg (MailBox msg) +myMailBox = asks acMailBox -data Handler = forall m . (Typeable m) - => Case (m -> ActorM ()) - | Default (ActorM ()) +-- }}} --- | Used to obtain an actor's own address inside the actor -self :: ActorM Address -self = do - c <- ask - i <- liftIO myThreadId - return $ Address i c +-- {{{ Receiving --- | Try to handle a message using a list of handlers. --- The first handler matching the type of the message --- is used. -receive :: [Handler] -> ActorM () -receive hs = do - ch <- asks ctxChan - msg <- liftIO . readChan $ ch - rec msg hs +-- | Reads a message from the actor's mail box. +-- If there are no messages, blocks until one is received. If you don't want +-- this, use `receiveMaybe` instead. +receive :: ActorMessage msg => ActorM msg (msg) +receive = do + chan <- myMailBox + -- Read from the channel, retrying if there is nothing to read. + liftIO $ atomically $ readTChan chan --- | Same as receive, but times out after a specified --- amount of time and runs a default action -receiveWithTimeout :: Int -> [Handler] -> ActorM () -> ActorM () -receiveWithTimeout n hs act = do - ch <- asks ctxChan - msg <- liftIO . timeout n . readChan $ ch - case msg of - Just m -> rec m hs - Nothing -> act +-- | Reads a message from the actor's mail box. +-- If there are no messages, returns `Nothing`. +receiveMaybe :: ActorMessage msg => ActorM msg (Maybe msg) +receiveMaybe = do + chan <- myMailBox + liftIO $ atomically $ tryReadTChan chan -rec :: Message -> [Handler] -> ActorM () -rec msg [] = liftIO . throwIO $ PatternMatchFail err where - err = "no handler for messages of type " ++ (show msg) -rec msg ((Case hdl):hs) = case fromMsg msg of - Just m -> hdl m - Nothing -> rec msg hs -rec _ ((Default act):_) = act +-- }}} +-- {{{ Sending --- | Sends a message from inside the 'ActorM' monad -send :: Typeable m => Address -> m -> ActorM () -send addr msg = do - let ch = ctxChan . addrContext $ addr - liftIO . writeChan ch . toMsg $ msg +-- | Sends a message to the given actor handle. +-- This is secretly just `sendIO` lifted into an actor monad. +send :: (ActorMessage msg, ActorMessage msg') => + ActorHandle msg -> msg -> ActorM msg' () +send hand msg = liftIO $ sendIO hand msg --- FIXME: This is a big and complicated function. It would be a good idea to --- break it up into smaller parts. --- | Prepares the given actor monad to be run as an actor. Returns a tuple --- containing the actor's context and an IO action to run the actor. -mkActor :: Actor -> [Flag] -> IO (IO (), Context) -mkActor actor flagList = do - chan <- liftIO newChan - linkedActors <- newMVar empty - flags <- newMVar $ foldl (flip setF) 0x00 flagList - let context = Context linkedActors chan flags - -- An IO action which executes the actual actor. - let actorFunc = actorInternal `catches` [ E.Handler linkedHandler - , E.Handler exceptionHandler] - -- The internal IO action for the actor. - -- This will be wrapped by exception handlers. - actorInternal = runReaderT actor context - -- Linked exception handler. Catches exceptions from linked actors and - -- handles them appropriately. - linkedHandler :: ActorException -> IO () - linkedHandler ex@(ActorException addr iex) = do - -- Remove the linked actor from our list. - modifyMVar_ linkedActors (return . delete addr) - me <- myThreadId - forward $ ActorException (Address me context) iex - throwIO ex - -- Exception handler. Catches all exceptions and forwards them to - -- monitoring actors. - exceptionHandler :: SomeException -> IO () - exceptionHandler ex = do - me <- myThreadId - forward $ ActorException (Address me context) ex - throwIO ex - -- Exception forwarding function. - -- Takes an `ActorException` and forwards it to all actors monitoring the - -- current actor. - forward :: ActorException -> IO () - forward ex = do - linkedSet <- readMVar linkedActors - mapM_ (forwardTo ex) $ elems linkedSet - -- Forwards the given exception to the given actor. - forwardTo :: ActorException -> Address -> IO () - forwardTo ex addr = do - let remoteFlags = ctxFlags . addrContext $ addr - remoteChan = ctxChan . addrContext $ addr - trap <- withMVar remoteFlags (return . isSetF TrapActorExceptions) - -- If the remote actor is trapping actor exceptions, send our - -- exception to it as a message. Otherwise, throw it to the actor's - -- thread. - if trap - then writeChan remoteChan $ toMsg ex - else throwTo (addrThread addr) ex - -- Return the context and the IO action. - return (actorFunc, context) +-- | Sends a message to the given actor handle from within the IO monad. +sendIO :: ActorMessage msg => ActorHandle msg -> msg -> IO () +sendIO hand msg = + atomically $ writeTChan mailBox $ msg + where + mailBox = handleMailBox hand +-- }}} --- | Spawn a new actor with default flags on a separate thread. -spawn :: Actor -> IO Address -spawn actor = do - (actorFunc, context) <- mkActor actor defaultFlags - -- Start the actor's thread. - threadId <- forkIO actorFunc - -- Return a handle. - return $ Address threadId context +-- {{{ Spawning --- | Run the given actor with default flags on the current thread. --- This can be useful for your program's "main actor". -runActor :: Actor -> IO () -runActor actor = do - -- Create the actor. - (actorFunc, _) <- mkActor actor defaultFlags - actorFunc +-- | Internal function for starting actors. +-- This takes an `ActorM` action, makes a channel for it, wraps it in exception +-- handling stuff, and turns it into an IO monad. The function returns a tuple +-- containing the actor's context and the IO action to execute the actor. +wrapActor :: ActorMessage msg => ActorM msg () -> IO (IO (), ActorContext msg) +wrapActor actorAction = do + -- TODO: Exception handling. + -- First, create a channel for the actor. + chan <- atomically newTChan + -- Next, create the context and run the ReaderT action. + let context = ActorContext chan + ioAction = runReaderT actorAction context + -- Return the information. + return (ioAction, context) --- | Monitors the actor at the specified address. --- If an exception is raised in the monitored actor's --- thread, it is wrapped in an 'ActorException' and --- forwarded to the monitoring actor. If the monitored --- actor terminates, an 'ActorException' is raised in --- the monitoring Actor -monitor :: Address -> ActorM () -monitor addr = do - me <- self - let mons = ctxMonitors . addrContext $ addr - liftIO $ modifyMVar_ mons (return . insert me) - --- | Like `monitor`, but bi-directional -link :: Address -> ActorM () -link addr = do - monitor addr - mons <- asks ctxMonitors - liftIO $ modifyMVar_ mons (return . insert addr) +-- | Spawns the given actor on another thread and returns a handle to it. +spawnActor :: ActorMessage msg => ActorM msg () -> IO (ActorHandle msg) +spawnActor actorAction = do + -- Wrap the actor action. + (ioAction, context) <- wrapActor actorAction + -- Fork the actor's IO action to another thread. + thread <- forkIO ioAction + -- Return the handle. + return $ ActorHandle context thread --- | Kill the actor at the specified address -kill :: Address -> ActorM () -kill = liftIO . killThread . addrThread +-- | Runs the given actor on the current thread. +-- This function effectively turns the current thread into the actor's thread. +-- Obviously, this means that this function will block until the actor exits. +-- You probably want to use this for your "main" actor. +runActor :: ActorMessage msg => ActorM msg () -> IO () +runActor actorAction = do + -- Wrap the actor action. We discard the context, because we won't be + -- returning a handle to this actor. + (ioAction, _) <- wrapActor actorAction + -- Execute the IO action on the current thread. + ioAction --- | The current status of an actor -status :: Address -> ActorM ThreadStatus -status = liftIO . threadStatus . addrThread +-- }}} --- | Sets the specified flag in the actor's environment -setFlag :: Flag -> ActorM () -setFlag flag = do - fs <- asks ctxFlags - liftIO $ modifyMVar_ fs (return . setF flag) +-- {{{ Utility functions --- | Clears the specified flag in the actor's environment -clearFlag :: Flag -> ActorM () -clearFlag flag = do - fs <- asks ctxFlags - liftIO $ modifyMVar_ fs (return . clearF flag) +-- | Gets the mail box for the given handle. +handleMailBox :: ActorMessage msg => ActorHandle msg -> MailBox msg +handleMailBox = acMailBox . ahContext --- | Toggles the specified flag in the actor's environment -toggleFlag :: Flag -> ActorM () -toggleFlag flag = do - fs <- asks ctxFlags - liftIO $ modifyMVar_ fs (return . toggleF flag) +-- | Gets the thread ID for the given actor handle. +actorThread :: ActorMessage msg => ActorHandle msg -> ThreadId +actorThread = ahThread --- | Checks if the specified flag is set in the actor's environment -testFlag :: Flag -> ActorM Bool -testFlag flag = do - fs <- asks ctxFlags - liftIO $ withMVar fs (return . isSetF flag) +-- }}} diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,30 +1,20 @@ -Copyright Alex Constandache & Forkk 2014 - -All rights reserved. - -Redistribution and use in source and binary forms, with or without -modification, are permitted provided that the following conditions are met: - - * Redistributions of source code must retain the above copyright - notice, this list of conditions and the following disclaimer. +Copyright (c) 2014 Forkk - * Redistributions in binary form must reproduce the above - copyright notice, this list of conditions and the following - disclaimer in the documentation and/or other materials provided - with the distribution. +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: - * Neither the names of Forkk, Alex Constandache, or any other - contributors may be used to endorse or promote products derived - from this software without specific prior written permission. +The above copyright notice and this permission notice shall be included +in all copies or substantial portions of the Software. -THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS -"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT -LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR -A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT -OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, -SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT -LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, -DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY -THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT -(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE -OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +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,10 @@ +Hactor +====== + +Hactor is a library which aims to provide lightweight actors similar to Erlang +processes. + +This project is inspired by, and was originally a fork of +[thespian](https://hackage.haskell.org/package/thespian), but I decided that it +would be better and easier to just rewrite it. + diff --git a/hactor.cabal b/hactor.cabal --- a/hactor.cabal +++ b/hactor.cabal @@ -1,66 +1,72 @@ -- The name of the package. -name: hactor +name: hactor --- The package version. See the Haskell package versioning policy --- (http://www.haskell.org/haskellwiki/Package_versioning_policy) for --- standards guiding when and how versions should be incremented. -version: 0.1.0.0 +-- The package version. See the Haskell package versioning policy (PVP) +-- for standards guiding when and how versions should be incremented. +-- http://www.haskell.org/haskellwiki/Package_versioning_policy +-- PVP summary: +-- +-+------- breaking API changes +-- | | +----- non-breaking API additions +-- | | | +--- code changes with no API change +version: 1.0.0.0 -- A short (one-line) description of the package. -synopsis: Lightweight Erlang-style actors for Haskell. +synopsis: Lightweight Erlang-style actors for Haskell. -- A longer description of the package. -description: This is a fork of Thespian, a library which aims to provide lightweight Erlang-style actors for Haskell. +description: Hactor is a library which aims to provide lightweight Erlang-style actors for Haskell. -- URL for the project homepage or repository. -homepage: https://github.com/Forkk/hactor +homepage: https://github.com/Forkk/hactor -- The license under which the package is released. -license: BSD3 +license: MIT -- The file containing the license text. -license-file: LICENSE +license-file: LICENSE -- The package author(s). -author: Alex Constandache, Forkk +author: Forkk -- An email address to which users can send suggestions, bug reports, -- and patches. -maintainer: forkk@forkk.net +maintainer: forkk@forkk.net -- A copyright notice. --- Copyright: +copyright: Copyright Forkk 2014 -- Stability of the pakcage (experimental, provisional, stable...) -stability: experimental - -category: Concurrency - -build-type: Simple +stability: experimental +category: Concurrency +build-type: Simple -- Extra files to be distributed with the package, such as examples or -- a README. --- Extra-source-files: +extra-source-files: README.md -- Constraint on the version of Cabal needed to build this package. -cabal-version: >=1.2 - +cabal-version: >=1.20 library -- Modules exported by the library. - exposed-modules: Control.Concurrent.Actor + exposed-modules: Control.Concurrent.Actor -- Packages needed in order to build this package. - build-depends: base >= 3 && < 5, - mtl >= 1.1, - containers + build-depends: + base >=4.7 && <4.8 + , mtl >=2.2 && <2.3 + , stm >=2.4 && <2.5 + , stm-chans >=3.0 && <3.1 + , containers >=0.5 && <0.6 -- Extra options for ghc - ghc-options: -Wall + ghc-options: -Wall -- Modules not exported by this package. - -- Other-modules: + -- other-modules: -- Extra tools (e.g. alex, hsc2hs, ...) needed to build the source. - -- Build-tools: + -- build-tools: + + default-language: Haskell2010