All the world's a stage... C++ Actors from Just::Thread Pro

Wednesday, 15 July 2015

Handling shared mutable state is probably the single hardest part of writing multithreaded code. There are lots of ways to address this problem; one of the common ones is the actors metaphor. Going back to Hoare's Communicating Sequential Processes, the idea is simple - you build your program out of a set of actors that send each other messages. Each actor runs normal sequential code, occasionally pausing to receive incoming messages from other actors. This means that you can analyse the behaviour of each actor independently; you only need to consider which messages might be received at each receive point. You could treat each actor as a state machine, with the messages triggering state transitions.

This is how Erlang processes work: each process is an actor, which runs independently from the other processes, except that they can send messages to each other. Just::thread Pro: Actors Edition adds library facilities to support this to C++. In the rest of this article I will describe how to write programs that take advantage of it. Though the details will differ, the approach can be used with other libraries that provide similar facilities, or with the actor support in other languages.

Simple Actors

Actors are embodied in the jss::actor class. You pass in a function or other callable object (such as a lambda function) to the constructor, and this function is then run on a background thread. This is exactly the same as for std::thread, except that the destructor waits for the actor thread to finish, rather than calling std::terminate.

void simple_function(){
    std::cout<<"simple actor\n";
}

int main(){
    jss::actor actor1(simple_function);
    jss::actor actor2([]{
        std::cout<<"lambda actor\n";
    });
}

The waiting destructor is nice, but it's really a side issue — the main benefit of actors is the ability to communicate using messages rather than shared state.

Sending and receiving messages

To send a message to an actor you just call the send() member function on the actor object, passing in whatever message you wish to send. send() is a function template, so you can send any type of message — there are no special requirements on the message type, just that it is a MoveConstructible type. You can also use the stream-insertion operator to send a message, which allows easy chaining e.g.

actor1.send(42);
actor2.send(MyMessage("some data"));
actor2<<Message1()<<Message2();

Sending a message to an actor just adds it to the actor's message queue. If the actor never checks the message queue then the message does nothing. To check the message queue, the actor function needs to call the receive() static member function of jss::actor. This is a static member function so that it always has access to the running actor, anywhere in the code — if it were a non-static member function then you would need to ensure that the appropriate object was passed around, which would complicate interfaces, and open up the possibility of the wrong object being passed around, and lifetime management issues.

The call to jss::actor::receive() will then block the actor's thread until a message that it can handle has been received. By default, the only message type that can be handled is jss::stop_actor. If a message of this type is sent to an actor then the receive() function will throw a jss::stop_actor exception. Uncaught, this exception will stop the actor running. In the following example, the only output will be "Actor running", since the actor will block at the receive() call until the stop message is sent, and when the message arrives, receive() will throw.

void stoppable_actor(){
    std::cout<<"Actor running"<<std::endl;
    jss::actor::receive();
    std::cout<<"This line is never run"<<std::endl;
}

int main(){
    jss::actor a1(stoppable_actor);
    std::this_thread::sleep_for(std::chrono::seconds(1));
    a1.send(jss::stop_actor());
}

Sending a "stop" message is common-enough that there's a special member function for that too: stop(). "a1.stop()" is thus equivalent to "a1.send(jss::stop_actor())".

Handling a message of another type requires that you tell the receive() call what types of message you can handle, which is done by chaining one or more calls to the match() function template. You must specify the type of the message to handle, and then provide a function to call if the message is received. Any messages other than jss::stop_actor not specified in a match() call will be removed from the queue, but otherwise ignored. In the following example, only messages of type "int" and "std::string" are accepted; the output is thus:

Waiting
42
Waiting
Hello
Waiting
Done

Here's the code:

void simple_receiver(){
    while(true){
        std::cout<<"Waiting"<<std::endl;
        jss::actor::receive()
            .match<int>([](int i){std::cout<<i<<std::endl;})
            .match<std::string>([](std::string const&s){std::cout<<s<<std::endl;});
    }
}

int main(){
    {
        jss::actor a(simple_receiver);
        a.send(true);
        a.send(42);
        a.send(std::string("Hello"));
        a.send(3.141);
        a.send(jss::stop_actor());
    } // wait for actor to finish
    std::cout<<"Done"<<std::endl;
}

It is important to note that the receive() call will block until it receives one of the messages you have told it to handle, or a jss::stop_actor message, and unexpected messages will be removed from the queue and discarded. This means the actors don't accumulate a backlog of messages they haven't yet said they can handle, and you don't have to worry about out-of-order messages messing up a receive() call.

These simple examples have just had main() sending messages to the actors. For a true actor-based system we need them to be able to send messages to each other, and reply to messages. Let's take a look at how we can do that.

