Реализация dining_philosophers с использованием арбитра.

/*
 * A simple implementation of arbiter-based solution dining philosophers
 * problem. See description of this problem in Wikipedia:
 * http://en.wikipedia.org/wiki/Dining_philosophers_problem
 */

#include <iostream>
#include <iterator>
#include <numeric>
#include <cstdlib>
#include <vector>
#include <mutex>
#include <tuple>

#include <so_5/all.hpp>

// This request will be sent by the hungry agent.
struct msg_start_eating_request : public so_5::rt::message_t
{
    // Agent identifier.
    std::size_t m_philosopher;

    msg_start_eating_request( std::size_t philosopher )
        :   m_philosopher( philosopher )
    {}
};

// This signal will be sent to the hungry agent to whom
// eating is allowed.
struct msg_start_eating : public so_5::rt::signal_t {};

// This is notification about end of eating session.
struct msg_eating_finished : public so_5::rt::message_t
{
    // Agent identifier.
    std::size_t m_philosopher;

    msg_eating_finished( std::size_t philosopher )
        :   m_philosopher( philosopher )
    {}
};

// The state of a fork.
struct fork_state_t
{
    // Indication that the fork is in use.
    // It is true if some agent is holding it and waiting
    // for the right fork.
    // Or if agent is eating (e.g. agent holds both forks).
    bool m_in_use = false;

    // Indication that someone is waiting on that fork.
    // It could be agent which waits for his left fork
    // (but in that case agent is waiting only for the left fork).
    // Or it could be agent which waits for his right fork
    // (in that case agent is already holding his left fork).
    //
    // Value false means that there is no any waiting agents.
    bool m_someone_is_waiting = false;
};

// An arbiter who allows philosophers to eat.
//
// It also finishes the sample after test_duration seconds.
class a_arbiter_t : public so_5::rt::agent_t
{
    struct msg_shutdown : public so_5::rt::signal_t {};

public :
    a_arbiter_t(
        so_5::rt::environment_t & env,
        std::size_t philosophers_count,
        std::chrono::seconds test_duration )
        :   so_5::rt::agent_t( env )
        ,   m_philosophers_count( philosophers_count )
        ,   m_test_duration( test_duration )
        ,   m_forks( philosophers_count, fork_state_t() )
    {
        m_philosophers.reserve( philosophers_count );
    }

    // This method must be subsequently called during
    // the creation of the philosophers.
    void
    add_philosopher(
        const so_5::rt::mbox_ref_t & mbox )
    {
        m_philosophers.emplace_back( mbox );
    }

    virtual void
    so_define_agent() override
    {
        so_subscribe( so_direct_mbox() )
            .event( so_5::signal< msg_shutdown >,
                    [this]() { so_environment().stop(); } );

        so_subscribe( so_direct_mbox() )
            .event( &a_arbiter_t::evt_start_eating_request );

        so_subscribe( so_direct_mbox() )
            .event( &a_arbiter_t::evt_eating_finished );
    }

    virtual void
    so_evt_start()
    {
        so_environment().single_timer< msg_shutdown >(
                so_direct_mbox(),
                m_test_duration );
    }

    // Some philosopher is hungry and wants to eat.
    // This request is fulfilled or philosopher will wait for
    // one of his forks.
    void
    evt_start_eating_request( const msg_start_eating_request & evt )
    {
        try_allow_philosopher_to_eat( evt.m_philosopher );
    }

    // Some philosopher completed eating.
    // The forks of this philosopher will be marked as free and
    // if there is someone who is waiting for them it will be
    // processed.
    void
    evt_eating_finished( const msg_eating_finished & evt )
    {
        // Free left fork and check if it is necessary
        // for the left neighbor as right fork...
        auto & left_fork = m_forks[ evt.m_philosopher ];
        left_fork.m_in_use = false;

        if( left_fork.m_someone_is_waiting )
        {
            // Left neighbor is waiting this fork as his right fork.
            left_fork.m_someone_is_waiting = false;
            left_fork.m_in_use = true;

            enable_eating_for_philosopher( left_neighbor( evt.m_philosopher ) );
        }

        // Free right fork and check if it is necessary
        // for the right neighbor as left fork...
        const auto right_fork_index = right_neighbor( evt.m_philosopher );
        auto & right_fork = m_forks[ right_fork_index ];
        right_fork.m_in_use = false;

        if( right_fork.m_someone_is_waiting )
        {
            // Right neighbor is waiting this fork as his left fork.
            right_fork.m_someone_is_waiting = false;
            try_allow_philosopher_to_eat( right_fork_index );
        }
    }

private :
    // Count of the philosophers in the test.
    const std::size_t m_philosophers_count;

    // Duration of the sample.
    const std::chrono::seconds m_test_duration;

    // States of the forks.
    std::vector< fork_state_t > m_forks;

    // Mboxes for the philosophers.
    // They are necessary to send msg_start_eating signals
    // for a philosopher when his right fork is taken to him.
    std::vector< so_5::rt::mbox_ref_t > m_philosophers;

