Commit a917f0f0 authored by neverlord's avatar neverlord

mock scheduler implemented

parent 9292a289
#all:
# /opt/local/bin/g++-mp-4.5 -std=c++0x -g -O0 main.cpp
CXX = /opt/local/bin/g++-mp-4.5
#CXX = /opt/local/bin/g++-mp-4.6
CXXFLAGS = -std=c++0x -pedantic -Wall -Wextra -g -O0 -I/opt/local/include/
LIBS = -L/opt/local/lib -lboost_thread-mt
INCLUDES = -I./
HEADERS = cppa/actor.hpp \
cppa/any_type.hpp \
cppa/config.hpp \
cppa/cow_ptr.hpp \
cppa/get.hpp \
cppa/intrusive_ptr.hpp \
cppa/invoke.hpp \
cppa/invoke_rules.hpp \
cppa/match.hpp \
cppa/message.hpp \
cppa/on.hpp \
cppa/ref_counted.hpp \
cppa/test.hpp \
cppa/tuple.hpp \
cppa/tuple_view.hpp \
cppa/uniform_type_info.hpp \
cppa/untyped_tuple.hpp \
cppa/util.hpp \
cppa/detail/abstract_tuple.hpp \
cppa/detail/decorated_tuple.hpp \
cppa/detail/intermediate.hpp \
cppa/detail/invokable.hpp \
cppa/detail/matcher.hpp \
cppa/detail/ref_counted_impl.hpp \
cppa/detail/scheduler.hpp \
cppa/detail/tdata.hpp \
cppa/detail/tuple_vals.hpp \
cppa/util/a_matches_b.hpp \
cppa/util/callable_trait.hpp \
cppa/util/compare_tuples.hpp \
cppa/util/concat_type_lists.hpp \
cppa/util/conjunction.hpp \
cppa/util/detach.hpp \
cppa/util/disjunction.hpp \
cppa/util/eval_type_lists.hpp \
cppa/util/filter_type_list.hpp \
cppa/util/has_copy_member_fun.hpp \
cppa/util/is_comparable.hpp \
cppa/util/is_copyable.hpp \
cppa/util/is_one_of.hpp \
cppa/util/remove_const_reference.hpp \
cppa/util/reverse_type_list.hpp \
cppa/util/type_at.hpp \
cppa/util/type_list.hpp \
cppa/util/type_list_apply.hpp \
cppa/util/type_list_pop_back.hpp \
cppa/util/utype_iterator.hpp \
cppa/util/void_type.hpp
SOURCES = src/decorated_tuple.cpp \
src/deserializer.cpp \
src/main.cpp \
src/ref_counted.cpp \
src/serializer.cpp \
src/test__a_matches_b.cpp \
src/test__atom.cpp \
src/test__intrusive_ptr.cpp \
src/test__queue_performance.cpp \
src/test__serialization.cpp \
src/test__spawn.cpp \
src/test__tuple.cpp \
src/test__type_list.cpp \
src/uniform_type_info.cpp \
src/untyped_tuple.cpp
OBJECTS = $(SOURCES:.cpp=.o)
EXECUTABLE = test
%.o : %.cpp $(HEADERS)
$(CXX) $(CXXFLAGS) $(INCLUDES) -c $< -o $@
$(EXECUTABLE) : $(OBJECTS) $(HEADERS)
$(CXX) $(LIBS) $(OBJECTS) -o $(EXECUTABLE)
all : $(SOURCES) $(EXECUTABLE)
all:
make -f libcppa.Makefile
make -f test.Makefile
clean:
rm -f $(OBJECTS) $(EXECUTABLE)
make -f libcppa.Makefile clean
make -f test.Makefile clean
......@@ -78,3 +78,10 @@ cppa/util/abstract_type_list.hpp
cppa/detail/serialize_tuple.hpp
src/test__atom.cpp
src/test__queue_performance.cpp
cppa/detail/actor_public.hpp
cppa/detail/comparable.hpp
cppa/util/single_reader_queue.hpp
cppa/util/singly_linked_list.hpp
cppa/detail/spawn_impl.hpp
cppa/detail/actor_private.hpp
cppa/reply.hpp
#ifndef ACTOR_HPP
#define ACTOR_HPP
#include "cppa/tuple.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/detail/actor_public.hpp"
#include "cppa/detail/actor_private.hpp"
namespace cppa {
struct actor
class actor
{
public:
typedef cppa::intrusive_ptr<detail::actor_public> ptr_type;
actor() = default;
inline actor(detail::actor_public* ptr) : m_ptr(ptr) { }
inline actor(const ptr_type& ptr) : m_ptr(ptr) { }
inline actor(ptr_type&& ptr) : m_ptr(ptr) { }
actor(const actor&) = default;
inline actor(actor&& other) : m_ptr(std::move(other.m_ptr)) { }
actor& operator=(const actor&) = default;
actor& operator=(actor&& other)
{
m_ptr = std::move(other.m_ptr);
return *this;
}
template<typename... Args>
void send(const Args&... args)
{
this_actor()->send(m_ptr.get(), tuple<Args...>(args...));
}
template<typename... Args>
void send(const Args&...)
void send_tuple(const tuple<Args...>& args)
{
this_actor()->send(m_ptr.get(), args);
}
private:
ptr_type m_ptr;
};
/*
template<typename... Args>
actor& operator<<(actor& whom, const tuple<Args...>& data)
{
whom.send_tuple(data);
}
*/
} // namespace cppa
#endif // ACTOR_HPP
#ifndef ACTOR_PRIVATE_HPP
#define ACTOR_PRIVATE_HPP
#include "cppa/invoke_rules.hpp"
#include "cppa/untyped_tuple.hpp"
#include "cppa/detail/actor_public.hpp"
namespace cppa { class message; }
namespace cppa { namespace detail {
// private part of the actor interface (callable only from this_actor())
struct actor_private : public actor_public
{
virtual const message& receive() = 0;
virtual const message& last_dequeued() const = 0;
virtual void receive(invoke_rules&) = 0;
virtual void send(actor_public* whom, untyped_tuple&& what) = 0;
};
} } // namespace cppa::detail
namespace cppa {
detail::actor_private* this_actor();
inline const message& receive()
{
return this_actor()->receive();
}
inline void receive(invoke_rules& rules)
{
this_actor()->receive(rules);
}
inline void receive(invoke_rules&& rules)
{
this_actor()->receive(rules);
}
inline const message& last_dequeued()
{
return this_actor()->last_dequeued();
}
} // namespace cppa
#endif // ACTOR_PRIVATE_HPP
#ifndef ACTOR_PUBLIC_HPP
#define ACTOR_PUBLIC_HPP
#include "cppa/ref_counted.hpp"
namespace cppa { class message; }
namespace cppa { namespace detail {
// public part of the actor interface
struct actor_public : ref_counted
{
virtual void enqueue_msg(const message& msg) = 0;
};
} } // namespace cppa::detail
#endif // ACTOR_PUBLIC_HPP
#ifndef COMPARABLE_HPP
#define COMPARABLE_HPP
namespace cppa { namespace detail {
template<typename Subclass, typename T = Subclass>
class comparable
{
friend bool operator==(const Subclass& lhs, const T& rhs)
{
return lhs.equal_to(rhs);
}
friend bool operator==(const T& lhs, const Subclass& rhs)
{
return rhs.equal_to(lhs);
}
friend bool operator!=(const Subclass& lhs, const T& rhs)
{
return !lhs.equal_to(rhs);
}
friend bool operator!=(const T& lhs, const Subclass& rhs)
{
return !rhs.equal_to(lhs);
}
};
template<typename Subclass>
class comparable<Subclass, Subclass>
{
friend bool operator==(const Subclass& lhs, const Subclass& rhs)
{
return lhs.equal_to(rhs);
}
friend bool operator!=(const Subclass& lhs, const Subclass& rhs)
{
return !lhs.equal_to(rhs);
}
};
} } // cppa::detail
#endif // COMPARABLE_HPP
#ifndef SCHEDULER_HPP
#define SCHEDULER_HPP
#include "cppa/actor.hpp"
#include "cppa/message.hpp"
namespace cppa { namespace detail {
// public part of the actor interface (callable from other actors)
struct actor_public
{
virtual void enqueue_msg(const message& msg) = 0;
};
// private part of the actor interface (callable only from this_actor())
struct actor_private : public actor_public
{
virtual message dequeue_msg() = 0;
};
actor_private* this_actor();
actor spawn_impl();
} } // namespace cppa::detail
#include "cppa/detail/spawn_impl.hpp"
#include "cppa/detail/actor_public.hpp"
#include "cppa/detail/actor_private.hpp"
#endif // SCHEDULER_HPP
#ifndef SPAWN_IMPL_HPP
#define SPAWN_IMPL_HPP
#include "cppa/actor.hpp"
#include "cppa/detail/scheduler.hpp"
namespace cppa { namespace detail {
struct behavior
{
virtual void act() = 0;
virtual void on_exit() = 0;
};
actor spawn_impl(behavior*);
} } // namespace cppa::detail
#endif // SPAWN_IMPL_HPP
......@@ -3,16 +3,19 @@
#include <algorithm>
#include <stdexcept>
#include <type_traits>
#include <cppa/detail/comparable.hpp>
namespace cppa {
template<typename T>
class intrusive_ptr
class intrusive_ptr : detail::comparable<intrusive_ptr<T>, T*>
{
T* m_ptr;
inline void init(T* raw_ptr)
inline void set_ptr(T* raw_ptr)
{
m_ptr = raw_ptr;
if (raw_ptr) raw_ptr->ref();
......@@ -23,21 +26,21 @@ class intrusive_ptr
intrusive_ptr() : m_ptr(0) { }
template<typename Y>
intrusive_ptr(Y* raw_ptr) { init(raw_ptr); }
intrusive_ptr(Y* raw_ptr) { set_ptr(raw_ptr); }
intrusive_ptr(T* raw_ptr) { init(raw_ptr); }
intrusive_ptr(T* raw_ptr) { set_ptr(raw_ptr); }
intrusive_ptr(intrusive_ptr&& other) : m_ptr(other.m_ptr)
{
other.m_ptr = 0;
}
intrusive_ptr(const intrusive_ptr& other) { init(other.m_ptr); }
intrusive_ptr(const intrusive_ptr& other) { set_ptr(other.m_ptr); }
template<typename Y>
intrusive_ptr(const intrusive_ptr<Y>& other)
{
init(const_cast<Y*>(other.get()));
set_ptr(const_cast<Y*>(other.get()));
}
~intrusive_ptr() { if (m_ptr && !m_ptr->deref()) delete m_ptr; }
......@@ -51,6 +54,20 @@ class intrusive_ptr
std::swap(m_ptr, other.m_ptr);
}
void reset(T* new_value = 0)
{
if (m_ptr && !m_ptr->deref()) delete m_ptr;
set_ptr(new_value);
}
template<typename Y>
void reset(Y* new_value)
{
static_assert(std::is_convertible<Y*, T*>::value,
"Y* is not assignable to T*");
reset(static_cast<T*>(new_value));
}
intrusive_ptr& operator=(const intrusive_ptr& other)
{
intrusive_ptr tmp(other);
......@@ -58,6 +75,13 @@ class intrusive_ptr
return *this;
}
intrusive_ptr& operator=(intrusive_ptr&& other)
{
reset();
swap(other);
return *this;
}
template<typename Y>
intrusive_ptr& operator=(const intrusive_ptr<Y>& other)
{
......@@ -76,79 +100,40 @@ class intrusive_ptr
inline explicit operator bool() const { return m_ptr != 0; }
};
template<typename X, typename Y>
bool operator==(const intrusive_ptr<X>& lhs, const Y* rhs)
{
return lhs.get() == rhs;
}
template<typename X, typename Y>
bool operator==(const X* lhs, const intrusive_ptr<Y>& rhs)
{
return lhs == rhs.get();
}
template<typename X, typename Y>
bool operator==(const intrusive_ptr<X>& lhs, const intrusive_ptr<Y>& rhs)
{
return lhs.get() == rhs.get();
}
template<typename X, typename Y>
bool operator!=(const intrusive_ptr<X>& lhs, const Y* rhs)
{
return !(lhs == rhs);
}
template<typename X, typename Y>
bool operator!=(const X* lhs, const intrusive_ptr<Y>& rhs)
{
return !(lhs == rhs);
}
template<typename X, typename Y>
bool operator!=(const intrusive_ptr<X>& lhs, const intrusive_ptr<Y>& rhs)
{
return !(lhs == rhs);
}
inline bool equal_to(const T* ptr) const
{
return get() == ptr;
}
template<typename X>
bool operator==(const intrusive_ptr<X>& lhs, const X* rhs)
{
return lhs.get() == rhs;
}
template<typename Y>
inline bool equal_to(const Y* ptr) const
{
return get() == ptr;
}
template<typename X>
bool operator==(const X* lhs, const intrusive_ptr<X>& rhs)
{
return lhs == rhs.get();
}
inline bool operator==(const intrusive_ptr& other) const
{
return equal_to(other.get());
}
template<typename X>
bool operator==(const intrusive_ptr<X>& lhs, const intrusive_ptr<X>& rhs)
{
return lhs.get() == rhs.get();
}
template<typename Y>
inline bool operator==(const intrusive_ptr<Y>& other) const
{
return equal_to(other.get());
}
template<typename X>
bool operator!=(const intrusive_ptr<X>& lhs, const X* rhs)
{
return !(lhs == rhs);
}
inline bool operator!=(const intrusive_ptr& other) const
{
return !(*this == other);
}
template<typename X>
bool operator!=(const X* lhs, const intrusive_ptr<X>& rhs)
{
return !(lhs == rhs);
}
template<typename Y>
inline bool operator!=(const intrusive_ptr<Y>& other) const
{
return !(*this == other);
}
template<typename X>
bool operator!=(const intrusive_ptr<X>& lhs, const intrusive_ptr<X>& rhs)
{
return !(lhs == rhs);
}
};
} // namespace cppa
......
#ifndef INVOKE_RULES_HPP
#define INVOKE_RULES_HPP
#include <list>
#include "cppa/invoke.hpp"
#include "cppa/intrusive_ptr.hpp"
......
#ifndef MESSAGE_HPP
#define MESSAGE_HPP
#include "cppa/actor.hpp"
#include "cppa/tuple.hpp"
#include "cppa/untyped_tuple.hpp"
#include "cppa/intrusive_ptr.hpp"
namespace cppa {
class message
{
public:
struct content : ref_counted
{
const actor sender;
const actor receiver;
const untyped_tuple data;
content(const actor& s, const actor& r, const untyped_tuple& ut)
: sender(s), receiver(r), data(ut)
{
}
content(const actor& s, const actor& r, untyped_tuple&& ut)
: sender(s), receiver(r), data(ut)
{
}
};
private:
intrusive_ptr<content> m_content;
public:
template<typename... Args>
message(const actor& from, const actor& to, const Args&... args)
: m_content(new content(from, to, tuple<Args...>(args...))) { }
message(const actor& from, const actor& to, const untyped_tuple& ut)
: m_content(new content(from, to, ut)) { }
message(const actor& from, const actor& to, untyped_tuple&& ut)
: m_content(new content(from, to, std::move(ut))) { }
message() : m_content(new content(actor(), actor(), tuple<int>(0))) { }
const actor& sender() const
{
return m_content->sender;
}
const actor& receiver() const
{
return m_content->receiver;
}
const untyped_tuple& data() const
{
return m_content->data;
}
};
} // namespace cppa
#include "cppa/reply.hpp"
#endif // MESSAGE_HPP
#ifndef REPLY_HPP
#define REPLY_HPP
#include "cppa/message.hpp"
namespace cppa {
template<typename... Args>
void reply(const Args&... args)
{
auto whom = this_actor()->last_dequeued().sender();
whom.send(args...);
}
} // namespace cppa
#endif // REPLY_HPP
......@@ -26,11 +26,11 @@ if (!(line_of_code)) \
++error_count; \
} ((void) 0)
#define CPPA_CHECK_EQUAL(line_of_code, expected) \
if ((line_of_code) != expected) \
#define CPPA_CHECK_EQUAL(lhs_loc, rhs_loc) \
if ((lhs_loc) != (rhs_loc)) \
{ \
std::cerr << "ERROR in file " << __FILE__ << " on line " << __LINE__ \
<< " => " << #line_of_code << " != " << expected << std::endl; \
<< " => " << #lhs_loc << " != " << #rhs_loc << std::endl; \
++error_count; \
} ((void) 0)
......
#ifndef SINGLE_READER_QUEUE_HPP
#define SINGLE_READER_QUEUE_HPP
#include <atomic>
#include <boost/thread.hpp>
namespace cppa { namespace util {
/**
* @brief An intrusive, thread safe queue implementation.
*/
template<typename T>
class single_reader_queue
{
typedef boost::unique_lock<boost::mutex> lock_type;
public:
typedef T element_type;
/**
* @warning call only from the reader (owner)
*/
element_type* pop()
{
wait_for_data();
element_type* result = take_head();
return result;
}
/**
* @warning call only from the reader (owner)
*/
element_type* try_pop()
{
return take_head();
}
/**
* @warning call only from the reader (owner)
*/
void push_front(element_type* element)
{
element->next = m_head;
m_head = element;
}
/**
* @warning call only from the reader (owner)
*/
void push_front(element_type* first, element_type* last)
{
last->next = m_head;
m_head = first;
}
/**
* @warning call only from the reader (owner)
*/
template<typename List>
void prepend(List&& list)
{
if (!list.empty())
{
auto p = list.take();
if (p.first)
{
if (p.first == p.second)
{
push_front(p.first);
}
else
{
push_front(p.first, p.second);
}
}
}
}
void push_back(element_type* new_element)
{
element_type* e = m_tail.load();
for (;;)
{
new_element->next = e;
if (!e)
{
lock_type guard(m_mtx);
if (m_tail.compare_exchange_weak(e, new_element))
{
m_cv.notify_one();
return;
}
}
else
{
if (m_tail.compare_exchange_weak(e, new_element))
{
return;
}
}
}
}
single_reader_queue() : m_tail(0), m_head(0) { }
private:
// exposed to "outside" access
atomic<element_type*> m_tail;
// accessed only by the owner
element_type* m_head;
// locked on enqueue/dequeue operations to/from an empty list
boost::mutex m_mtx;
boost::condition_variable m_cv;
void wait_for_data()
{
if (!m_head && !(m_tail.load()))
{
lock_type guard(m_mtx);
while (!(m_tail.load())) m_cv.wait(guard);
}
}
// atomically set public_tail to nullptr and enqueue all
bool fetch_new_data()
{
element_type* e = m_tail.load();
while (e)
{
if (m_tail.compare_exchange_weak(e, 0))
{
// 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;
// next iteration
e = next;
}
return true;
}
// next iteration
}
// !public_tail
return false;
}
element_type* take_head()
{
if (m_head || fetch_new_data())
{
element_type* result = m_head;
m_head = result->next;
return result;
}
return 0;
}
};
} } // namespace cppa::util
#endif // SINGLE_READER_QUEUE_HPP
#ifndef SINGLY_LINKED_LIST_HPP
#define SINGLY_LINKED_LIST_HPP
#include <utility>
namespace cppa { namespace util {
template<typename T>
class singly_linked_list
{
T* m_head;
T* m_tail;
public:
typedef T element_type;
singly_linked_list() : m_head(0), m_tail(0) { }
inline bool empty() const { return m_head == 0; }
void push_back(element_type* what)
{
what->next = 0;
if (m_tail)
{
m_tail->next = what;
m_tail = what;
}
else
{
m_head = m_tail = what;
}
}
std::pair<element_type*, element_type*> take()
{
element_type* first = m_head;
element_type* last = m_tail;
m_head = m_tail = 0;
return { first, last };
}
};
} } // namespace cppa::util
#endif // SINGLY_LINKED_LIST_HPP
CXX = /opt/local/bin/g++-mp-4.5
#CXXFLAGS = -std=c++0x -pedantic -Wall -Wextra -g -O0 -I/opt/local/include/
CXXFLAGS = -std=c++0x -pedantic -Wall -Wextra -O2 -I/opt/local/include/ -fPIC
LIBS = -L/opt/local/lib -lboost_thread-mt
INCLUDES = -I./
HEADERS = cppa/actor.hpp \
cppa/any_type.hpp \
cppa/config.hpp \
cppa/cow_ptr.hpp \
cppa/get.hpp \
cppa/intrusive_ptr.hpp \
cppa/invoke.hpp \
cppa/invoke_rules.hpp \
cppa/match.hpp \
cppa/message.hpp \
cppa/on.hpp \
cppa/ref_counted.hpp \
cppa/tuple.hpp \
cppa/tuple_view.hpp \
cppa/uniform_type_info.hpp \
cppa/untyped_tuple.hpp \
cppa/util.hpp \
cppa/detail/abstract_tuple.hpp \
cppa/detail/decorated_tuple.hpp \
cppa/detail/intermediate.hpp \
cppa/detail/invokable.hpp \
cppa/detail/matcher.hpp \
cppa/detail/ref_counted_impl.hpp \
cppa/detail/scheduler.hpp \
cppa/detail/tdata.hpp \
cppa/detail/tuple_vals.hpp \
cppa/util/a_matches_b.hpp \
cppa/util/callable_trait.hpp \
cppa/util/compare_tuples.hpp \
cppa/util/concat_type_lists.hpp \
cppa/util/conjunction.hpp \
cppa/util/detach.hpp \
cppa/util/disjunction.hpp \
cppa/util/eval_type_lists.hpp \
cppa/util/filter_type_list.hpp \
cppa/util/has_copy_member_fun.hpp \
cppa/util/is_comparable.hpp \
cppa/util/is_copyable.hpp \
cppa/util/is_one_of.hpp \
cppa/util/remove_const_reference.hpp \
cppa/util/reverse_type_list.hpp \
cppa/util/single_reader_queue.hpp \
cppa/util/type_at.hpp \
cppa/util/type_list.hpp \
cppa/util/type_list_apply.hpp \
cppa/util/type_list_pop_back.hpp \
cppa/util/utype_iterator.hpp \
cppa/util/void_type.hpp
SOURCES = src/decorated_tuple.cpp \
src/deserializer.cpp \
src/mock_scheduler.cpp \
src/ref_counted.cpp \
src/serializer.cpp \
src/uniform_type_info.cpp \
src/untyped_tuple.cpp
OBJECTS = $(SOURCES:.cpp=.o)
LIB_NAME = libcppa.dylib
%.o : %.cpp $(HEADERS)
$(CXX) $(CXXFLAGS) $(INCLUDES) -fPIC -c $< -o $@
$(LIB_NAME) : $(OBJECTS) $(HEADERS)
$(CXX) $(LIBS) -dynamiclib -o $(LIB_NAME) $(OBJECTS)
all : $(LIB_NAME) $(OBJECTS)
clean:
rm -f $(LIB_NAME) $(OBJECTS)
......@@ -2,6 +2,7 @@
#include <atomic>
#include <string>
#include <limits>
#include <vector>
#include <cassert>
#include <cstddef>
#include <cstdint>
......@@ -13,52 +14,54 @@
#include "cppa/config.hpp"
#include "cppa/uniform_type_info.hpp"
#ifdef CPPA_GCC
#include <cxxabi.h>
#endif
#define RUN_TEST(fun_name) \
std::cout << "run " << #fun_name << " ..." << std::endl; \
errors += fun_name (); \
std::cout << std::endl
using std::cout;
using std::cerr;
using std::endl;
int main()
int main(int argc, char** c_argv)
{
std::vector<std::string> argv;
for (int i = 1; i < argc; ++i)
{
argv.push_back(c_argv[i]);
}
std::cout << std::boolalpha;
/*
std::atomic<std::int32_t> a(42);
std::int32_t b, c;
b = 42;
c = 10;
cout << "a.compare_exchange_weak(b, c, std::memory_order_acquire) = "
<< a.compare_exchange_weak(b, c, std::memory_order_acquire) << endl;
cout << "a = " << a << endl;
*/
std::size_t errors = 0;
RUN_TEST(test__a_matches_b);
RUN_TEST(test__intrusive_ptr);
RUN_TEST(test__spawn);
RUN_TEST(test__tuple);
RUN_TEST(test__type_list);
RUN_TEST(test__serialization);
RUN_TEST(test__atom);
cout << endl
<< "error(s) in all tests: " << errors
<< endl;
cout << endl << "run queue performance test ... " << endl;
test__queue_performance();
if (!argv.empty())
{
if (argv.size() == 1 && argv.front() == "performance_test")
{
cout << endl << "run queue performance test ... " << endl;
test__queue_performance();
}
else
{
cerr << "unrecognized options"
<< endl
<< "no options:\n\tunit tests"
<< endl
<< "performance_test:\n\trun single reader queue tests"
<< endl;
}
}
else
{
std::cout << std::boolalpha;
std::size_t errors = 0;
RUN_TEST(test__a_matches_b);
RUN_TEST(test__intrusive_ptr);
RUN_TEST(test__spawn);
RUN_TEST(test__tuple);
RUN_TEST(test__type_list);
RUN_TEST(test__serialization);
RUN_TEST(test__atom);
cout << endl
<< "error(s) in all tests: " << errors
<< endl;
}
return 0;
}
#include <boost/thread.hpp>
#include "cppa/detail/scheduler.hpp"
#include "cppa/util/singly_linked_list.hpp"
#include "cppa/util/single_reader_queue.hpp"
namespace {
struct actor_message
{
actor_message* next;
cppa::message msg;
actor_message(const cppa::message& from) : msg(from) { }
};
struct actor_impl;
void cleanup_fun(actor_impl* what);
boost::thread_specific_ptr<actor_impl> m_this_actor(cleanup_fun);
struct actor_impl : cppa::detail::actor_private
{
cppa::message m_last_dequeued;
cppa::util::single_reader_queue<actor_message> mailbox;
cppa::detail::behavior* m_behavior;
actor_impl(cppa::detail::behavior* b = 0) : m_behavior(b) { }
virtual void enqueue_msg(const cppa::message& msg)
{
mailbox.push_back(new actor_message(msg));
}
virtual const cppa::message& receive()
{
actor_message* msg = mailbox.pop();
m_last_dequeued = std::move(msg->msg);
delete msg;
return m_last_dequeued;
}
virtual const cppa::message& last_dequeued() const
{
return m_last_dequeued;
}
virtual void receive(cppa::invoke_rules& rules)
{
cppa::util::singly_linked_list<actor_message> buffer;
actor_message* amsg = mailbox.pop();
cppa::intrusive_ptr<cppa::detail::intermediate> imd;
while (!(imd = rules.get_intermediate(amsg->msg.data())))
{
buffer.push_back(amsg);
amsg = mailbox.pop();
}
m_last_dequeued = amsg->msg;
delete amsg;
imd->invoke();
mailbox.prepend(std::move(buffer));
}
virtual void send(cppa::detail::actor_public* whom,
cppa::untyped_tuple&& what)
{
if (whom) whom->enqueue_msg(cppa::message(this, whom, std::move(what)));
}
void operator()()
{
if (m_behavior)
{
try
{
m_behavior->act();
}
catch (...) { }
m_behavior->on_exit();
}
}
};
void cleanup_fun(actor_impl* what)
{
if (what)
{
if (!what->deref()) delete what;
}
}
struct actor_ptr
{
cppa::intrusive_ptr<actor_impl> m_impl;
actor_ptr(actor_impl* ai) : m_impl(ai) { }
actor_ptr(actor_ptr&& other) : m_impl(std::move(other.m_impl)) { }
actor_ptr(const actor_ptr&) = default;
void operator()()
{
m_this_actor.reset(m_impl.get());
(*m_impl)();
}
};
} // namespace <anonymous>
namespace cppa {
detail::actor_private* this_actor()
{
actor_impl* res = m_this_actor.get();
if (!res)
{
res = new actor_impl;
res->ref();
m_this_actor.reset(res);
}
return res;
}
} // namespace cppa
namespace cppa { namespace detail {
actor spawn_impl(behavior* actor_behavior)
{
actor_ptr aptr(new actor_impl(actor_behavior));
boost::thread(aptr).detach();
return actor(std::move(aptr.m_impl));
}
} } // namespace cppa::detail
......@@ -8,27 +8,40 @@
using namespace cppa;
namespace { int rc_instances = 0; }
namespace {
struct test_rc : public cppa::detail::ref_counted_impl<std::size_t>
int class0_instances = 0;
int class1_instances = 0;
}
struct class0 : cppa::detail::ref_counted_impl<std::size_t>
{
class0() { ++class0_instances; }
test_rc() { ++rc_instances; }
virtual ~class0() { --class0_instances; }
virtual ~test_rc() { --rc_instances; }
virtual class0* create() const { return new class0; }
};
test_rc* create() { return new test_rc; }
struct class1 : class0
{
class1() { ++class1_instances; }
virtual ~class1() { --class1_instances; }
virtual class1* create() const { return new class1; }
};
typedef intrusive_ptr<test_rc> test_ptr;
typedef intrusive_ptr<class0> class0_ptr;
typedef intrusive_ptr<class1> class1_ptr;
test_rc* get_test_rc()
class0* get_test_rc()
{
return new test_rc;
return new class0;
}
test_ptr get_test_ptr()
class0_ptr get_test_ptr()
{
return get_test_rc();
}
......@@ -39,38 +52,52 @@ std::size_t test__intrusive_ptr()
CPPA_TEST(test__intrusive_ptr);
{
test_ptr p(new test_rc);
CPPA_CHECK_EQUAL(rc_instances, 1);
class0_ptr p(new class0);
CPPA_CHECK_EQUAL(class0_instances, 1);
CPPA_CHECK(p->unique());
}
CPPA_CHECK_EQUAL(rc_instances, 0);
CPPA_CHECK_EQUAL(class0_instances, 0);
{
test_ptr p;
p = new test_rc;
CPPA_CHECK_EQUAL(rc_instances, 1);
class0_ptr p;
p = new class0;
CPPA_CHECK_EQUAL(class0_instances, 1);
CPPA_CHECK(p->unique());
}
CPPA_CHECK_EQUAL(rc_instances, 0);
CPPA_CHECK_EQUAL(class0_instances, 0);
{
test_ptr p1;
class0_ptr p1;
p1 = get_test_rc();
test_ptr p2 = p1;
CPPA_CHECK_EQUAL(rc_instances, 1);
class0_ptr p2 = p1;
CPPA_CHECK_EQUAL(class0_instances, 1);
CPPA_CHECK_EQUAL(p1->unique(), false);
}
CPPA_CHECK_EQUAL(rc_instances, 0);
CPPA_CHECK_EQUAL(class0_instances, 0);
{
std::list<test_ptr> pl;
std::list<class0_ptr> pl;
pl.push_back(get_test_ptr());
pl.push_back(get_test_rc());
pl.push_back(pl.front()->create());
CPPA_CHECK(pl.front()->unique());
CPPA_CHECK_EQUAL(rc_instances, 3);
CPPA_CHECK_EQUAL(class0_instances, 3);
}
CPPA_CHECK_EQUAL(class0_instances, 0);
{
class0_ptr p1(new class0);
p1 = new class1;
CPPA_CHECK_EQUAL(class0_instances, 1);
CPPA_CHECK_EQUAL(class1_instances, 1);
class1_ptr p2(new class1);
p1 = p2;
CPPA_CHECK_EQUAL(class0_instances, 1);
CPPA_CHECK_EQUAL(class1_instances, 1);
CPPA_CHECK_EQUAL(p1, p2);
}
CPPA_CHECK_EQUAL(rc_instances, 0);
CPPA_CHECK_EQUAL(class0_instances, 0);
CPPA_CHECK_EQUAL(class1_instances, 0);
return CPPA_TEST_RESULT;
......
......@@ -7,6 +7,10 @@
#include <boost/thread.hpp>
#include <boost/progress.hpp>
#include "cppa/util/single_reader_queue.hpp"
using cppa::util::single_reader_queue;
using std::cout;
using std::cerr;
using std::endl;
......@@ -18,130 +22,55 @@ struct queue_element
queue_element(std::size_t val) : next(0), value(val) { }
};
/**
* @brief An intrusive, thread safe queue implementation.
*/
template<typename T>
class single_reader_queue
class singly_linked_list
{
typedef boost::unique_lock<boost::mutex> lock_type;
public:
typedef T element_type;
element_type* pop()
{
wait_for_data();
element_type* result = take_head();
return result;
}
element_type* m_head;
element_type* m_tail;
element_type* try_pop()
{
return take_head();
}
public:
/*
element_type* front()
{
return (m_head || fetch_new_data()) ? m_head : 0;
}
*/
singly_linked_list() : m_head(0), m_tail(0) { }
void push(element_type* new_element)
singly_linked_list& operator=(singly_linked_list&& other)
{
element_type* e = m_tail.load();
for (;;)
{
// element_type* e = const_cast<element_type*>(m_tail);
new_element->next = e;
if (!e)
{
lock_type guard(m_mtx);
// if (atomic_cas(&m_tail, (element_type*) 0, new_element))
if (m_tail.compare_exchange_weak(e, new_element))
{
m_cv.notify_one();
return;
}
}
else
{
// if (atomic_cas(&m_tail, e, new_element))
if (m_tail.compare_exchange_weak(e, new_element))
{
return;
}
}
}
m_head = other.m_head;
m_tail = other.m_tail;
other.m_head = 0;
other.m_tail = 0;
return *this;
}
single_reader_queue() : m_tail(0), m_head(0) { }
private:
inline bool empty() const { return m_head == 0; }
// exposed to "outside" access
atomic<element_type*> m_tail;
// volatile element_type* m_tail;
// accessed only by the owner
element_type* m_head;
// locked on enqueue/dequeue operations to/from an empty list
boost::mutex m_mtx;
boost::condition_variable m_cv;
void wait_for_data()
void push_back(element_type* e)
{
if (!m_head && !(m_tail.load()))
if (!m_head)
{
lock_type guard(m_mtx);
while (!(m_tail.load())) m_cv.wait(guard);
m_head = m_tail = e;
}
}
// atomically set public_tail to nullptr and enqueue all
bool fetch_new_data()
{
element_type* e = m_tail.load();
while (e)
else
{
if (m_tail.compare_exchange_weak(e, 0))
// if (atomic_cas(&m_tail, e, (element_type*) 0))
{
// 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;
// next iteration
e = next;
}
return true;
}
// next iteration
// e = const_cast<element_type*>(m_tail);
m_tail->next = e;
m_tail = e;
}
// !public_tail
return false;
}
element_type* take_head()
element_type* pop_front()
{
if (m_head || fetch_new_data())
element_type* result = m_head;
if (result)
{
element_type* result = m_head;
m_head = result->next;
return result;
if (!m_head)
{
m_tail = 0;
}
}
return 0;
return result;
}
};
......@@ -158,32 +87,42 @@ class locked_queue
element_type* pop()
{
lock_type guard(m_mtx);
element_type* result;
while (m_impl.empty())
if (!m_priv.empty())
{
m_cv.wait(guard);
return m_priv.pop_front();
}
else
{
// lifetime scope of guard
{
lock_type guard(m_mtx);
while (m_pub.empty())
{
m_cv.wait(guard);
}
m_priv = std::move(m_pub);
}
// tail recursion
return pop();
}
result = m_impl.front();
m_impl.pop_front();
return result;
}
void push(element_type* new_element)
{
lock_type guard(m_mtx);
if (m_impl.empty())
if (m_pub.empty())
{
m_cv.notify_one();
}
m_impl.push_back(new_element);
m_pub.push_back(new_element);
}
private:
boost::mutex m_mtx;
boost::condition_variable m_cv;
std::list<element_type*> m_impl;
singly_linked_list<element_type> m_pub;
singly_linked_list<element_type> m_priv;
};
......@@ -206,7 +145,7 @@ void slave(Queue& q, std::size_t from, std::size_t to)
{
for (std::size_t x = from; x < to; ++x)
{
q.push(new queue_element(x));
q.push_back(new queue_element(x));
}
}
......@@ -243,39 +182,64 @@ void master(Queue& q)
<< "min: " << min_val << endl
<< "max: " << max_val << endl;
}
cout << t0.elapsed() << " " << num_msgs << endl;
cout << t0.elapsed() << " " << num_slaves << endl;
}
template<typename Queue>
void test_q_impl()
namespace { const std::size_t slave_messages = 1000000; }
template<std::size_t Pos, std::size_t Max, std::size_t Step,
template<std::size_t> class Stmt>
struct static_for
{
typedef Queue queue_type;
template<typename... Args>
static inline void _(const Args&... args)
{
queue_type q0;
boost::thread t0(master<queue_type, 10, 10000>, boost::ref(q0));
t0.join();
}
{
queue_type q0;
boost::thread t0(master<queue_type, 100, 10000>, boost::ref(q0));
t0.join();
Stmt<Pos>::_(args...);
static_for<Pos + Step, Max, Step, Stmt>::_(args...);
}
};
template<std::size_t Max, std::size_t Step,
template<std::size_t> class Stmt>
struct static_for<Max, Max, Step, Stmt>
{
template<typename... Args>
static inline void _(const Args&... args)
{
queue_type q0;
boost::thread t0(master<queue_type, 1000, 10000>, boost::ref(q0));
t0.join();
Stmt<Max>::_(args...);
}
};
template<typename What>
struct type_token
{
typedef What type;
};
template<std::size_t NumThreads>
struct test_step
{
template<typename QueueToken>
static inline void _(QueueToken)
{
queue_type q0;
boost::thread t0(master<queue_type, 10000, 10000>, boost::ref(q0));
typename QueueToken::type q;
boost::thread t0(master<typename QueueToken::type, NumThreads, slave_messages>, boost::ref(q));
t0.join();
}
};
template<typename Queue>
void test_q_impl()
{
typedef type_token<Queue> queue_token;
static_for<10, 50, 5, test_step>::_(queue_token());
}
void test__queue_performance()
{
cout << "locked_queue:" << endl;
// test_q_impl<locked_queue<queue_element>>();
cout << endl;
cout << "single_reader_queue:" << endl;
test_q_impl<single_reader_queue<queue_element>>();
cout << endl << "locked_queue:" << endl;
test_q_impl<locked_queue<queue_element>>();
}
#include <iostream>
#include <functional>
#include "cppa/on.hpp"
#include "cppa/test.hpp"
#include "cppa/actor.hpp"
//#include "cppa/spawn.hpp"
#include "cppa/detail/scheduler.hpp"
using std::cout;
using std::endl;
using cppa::actor;
using namespace cppa;
template<typename F>
actor spawn(F)
actor spawn(F act_fun)
{
return actor();
struct bhv : cppa::detail::behavior
{
std::function<void ()> m_act;
bhv(const F& invokable) : m_act(invokable) { }
virtual void act()
{
m_act();
}
virtual void on_exit()
{
}
};
return cppa::detail::spawn_impl(new bhv(act_fun));
// return actor();
}
void pong()
{
receive(on<int>() >> [](int value) {
reply((value * 20) + 2);
});
}
std::size_t test__spawn()
......@@ -15,15 +42,14 @@ std::size_t test__spawn()
CPPA_TEST(test__spawn);
actor a0 = spawn([](){
// receive(on<int>() >> [](int i){
// reply(i + 2);
// });
});
a0.send(42);
CPPA_CHECK(true);
{
actor sl = spawn(pong);
sl.send(23.f);
sl.send(2);
receive(on<int>() >> [&](int value) {
CPPA_CHECK_EQUAL(value, 42);
});
}
return CPPA_TEST_RESULT;
......
CXX = /opt/local/bin/g++-mp-4.5
#CXX = /opt/local/bin/g++-mp-4.6
#CXXFLAGS = -std=c++0x -pedantic -Wall -Wextra -g -O0 -I/opt/local/include/
CXXFLAGS = -std=c++0x -pedantic -Wall -Wextra -O2 -I/opt/local/include/
LIBS = -L/opt/local/lib -lboost_thread-mt
INCLUDES = -I./
EXECUTABLE = test
HEADERS = cppa/test.hpp
SOURCES = src/main.cpp \
src/test__a_matches_b.cpp \
src/test__atom.cpp \
src/test__intrusive_ptr.cpp \
src/test__queue_performance.cpp \
src/test__serialization.cpp \
src/test__spawn.cpp \
src/test__tuple.cpp \
src/test__type_list.cpp
OBJECTS = $(SOURCES:.cpp=.o)
%.o : %.cpp $(HEADERS) $(HEADERS)
$(CXX) $(CXXFLAGS) $(INCLUDES) -c $< -o $@
$(EXECUTABLE) : $(OBJECTS) $(HEADERS)
$(CXX) $(LIBS) -L./ -lcppa $(OBJECTS) -o $(EXECUTABLE)
all : $(EXECUTABLE)
clean:
rm -f $(OBJECTS) $(EXECUTABLE)
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