Referencing one actor from another

Suppose we want to write a simple time service actor, that sends the current time back to any other actor that asks it for the time. At first thought it looks rather simple: write a simple loop that handles a "time request" message, gets the time, and sends a response. It won't be that much different from our simple_receiver() function above:

struct time_request{};

void time_server(){
    while(true){
        jss::actor::receive()
            .match<time_request>([](time_request r){
                auto now=std::chrono::system_clock::now();
                ????.send(now);
        });
    }
}

The problem is, we don't know which actor to send the response to — the whole point of this time server is that it will respond to a message from any other actor. The solution is to pass the sender as part of the message. We could just pass a pointer or reference to the jss::actor instance, but that requires that the actor knows the location of its own controlling object, which makes it more complicated — none of the examples we've had so far could know that, since the controlling object is a local variable declared in a separate function. What is needed instead is a simple means of identifying an actor, which the actor code can query — an actor reference. The type of an actor reference is jss::actor_ref, which is implicitly constructible from a jss::actor. An actor can also obtain a reference to itself by calling jss::actor::self(). jss::actor_ref has a send() member function and stream insertion operator for sending messages, just like jss::actor. So, we can put the sender of our time_request message in the message itself as a jss::actor_ref data member, and use that when sending the response.

struct time_request{
    jss::actor_ref sender;
};

void time_server(){
    while(true){
        jss::actor::receive()
            .match<time_request>([](time_request r){
                auto now=std::chrono::system_clock::now();
                r.sender<<now;
            });
    }
}

void query(jss::actor_ref server){
    server<<time_request{jss::actor::self()};
    jss::actor::receive()
        .match<std::chrono::system_clock::time_point>(
            [](std::chrono::system_clock::time_point){
                std::cout<<"time received"<<std::endl;
        });
}

Dangling references

If you use jss::actor_ref then you have to be prepared for the case that the referenced actor might have stopped executing by the time you send the message. In this case, any attempts to send a message through the jss::actor_ref instance will throw an exception of type [jss::no_actor]http://www.stdthread.co.uk/prodoc/headers/actor/no_actor.html. To be robust, our time server really ought to handle that too — if an unhandled exception of any type other than jss::stop_actor escapes the actor function then the library will call std::terminate. We should therefore wrap the attempt to send the message in a try-catch block.

void time_server(){
    while(true){
        jss::actor::receive()
            .match<time_request>([](time_request r){
                auto now=std::chrono::system_clock::now();
                try{
                    r.sender<<now;
                } catch(jss::no_actor&){}
            });
    }
}

We can now set up a pair of actors that play ping-pong:

struct pingpong{
    jss::actor_ref sender;
};

void pingpong_player(std::string message){
    while(true){
        try{
            jss::actor::receive()
                .match<pingpong>([&](pingpong msg){
                    std::cout<<message<<std::endl;
                    std::this_thread::sleep_for(std::chrono::milliseconds(50));
                    msg.sender<<pingpong{jss::actor::self()};
                });
        }
        catch(jss::no_actor&){
            std::cout<<"Partner quit"<<std::endl;
            break;
        }
    }
}

int main(){
    jss::actor ping(pingpong_player,"ping");
    jss::actor pong(pingpong_player,"pong");
    ping<<pingpong{pong};
    std::this_thread::sleep_for(std::chrono::seconds(1));
    ping.stop();
    pong.stop();
}

This will give output along the lines of the following:

ping
pong
ping
pong
ping
pong
ping
pong
ping
pong
ping
pong
ping
pong
ping
pong
ping
pong
ping
pong
Partner quit

The sleep in the player's message handler is to slow everything down — if you take it out then messages will go back and forth as fast as the system can handle, and you'll get thousands of lines of output. However, even at full speed the pings and pongs will be interleaved, because sending a message synchronizes with the receive() call that receives it.

That's essentially all there is to it — the rest is just application design. As an example of how it can all be put together, let's look at an implementation of the classic sleeping barber problem.

The Lazy Barber

For those that haven't met it before, the problem goes like this: Mr Todd runs a barber shop, but he's very lazy. If there are no customers in the shop then he likes to go to sleep. When a customer comes in they have to wake him up if he's asleep, take a seat if there is one, or come back later if there are no free seats. When Mr Todd has cut someone's hair, he must move on to the next customer if there is one, otherwise he can go back to sleep.

The barber actor

Let's start with the barber. He sleeps in his chair until a customer comes in, then wakes up and cuts the customer's hair. When he's done, if there is a waiting customer he cuts that customer's hair. If there are no customers, he goes back to sleep, and finally at closing time he goes home. This is shown as a state machine in figure 1.

void barber_func()
{
    bool go_home=false;
    unsigned haircuts=0;

    while(!go_home)
    {
        logger<<std::string("barber is sleeping");
        bool can_sleep=false;
        do
        {
            jss::actor::receive()
                .match<customer_waiting>(
                    [&](customer_waiting c){
                        logger<<std::string("barber is cutting hair");
                        c.customer<<start_haircut();
                        std::this_thread::sleep_for(std::chrono::milliseconds(1000*(1+(rand()%5))));
                        c.customer<<done_haircut();
                        ++haircuts;
                    })
                .match<no_customers>(
                    [&](no_customers){
                        can_sleep=true;
                    })
                .match<closing_time>(
                    [&](closing_time){
                        go_home=true;
                    });
        }
        while(!can_sleep && !go_home);
    }
    std::ostringstream os;
    os<<"barber is going home. He did "<<haircuts<<" haircuts today";
    logger<<os.str();
}

This translates into code as shown in listing 1.The wait loops for "sleeping" and "cutting hair" have been combined, since almost the same set of messages is being handled in each case — the only difference is that the "cutting hair" state also has the option of "no customers", which cannot be received in the "sleeping" state, and would be a no-op if it was. This allows the action associated with the "cutting hair" state to be entirely handled in the lambda associated with the customer_waiting message; splitting the wait loops would require that the code was extracted out to a separate function, which would make it harder to keep count of the haircuts. Of course, if you don't have a compiler with lambda support then you'll need to do that anyway. The logger is a global actor that receives std::strings as messages and writes them to std::cout. This avoids any synchronization issues with multiple threads trying to write out at once, but it does mean that you have to pre-format the strings, such as when logging the number of haircuts done in the day. The code for this is shown in listing 2.

void logger_func(){
    for(;;){
        jss::actor::receive()
            .match<std::string>([](std::string s){
                    std::cout<<s<<std::endl;
                });
    }
}

jss::actor logger(logger_func);

The customer actor

Let's look at things from the other side: the customer. The customer goes to town, and does some shopping. Each customer periodically goes into the barber shop to try and get a hair cut. If they manage, or the shop is closed, then they go home, otherwise they do some more shopping and go back to the barber shop later. This is shown in the state machine in figure 2.

enum class haircut_status{
    had_haircut,no_room,shop_closed
        };

haircut_status try_and_get_hair_cut(unsigned customer,jss::actor_ref shop){
    std::ostringstream os;
    os<<"customer "<<customer<<" goes into barber shop";
    logger<<os.str();
    try{
        shop<<customer_enters{jss::actor::self()};
    }
    catch(jss::no_actor){
        os.str("");
        os<<"customer "<<customer<<" finds barber shop is closed";
        logger<<os.str();
        return haircut_status::shop_closed;
    }

    haircut_status status=haircut_status::no_room;

    jss::actor::receive()
        .match<start_haircut>(
            [&](start_haircut)
            {
                os.str("");
                os<<"customer "<<customer<<" is having a haircut";
                logger<<os.str();
                jss::actor::receive()
                    .match<done_haircut>(
                        [&](done_haircut)
                        {
                            os.str("");
                            os<<"customer "<<customer<<" is done having a haircut";
                            logger<<os.str();
                        }
                        );
                status=haircut_status::had_haircut;
            }
            )
        .match<no_room>(
            [&](no_room)
            {
                os.str("");
                os<<"customer "<<customer<<" leaves because there is no room";
                logger<<os.str();
                status=haircut_status::no_room;
            }
            )
        .match<shop_closed>(
            [&](shop_closed)
            {
                os.str("");
                os<<"customer "<<customer<<" finds barber shop is closed";
                logger<<os.str();
                status=haircut_status::shop_closed;
            }
            );
    os.str("");
    os<<"customer "<<customer<<" leaves barber shop";
    logger<<os.str();
    try{
        shop<<customer_leaves();
    }
    catch(jss::no_actor){
    }
    return status;
}

void customer_func(unsigned i,jss::actor_ref shop)
{
    std::ostringstream os;
    os<<"customer "<<i<<" goes to town";
    logger<<os.str();

    haircut_status status;
    do{
        os.str("");
        os<<"customer "<<i<<" is shopping";
        logger<<os.str();
        std::this_thread::sleep_for(std::chrono::milliseconds(500*(rand()%20)));
    }
    while((status=try_and_get_hair_cut(i,shop))==haircut_status::no_room);
    os.str("");
    os<<"customer "<<i<<" is going home";
    logger<<os.str();
}

This translates into the code in listing 3. Note that the customer interacts with a "shop" actor that I haven't mentioned yet. It is often convenient to have an actor that represents shared state, since this allows access to the shared state from other actors to be serialized without needing an explicit mutex. In this case, the shop holds the number of waiting customers, which must be shared with any customers that come in, so they know whether there is a free chair or not. Rather than have the barber have to deal with messages from new customers while he is cutting hair, the shop acts as an intermediary. The customer also has to handle the case that the shop has already closed, so the shop reference might refer to an actor that has finished executing, and thus get a jss::no_actor exception when trying to send messages.

void shop_func(jss::actor_ref barber)
{
    bool closed=false;
    unsigned waiting_customers=0;

    unsigned const max_waiting_customers=3;

    std::ostringstream os;
    logger<<std::string("shop opens");

    while(!closed)
    {
        jss::actor::receive()
            .match<customer_enters>(
                [&](customer_enters c){
                    ++waiting_customers;
                    os.str("");
                    os<<"shop has "<<waiting_customers<<" customers";
                    logger<<os.str();
                    if(waiting_customers<=max_waiting_customers){
                        barber<<customer_waiting{c.customer};
                    } else
                        c.customer<<no_room();
                })
            .match<customer_leaves>(
                [&](customer_leaves){
                    if(!--waiting_customers)
                    {
                        logger<<"last customer left shop";
                        barber<<no_customers();
                    } else{
                        os.str("");
                        os<<"shop has "<<waiting_customers<<" customers";
                        logger<<os.str();
                    }
                })
            .match<closing_time>(
                [&](closing_time c){
                    logger<<std::string("shop closing");
                    closed=true;
                    barber<<c;
                });
    }
    while(waiting_customers){
        os.str("");
        os<<"shop has "<<waiting_customers<<" customers";
                        logger<<os.str();
        jss::actor::receive()
            .match<customer_enters>(
                [&](customer_enters c){
                    ++waiting_customers;
                    logger<<"customer turned away because shop closed";
                    c.customer<<shop_closed();
                })
            .match<customer_leaves>(
                [&](customer_leaves){
                    if(!--waiting_customers)
                    {
                        logger<<"last customer left shop";
                    }
                });
    }
    logger<<std::string("shop closed");
}

The message handlers for the shop are short, and just send out further messages to the barber or the customer, which is ideal for a simple state-manager — you don't want other actors waiting to perform simple state checks because the state manager is performing a lengthy operation; this is why we separated the shop from the barber. The shop has 2 states: open, where new customers are accepted provided there are fewer than the remaining spaces, and closed, where new customers are turned away, and the shop is just waiting for the last customer to leave. If a customer comes in, and there is a free chair then a message is sent to the barber that there is a customer waiting; if there is no space then a message is sent back to the customer to say so. When it's closing time then we switch to the "closing" state — in the code we exit the first while loop and enter the second. This is all shown in listing 4.

The messages are shown in listing 5, and the main() function that drives it all is in listing 6.

struct customer_waiting
{
    jss::actor_ref customer;
};

struct customer_enters
{
    jss::actor_ref customer;
};

struct customer_leaves
{};

struct start_haircut
{};

struct done_haircut
{};

struct closing_time
{};

struct no_room
{};

struct shop_closed
{};

struct no_customers
{};

int main()
{
    {
        jss::actor barber(barber_func);
        jss::actor barbershop(shop_func,jss::actor_ref(barber));

        unsigned const count=20;
        jss::actor customers[count];

        for(unsigned i=0;i<count;++i)
        {
            std::ostringstream os;

            os<<"Starting customer "<<i;
            logger<<os.str();

            customers[i]=jss::actor(customer_func,i,jss::actor_ref(barbershop));
        }

        std::this_thread::sleep_for(std::chrono::seconds(20));
        barbershop<<closing_time();
    }
    logger.stop();
}

Exit stage left

There are of course other ways of writing code to deal with any particular scenario, even if you stick to using actors. This article has shown some of the issues that you need to think about when using an actor-based approach, as well as demonstrating how it all fits together with the Just::Thread Pro actors library. Though the details will be different, the larger issues will be common to any implementation of the actor model.

Get the code

If you want to download the code for a better look, or to try it out, you can download it here.

Get your copy of Just:::Thread Pro

If you like the idea of working with actors in your code, now is the ideal time to get Just::Thread Pro. Get your copy now.

This blog post is based on an article that was printed in the July 2013 issue of CVu, the Journal of the ACCU.

Posted by Anthony Williams
[/ threading /] permanent link
Tags: cplusplus, actors, concurrency, multithreading
Stumble It! | Submit to Reddit | Submit to DZone

Comment on this post

If you liked this post, why not subscribe to the RSS feed or Follow me on Twitter? You can also subscribe to this blog by email using the form on the left.

Design and Content Copyright © 2005-2024 Just Software Solutions Ltd. All rights reserved. | Privacy Policy