Control.Concurrent.Actors

Documentation

Here we demonstrate a binary tree of actors that supports concurrent insert and query operations:

 import Control.Concurrent.Actors
 import Control.Applicative
 import Control.Concurrent.MVar
 
 -- the actor equivalent of a Nil leaf node:
 nil :: Behavior Operation
 nil = Receive $ do
     (Query _ var) <- received 
     send var False -- signal Int is not present in tree
     return nil     -- await next message
 
    <|> do          -- else, Insert received
     l <- spawn nil -- spawn child nodes
     r <- spawn nil
     branch l r . val <$> received  -- create branch from inserted val
     
 -- a branch node with a value 'v' and two children
 branch :: Node -> Node -> Int -> Behavior Operation    
 branch l r v = loop where
     loop = Receive $ do
         m <- received 
         case compare (val m) v of
              LT -> send l m
              GT -> send r m
              EQ -> case m of -- signal Int present in tree:
                         (Query _ var) -> send var True
                         _             -> return ()
         return loop
 
 type Node = Mailbox Operation
 
 -- operations supported by the network:
 data Operation = Insert { val :: Int }
                | Query { val :: Int
                        , sigVar :: Mailbox Bool }
 
 insert :: Node -> Int -> IO ()
 insert t = send t . Insert
 
 query :: Node -> Int -> IO Bool
 query t a = do
     -- turn an MVar into a Mailbox actors can send to with 'out'
     v <- newEmptyMVar
     send t (Query a $ out v)
     takeMVar v

You can use the tree defined above in GHCi:

>>> :l TreeExample.hs
Ok
>>> t <- spawn nil
>>> query t 7
False
>>> insert t 7
>>> query t 7
True

Actor Behaviors

data Action i a

In the Actor Model, at each step an actor...

• processes a single received message
• may spawn new actors
• may send messages to other actors
• returns the Behavior for processing the next message

These actions take place within the Action i monad, where i is the type of the input message the actor receives.

N.B.: the MonadIO instance here is an abstraction leak. An example of a good use of liftIO might be to give an Action access to a source of randomness.

Instances

Arrow Action

ArrowZero Action

ArrowPlus Action

ArrowChoice Action

ArrowApply Action

ArrowLoop Action

Category Action

MonadReader i (Action i)

Monad (Action i)

Functor (Action i)

MonadFix (Action i)

MonadPlus (Action i)

Applicative (Action i)

Alternative (Action i)

MonadIO (Action i)

newtype Behavior i

An actor is created by spawning a Behavior. Behaviors consist of a composed Action that is executed when a message is received and returns the Behavior for processing the next input.

Constructors

Receive

Fields

headAction :: Action i (Behavior i)

Instances

Contravariant Behavior

Monoid (Behavior i)

b1 mplus b2 has the headAction of b2 begin where the yield occured in b1, i.e. b2's first input will be the final input handed to b1.

Composing and Transforming Behaviors

(<.|>) :: Behavior i -> Behavior i -> Behavior i

Sequence two Behaviors. After the first yields the second takes over, discarding the message the former was processing. See also the Monoid instance for Behavior.

 b <.|> b' = b `mappend` constB b'

Available actions

Message passing

data Mailbox a

One can send a messages to a Mailbox where it will be processed according to an actor's defined Behavior

 type Joined (Mailbox a) = a

Instances

Contravariant Mailbox

SplitChan Mailbox Messages

NewSplitChan Mailbox Messages

Sources (Mailbox a)

out :: SplitChan i x => i a -> Mailbox a

Convert the input side of a SplitChan to a Mailbox. Useful for sending data out from an actor system via a channel created in IO.

send :: MonadIO m => Mailbox a -> a -> m ()

Send a message asynchronously to an actor receiving from Mailbox. See also out for converting other types of chans to Mailbox.

 send b = liftIO . writeChan b

send' :: MonadIO m => Mailbox a -> a -> m ()

A strict send:

 send' b a = a `seq` send b a

(<->) :: MonadIO m => a -> m (Mailbox a) -> m (Mailbox a)

Like send but supports chaining sends by returning the Mailbox. Convenient for initializing an Actor with its first input after spawning, e.g.

     do mb <- 0 <-> spawn foo

received :: Action i i

Return the message received to start this Action block. N.B the value returned here does not change between calls in the same Action.

 received = ask

guardReceived :: (i -> Bool) -> Action i i

Return received message matching predicate, otherwise yield.

 guardReceived p = ask >>= \i-> guard (p i) >> return i

Spawning actors

class Sources s

We extend the actor model to support joining (or synchronizing) multiple Mailboxes to a single Behavior input type, using a new class with an associated type. Functionality is best explained by example:

Spawn an actor returning it's Mailbox, and send it its first message:

 sumTuple :: Behavior (Int, Int)

 do b <- spawn sumTuple
    send b (4, 1) 
    ...

But now we would like our sumTuple actor to receive each number from a different concurrent actor:

 do (b1, b2) <- spawn sumTuple
    b3 <- spawn (multipliesBy2AndSendsTo b1)
    send b3 2
    send b2 1
    ...

Lastly spawn an actor that starts immediately on an infinite supply of ()s, and supplies an endless stream of Ints to sumTuple

 do (b1, b2) <- spawn sumTuple
    () <- spawn (sendsIntsTo b2)
    send b1 4
    ...

Associated Types

type Joined s

Instances

Sources ()

 type Joined () = ()

Represents an endless supply of ()s. Allows spawn-ing a Behavior () that starts immediately and loops until it yield-s, e.g.

 do () <- spawn startsImmediately -- :: Behavior ()

Sources (Mailbox a)

(Sources a, Sources b) => Sources (a, b)

(Sources a, Sources b, Sources c) => Sources (a, b, c)

(Sources a, Sources b, Sources c, Sources d) => Sources (a, b, c, d)

(Sources a, Sources b, Sources c, Sources d, Sources e) => Sources (a, b, c, d, e)

(Sources a, Sources b, Sources c, Sources d, Sources e, Sources f) => Sources (a, b, c, d, e, f)

(Sources a, Sources b, Sources c, Sources d, Sources e, Sources f, Sources g) => Sources (a, b, c, d, e, f, g)

spawn :: (MonadIO m, Sources s) => Behavior (Joined s) -> m s

Fork an actor performing the specified Behavior. N.B. an actor begins execution of its headBehavior only after a message becomes available to process; for sending an initial message to an actor right after spawning it, (<|>) can be convenient.

Mailboxes and scoping

Straightforward use of the spawn function will be sufficient for forking actors in most cases, but launching mutually-communicating actors presents a problem.

In cases where a Behavior needs access to its own Mailbox or that of an actor that must be forked later, the MonadFix instance should be used. GHC's "Recursive Do" notation make this especially easy:

 
 beh = Receive $ do
     i <- received
     -- similar to the scoping in a "let" block:
     rec b1 <- spawn (senderTo b2)
         b2 <- spawn (senderTo b1)
         b3 <- spawn (senderTo b3)
     -- send initial messages to actors spawned above:
     send b3 i
     send b2 "first"
     yield

Building an actor computation

An actor computation can be halted immediately by calling yield, a synonym for mzero. When an Action calling yield is composed with another using <|> the second takes over processing the same input which the former yield-ed on.

Here is an example of a computation using guard which returns mzero if the test is false:

 foo c n = Receive $ 
       do i <- received
          guard (n<10)
          send c i
          return (foo c $ n+1)

   <|> do i <- received -- same as the 'i' above
          send c $ "TENTH INPUT: "++i
          return (foo c 0)

The Monoid instance for Behavior works on the same principle.

yield :: Action i a

Immediately give up processing an input, perhaps relinquishing the input to an Alternative computation or exiting the actor.

 yield = mzero

receive :: Action i (Behavior i) -> Action i (Behavior i)

Useful to make defining a continuing Behavior more readable as a "receive block", e.g.

 pairUpAndSendTo mb = Receive $ do
     a <- received
     receive $ do
         b <- received
         send mb (b,a)
         return (pairUpAndSendTo mb)

Defined as:

 receive = return . Receive

Composing and Transforming Mailboxes

We offer some operations to split and combine Mailboxes of sum and product types.

coproductMb :: Mailbox a -> Mailbox b -> Mailbox (Either a b)

contraProduct :: Contravariant f => f (Either a b) -> (f a, f b)

 contraProduct = contramap Left &&& contramap Right

zipMb :: Mailbox a -> Mailbox b -> Mailbox (a, b)

contraFanin :: Contravariant f => (b -> a) -> (c -> a) -> f a -> f (Either b c)

 contraFanin f g = contramap (f ||| g)

contraFanout :: Contravariant f => (a -> b) -> (a -> c) -> f (b, c) -> f a

 contraFanout f g = contramap (f &&& g)

Utility functions

These are useful for debugging Behaviors

runBehavior_ :: Behavior () -> IO ()

Run a Behavior () in the main thread, returning when the computation exits.

runBehavior :: Behavior a -> [a] -> IO ()

run a Behavior in the IO monad, taking its "messages" from the list.

Useful predefined Behaviors

printB :: (Show s, Eq n, Num n) => n -> Behavior s

Prints all messages to STDOUT in the order they are received, yield-ing immediately after n inputs are printed.

putStrB :: (Eq n, Num n) => n -> Behavior String

Like printB but using putStr.

signalB :: Mailbox () -> Behavior i

Sends a () to the passed chan. This is useful with mappend for signalling the end of some other Behavior.

 signalB c = Receive (send c () >> yield)

constB :: Behavior i -> Behavior i

A Behavior that discard its first input, returning the passed Behavior for processing subsequent inputs. Useful with Alternative or Monoid compositions when one wants to ignore the leftover yielded message.

 constB = Receive . return