Commit e9731720 authored by Dominik Charousset's avatar Dominik Charousset

maintenance

parent 3f66a8d1
......@@ -33,7 +33,6 @@ set(LIBCPPA_SRC
src/group_manager.cpp
src/local_actor.cpp
src/mailman.cpp
src/mock_scheduler.cpp
src/native_socket.cpp
src/network_manager.cpp
src/object.cpp
......
......@@ -38,7 +38,6 @@
#include "cppa/cppa.hpp"
#include "cppa/match.hpp"
#include "cppa/sb_actor.hpp"
#include "cppa/detail/mock_scheduler.hpp"
#include "cppa/context_switching_actor.hpp"
using std::cout;
......
......@@ -60,7 +60,6 @@ cppa/detail/list_member.hpp
cppa/detail/mailman.hpp
cppa/detail/map_member.hpp
cppa/detail/matches.hpp
cppa/detail/mock_scheduler.hpp
cppa/detail/native_socket.hpp
cppa/detail/network_manager.hpp
cppa/detail/object_array.hpp
......@@ -212,7 +211,6 @@ src/group.cpp
src/group_manager.cpp
src/local_actor.cpp
src/mailman.cpp
src/mock_scheduler.cpp
src/native_socket.cpp
src/network_manager.cpp
src/object.cpp
......
......@@ -62,8 +62,6 @@ class actor : public channel {
public:
~actor();
/**
* @brief Enqueues @p msg to the actor's mailbox and returns true if
* this actor is an scheduled actor that successfully changed
......@@ -109,22 +107,6 @@ class actor : public channel {
std::is_base_of<attachable,T>::value
>::type* = 0);
/**
* @brief Forces this actor to subscribe to the group @p what.
*
* The group will be unsubscribed if the actor finishes execution.
* @param what Group instance that should be joined.
*/
void join(group_ptr& what);
/**
* @brief Forces this actor to leave the group @p what.
* @param what Joined group that should be leaved.
* @note Groups are leaved automatically if the Actor finishes
* execution.
*/
void leave(const group_ptr& what);
/**
* @brief Links this actor to @p other.
* @param other Actor instance that whose execution is coupled to the
......
......@@ -85,6 +85,8 @@ class context_switching_actor : public detail::stacked_actor_mixin<
resume_result resume(util::fiber* from); //override
scheduled_actor_type impl_type();
protected:
context_switching_actor();
......
......@@ -424,11 +424,11 @@ namespace cppa {
/**
* @ingroup ActorManagement
* @brief Spawns a new context-switching actor.
* @param whom Instance of an event-based actor.
* @param impl Instance of an event-based actor.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
inline actor_ptr spawn(scheduled_actor* whom) {
return get_scheduler()->spawn(whom);
inline actor_ptr spawn(scheduled_actor* impl) {
return get_scheduler()->spawn(impl);
}
/**
......@@ -439,8 +439,8 @@ inline actor_ptr spawn(scheduled_actor* whom) {
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<scheduling_hint Hint>
inline actor_ptr spawn(std::function<void()> bhvr) {
return get_scheduler()->spawn(std::move(bhvr), Hint);
inline actor_ptr spawn(void_function fun) {
return get_scheduler()->spawn(std::move(fun), Hint);
}
/**
......@@ -450,8 +450,8 @@ inline actor_ptr spawn(std::function<void()> bhvr) {
* @param bhvr A function implementing the actor's behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
inline actor_ptr spawn(std::function<void()> bhvr) {
return get_scheduler()->spawn(std::move(bhvr), scheduled);
inline actor_ptr spawn(void_function fun) {
return get_scheduler()->spawn(std::move(fun), scheduled);
}
// forwards self_type as actor_ptr
......@@ -477,11 +477,11 @@ struct spawn_fwd_<self_type> {
* @param args Further argumetns to @p bhvr.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<scheduling_hint Hint, typename F, typename Arg0, typename... Args>
inline actor_ptr spawn(F bhvr, Arg0&& arg0, Args&&... args) {
template<scheduling_hint Hint, typename Fun, typename Arg0, typename... Args>
inline actor_ptr spawn(Fun fun, Arg0&& arg0, Args&&... args) {
return spawn<Hint>(
std::bind(
std::move(bhvr),
std::move(fun),
spawn_fwd_<typename util::rm_ref<Arg0>::type>::_(arg0),
spawn_fwd_<typename util::rm_ref<Args>::type>::_(args)...));
}
......@@ -496,9 +496,46 @@ inline actor_ptr spawn(F bhvr, Arg0&& arg0, Args&&... args) {
* @param args Further argumetns to @p bhvr.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<typename F, typename Arg0, typename... Args>
inline actor_ptr spawn(F&& bhvr, Arg0&& arg0, Args&&... args) {
return spawn<scheduled>(std::forward<F>(bhvr),
template<typename Fun, typename Arg0, typename... Args>
inline actor_ptr spawn(Fun&& fun, Arg0&& arg0, Args&&... args) {
return spawn<scheduled>(std::forward<Fun>(fun),
std::forward<Arg0>(arg0),
std::forward<Args>(args)...);
}
inline actor_ptr spawn_in_group(const group_ptr& grp, event_based_actor* impl) {
return get_scheduler()->spawn(impl, [grp](local_actor* ptr) {
ptr->join(grp);
});
}
template<scheduling_hint Hint>
inline actor_ptr spawn_in_group(const group_ptr& grp, void_function fun) {
return get_scheduler()->spawn(std::move(fun), Hint, [grp](local_actor* ptr){
ptr->join(grp);
});
}
inline actor_ptr spawn_in_group(const group_ptr& grp, void_function fun) {
return spawn_in_group<scheduled>(grp, std::move(fun));
}
template<scheduling_hint Hint, typename Fun, typename Arg0, typename... Args>
inline actor_ptr spawn_in_group(const group_ptr& grp,
Fun fun, Arg0&& arg0, Args&&... args) {
return spawn_in_group<Hint>(
grp,
std::bind(
std::move(fun),
spawn_fwd_<typename util::rm_ref<Arg0>::type>::_(arg0),
spawn_fwd_<typename util::rm_ref<Args>::type>::_(args)...));
}
template<typename Fun, typename Arg0, typename... Args>
inline actor_ptr spawn_in_group(const group_ptr& grp,
Fun&& fun, Arg0&& arg0, Args&&... args) {
return spawn_in_group<scheduled>(grp,
std::forward<Fun>(fun),
std::forward<Arg0>(arg0),
std::forward<Args>(args)...);
}
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_MOCK_SCHEDULER_HPP
#define CPPA_MOCK_SCHEDULER_HPP
#include <thread>
#include <utility>
#include "cppa/scheduler.hpp"
#include "cppa/detail/tdata.hpp"
namespace cppa { namespace detail {
class mock_scheduler : public scheduler {
public:
actor_ptr spawn(scheduled_actor* what);
actor_ptr spawn(std::function<void()> what, scheduling_hint);
static actor_ptr spawn_impl(std::function<void()> what);
static std::thread spawn_hidden_impl(std::function<void()> what, local_actor_ptr ctx);
void enqueue(scheduled_actor* what);
};
} } // namespace cppa::detail
#endif // CPPA_MOCK_SCHEDULER_HPP
......@@ -49,6 +49,7 @@ struct scheduled_actor_dummy : abstract_scheduled_actor {
void unbecome();
void do_become(behavior*, bool, bool);
bool has_behavior();
scheduled_actor_type impl_type();
};
} } // namespace cppa::detail
......
......@@ -34,6 +34,7 @@
#include <thread>
#include "cppa/scheduler.hpp"
#include "cppa/context_switching_actor.hpp"
#include "cppa/util/producer_consumer_list.hpp"
#include "cppa/detail/scheduled_actor_dummy.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
......@@ -56,7 +57,13 @@ class thread_pool_scheduler : public scheduler {
actor_ptr spawn(scheduled_actor* what);
actor_ptr spawn(std::function<void()> what, scheduling_hint hint);
actor_ptr spawn(scheduled_actor* what, init_callback init_cb);
actor_ptr spawn(void_function fun, scheduling_hint hint);
actor_ptr spawn(void_function what,
scheduling_hint hint,
init_callback init_cb);
private:
......@@ -67,11 +74,27 @@ class thread_pool_scheduler : public scheduler {
scheduled_actor_dummy m_dummy;
std::thread m_supervisor;
actor_ptr spawn_impl(scheduled_actor_ptr what, bool push_to_queue = true);
static void worker_loop(worker*);
static void supervisor_loop(job_queue*, scheduled_actor*);
actor_ptr spawn_impl(scheduled_actor_ptr what);
actor_ptr spawn_as_thread(void_function fun, init_callback cb, bool hidden);
# ifndef CPPA_DISABLE_CONTEXT_SWITCHING
template<typename F>
actor_ptr spawn_impl(void_function fun, scheduling_hint sh, F cb) {
if (sh == scheduled) {
scheduled_actor_ptr ptr{new context_switching_actor(std::move(fun))};
ptr->attach_to_scheduler(this);
cb(ptr.get());
return spawn_impl(std::move(ptr));
}
else {
return spawn_as_thread(std::move(fun), std::move(cb));
}
}
# endif
};
} } // namespace cppa::detail
......
......@@ -82,12 +82,14 @@ class event_based_actor : public detail::abstract_scheduled_actor {
void unbecome();
bool has_behavior();
scheduled_actor_type impl_type();
protected:
event_based_actor();
virtual bool has_behavior();
// provoke compiler errors for usage of receive() and related functions
/**
......
......@@ -77,6 +77,22 @@ class local_actor : public actor {
public:
/**
* @brief Causes this actor to subscribe to the group @p what.
*
* The group will be unsubscribed if the actor finishes execution.
* @param what Group instance that should be joined.
*/
void join(const group_ptr& what);
/**
* @brief Causes this actor to leave the group @p what.
* @param what Joined group that should be leaved.
* @note Groups are leaved automatically if the Actor finishes
* execution.
*/
void leave(const group_ptr& what);
/**
* @brief Finishes execution of this actor.
*
......
......@@ -44,6 +44,11 @@ enum class resume_result {
actor_done
};
enum scheduled_actor_type {
context_switching_impl,
event_based_impl
};
/**
* @brief A base class for cooperatively scheduled actors.
*/
......@@ -61,10 +66,12 @@ class scheduled_actor : public local_actor {
// called from worker thread
virtual resume_result resume(util::fiber* from) = 0;
scheduled_actor* attach_to_scheduler(scheduler* sched);
void attach_to_scheduler(scheduler* sched);
virtual bool has_behavior() = 0;
virtual scheduled_actor_type impl_type() = 0;
protected:
scheduler* m_scheduler;
......
......@@ -51,6 +51,10 @@ namespace cppa {
class scheduled_actor;
class scheduler_helper;
typedef std::function<void()> void_function;
typedef std::function<void(local_actor*)> init_callback;
/**
* @brief
*/
......@@ -81,17 +85,33 @@ class scheduler {
virtual void enqueue(scheduled_actor*) = 0;
/**
* @brief Spawns a new actor that executes <code>behavior->act()</code>
* @brief Spawns a new actor that executes <code>fun()</code>
* with the scheduling policy @p hint if possible.
*/
virtual actor_ptr spawn(std::function<void()> behavior,
scheduling_hint hint) = 0;
virtual actor_ptr spawn(void_function fun, scheduling_hint hint) = 0;
/**
* @brief Spawns a new actor that executes <code>behavior()</code>
* with the scheduling policy @p hint if possible and calls
* <code>init_cb</code> after the actor is initialized but before
* it starts execution.
*/
virtual actor_ptr spawn(void_function fun,
scheduling_hint hint,
init_callback init_cb) = 0;
/**
* @brief Spawns a new event-based actor.
*/
virtual actor_ptr spawn(scheduled_actor* what) = 0;
/**
* @brief Spawns a new event-based actor and calls
* <code>init_cb</code> after the actor is initialized but before
* it starts execution.
*/
virtual actor_ptr spawn(scheduled_actor* what, init_callback init_cb) = 0;
/**
* @brief Informs the scheduler about a converted context
* (a thread that acts as actor).
......
......@@ -47,7 +47,13 @@ enum scheduling_hint {
/**
* @brief Indicates that an actor should run in its own thread.
*/
detached
detached,
/**
* @brief Indicates that an actor should run in its own thread but should
* be ignored by {@link await_others_done()}.
*/
detached_and_hidden
};
......
......@@ -47,6 +47,7 @@
#include "cppa/pattern.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/detail/abstract_actor.hpp"
#include "cppa/intrusive/singly_linked_list.hpp"
......@@ -141,6 +142,8 @@ class thread_mapped_actor : public detail::stacked_actor_mixin<
};
typedef intrusive_ptr<thread_mapped_actor> thread_mapped_actor_ptr;
#endif // CPPA_DOCUMENTATION
} // namespace cppa
......
......@@ -185,8 +185,10 @@ int main(int, char**) {
std::vector<std::string> names = { "Plato", "Hume", "Kant",
"Nietzsche", "Descartes" };
for (size_t i = 0; i < 5; ++i) {
spawn(new philosopher(names[i], chopsticks[i],
chopsticks[(i+1)%5]) )->join(dinner_club);
spawn_in_group(dinner_club,
new philosopher(names[i],
chopsticks[i],
chopsticks[(i+1)%5]));
}
// tell philosophers to start thinking
send(dinner_club, atom("think"));
......
......@@ -73,20 +73,6 @@ actor::actor(const process_information_ptr& pptr)
}
}
actor::~actor() {
}
void actor::join(group_ptr& what) {
if (!what) return;
attach(what->subscribe(this));
}
void actor::leave(const group_ptr& what) {
if (!what) return;
attachable::token group_token(typeid(group::unsubscriber), what.get());
detach(group_token);
}
void actor::link_to(intrusive_ptr<actor>&& other) {
intrusive_ptr<actor> tmp(std::move(other));
link_to(tmp);
......
......@@ -84,6 +84,10 @@ detail::recursive_queue_node* context_switching_actor::receive_node() {
return e;
}
scheduled_actor_type context_switching_actor::impl_type() {
return context_switching_impl;
}
resume_result context_switching_actor::resume(util::fiber* from) {
using namespace detail;
scoped_self_setter sss{this};
......
......@@ -121,4 +121,8 @@ void event_based_actor::unbecome() {
m_bhvr_stack.pop_back();
}
scheduled_actor_type event_based_actor::impl_type() {
return event_based_impl;
}
} // namespace cppa
......@@ -84,4 +84,15 @@ void local_actor::on_exit() { }
void local_actor::init() { }
void local_actor::join(const group_ptr& what) {
if (!what) return;
attach(what->subscribe(this));
}
void local_actor::leave(const group_ptr& what) {
if (!what) return;
attachable::token group_token(typeid(group::unsubscriber), what.get());
detach(group_token);
}
} // namespace cppa
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011, 2012 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include "cppa/config.hpp"
#include <thread>
#include <atomic>
#include <iostream>
#include "cppa/self.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/attachable.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/thread_mapped_actor.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/mock_scheduler.hpp"
#include "cppa/detail/to_uniform_name.hpp"
using std::cout;
using std::cerr;
using std::endl;
namespace cppa { namespace detail {
typedef intrusive_ptr<thread_mapped_actor> thread_mapped_actor_ptr;
namespace {
void run_actor(thread_mapped_actor_ptr m_self) {
self.set(m_self.get());
try {
m_self->init();
m_self->initialized(true);
m_self->run();
m_self->on_exit();
}
catch (...) { }
self.set(nullptr);
std::atomic_thread_fence(std::memory_order_seq_cst);
dec_actor_count();
}
void run_hidden_actor(local_actor_ptr m_self,
std::function<void()> what) {
self.set(m_self.get());
try { what(); }
catch (...) { }
self.set(nullptr);
}
} // namespace <anonymous>
std::thread mock_scheduler::spawn_hidden_impl(std::function<void()> what,
local_actor_ptr ctx) {
return std::thread{run_hidden_actor, std::move(ctx), std::move(what)};
}
actor_ptr mock_scheduler::spawn_impl(std::function<void()> what) {
inc_actor_count();
std::atomic_thread_fence(std::memory_order_seq_cst);
thread_mapped_actor_ptr ctx{new thread_mapped_actor(std::move(what))};
std::thread{run_actor, ctx}.detach();
return std::move(ctx);
}
actor_ptr mock_scheduler::spawn(scheduled_actor*) {
cerr << "mock_scheduler::spawn(scheduled_actor*)" << endl;
abort();
return nullptr;
}
actor_ptr mock_scheduler::spawn(std::function<void()> what, scheduling_hint) {
return spawn_impl(what);
}
void mock_scheduler::enqueue(scheduled_actor*) {
cerr << "mock_scheduler::enqueue(scheduled_actor)" << endl;
abort();
}
} } // namespace detail
......@@ -38,11 +38,12 @@
#include <sys/time.h>
#include <sys/types.h>
#include "cppa/self.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/thread_mapped_actor.hpp"
#include "cppa/detail/mailman.hpp"
#include "cppa/detail/post_office.hpp"
#include "cppa/detail/mock_scheduler.hpp"
#include "cppa/detail/network_manager.hpp"
namespace {
......@@ -64,12 +65,22 @@ struct network_manager_impl : network_manager {
if (pipe(pipe_fd) != 0) {
CPPA_CRITICAL("cannot create pipe");
}
// create actors
m_post_office.reset(new thread_mapped_actor);
m_post_office_thread = mock_scheduler::spawn_hidden_impl(std::bind(post_office_loop, pipe_fd[0]), m_post_office);
m_mailman.reset(new thread_mapped_actor);
m_mailman_thread = mock_scheduler::spawn_hidden_impl(mailman_loop, m_mailman);
// store some data in local variables for lambdas
int pipe_fd0 = pipe_fd[0];
auto po_ptr = m_post_office;
auto mm_ptr = m_mailman;
// start threads
m_post_office_thread = std::thread([pipe_fd0, po_ptr]() {
scoped_self_setter sss{po_ptr.get()};
post_office_loop(pipe_fd0);
});
m_mailman_thread = std::thread([mm_ptr]() {
scoped_self_setter sss{mm_ptr.get()};
mailman_loop();
});
}
void stop() { // override
......
......@@ -36,7 +36,7 @@ namespace cppa {
scheduled_actor::scheduled_actor(bool enable_chained_send)
: local_actor(enable_chained_send), next(nullptr), m_scheduler(nullptr) { }
scheduled_actor* scheduled_actor::attach_to_scheduler(scheduler* sched) {
void scheduled_actor::attach_to_scheduler(scheduler* sched) {
CPPA_REQUIRE(sched != nullptr);
// init is called by the spawning actor, manipulate self to
// point to this actor
......@@ -47,7 +47,6 @@ scheduled_actor* scheduled_actor::attach_to_scheduler(scheduler* sched) {
// make sure scheduler is not set until init() is done
std::atomic_thread_fence(std::memory_order_seq_cst);
m_scheduler = sched;
return this;
}
bool scheduled_actor::initialized() {
......
......@@ -32,15 +32,20 @@
namespace cppa { namespace detail {
resume_result scheduled_actor_dummy::resume(util::fiber*) {
return resume_result::actor_blocked;
}
void scheduled_actor_dummy::quit(std::uint32_t) { }
void scheduled_actor_dummy::dequeue(behavior&) { }
void scheduled_actor_dummy::dequeue(partial_function&) { }
void scheduled_actor_dummy::link_to(intrusive_ptr<actor>&) { }
void scheduled_actor_dummy::unlink_from(intrusive_ptr<actor>&) { }
void scheduled_actor_dummy::detach(const attachable::token&) { }
bool scheduled_actor_dummy::attach(attachable*) { return false; }
void scheduled_actor_dummy::unbecome() { }
void scheduled_actor_dummy::do_become(behavior*, bool, bool) { }
bool scheduled_actor_dummy::has_behavior() { return false; }
resume_result scheduled_actor_dummy::resume(util::fiber*) {
return resume_result::actor_blocked;
}
bool scheduled_actor_dummy::establish_backlink(intrusive_ptr<actor>&) {
return false;
......@@ -50,10 +55,8 @@ bool scheduled_actor_dummy::remove_backlink(intrusive_ptr<actor>&) {
return false;
}
void scheduled_actor_dummy::detach(const attachable::token&) { }
bool scheduled_actor_dummy::attach(attachable*) { return false; }
void scheduled_actor_dummy::unbecome() { }
void scheduled_actor_dummy::do_become(behavior*, bool, bool) { }
bool scheduled_actor_dummy::has_behavior() { return false; }
scheduled_actor_type scheduled_actor_dummy::impl_type() {
return event_based_impl;
}
} } // namespace cppa::detail
......@@ -42,7 +42,6 @@
#include "cppa/thread_mapped_actor.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/mock_scheduler.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/thread_pool_scheduler.hpp"
......
......@@ -35,9 +35,9 @@
#include <iostream>
#include "cppa/event_based_actor.hpp"
#include "cppa/thread_mapped_actor.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/mock_scheduler.hpp"
#include "cppa/context_switching_actor.hpp"
#include "cppa/detail/thread_pool_scheduler.hpp"
......@@ -195,12 +195,35 @@ void thread_pool_scheduler::enqueue(scheduled_actor* what) {
m_queue.push_back(what);
}
actor_ptr thread_pool_scheduler::spawn_impl(scheduled_actor_ptr what,
bool push_to_queue) {
actor_ptr thread_pool_scheduler::spawn_as_thread(void_function fun,
init_callback cb,
bool hidden) {
if (!hidden) inc_actor_count();
thread_mapped_actor_ptr ptr{new thread_mapped_actor(std::move(fun))};
ptr->init();
ptr->initialized(true);
cb(ptr.get());
std::thread([hidden, ptr]() {
scoped_self_setter sss{ptr.get()};
try {
ptr->run();
ptr->on_exit();
}
catch (...) { }
std::atomic_thread_fence(std::memory_order_seq_cst);
if (!hidden) dec_actor_count();
}).detach();
return ptr;
}
actor_ptr thread_pool_scheduler::spawn_impl(scheduled_actor_ptr what) {
if (what->has_behavior()) {
inc_actor_count();
what->ref();
if (push_to_queue) m_queue.push_back(what.get());
// event-based actors are not pushed to the job queue on startup
if (what->impl_type() == context_switching_impl) {
m_queue.push_back(what.get());
}
}
else {
what->on_exit();
......@@ -208,29 +231,53 @@ actor_ptr thread_pool_scheduler::spawn_impl(scheduled_actor_ptr what,
return std::move(what);
}
actor_ptr thread_pool_scheduler::spawn(scheduled_actor* ptr) {
// do NOT push event-based actors to the queue on startup
scheduled_actor_ptr what{ptr};
what->attach_to_scheduler(this);
return spawn_impl(std::move(what), false);
actor_ptr thread_pool_scheduler::spawn(scheduled_actor* raw) {
scheduled_actor_ptr ptr{raw};
ptr->attach_to_scheduler(this);
return spawn_impl(std::move(ptr));
}
actor_ptr thread_pool_scheduler::spawn(scheduled_actor* raw, init_callback cb) {
scheduled_actor_ptr ptr{raw};
ptr->attach_to_scheduler(this);
cb(ptr.get());
return spawn_impl(std::move(ptr));
}
#ifndef CPPA_DISABLE_CONTEXT_SWITCHING
actor_ptr thread_pool_scheduler::spawn(std::function<void()> bhvr,
scheduling_hint hint) {
if (hint == scheduled) {
scheduled_actor_ptr ptr{new context_switching_actor(std::move(bhvr))};
actor_ptr thread_pool_scheduler::spawn(void_function fun, scheduling_hint sh) {
if (sh == scheduled) {
scheduled_actor_ptr ptr{new context_switching_actor(std::move(fun))};
ptr->attach_to_scheduler(this);
return spawn_impl(std::move(ptr));
}
else {
return mock_scheduler::spawn_impl(std::move(bhvr));
return spawn_as_thread(std::move(fun),
[](local_actor*) { },
sh == detached_and_hidden);
}
}
actor_ptr thread_pool_scheduler::spawn(void_function fun,
scheduling_hint sh,
init_callback init_cb) {
if (sh == scheduled) {
scheduled_actor_ptr ptr{new context_switching_actor(std::move(fun))};
ptr->attach_to_scheduler(this);
init_cb(ptr.get());
return spawn_impl(std::move(ptr));
}
else {
return spawn_as_thread(std::move(fun), std::move(init_cb), sh == detached_and_hidden);
}
}
#else
actor_ptr thread_pool_scheduler::spawn(std::function<void()> what,
scheduling_hint) {
return mock_scheduler::spawn_impl(std::move(what));
actor_ptr thread_pool_scheduler::spawn(void_function what, scheduling_hint) {
return spawn_as_thread(std::move(what), [](local_actor*) { });
}
actor_ptr thread_pool_scheduler::spawn(void_function what,
scheduling_hint,
init_callback init_cb) {
return spawn_as_thread(std::move(what), std::move(init_cb));
}
#endif
......
......@@ -23,7 +23,6 @@
#include "cppa/detail/demangle.hpp"
#include "cppa/uniform_type_info.hpp"
#include "cppa/process_information.hpp"
#include "cppa/detail/mock_scheduler.hpp"
#include "cppa/detail/thread_pool_scheduler.hpp"
#define CPPA_TEST_CATCH_BLOCK() \
......@@ -151,10 +150,6 @@ int main(int argc, char** argv) {
cout << "using thread_pool_scheduler" << endl;
set_scheduler(new cppa::detail::thread_pool_scheduler);
}
else if (sched == "mock_scheduler") {
cout << "using mock_scheduler" << endl;
set_scheduler(new cppa::detail::mock_scheduler);
}
else {
cerr << "unknown scheduler: " << sched << endl;
exit(1);
......
......@@ -28,12 +28,11 @@ size_t test__local_group() {
actor_ptr master = self;
for (int i = 0; i < 5; ++i) {
// spawn five workers and let them join local/foo
auto w = spawn([&master]() {
receive(on<int>() >> [&master](int v) {
spawn_in_group(foo_group, [master]() {
receive(on<int>() >> [master](int v) {
send(master, v);
});
});
w->join(foo_group);
}
send(foo_group, 2);
int result = 0;
......
......@@ -302,6 +302,107 @@ std::string behavior_test(actor_ptr et) {
return result;
}
#ifdef __clang__
class fixed_stack : public sb_actor<fixed_stack> {
friend class sb_actor<fixed_stack>;
static constexpr size_t max_size = 10;
std::vector<int> data;
behavior full = (
on(atom("push"), arg_match) >> [=](int) { },
on(atom("pop")) >> [=]() {
reply(atom("ok"), data.back());
data.pop_back();
become(&filled);
}
);
behavior filled = (
on(atom("push"), arg_match) >> [=](int what) {
data.push_back(what);
if (data.size() == max_size)
become(&full);
},
on(atom("pop")) >> [=]() {
reply(atom("ok"), data.back());
data.pop_back();
if (data.empty())
become(&empty);
}
);
behavior empty = (
on(atom("push"), arg_match) >> [=](int what) {
data.push_back(what);
become(&filled);
},
on(atom("pop")) >> [=]() {
reply(atom("failure"));
}
);
behavior& init_state = empty;
};
#else
class fixed_stack : public sb_actor<fixed_stack> {
friend class sb_actor<fixed_stack>;
static constexpr size_t max_size = 10;
std::vector<int> data;
behavior full;
behavior filled;
behavior empty;
behavior& init_state;
public:
fixed_stack() : init_state(empty) {
full = (
on(atom("push"), arg_match) >> [=](int) { },
on(atom("pop")) >> [=]() {
reply(atom("ok"), data.back());
data.pop_back();
become(&filled);
}
);
filled = (
on(atom("push"), arg_match) >> [=](int what) {
data.push_back(what);
if (data.size() == max_size)
become(&full);
},
on(atom("pop")) >> [=]() {
reply(atom("ok"), data.back());
data.pop_back();
if (data.empty())
become(&empty);
}
);
empty = (
on(atom("push"), arg_match) >> [=](int what) {
data.push_back(what);
become(&filled);
},
on(atom("pop")) >> [=]() {
reply(atom("failure"));
}
);
}
};
#endif
size_t test__spawn() {
using std::string;
CPPA_TEST(test__spawn);
......@@ -401,6 +502,36 @@ size_t test__spawn() {
await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl);
CPPA_IF_VERBOSE(cout << "test fixed_stack ... " << std::flush);
auto st = spawn(new fixed_stack);
// push 20 values
for (int i = 0; i < 20; ++i) send(st, atom("push"), i);
// pop 20 times
for (int i = 0; i < 20; ++i) send(st, atom("pop"));
// expect 10 failure messages
{
int i = 0;
receive_for(i, 10) (
on(atom("failure")) >> []() { }
);
}
// expect 10 {'ok', value} messages
{
std::vector<int> values;
int i = 0;
receive_for(i, 10) (
on(atom("ok"), arg_match) >> [&](int value) {
values.push_back(value);
}
);
std::vector<int> expected{9, 8, 7, 6, 5, 4, 3, 2, 1, 0};
CPPA_CHECK(values == expected);
}
// terminate st
send(st, atom("EXIT"), exit_reason::user_defined);
await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl);
int zombie_init_called = 0;
int zombie_on_exit_called = 0;
factory::event_based([&]() {
......@@ -434,6 +565,32 @@ size_t test__spawn() {
CPPA_CHECK_EQUAL(3, zombie_init_called);
CPPA_CHECK_EQUAL(3, zombie_on_exit_called);
auto f = factory::event_based([](std::string* name) {
self->become (
on(atom("get_name")) >> [name]() {
reply(atom("name"), *name);
}
);
});
auto a1 = f.spawn("alice");
auto a2 = f.spawn("bob");
send(a1, atom("get_name"));
receive (
on(atom("name"), arg_match) >> [&](const std::string& name) {
CPPA_CHECK_EQUAL("alice", name);
}
);
send(a2, atom("get_name"));
receive (
on(atom("name"), arg_match) >> [&](const std::string& name) {
CPPA_CHECK_EQUAL("bob", name);
}
);
auto kill_msg = make_any_tuple(atom("EXIT"), exit_reason::user_defined);
a1 << kill_msg;
a2 << kill_msg;
await_all_others_done();
CPPA_CHECK_EQUAL(behavior_test<testee_actor>(spawn(testee_actor{})), "wait4int");
CPPA_CHECK_EQUAL(behavior_test<event_testee>(spawn(new event_testee)), "wait4int");
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment