Commit 55f9f4d3 authored by neverlord's avatar neverlord

maintenance (removed delegate, streamlined scheduling interface)

parent 46d6207a
......@@ -20,7 +20,6 @@ libcppa_la_SOURCES = \
src/channel.cpp \
src/converted_thread_context.cpp \
src/cppa.cpp \
src/delegate.cpp \
src/demangle.cpp \
src/deserializer.cpp \
src/duration.cpp \
......@@ -100,7 +99,6 @@ nobase_library_include_HEADERS = \
cppa/detail/converted_thread_context.hpp \
cppa/detail/decorated_tuple.hpp \
cppa/detail/default_uniform_type_info_impl.hpp \
cppa/detail/delegate.hpp \
cppa/detail/demangle.hpp \
cppa/detail/disablable_delete.hpp \
cppa/detail/empty_tuple.hpp \
......
......@@ -55,7 +55,6 @@ struct pool_job
{
abstract_event_based_actor* job;
handler(abstract_event_based_actor* mjob) : job(mjob) { }
bool still_ready() { return true; }
void exec_done()
{
if (!job->deref()) delete job;
......@@ -67,11 +66,6 @@ struct pool_job
}
};
class boost_threadpool_scheduler;
void enqueue_to_bts(boost_threadpool_scheduler* where,
abstract_scheduled_actor *what);
class boost_threadpool_scheduler : public scheduler
{
......@@ -80,17 +74,16 @@ class boost_threadpool_scheduler : public scheduler
public:
void start() /*override*/
void start()
{
m_pool.size_controller().resize(std::max(num_cores(), 4));
}
void stop() /*override*/
void stop()
{
m_pool.wait();
}
void schedule(abstract_scheduled_actor* what) /*override*/
void enqueue(abstract_scheduled_actor* what)
{
auto job = static_cast<abstract_event_based_actor*>(what);
boost::threadpool::schedule(m_pool, pool_job{job});
......@@ -98,7 +91,7 @@ class boost_threadpool_scheduler : public scheduler
actor_ptr spawn(abstract_event_based_actor* what)
{
what->attach_to_scheduler(enqueue_to_bts, this);
what->attach_to_scheduler(this);
inc_actor_count();
CPPA_MEMORY_BARRIER();
intrusive_ptr<abstract_event_based_actor> ctx(what);
......@@ -113,12 +106,6 @@ class boost_threadpool_scheduler : public scheduler
};
void enqueue_to_bts(boost_threadpool_scheduler* where,
abstract_scheduled_actor* what)
{
where->schedule(what);
}
} } // namespace cppa::detail
using std::cout;
......
......@@ -154,8 +154,6 @@ src/abstract_scheduled_actor.cpp
src/invokable.cpp
cppa/detail/thread_pool_scheduler.hpp
src/thread_pool_scheduler.cpp
cppa/detail/delegate.hpp
src/delegate.cpp
cppa/detail/receive_loop_helper.hpp
cppa/util/wrapped.hpp
cppa/detail/boxed.hpp
......
......@@ -55,11 +55,11 @@ class abstract_event_based_actor : public detail::abstract_scheduled_actor
public:
void dequeue(behavior&) /*override*/;
void dequeue(behavior&); //override
void dequeue(partial_function&) /*override*/;
void dequeue(partial_function&); //override
void resume(util::fiber*, resume_callback* callback) /*override*/;
void resume(util::fiber*, resume_callback* callback); //override
/**
* @brief Initializes the actor by defining an initial behavior.
......@@ -71,13 +71,11 @@ class abstract_event_based_actor : public detail::abstract_scheduled_actor
*/
virtual void on_exit();
template<typename Scheduler>
abstract_event_based_actor*
attach_to_scheduler(void (*enq_fun)(Scheduler*, abstract_scheduled_actor*),
Scheduler* sched)
inline abstract_event_based_actor* attach_to_scheduler(scheduler* sched)
{
m_enqueue_to_scheduler.reset(enq_fun, sched, this);
this->init();
CPPA_REQUIRE(sched != nullptr);
m_scheduler = sched;
init();
return this;
}
......
......@@ -31,16 +31,17 @@
#ifndef SCHEDULED_ACTOR_HPP
#define SCHEDULED_ACTOR_HPP
#include "cppa/scheduler.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/abstract_actor.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/util/fiber.hpp"
#include "cppa/intrusive/singly_linked_list.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
#include "cppa/detail/delegate.hpp"
namespace cppa { class scheduler; }
namespace cppa { namespace detail {
......@@ -57,7 +58,7 @@ class abstract_scheduled_actor : public abstract_actor<local_actor>
protected:
std::atomic<int> m_state;
delegate m_enqueue_to_scheduler;
scheduler* m_scheduler;
typedef abstract_actor super;
typedef super::queue_node_guard queue_node_guard;
......@@ -113,17 +114,7 @@ class abstract_scheduled_actor : public abstract_actor<local_actor>
abstract_scheduled_actor(int state = done);
template<typename Scheduler>
abstract_scheduled_actor(void (*enqueue_fun)(Scheduler*,
abstract_scheduled_actor*),
Scheduler* sched)
: next(nullptr)
, m_state(ready)
, m_enqueue_to_scheduler(enqueue_fun, sched, this)
, m_has_pending_timeout_request(false)
, m_active_timeout_id(0)
{
}
abstract_scheduled_actor(scheduler* sched);
void quit(std::uint32_t reason);
......@@ -136,10 +127,6 @@ class abstract_scheduled_actor : public abstract_actor<local_actor>
struct resume_callback
{
virtual ~resume_callback();
// actor could continue computation
// - if false: actor interrupts
// - if true: actor continues
virtual bool still_ready() = 0;
// called if an actor finished execution
virtual void exec_done() = 0;
};
......
......@@ -69,15 +69,15 @@ class converted_thread_context : public abstract_actor<local_actor>
// called if the converted thread finished execution
void cleanup(std::uint32_t reason = exit_reason::normal);
void quit(std::uint32_t reason) /*override*/;
void quit(std::uint32_t reason); //override
void enqueue(actor* sender, any_tuple&& msg) /*override*/;
void enqueue(actor* sender, any_tuple&& msg); //override
void enqueue(actor* sender, any_tuple const& msg) /*override*/;
void enqueue(actor* sender, any_tuple const& msg); //override
void dequeue(behavior& rules) /*override*/;
void dequeue(behavior& rules); //override
void dequeue(partial_function& rules) /*override*/;
void dequeue(partial_function& rules) ; //override
inline decltype(m_mailbox)& mailbox()
{
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 DELEGATE_HPP
#define DELEGATE_HPP
namespace cppa { namespace detail {
class delegate
{
typedef void (*void_fun)(void*, void*);
void_fun m_fun;
void* m_arg1;
void* m_arg2;
public:
template<typename Arg1, typename Arg2, typename Function>
delegate(Function* fun, Arg1* a1, Arg2* a2)
: m_fun(reinterpret_cast<void_fun>(fun))
, m_arg1(reinterpret_cast<void*>(a1))
, m_arg2(reinterpret_cast<void*>(a2))
{
}
template<typename Arg1, typename Arg2, typename Function>
void reset(Function* fun, Arg1* a1, Arg2* a2)
{
m_fun = reinterpret_cast<void_fun>(fun);
m_arg1 = reinterpret_cast<void*>(a1);
m_arg2 = reinterpret_cast<void*>(a2);
}
void operator()();
};
} } // namespace cppa::detail
#endif // DELEGATE_HPP
......@@ -46,6 +46,8 @@ class mock_scheduler : public scheduler
static actor_ptr spawn(scheduled_actor*);
void enqueue(detail::abstract_scheduled_actor*);
};
} } // namespace cppa::detail
......
......@@ -47,15 +47,15 @@ class thread_pool_scheduler : public scheduler
struct worker;
void start() /*override*/;
void start();// override
void stop() /*override*/;
void stop();// override
void schedule(abstract_scheduled_actor* what) /*override*/;
void enqueue(abstract_scheduled_actor* what);// override
actor_ptr spawn(abstract_event_based_actor* what);
actor_ptr spawn(abstract_event_based_actor* what); //override
actor_ptr spawn(scheduled_actor* behavior, scheduling_hint hint);
actor_ptr spawn(scheduled_actor* behavior, scheduling_hint hint); //override
private:
......
......@@ -40,7 +40,6 @@
#include "cppa/either.hpp"
#include "cppa/pattern.hpp"
#include "cppa/detail/delegate.hpp"
#include "cppa/detail/yield_interface.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
......@@ -64,23 +63,15 @@ class yielding_actor : public abstract_scheduled_actor
public:
template<typename Scheduler>
yielding_actor(scheduled_actor* behavior,
void (*enqueue_fun)(Scheduler*, abstract_scheduled_actor*),
Scheduler* sched)
: super(enqueue_fun, sched)
, m_fiber(&yielding_actor::run, this)
, m_behavior(behavior)
{
}
yielding_actor(scheduled_actor* behavior, scheduler* sched);
~yielding_actor() /*override*/;
~yielding_actor(); //override
void dequeue(behavior& bhvr) /*override*/;
void dequeue(behavior& bhvr); //override
void dequeue(partial_function& fun) /*override*/;
void dequeue(partial_function& fun); //override
void resume(util::fiber* from, resume_callback* callback) /*override*/;
void resume(util::fiber* from, resume_callback* callback); //override
};
......
......@@ -37,20 +37,22 @@
#include "cppa/self.hpp"
#include "cppa/atom.hpp"
#include "cppa/cow_tuple.hpp"
#include "cppa/actor.hpp"
#include "cppa/channel.hpp"
#include "cppa/cow_tuple.hpp"
#include "cppa/attachable.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/scheduling_hint.hpp"
#include "cppa/abstract_event_based_actor.hpp"
#include "cppa/util/duration.hpp"
namespace cppa {
//class local_actor;
class scheduled_actor;
class scheduler_helper;
class abstract_event_based_actor;
namespace detail { class abstract_scheduled_actor; }
/**
* @brief
......@@ -80,6 +82,8 @@ class scheduler
*/
virtual void stop();
virtual void enqueue(detail::abstract_scheduled_actor*) = 0;
/**
* @brief Spawns a new actor that executes <code>behavior->act()</code>
* with the scheduling policy @p hint if possible.
......
......@@ -28,19 +28,15 @@
\******************************************************************************/
#include <iostream>
#include "cppa/cppa.hpp"
#include "cppa/config.hpp"
#include "cppa/to_string.hpp"
#include "cppa/exception.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/detail/yield_interface.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
using std::cout;
using std::endl;
#include <iostream>
namespace cppa { namespace detail {
namespace {
......@@ -48,10 +44,20 @@ void dummy_enqueue(void*, abstract_scheduled_actor*) { }
types_array<atom_value, std::uint32_t> t_atom_ui32_types;
}
abstract_scheduled_actor::abstract_scheduled_actor(scheduler* sched)
: next(nullptr)
, m_state(ready)
, m_scheduler(sched)
, m_has_pending_timeout_request(false)
, m_active_timeout_id(0)
{
CPPA_REQUIRE(sched != nullptr);
}
abstract_scheduled_actor::abstract_scheduled_actor(int state)
: next(nullptr)
, m_state(state)
, m_enqueue_to_scheduler(dummy_enqueue, static_cast<void*>(nullptr), this)
, m_scheduler(nullptr)
, m_has_pending_timeout_request(false)
, m_active_timeout_id(0)
{
......@@ -80,7 +86,8 @@ void abstract_scheduled_actor::enqueue_node(queue_node* node)
{
if (m_state.compare_exchange_weak(state, ready))
{
m_enqueue_to_scheduler();
CPPA_REQUIRE(m_scheduler != nullptr);
m_scheduler->enqueue(this);
return;
}
break;
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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/detail/delegate.hpp"
namespace cppa { namespace detail {
void delegate::operator()()
{
m_fun(m_arg1, m_arg2);
}
} } // namespace cppa::detail
......@@ -39,6 +39,7 @@
#include "cppa/attachable.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/abstract_event_based_actor.hpp"
#include "cppa/detail/thread.hpp"
#include "cppa/detail/actor_count.hpp"
......@@ -47,6 +48,7 @@
#include "cppa/detail/converted_thread_context.hpp"
using std::cout;
using std::cerr;
using std::endl;
namespace {
......@@ -92,4 +94,10 @@ actor_ptr mock_scheduler::spawn(scheduled_actor* behavior, scheduling_hint)
return spawn(behavior);
}
void mock_scheduler::enqueue(detail::abstract_scheduled_actor*)
{
cerr << "mock_scheduler::enqueue" << endl;
abort();
}
} } // namespace detail
......@@ -32,6 +32,9 @@
#include <cstddef>
#include <iostream>
#include "cppa/config.hpp"
#include "cppa/abstract_event_based_actor.hpp"
#include "cppa/detail/invokable.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/mock_scheduler.hpp"
......@@ -45,12 +48,6 @@ namespace cppa { namespace detail {
namespace {
void enqueue_fun(cppa::detail::thread_pool_scheduler* where,
cppa::detail::abstract_scheduled_actor* what)
{
where->schedule(what);
}
typedef unique_lock<mutex> guard_type;
typedef std::unique_ptr<thread_pool_scheduler::worker> worker_ptr;
typedef intrusive::single_reader_queue<thread_pool_scheduler::worker> worker_queue;
......@@ -142,9 +139,9 @@ struct thread_pool_scheduler::worker
{
abstract_scheduled_actor* job;
handler() : job(nullptr) { }
bool still_ready() { return true; }
void exec_done()
{
CPPA_REQUIRE(job != nullptr);
if (!job->deref()) delete job;
dec_actor_count();
job = nullptr;
......@@ -186,7 +183,7 @@ void thread_pool_scheduler::supervisor_loop(job_queue* jqueue,
abstract_scheduled_actor* dummy)
{
std::vector<worker_ptr> workers;
size_t num_workers = std::max<size_t>(thread::hardware_concurrency() * 2, 8);
auto num_workers = std::max<size_t>(thread::hardware_concurrency() * 2, 8);
for (size_t i = 0; i < num_workers; ++i)
{
worker_ptr wptr(new worker(jqueue, dummy));
......@@ -214,7 +211,7 @@ void thread_pool_scheduler::stop()
super::stop();
}
void thread_pool_scheduler::schedule(abstract_scheduled_actor* what)
void thread_pool_scheduler::enqueue(abstract_scheduled_actor* what)
{
m_queue.push_back(what);
}
......@@ -234,10 +231,15 @@ actor_ptr thread_pool_scheduler::spawn_impl(abstract_scheduled_actor* what,
actor_ptr thread_pool_scheduler::spawn(abstract_event_based_actor* what)
{
// do NOT push event-based actors to the queue on startup
return spawn_impl(what->attach_to_scheduler(enqueue_fun, this), false);
return spawn_impl(what->attach_to_scheduler(this));
}
#ifndef CPPA_DISABLE_CONTEXT_SWITCHING
#ifdef CPPA_DISABLE_CONTEXT_SWITCHING
actor_ptr thread_pool_scheduler::spawn(scheduled_actor* bhvr, scheduling_hint)
{
return mock_scheduler::spawn(bhvr);
}
#else
actor_ptr thread_pool_scheduler::spawn(scheduled_actor* bhvr,
scheduling_hint hint)
{
......@@ -247,14 +249,9 @@ actor_ptr thread_pool_scheduler::spawn(scheduled_actor* bhvr,
}
else
{
return spawn_impl(new yielding_actor(bhvr, enqueue_fun, this));
return spawn_impl(new yielding_actor(bhvr, this));
}
}
#else
actor_ptr thread_pool_scheduler::spawn(scheduled_actor* bhvr, scheduling_hint)
{
return mock_scheduler::spawn(bhvr);
}
#endif
} } // namespace cppa::detail
......@@ -39,6 +39,13 @@
namespace cppa { namespace detail {
yielding_actor::yielding_actor(scheduled_actor* behavior, scheduler* sched)
: super(sched)
, m_fiber(&yielding_actor::run, this)
, m_behavior(behavior)
{
}
yielding_actor::~yielding_actor()
{
delete m_behavior;
......@@ -159,8 +166,6 @@ void yielding_actor::resume(util::fiber* from, resume_callback* callback)
case yield_state::ready:
{
break;
//if (callback->still_ready()) break;
//else return;
}
case yield_state::blocked:
{
......@@ -170,8 +175,6 @@ void yielding_actor::resume(util::fiber* from, resume_callback* callback)
case abstract_scheduled_actor::ready:
{
break;
//if (callback->still_ready()) break;
//else return;
}
case abstract_scheduled_actor::blocked:
{
......
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