Commit b363feae authored by neverlord's avatar neverlord

maintenance

parent 1e561870
......@@ -54,7 +54,6 @@ libcppa_la_SOURCES = \
src/singleton_manager.cpp \
src/stacked_event_based_actor.cpp \
src/string_serialization.cpp \
src/task_scheduler.cpp \
src/thread_pool_scheduler.cpp \
src/to_uniform_name.cpp \
src/unicast_network.cpp \
......@@ -127,7 +126,6 @@ nobase_library_include_HEADERS = \
cppa/detail/serialize_tuple.hpp \
cppa/detail/singleton_manager.hpp \
cppa/detail/swap_bytes.hpp \
cppa/detail/task_scheduler.hpp \
cppa/detail/tdata.hpp \
cppa/detail/thread.hpp \
cppa/detail/thread_pool_scheduler.hpp \
......
......@@ -153,8 +153,6 @@ src/native_socket.cpp
cppa/detail/post_office.hpp
src/post_office.cpp
cppa/detail/buffer.hpp
cppa/detail/task_scheduler.hpp
src/task_scheduler.cpp
cppa/detail/actor_count.hpp
src/actor_count.cpp
cppa/util/fiber.hpp
......
......@@ -44,13 +44,10 @@
namespace cppa { namespace detail {
class task_scheduler;
// A spawned, scheduled Actor.
class abstract_scheduled_actor : public abstract_actor<local_actor>
{
friend class task_scheduler;
friend class intrusive::single_reader_queue<abstract_scheduled_actor>;
abstract_scheduled_actor* next; // intrusive next pointer
......
......@@ -65,17 +65,17 @@ class invokable
inline invokable() : next(nullptr) { }
virtual ~invokable();
virtual bool invoke(any_tuple const&) const = 0;
virtual bool invoke(any_tuple const&) const;
// Suppress type checking.
virtual bool unsafe_invoke(any_tuple const& value) const = 0;
virtual bool unsafe_invoke(any_tuple const& value) const;
// Checks whether the types of @p value match the pattern.
virtual bool types_match(any_tuple const& value) const = 0;
virtual bool types_match(any_tuple const& value) const;
// Checks whether this invokable could be invoked with @p value.
virtual bool could_invoke(any_tuple const& value) const = 0;
virtual bool could_invoke(any_tuple const& value) const;
// Prepare this invokable.
virtual intermediate* get_intermediate(any_tuple const& value) = 0;
virtual intermediate* get_intermediate(any_tuple const& value);
// Prepare this invokable and suppress type checking.
virtual intermediate* get_unsafe_intermediate(any_tuple const& value) = 0;
virtual intermediate* get_unsafe_intermediate(any_tuple const& value);
};
......
......@@ -67,17 +67,21 @@ struct mailman_add_peer
class mailman_job
{
friend class intrusive::singly_linked_list<mailman_job>;
friend class intrusive::single_reader_queue<mailman_job>;
public:
enum job_type
{
invalid_type,
send_job_type,
add_peer_type,
kill_type
};
inline mailman_job() : next(nullptr), m_type(invalid_type) { }
mailman_job(process_information_ptr piptr,
actor_ptr const& from,
channel_ptr const& to,
......
......@@ -43,12 +43,14 @@ namespace cppa { namespace detail {
class post_office_msg
{
friend class intrusive::singly_linked_list<post_office_msg>;
friend class intrusive::single_reader_queue<post_office_msg>;
public:
enum msg_type
{
invalid_type,
add_peer_type,
add_server_socket_type,
proxy_exited_type
......@@ -85,6 +87,8 @@ class post_office_msg
inline proxy_exited(actor_proxy_ptr const& who) : proxy_ptr(who) { }
};
inline post_office_msg() : next(nullptr), m_type(invalid_type) { }
post_office_msg(native_socket_type arg0,
process_information_ptr const& arg1,
actor_proxy_ptr const& arg2,
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 TASK_SCHEDULER_HPP
#define TASK_SCHEDULER_HPP
#include "cppa/scheduler.hpp"
#include "cppa/detail/thread.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa { namespace detail {
class task_scheduler : public scheduler
{
typedef scheduler super;
typedef intrusive::single_reader_queue<abstract_scheduled_actor> job_queue;
job_queue m_queue;
scheduled_actor_dummy m_dummy;
thread m_worker;
static void worker_loop(job_queue*, abstract_scheduled_actor* dummy);
public:
virtual void start();
virtual void stop();
void schedule(abstract_scheduled_actor* what);
actor_ptr spawn(abstract_event_based_actor* what);
actor_ptr spawn(scheduled_actor*, scheduling_hint);
private:
actor_ptr spawn_impl(abstract_scheduled_actor* what);
};
} } // namespace cppa::detail
#endif // TASK_SCHEDULER_HPP
......@@ -35,6 +35,9 @@
namespace cppa { namespace intrusive {
/**
* @brief A forward iterator for intrusive lists.
*/
template<class T>
class iterator // : std::iterator<forward_iterator_tag, T>
{
......@@ -67,6 +70,10 @@ class iterator // : std::iterator<forward_iterator_tag, T>
return tmp;
}
inline explicit operator bool() { return m_ptr != nullptr; }
inline bool operator!() { return m_ptr != nullptr; }
inline reference operator*() { return *m_ptr; }
inline const_reference operator*() const { return *m_ptr; }
......
......@@ -40,6 +40,8 @@ namespace cppa { namespace intrusive {
/**
* @brief An intrusive, thread safe queue implementation.
* @note For implementation details see
* http://libcppa.blogspot.com/2011/04/mailbox-part-1.html
*/
template<typename T>
class single_reader_queue
......@@ -49,75 +51,66 @@ class single_reader_queue
public:
typedef T element_type;
typedef T value_type;
typedef size_t size_type;
typedef ptrdiff_t difference_type;
typedef value_type& reference;
typedef value_type const& const_reference;
typedef value_type* pointer;
typedef value_type const* const_pointer;
typedef singly_linked_list<value_type> cache_type;
/**
* @warning call only from the reader (owner)
*/
element_type* pop()
pointer pop()
{
wait_for_data();
element_type* result = take_head();
return result;
return take_head();
}
/**
* @warning call only from the reader (owner)
*/
element_type* try_pop()
pointer try_pop()
{
return take_head();
}
template<typename TimePoint>
element_type* try_pop(TimePoint const& abs_time)
{
return (timed_wait_for_data(abs_time)) ? take_head() : nullptr;
}
/**
* @warning call only from the reader (owner)
*/
void push_front(element_type* element)
template<typename TimePoint>
pointer try_pop(TimePoint const& abs_time)
{
element->next = m_head;
m_head = element;
return (timed_wait_for_data(abs_time)) ? take_head() : nullptr;
}
/**
* @warning call only from the reader (owner)
*/
void push_front(element_type* first, element_type* last)
void push_front(pointer element)
{
last->next = m_head;
m_head = first;
m_cache.push_front(element);
}
/**
* @warning call only from the reader (owner)
*/
//template<typename List>
//void push_front(List&& list)
void push_front(singly_linked_list<T>&& list)
void push_front(cache_type&& list)
{
if (!list.empty())
{
auto p = list.take();
if (p.first)
{
push_front(p.first, p.second);
}
}
m_cache.splice_after(m_cache.before_begin(), std::move(list));
}
// returns true if the queue was empty
bool _push_back(element_type* new_element)
bool _push_back(pointer new_element)
{
element_type* e = m_tail.load();
pointer e = m_stack.load();
for (;;)
{
new_element->next = e;
if (m_tail.compare_exchange_weak(e, new_element))
if (m_stack.compare_exchange_weak(e, new_element))
{
return (e == nullptr);
}
......@@ -127,16 +120,16 @@ class single_reader_queue
/**
* @reentrant
*/
void push_back(element_type* new_element)
void push_back(pointer new_element)
{
element_type* e = m_tail.load();
pointer e = m_stack.load();
for (;;)
{
new_element->next = e;
if (!e)
{
lock_type guard(m_mtx);
if (m_tail.compare_exchange_weak(e, new_element))
if (m_stack.compare_exchange_weak(e, new_element))
{
m_cv.notify_one();
return;
......@@ -144,7 +137,7 @@ class single_reader_queue
}
else
{
if (m_tail.compare_exchange_weak(e, new_element))
if (m_stack.compare_exchange_weak(e, new_element))
{
return;
}
......@@ -152,37 +145,38 @@ class single_reader_queue
}
}
inline cache_type& cache() { return m_cache; }
/**
* @warning call only from the reader (owner)
*/
bool empty()
inline bool empty() const
{
return m_cache.empty() && m_stack.load() == nullptr;
}
inline bool not_empty() const
{
return !m_head && !(m_tail.load());
return !empty();
}
single_reader_queue() : m_tail(nullptr), m_head(nullptr)
single_reader_queue() : m_stack(nullptr)
{
}
~single_reader_queue()
{
fetch_new_data();
element_type* e = m_head;
while (e)
{
element_type* tmp = e;
e = e->next;
delete tmp;
}
// empty the stack
(void) fetch_new_data();
}
private:
// exposed to "outside" access
std::atomic<element_type*> m_tail;
std::atomic<pointer> m_stack;
// accessed only by the owner
element_type* m_head;
cache_type m_cache;
// locked on enqueue/dequeue operations to/from an empty list
detail::mutex m_mtx;
......@@ -191,10 +185,10 @@ class single_reader_queue
template<typename TimePoint>
bool timed_wait_for_data(TimePoint const& timeout)
{
if (!m_head && !(m_tail.load()))
if (m_cache.empty() && !(m_stack.load()))
{
lock_type guard(m_mtx);
while (!(m_tail.load()))
while (!(m_stack.load()))
{
if (detail::wait_until(guard, m_cv, timeout) == false)
{
......@@ -207,30 +201,31 @@ class single_reader_queue
void wait_for_data()
{
if (!m_head && !(m_tail.load()))
if (m_cache.empty() && !(m_stack.load()))
{
lock_type guard(m_mtx);
while (!(m_tail.load())) m_cv.wait(guard);
while (!(m_stack.load())) m_cv.wait(guard);
}
}
// atomically set public_tail to nullptr and enqueue all
// atomically sets m_stack to nullptr and enqueues all elements to the cache
bool fetch_new_data()
{
element_type* e = m_tail.load();
pointer e = m_stack.load();
while (e)
{
if (m_tail.compare_exchange_weak(e, 0))
if (m_stack.compare_exchange_weak(e, 0))
{
// iterator to the last element before insertions begin
auto iter = m_cache.before_end();
// public_tail (e) has LIFO order,
// but private_head requires FIFO order
while (e)
{
// next iteration element
element_type* next = e->next;
// enqueue e to private_head
e->next = m_head;
m_head = e;
pointer next = e->next;
// insert e to private cache (convert to LIFO order)
m_cache.insert_after(iter, e);
// next iteration
e = next;
}
......@@ -242,15 +237,13 @@ class single_reader_queue
return false;
}
element_type* take_head()
pointer take_head()
{
if (m_head || fetch_new_data())
if (m_cache.not_empty() || fetch_new_data())
{
element_type* result = m_head;
m_head = result->next;
return result;
return m_cache.take_after(m_cache.before_begin());
}
return 0;
return nullptr;
}
};
......
......@@ -39,6 +39,7 @@
namespace cppa { namespace intrusive {
// like std::forward_list but intrusive and supports push_back()
template<class T>
class singly_linked_list
{
......@@ -59,25 +60,30 @@ class singly_linked_list
typedef ::cppa::intrusive::iterator<value_type> iterator;
typedef ::cppa::intrusive::iterator<value_type const> const_iterator;
singly_linked_list() : m_head(nullptr), m_tail(nullptr) { }
singly_linked_list() : m_head(), m_tail(&m_head) { }
singly_linked_list(singly_linked_list&& other)
: m_head(other.m_head), m_tail(other.m_tail)
singly_linked_list(singly_linked_list&& other) : m_head(), m_tail(&m_head)
{
other.m_head = other.m_tail = nullptr;
*this = std::move(other);
}
singly_linked_list& operator=(singly_linked_list&& other)
{
std::swap(m_head, other.m_head);
std::swap(m_tail, other.m_tail);
clear();
if (other.not_empty())
{
m_head.next = other.m_head.next;
m_tail = other.m_tail;
other.m_head.next = nullptr;
other.m_tail = &(other.m_head);
}
return *this;
}
static singly_linked_list from(std::pair<pointer, pointer> const& p)
{
singly_linked_list result;
result.m_head = p.first;
result.m_head.next = p.first;
result.m_tail = p.second;
return result;
}
......@@ -86,17 +92,21 @@ class singly_linked_list
// element access
inline reference front() { return *m_head; }
inline const_reference front() const { return *m_head; }
inline reference front() { return *(m_head.next); }
inline const_reference front() const { return *(m_head.next); }
inline reference back() { return *m_tail; }
inline const_reference back() const { return *m_tail; }
// iterators
inline iterator begin() { return m_head; }
inline const_iterator begin() const { return m_head; }
inline const_iterator cbegin() const { return m_head; }
inline iterator before_begin() { return &m_head; }
inline const_iterator before_begin() const { return &m_head; }
inline const_iterator before_cbegin() const { return &m_head; }
inline iterator begin() { return m_head.next; }
inline const_iterator begin() const { return m_head.next; }
inline const_iterator cbegin() const { return m_head.next; }
inline iterator before_end() { return m_tail; }
inline const_iterator before_end() const { return m_tail; }
......@@ -108,26 +118,34 @@ class singly_linked_list
// capacity
inline bool empty() const { return m_head == nullptr; }
inline bool empty() const { return m_head.next == nullptr; }
inline bool not_empty() const { return m_head.next != nullptr; }
// no size member function because it would have O(n) complexity
// modifiers
void clear()
{
while (m_head)
if (not_empty())
{
pointer next = m_head->next;
delete m_head;
m_head = next;
auto h = m_head.next;
while (h)
{
pointer next = h->next;
delete h;
h = next;
}
m_head.next = nullptr;
m_tail = &m_head;
}
m_tail = nullptr;
}
iterator insert_after(iterator i, pointer what)
{
what->next = i->next;
i->next = what;
if (i == m_tail) m_tail = what;
return what;
}
......@@ -143,24 +161,31 @@ class singly_linked_list
auto next = pos->next;
if (next)
{
if (next == m_tail) m_tail = pos.ptr();
pos->next = next->next;
delete next;
}
return pos->next;
}
void push_back(pointer what)
// removes the element after pos from the list and returns it
pointer take_after(iterator pos)
{
what->next = nullptr;
if (empty())
{
m_tail = m_head = what;
}
else
CPPA_REQUIRE(pos != nullptr);
auto next = pos->next;
if (next)
{
m_tail->next = what;
m_tail = what;
if (next == m_tail) m_tail = pos.ptr();
pos->next = next->next;
}
return next;
}
void push_back(pointer what)
{
what->next = nullptr;
m_tail->next = what;
m_tail = what;
}
template<typename... Args>
......@@ -169,19 +194,16 @@ class singly_linked_list
push_back(new value_type(std::forward<Args>(args)...));
}
// pop_back would have complexity O(n)
void push_front(pointer what)
{
if (empty())
{
what->next = nullptr;
m_tail = m_head = what;
push_back(what);
}
else
{
what->next = m_head;
m_head = what;
what->next = m_head.next;
m_head.next = what;
}
}
......@@ -193,27 +215,46 @@ class singly_linked_list
void pop_front()
{
if (!empty())
auto x = m_head.next;
if (x == nullptr)
{
// list is empty
return;
}
else if (x == m_tail)
{
auto next = m_head->next;
delete m_head;
m_head = next;
if (m_head == nullptr) m_tail = nullptr;
m_tail = &m_head;
m_head.next = nullptr;
}
else
{
m_head.next = x->next;
}
delete x;
}
// pop_back would have complexity O(n)
std::pair<pointer, pointer> take()
{
auto result = std::make_pair(m_head, m_tail);
m_head = m_tail = nullptr;
return result;
if (empty())
{
return {nullptr, nullptr};
}
else
{
auto result = std::make_pair(m_head.next, m_tail);
m_head.next = nullptr;
m_tail = &m_head;
return result;
}
}
void splice_after(iterator pos, singly_linked_list&& other)
{
CPPA_REQUIRE(pos != nullptr);
CPPA_REQUIRE(this != &other);
if (other.empty() == false)
if (other.not_empty())
{
auto next = pos->next;
auto pair = other.take();
......@@ -230,24 +271,15 @@ class singly_linked_list
template<typename UnaryPredicate>
void remove_if(UnaryPredicate p)
{
auto i = begin();
while (!empty() && p(*i))
{
pop_front();
i = begin();
}
if (empty()) return;
auto predecessor = i;
++i;
while (i != nullptr)
auto i = before_begin();
while (i->next != nullptr)
{
if (p(*i))
if (p(*(i->next)))
{
i = erase_after(predecessor);
erase_after(i);
}
else
{
predecessor = i;
++i;
}
}
......@@ -260,7 +292,7 @@ class singly_linked_list
private:
pointer m_head;
value_type m_head;
pointer m_tail;
};
......
......@@ -70,8 +70,8 @@ class partial_function
template<class... Args>
partial_function& splice(partial_function&& arg0, Args&&... args)
{
if (m_funs.empty()) m_funs = std::move(arg0.m_funs);
else m_funs.splice_after(m_funs.before_end(), std::move(arg0.m_funs));
m_funs.splice_after(m_funs.before_end(), std::move(arg0.m_funs));
arg0.m_cache.clear();
return splice(std::forward<Args>(args)...);
}
......
......@@ -33,7 +33,7 @@
namespace cppa { namespace util {
template<bool Stmt, typename T>
template<bool Stmt, typename T = void>
struct disable_if_c { };
template<typename T>
......
......@@ -16,18 +16,33 @@ class producer_consumer_list
public:
typedef T* value_ptr;
typedef T value_type;
typedef size_t size_type;
typedef ptrdiff_t difference_type;
typedef value_type& reference;
typedef value_type const& const_reference;
typedef value_type* pointer;
typedef value_type const* const_pointer;
struct node
{
value_ptr value;
pointer value;
std::atomic<node*> next;
node(value_ptr val) : value(val), next(nullptr) { }
char pad[CPPA_CACHE_LINE_SIZE - sizeof(value_ptr)- sizeof(std::atomic<node*>)];
node(pointer val) : value(val), next(nullptr) { }
static constexpr size_type payload_size =
sizeof(pointer) + sizeof(std::atomic<node*>);
static constexpr size_type pad_size =
(CPPA_CACHE_LINE_SIZE * ((payload_size / CPPA_CACHE_LINE_SIZE) + 1)) - payload_size;
// avoid false sharing
char pad[pad_size];
};
private:
static_assert(sizeof(node*) < CPPA_CACHE_LINE_SIZE,
"sizeof(node*) >= CPPA_CACHE_LINE_SIZE");
// for one consumer at a time
node* m_first;
char m_pad1[CPPA_CACHE_LINE_SIZE - sizeof(node*)];
......@@ -73,16 +88,16 @@ class producer_consumer_list
}
}
inline void push_back(value_ptr value)
inline void push_back(pointer value)
{
assert(value != nullptr);
push_impl(new node(value));
}
// returns nullptr on failure
value_ptr try_pop()
pointer try_pop()
{
value_ptr result = nullptr;
pointer result = nullptr;
while (m_consumer_lock.exchange(true))
{
detail::this_thread::yield();
......
......@@ -34,7 +34,6 @@
#include "cppa/to_string.hpp"
#include "cppa/exception.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/detail/task_scheduler.hpp"
#include "cppa/detail/yield_interface.hpp"
#include "cppa/detail/abstract_scheduled_actor.hpp"
......
......@@ -36,4 +36,16 @@ invokable::~invokable()
{
}
bool invokable::invoke(any_tuple const&) const { return false; }
bool invokable::unsafe_invoke(any_tuple const&) const { return false; }
bool invokable::types_match(any_tuple const&) const { return false; }
bool invokable::could_invoke(any_tuple const&) const { return false; }
intermediate* invokable::get_intermediate(any_tuple const&) { return 0; }
intermediate* invokable::get_unsafe_intermediate(any_tuple const&) { return 0; }
} } // namespace cppa::detail
......@@ -82,7 +82,7 @@ mailman_job::~mailman_job()
m_add_socket.~mailman_add_peer();
break;
}
case kill_type:
default:
{
// union doesn't contain a valid object
break;
......
......@@ -93,7 +93,7 @@ post_office_msg::~post_office_msg()
m_proxy_exited.~proxy_exited();
break;
}
default: throw std::logic_error("invalid post_office_msg type");
default: 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 <iostream>
#include "cppa/config.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/util/fiber.hpp"
#include "cppa/scheduled_actor.hpp"
#include "cppa/detail/invokable.hpp"
#include "cppa/detail/actor_count.hpp"
#include "cppa/detail/mock_scheduler.hpp"
#include "cppa/detail/task_scheduler.hpp"
#include "cppa/detail/yielding_actor.hpp"
#include "cppa/detail/yield_interface.hpp"
#include "cppa/detail/converted_thread_context.hpp"
using std::cout;
using std::cerr;
using std::endl;
namespace {
void enqueue_fun(cppa::detail::task_scheduler* where,
cppa::detail::abstract_scheduled_actor* what)
{
where->schedule(what);
}
} // namespace <anonmyous>
namespace cppa { namespace detail {
void task_scheduler::worker_loop(job_queue* jq, abstract_scheduled_actor* dummy)
{
cppa::util::fiber fself;
abstract_scheduled_actor* job = nullptr;
struct handler : abstract_scheduled_actor::resume_callback
{
abstract_scheduled_actor** job_ptr;
handler(abstract_scheduled_actor** jptr) : job_ptr(jptr) { }
bool still_ready()
{
return true;
}
void exec_done()
{
auto job = *job_ptr;
if (!job->deref()) delete job;
CPPA_MEMORY_BARRIER();
dec_actor_count();
}
};
handler h(&job);
for (;;)
{
job = jq->pop();
if (job == dummy)
{
return;
}
job->resume(&fself, &h);
}
}
void task_scheduler::start()
{
super::start();
m_worker = thread(worker_loop, &m_queue, &m_dummy);
}
void task_scheduler::stop()
{
m_queue.push_back(&m_dummy);
m_worker.join();
super::stop();
}
void task_scheduler::schedule(abstract_scheduled_actor* what)
{
if (what)
{
if (this_thread::get_id() == m_worker.get_id())
{
m_queue._push_back(what);
}
else
{
m_queue.push_back(what);
}
}
}
actor_ptr task_scheduler::spawn_impl(abstract_scheduled_actor* what)
{
inc_actor_count();
CPPA_MEMORY_BARRIER();
intrusive_ptr<abstract_scheduled_actor> ctx(what);
// add an implicit reference to ctx
ctx->ref();
m_queue.push_back(ctx.get());
return std::move(ctx);
}
actor_ptr task_scheduler::spawn(abstract_event_based_actor* what)
{
return spawn_impl(what->attach_to_scheduler(enqueue_fun, this));
}
#ifndef CPPA_DISABLE_CONTEXT_SWITCHING
actor_ptr task_scheduler::spawn(scheduled_actor* bhvr, scheduling_hint hint)
{
if (hint == detached) return mock_scheduler::spawn(bhvr);
return spawn_impl(new yielding_actor(bhvr, enqueue_fun, this));
}
#else
actor_ptr task_scheduler::spawn(scheduled_actor* bhvr, scheduling_hint)
{
return mock_scheduler::spawn(bhvr);
}
#endif
} } // namespace cppa::detail
......@@ -186,7 +186,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, 4);
size_t 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));
......
......@@ -21,7 +21,6 @@
#include "cppa/uniform_type_info.hpp"
#include "cppa/process_information.hpp"
#include "cppa/detail/mock_scheduler.hpp"
#include "cppa/detail/task_scheduler.hpp"
#include "cppa/detail/thread_pool_scheduler.hpp"
#define CPPA_TEST_CATCH_BLOCK() \
......@@ -133,12 +132,7 @@ int main(int argc, char** argv)
else if (found_key(i, args, "scheduler"))
{
auto& sched = i->second;
if (sched == "task_scheduler")
{
cout << "using task_scheduler" << endl;
cppa::set_scheduler(new cppa::detail::task_scheduler);
}
else if (sched == "thread_pool_scheduler")
if (sched == "thread_pool_scheduler")
{
cout << "using thread_pool_scheduler" << endl;
cppa::set_scheduler(new cppa::detail::thread_pool_scheduler);
......
......@@ -5,6 +5,40 @@
#include <string>
#include <cstddef>
#include <iostream>
#include <type_traits>
#include "cppa/util/enable_if.hpp"
#include "cppa/util/disable_if.hpp"
template<typename T1, typename T2>
inline bool cppa_check_value_fun_eq(T1 const& value1, T2 const& value2,
typename cppa::util::disable_if_c<std::is_integral<T1>::value && std::is_integral<T2>::value>::type* = 0)
{
return value1 == value2;
}
template<typename T1, typename T2>
inline bool cppa_check_value_fun_eq(T1 value1, T2 value2,
typename cppa::util::enable_if_c<std::is_integral<T1>::value && std::is_integral<T2>::value>::type* = 0)
{
return value1 == static_cast<T1>(value2);
}
template<typename T1, typename T2>
inline void cppa_check_value_fun(T1 const& value1, T2 const& value2,
char const* file_name,
int line_number,
size_t& error_count)
{
if (cppa_check_value_fun_eq(value1, value2) == false)
{
std::cerr << "ERROR in file " << file_name << " on line " << line_number
<< " => expected value: " << value1
<< ", found: " << value2
<< std::endl;
++error_count;
}
}
#define CPPA_TEST(name) \
struct cppa_test_scope \
......@@ -20,6 +54,7 @@ struct cppa_test_scope \
#define CPPA_TEST_RESULT cppa_ts.error_count
#ifdef CPPA_VERBOSE_CHECK
#define CPPA_IF_VERBOSE(line_of_code) line_of_code
#define CPPA_CHECK(line_of_code) \
if (!(line_of_code)) \
{ \
......@@ -33,6 +68,7 @@ else \
} \
((void) 0)
#else
#define CPPA_IF_VERBOSE(line_of_code) ((void) 0)
#define CPPA_CHECK(line_of_code) \
if (!(line_of_code)) \
{ \
......@@ -46,7 +82,10 @@ if (!(line_of_code)) \
std::cerr << err_msg << std::endl; \
++cppa_ts.error_count
#define CPPA_CHECK_EQUAL(lhs_loc, rhs_loc) CPPA_CHECK(((lhs_loc) == (rhs_loc)))
#define CPPA_CHECK_EQUAL(lhs_loc, rhs_loc) \
cppa_check_value_fun((lhs_loc), (rhs_loc), __FILE__, __LINE__, \
cppa_ts.error_count)
#define CPPA_CHECK_NOT_EQUAL(lhs_loc, rhs_loc) CPPA_CHECK(((lhs_loc) != (rhs_loc)))
size_t test__yield_interface();
......
......@@ -13,7 +13,16 @@ using std::string;
using namespace cppa;
using namespace cppa::util;
namespace { constexpr auto s_foo = atom("FooBar"); }
namespace {
constexpr auto s_foo = atom("FooBar");
inline std::ostream& operator<<(std::ostream& out, atom_value const& a)
{
return (out << to_string(a));
}
} // namespace <anonymous>
template<atom_value AtomValue, typename... Types>
void foo()
......
......@@ -23,12 +23,12 @@ size_t test__fixed_vector()
fixed_vector<int, 2> vec5 {3, 4};
vec4.insert(vec4.end(), vec5.begin(), vec5.end());
auto vec6 = vec4;
CPPA_CHECK_EQUAL(vec1.size(), 4);
CPPA_CHECK_EQUAL(vec2.size(), 4);
CPPA_CHECK_EQUAL(vec3.size(), 4);
CPPA_CHECK_EQUAL(vec4.size(), 4);
CPPA_CHECK_EQUAL(vec5.size(), 2);
CPPA_CHECK_EQUAL(vec6.size(), 4);
CPPA_CHECK_EQUAL(vec1.size(), static_cast<size_t>(4));
CPPA_CHECK_EQUAL(vec2.size(), static_cast<size_t>(4));
CPPA_CHECK_EQUAL(vec3.size(), static_cast<size_t>(4));
CPPA_CHECK_EQUAL(vec4.size(), static_cast<size_t>(4));
CPPA_CHECK_EQUAL(vec5.size(), static_cast<size_t>(2));
CPPA_CHECK_EQUAL(vec6.size(), static_cast<size_t>(4));
CPPA_CHECK_EQUAL(vec1.full(), true);
CPPA_CHECK_EQUAL(vec2.full(), false);
CPPA_CHECK_EQUAL(vec3.full(), true);
......
......@@ -32,6 +32,7 @@
#include "test.hpp"
#include "cppa/intrusive/singly_linked_list.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
using std::begin;
using std::end;
......@@ -62,6 +63,7 @@ inline bool operator==(int lhs, iint const& rhs)
}
typedef cppa::intrusive::singly_linked_list<iint> iint_list;
typedef cppa::intrusive::single_reader_queue<iint> iint_queue;
size_t test__intrusive_containers()
{
......@@ -84,9 +86,9 @@ size_t test__intrusive_containers()
ilist1.push_front(new iint(0));
auto i = ilist1.erase_after(ilist1.begin());
// i points to second element
CPPA_CHECK_EQUAL(*i, 2);
CPPA_CHECK_EQUAL(i->value, 2);
i = ilist1.insert_after(i, new iint(20));
CPPA_CHECK_EQUAL(*i, 20);
CPPA_CHECK_EQUAL(i->value, 20);
int iarr2[] = { 0, 2, 20, 3, 4, 5 };
CPPA_CHECK((std::equal(begin(iarr2), end(iarr2), begin(ilist1))));
......@@ -95,22 +97,59 @@ size_t test__intrusive_containers()
auto ilist2 = iint_list::from(p);
ilist2.emplace_front(1); // 1 0 2 20 3 4 5
i = ilist2.erase_after(ilist2.begin()); // 1 2 20 3 4 5
CPPA_CHECK_EQUAL(*i, 2);
CPPA_CHECK_EQUAL(i->value, 2);
ilist2.erase_after(i); // 1 2 3 4 5
CPPA_CHECK((std::equal(begin(iarr1), end(iarr1), begin(ilist2))));
CPPA_CHECK_EQUAL(s_iint_instances, 5);
// five elements + two dummies
CPPA_CHECK_EQUAL(s_iint_instances, 7);
ilist2.remove_if([](iint const& val) { return (val.value % 2) != 0; });
CPPA_CHECK_EQUAL(s_iint_instances, 2);
// two elements + two dummies
CPPA_CHECK_EQUAL(s_iint_instances, 4);
int iarr3[] = { 2, 4 };
CPPA_CHECK((std::equal(begin(iarr3), end(iarr3), begin(ilist2))));
auto x = ilist2.take_after(ilist2.before_begin());
CPPA_CHECK_EQUAL(x->value, 2);
delete x;
ilist2.clear();
CPPA_CHECK_EQUAL(s_iint_instances, 0);
// two dummies
CPPA_CHECK_EQUAL(s_iint_instances, 2);
CPPA_CHECK(ilist2.empty());
{
iint_queue iq;
for (int i = 0; i < 20; ++i) iq._push_back(new iint(i));
iint_list tmp;
for (int i = 0; i < 9; ++i)
{
tmp.push_back(iq.pop());
}
delete iq.pop();
iq.push_front(std::move(tmp));
CPPA_CHECK(tmp.empty());
CPPA_CHECK_EQUAL(std::distance(iq.cache().begin(), iq.cache().end()), 19);
std::unique_ptr<iint> iptr;
for (int i = 0; i < 9; ++i)
{
iptr.reset(iq.pop());
CPPA_CHECK(iptr);
if (iptr) CPPA_CHECK_EQUAL(iptr->value, i);
}
for (int i = 10; i < 20; ++i)
{
iptr.reset(iq.pop());
CPPA_CHECK(iptr);
if (iptr) CPPA_CHECK_EQUAL(iptr->value, i);
}
}
// two dummies
CPPA_CHECK_EQUAL(s_iint_instances, 2);
return CPPA_TEST_RESULT;
}
......@@ -96,7 +96,7 @@ size_t test__intrusive_ptr()
p1 = p2;
CPPA_CHECK_EQUAL(class0_instances, 1);
CPPA_CHECK_EQUAL(class1_instances, 1);
CPPA_CHECK_EQUAL(p1, p2);
CPPA_CHECK(p1 == p2);
}
CPPA_CHECK_EQUAL(class0_instances, 0);
CPPA_CHECK_EQUAL(class1_instances, 0);
......
......@@ -4,6 +4,29 @@
using namespace cppa;
namespace {
struct streamer
{
std::ostream& o;
streamer(std::ostream& mo) : o(mo) { }
template<typename T>
void operator()(T const& value)
{
o << value;
}
};
inline std::ostream& operator<<(std::ostream& o,
cppa::primitive_variant const& pv)
{
streamer s{o};
pv.apply(s);
return o;
}
} // namespace <anonymous>
size_t test__primitive_variant()
{
CPPA_TEST(test__primitive_variant);
......
......@@ -32,7 +32,7 @@ struct queue_element
{
queue_element* next;
size_t value;
queue_element(size_t val) : next(0), value(val) { }
queue_element(size_t val = 0) : next(0), value(val) { }
};
template<typename T>
......
......@@ -129,8 +129,8 @@ size_t test__serialization()
if (opt)
{
auto& tup = *opt;
CPPA_CHECK_EQUAL(tup.size(), 2);
CPPA_CHECK_EQUAL(get<0>(tup), 42);
CPPA_CHECK_EQUAL(tup.size(), static_cast<size_t>(2));
CPPA_CHECK_EQUAL(get<0>(tup), static_cast<std::uint32_t>(42));
CPPA_CHECK_EQUAL(get<1>(tup), "foo");
}
}
......@@ -146,7 +146,7 @@ size_t test__serialization()
binary_deserializer bd(bs.data(), bs.size());
any_tuple ttup2;
uniform_typeid<any_tuple>()->deserialize(&ttup2, &bd);
CPPA_CHECK_EQUAL(ttup, ttup2);
CPPA_CHECK(ttup == ttup2);
}
{
......@@ -189,7 +189,7 @@ size_t test__serialization()
object obj1;
bd >> obj1;
object obj2 = from_string(to_string(msg1));
CPPA_CHECK_EQUAL(obj1, obj2);
CPPA_CHECK(obj1 == obj2);
if (typeid(any_tuple) == *(obj1.type()) && obj2.type() == obj1.type())
{
auto& content1 = get<any_tuple>(obj1);
......@@ -261,7 +261,7 @@ size_t test__serialization()
b2 = get<struct_b>(res);
}
// verify result of serialization / deserialization
CPPA_CHECK_EQUAL(b1, b2);
CPPA_CHECK(b1 == b2);
CPPA_CHECK_EQUAL(to_string(b2), b1str);
// deserialize b3 from string
{
......@@ -269,7 +269,7 @@ size_t test__serialization()
CPPA_CHECK_EQUAL(res.type()->name(), "struct_b");
b3 = get<struct_b>(res);
}
CPPA_CHECK_EQUAL(b1, b3);
CPPA_CHECK(b1 == b3);
}
// test serializers / deserializers with struct_c
{
......@@ -290,7 +290,7 @@ size_t test__serialization()
c2 = get<struct_c>(res);
}
// verify result of serialization / deserialization
CPPA_CHECK_EQUAL(c1, c2);
CPPA_CHECK(c1 == c2);
}
return CPPA_TEST_RESULT;
}
......@@ -138,7 +138,7 @@ abstract_event_based_actor* event_testee2()
after(std::chrono::milliseconds(50)) >> [=]()
{
if (remaining == 1) become_void();
else become(wait4timeout(remaining-1));
else become(wait4timeout(remaining - 1));
}
);
}
......@@ -330,9 +330,22 @@ size_t test__spawn()
{
CPPA_TEST(test__spawn);
CPPA_IF_VERBOSE(cout << "test future_send() ... " << std::flush);
future_send(self, std::chrono::seconds(1), 1, 2, 3);
receive(on(1, 2, 3) >> []() { });
CPPA_IF_VERBOSE(cout << "ok" << endl);
CPPA_IF_VERBOSE(cout << "test timeout ... " << std::flush);
receive(after(std::chrono::seconds(1)) >> []() { });
CPPA_IF_VERBOSE(cout << "ok" << endl);
CPPA_IF_VERBOSE(cout << "testee1 & event_testee2 ... " << std::flush);
spawn(testee1);
spawn(event_testee2());
await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl);
CPPA_IF_VERBOSE(cout << "chopstick ... " << std::flush);
auto cstk = spawn(new chopstick);
send(cstk, atom("take"), self);
receive
......@@ -343,8 +356,8 @@ size_t test__spawn()
send(cstk, atom("break"));
}
);
await_all_others_done();
CPPA_IF_VERBOSE(cout << "ok" << endl);
CPPA_CHECK_EQUAL(behavior_test<testee_actor>(), "wait4int");
CPPA_CHECK_EQUAL(behavior_test<event_testee>(), "wait4int");
......
......@@ -63,8 +63,8 @@ size_t test__tuple()
// check operator==
auto lhs = make_tuple(1,2,3,4);
auto rhs = make_tuple(static_cast<std::uint8_t>(1), 2.0, 3, 4);
CPPA_CHECK_EQUAL(lhs, rhs);
CPPA_CHECK_EQUAL(rhs, lhs);
CPPA_CHECK(lhs == rhs);
CPPA_CHECK(rhs == lhs);
}
any_tuple at1 = make_tuple("one", 2, 3.f, 4.0);
{
......@@ -73,7 +73,7 @@ size_t test__tuple()
CPPA_CHECK(opt0);
if (opt0)
{
CPPA_CHECK_EQUAL(*opt0, make_tuple("one", 2, 3.f, 4.0));
CPPA_CHECK((*opt0 == make_tuple("one", 2, 3.f, 4.0)));
CPPA_CHECK_EQUAL(&get<0>(*opt0), at1.at(0));
CPPA_CHECK_EQUAL(&get<1>(*opt0), at1.at(1));
CPPA_CHECK_EQUAL(&get<2>(*opt0), at1.at(2));
......@@ -100,7 +100,7 @@ size_t test__tuple()
CPPA_CHECK(opt3);
if (opt3)
{
CPPA_CHECK_EQUAL(*opt3, make_tuple("one", 4.0));
CPPA_CHECK((*opt3 == make_tuple("one", 4.0)));
CPPA_CHECK_EQUAL(get<0>(*opt3), "one");
CPPA_CHECK_EQUAL(get<1>(*opt3), 4.0);
CPPA_CHECK_EQUAL(&get<0>(*opt3), at1.at(0));
......
......@@ -59,7 +59,7 @@ size_t test__uniform_type()
//bar.create_object();
object obj1 = uniform_typeid<foo>()->create();
object obj2(obj1);
CPPA_CHECK_EQUAL(obj1, obj2);
CPPA_CHECK(obj1 == obj2);
get_ref<foo>(obj1).value = 42;
CPPA_CHECK(obj1 != obj2);
CPPA_CHECK_EQUAL(get<foo>(obj1).value, 42);
......
......@@ -12,6 +12,27 @@ using namespace cppa;
using namespace cppa::util;
using namespace cppa::detail;
namespace {
std::ostream& operator<<(std::ostream& o, yield_state ys)
{
switch (ys)
{
case yield_state::invalid:
return (o << "yield_state::invalid");
case yield_state::ready:
return (o << "yield_state::ready");
case yield_state::blocked:
return (o << "yield_state::blocked");
case yield_state::done:
return (o << "yield_state::done");
default:
return (o << "{{{invalid yield_state}}}");
}
}
} // namespace <anonymous>
struct pseudo_worker
{
......
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