    void
    try_allow_philosopher_to_eat( std::size_t philosopher )
    {
        // Left fork must be free to start the process.
        auto & left_fork = m_forks[ philosopher ];
        if( left_fork.m_in_use )
        {
            // Just mark that there is a waiting philosopher for this fork.
            // No more can be done now.
            left_fork.m_someone_is_waiting = true;
        }
        else
        {
            // This philosopher acquired his left fork.
            left_fork.m_in_use = true;

            // Checking availability of this right fork.
            const auto right_fork_index = right_neighbor( philosopher );
            auto & right_fork = m_forks[ right_fork_index ];
            if( right_fork.m_in_use )
            {
                // Just mark that there is a waiting philosopher for this fork.
                // No more can be done now.
                right_fork.m_someone_is_waiting = true;
            }
            else
            {
                // This philosopher acquired his right fork.
                right_fork.m_in_use = true;

                // Philosopher can start eating.
                enable_eating_for_philosopher( philosopher );
            }
        }
    }

    std::size_t
    left_neighbor( std::size_t index ) const
    {
        if( index ) return index - 1;
        else return m_philosophers_count - 1;
    }

    std::size_t
    right_neighbor( std::size_t index ) const
    {
        return (index + 1) % m_philosophers_count;
    }

    void
    enable_eating_for_philosopher( std::size_t philosopher )
    {
        m_philosophers[ philosopher ]->deliver_signal< msg_start_eating >();
    }
};

// A philosopher agent.
// Does the infinite loop of think()/eat() methods.
//
// The switch from thinking to eating is done automatically
// when think() method finishes. As opposite the switch from
// eating to thinking is done automatically after return
// from eat() method.
class a_philosopher_t : public so_5::rt::agent_t
{
    struct msg_start_thinking : public so_5::rt::signal_t {};

public :
    a_philosopher_t(
        so_5::rt::environment_t & env,
        std::size_t index,
        const so_5::rt::mbox_ref_t & arbiter_mbox )
        :   so_5::rt::agent_t( env )
        ,   m_index( index )
        ,   m_arbiter_mbox( arbiter_mbox )
    {}

    virtual void
    so_define_agent()
    {
        so_subscribe( so_direct_mbox() )
            .event( so_5::signal< msg_start_thinking >,
                    &a_philosopher_t::evt_start_thinking );

        so_subscribe( so_direct_mbox() )
            .event( so_5::signal< msg_start_eating >,
                    &a_philosopher_t::evt_start_eating );
    }

    virtual void
    so_evt_start()
    {
        initiate_thinking();
    }

    void
    evt_start_thinking()
    {
        think();

        m_arbiter_mbox->deliver_message(
                new msg_start_eating_request( m_index ) );
    }

    void
    evt_start_eating()
    {
        eat();

        m_arbiter_mbox->deliver_message(
                new msg_eating_finished( m_index ) );

        initiate_thinking();
    }

protected :
    virtual void
    think()
    {
        const auto p = std::chrono::milliseconds( std::rand() % 250 );

        std::this_thread::sleep_for( p );
    }

    virtual void
    eat()
    {
        const auto p = std::chrono::milliseconds( std::rand() % 250 );

        std::this_thread::sleep_for( p );
    }

private :
    // Agent identifier.
    const std::size_t m_index;

    // Arbiter mbox. Necessary for sending requests and notifications.
    const so_5::rt::mbox_ref_t m_arbiter_mbox;

    void
    initiate_thinking()
    {
        so_direct_mbox()->deliver_signal< msg_start_thinking >();
    }
};

void
init( so_5::rt::environment_t & env,
    const std::size_t philosophers_count,
    const std::chrono::seconds test_duration )
{
    auto coop = env.create_coop( "dining_philosophers_with_arbiter",
            // All philosophers will be active objects.
            so_5::disp::active_obj::create_disp_binder( "active_obj" ) );

    auto arbiter = coop->add_agent(
            new a_arbiter_t( env, philosophers_count, test_duration ),
            // But the arbiter will work on different context.
            so_5::rt::create_default_disp_binder() );

    for( std::size_t i = 0; i != philosophers_count; ++i )
    {
        auto p = coop->add_agent(
                new a_philosopher_t(
                    env,
                    i,
                    arbiter->so_direct_mbox() ) );

        arbiter->add_philosopher( p->so_direct_mbox() );
    }

    env.register_coop( std::move( coop ) );
}

std::tuple< std::size_t, std::chrono::seconds >
process_command_line_args( int argc, char ** argv )
{
    std::size_t philosophers = 5;
    unsigned int test_duration = 20;

    if( argc > 1 )
    {
        philosophers = std::atoi( argv[ 1 ] );
        if( philosophers < 2 || philosophers > 10000 )
            throw std::invalid_argument(
                    "philosophers count must be in [2..10000]" );
    }
    if( argc > 2 )
    {
        test_duration = std::atoi( argv[ 2 ] );
        if( !test_duration || test_duration > 3600 )
            throw std::invalid_argument(
                    "philosophers count must be in [1..3600] seconds" );
    }

    return std::make_tuple( philosophers, std::chrono::seconds( test_duration ) );
}

int
main( int argc, char ** argv )
{
    try
    {
        auto params = process_command_line_args( argc, argv );

        so_5::launch(
                [params]( so_5::rt::environment_t & env )
                {
                    init( env, std::get<0>(params), std::get<1>(params) );
                },
                []( so_5::rt::environment_params_t & p ) {
                    p.add_named_dispatcher( "active_obj",
                            so_5::disp::active_obj::create_disp() );
                } );
    }
    catch( const std::exception & ex )
    {
        std::cerr << "Error: " << ex.what() << std::endl;
        return 1;
    }

    return 0;
}