Commit db465f8c authored by neverlord's avatar neverlord

new singleton management

parent bd1c9fbf
...@@ -167,8 +167,71 @@ class pattern_iterator ...@@ -167,8 +167,71 @@ class pattern_iterator
}; };
template<typename VectorType> template<class T, class VectorType>
class tuple_iterator_arg class tuple_iterator_arg
{
size_t pos;
std::type_info const* element_type;
typename T::const_iterator i;
typename T::const_iterator end;
VectorType* mapping;
public:
inline tuple_iterator_arg(T const& iterable, VectorType* mv = nullptr)
: pos(0), element_type(&typeid(typename T::value_type)), i(iterable.begin()), end(iterable.end()), mapping(mv)
{
}
inline tuple_iterator_arg(tuple_iterator_arg const& other, VectorType* mv)
: pos(other.pos), element_type(other.element_type), i(other.i), end(other.end), mapping(mv)
{
}
inline bool at_end() const
{
return i == end;
}
inline void next()
{
++pos;
++i;
}
inline bool has_mapping() const
{
return mapping != nullptr;
}
inline void push_mapping()
{
if (mapping) mapping->push_back(pos);
}
inline void push_mapping(VectorType const& what)
{
if (mapping)
{
mapping->insert(mapping->end(), what.begin(), what.end());
}
}
inline std::type_info const& type() const
{
return *element_type;
}
inline void const* value() const
{
return &(*i);
}
};
template<class VectorType>
class tuple_iterator_arg<any_tuple, VectorType>
{ {
util::any_tuple_iterator iter; util::any_tuple_iterator iter;
...@@ -214,9 +277,9 @@ class tuple_iterator_arg ...@@ -214,9 +277,9 @@ class tuple_iterator_arg
} }
} }
inline uniform_type_info const* type() const inline auto type() const -> decltype(iter.type())
{ {
return &(iter.type()); return iter.type();
} }
inline void const* value() const inline void const* value() const
...@@ -226,8 +289,8 @@ class tuple_iterator_arg ...@@ -226,8 +289,8 @@ class tuple_iterator_arg
}; };
template<typename VectorType> template<class T, class VectorType>
bool do_match(pattern_iterator& iter, tuple_iterator_arg<VectorType>& targ) bool do_match(pattern_iterator& iter, tuple_iterator_arg<T, VectorType>& targ)
{ {
for ( ; !(iter.at_end() && targ.at_end()); iter.next(), targ.next()) for ( ; !(iter.at_end() && targ.at_end()); iter.next(), targ.next())
{ {
...@@ -250,7 +313,7 @@ bool do_match(pattern_iterator& iter, tuple_iterator_arg<VectorType>& targ) ...@@ -250,7 +313,7 @@ bool do_match(pattern_iterator& iter, tuple_iterator_arg<VectorType>& targ)
for ( ; targ.at_end() == false; mv.clear(), targ.next()) for ( ; targ.at_end() == false; mv.clear(), targ.next())
{ {
auto iter_cpy = iter; auto iter_cpy = iter;
tuple_iterator_arg<VectorType> targ_cpy(targ, mv_ptr); tuple_iterator_arg<T,VectorType> targ_cpy(targ, mv_ptr);
if (do_match(iter_cpy, targ_cpy)) if (do_match(iter_cpy, targ_cpy))
{ {
targ.push_mapping(mv); targ.push_mapping(mv);
...@@ -260,7 +323,7 @@ bool do_match(pattern_iterator& iter, tuple_iterator_arg<VectorType>& targ) ...@@ -260,7 +323,7 @@ bool do_match(pattern_iterator& iter, tuple_iterator_arg<VectorType>& targ)
return false; // no submatch found return false; // no submatch found
} }
// compare types // compare types
else if (targ.at_end() == false && iter.type() == targ.type()) else if (targ.at_end() == false && *(iter.type()) == targ.type())
{ {
// compare values if needed // compare values if needed
if ( iter.has_value() == false if ( iter.has_value() == false
......
...@@ -101,7 +101,7 @@ class pattern<T0, Tn...> ...@@ -101,7 +101,7 @@ class pattern<T0, Tn...>
mapping_vector* mapping = nullptr) const mapping_vector* mapping = nullptr) const
{ {
detail::pattern_iterator arg0(size, m_data_ptr, m_utis); detail::pattern_iterator arg0(size, m_data_ptr, m_utis);
detail::tuple_iterator_arg<mapping_vector> arg1(msg, mapping); detail::tuple_iterator_arg<any_tuple,mapping_vector> arg1(msg, mapping);
return detail::do_match(arg0, arg1); return detail::do_match(arg0, arg1);
} }
......
...@@ -110,7 +110,6 @@ class tuple ...@@ -110,7 +110,6 @@ class tuple
inline size_t size() const inline size_t size() const
{ {
return sizeof...(ElementTypes); return sizeof...(ElementTypes);
//return m_vals->size();
} }
/** /**
......
...@@ -113,7 +113,6 @@ class tuple_view ...@@ -113,7 +113,6 @@ class tuple_view
inline size_t size() const inline size_t size() const
{ {
return sizeof...(ElementTypes); return sizeof...(ElementTypes);
//return m_vals->size();
} }
private: private:
......
...@@ -51,67 +51,79 @@ namespace { ...@@ -51,67 +51,79 @@ namespace {
using namespace cppa; using namespace cppa;
using namespace cppa::detail; using namespace cppa::detail;
struct singleton_container std::atomic<uniform_type_info_map*> s_uniform_type_info_map(nullptr);
std::atomic<network_manager*> s_network_manager(nullptr);
std::atomic<actor_registry*> s_actor_registry(nullptr);
std::atomic<group_manager*> s_group_manager(nullptr);
std::atomic<empty_tuple*> s_empty_tuple(nullptr);
std::atomic<scheduler*> s_scheduler(nullptr);
template<typename T>
void stop_and_kill(std::atomic<T*>& ptr)
{ {
auto p = ptr.load();
std::atomic<scheduler*> m_scheduler; if (p)
network_manager* m_network_manager;
uniform_type_info_map* m_uniform_type_info_map;
actor_registry* m_actor_registry;
group_manager* m_group_manager;
empty_tuple* m_empty_tuple;
singleton_container() : m_scheduler(nullptr), m_network_manager(nullptr)
{ {
m_uniform_type_info_map = new uniform_type_info_map; if (ptr.compare_exchange_weak(p, nullptr))
m_actor_registry = new actor_registry; {
m_group_manager = new group_manager; p->stop();
m_empty_tuple = new empty_tuple; delete p;
m_empty_tuple->ref(); }
else
{
stop_and_kill(ptr);
}
} }
}
~singleton_container() struct singleton_cleanup_helper
{
~singleton_cleanup_helper()
{ {
// disconnect all peers if (self.unchecked() != nullptr)
if (m_network_manager)
{ {
m_network_manager->stop(); try { self.unchecked()->quit(exit_reason::normal); }
delete m_network_manager; catch (actor_exited&) { }
} }
// run actor specific cleanup if needed auto rptr = s_actor_registry.load();
if (self.unchecked() != nullptr) if (rptr)
{ {
try rptr->await_running_count_equal(0);
{
self.unchecked()->quit(exit_reason::normal);
}
catch (actor_exited&)
{
}
} }
// wait for all running actors to quit delete s_actor_registry.load();
m_actor_registry->await_running_count_equal(0);
// shutdown scheduler
// TODO: figure out why the ... Mac OS dies with a segfault here // TODO: figure out why the ... Mac OS dies with a segfault here
# ifndef __APPLE__ //# ifndef __APPLE__
auto s = m_scheduler.load(); stop_and_kill(s_scheduler);
if (s) //# endif
stop_and_kill(s_network_manager);
// it's safe now to delete all other singletons now
delete s_group_manager.load();
auto et = s_empty_tuple.load();
if (et && !et->deref()) delete et;
delete s_actor_registry.load();
delete s_uniform_type_info_map.load();
}
}
;//s_cleanup_helper;
template<typename T>
T* lazy_get(std::atomic<T*>& ptr)
{
T* result = ptr.load();
if (result == nullptr)
{
auto tmp = new T();
if (ptr.compare_exchange_weak(result, tmp) == false)
{
delete tmp;
}
else
{ {
s->stop(); return tmp;
delete s;
} }
# endif
// it's safe now to delete all other singletons
delete m_group_manager;
if (!m_empty_tuple->deref()) delete m_empty_tuple;
delete m_actor_registry;
delete m_uniform_type_info_map;
} }
return result;
}; }
singleton_container s_container;
} // namespace <anonymous> } // namespace <anonymous>
...@@ -119,46 +131,59 @@ namespace cppa { namespace detail { ...@@ -119,46 +131,59 @@ namespace cppa { namespace detail {
actor_registry* singleton_manager::get_actor_registry() actor_registry* singleton_manager::get_actor_registry()
{ {
return s_container.m_actor_registry; return lazy_get(s_actor_registry);
} }
uniform_type_info_map* singleton_manager::get_uniform_type_info_map() uniform_type_info_map* singleton_manager::get_uniform_type_info_map()
{ {
return s_container.m_uniform_type_info_map; return lazy_get(s_uniform_type_info_map);
} }
group_manager* singleton_manager::get_group_manager() group_manager* singleton_manager::get_group_manager()
{ {
return s_container.m_group_manager; return lazy_get(s_group_manager);
} }
scheduler* singleton_manager::get_scheduler() scheduler* singleton_manager::get_scheduler()
{ {
return s_container.m_scheduler.load(); return s_scheduler.load();
} }
bool singleton_manager::set_scheduler(scheduler* ptr) bool singleton_manager::set_scheduler(scheduler* ptr)
{ {
scheduler* expected = nullptr; scheduler* expected = nullptr;
if (s_container.m_scheduler.compare_exchange_weak(expected, ptr)) if (s_scheduler.compare_exchange_weak(expected, ptr))
{ {
ptr->start(); ptr->start();
s_container.m_network_manager = network_manager::create_singleton(); auto nm = network_manager::create_singleton();
s_container.m_network_manager->start(); network_manager* nm_expected = nullptr;
if (s_network_manager.compare_exchange_weak(nm_expected, nm))
{
nm->start();
}
else
{
delete nm;
}
return true; return true;
} }
return false; else
{
delete ptr;
return false;
}
} }
network_manager* singleton_manager::get_network_manager() network_manager* singleton_manager::get_network_manager()
{ {
network_manager* result = s_container.m_network_manager; network_manager* result = s_network_manager.load();
if (result == nullptr) if (result == nullptr)
{ {
scheduler* s = new thread_pool_scheduler; scheduler* s = new thread_pool_scheduler;
// set_scheduler sets s_network_manager
if (set_scheduler(s) == false) if (set_scheduler(s) == false)
{ {
delete s; //delete s;
} }
return get_network_manager(); return get_network_manager();
} }
...@@ -167,7 +192,21 @@ network_manager* singleton_manager::get_network_manager() ...@@ -167,7 +192,21 @@ network_manager* singleton_manager::get_network_manager()
empty_tuple* singleton_manager::get_empty_tuple() empty_tuple* singleton_manager::get_empty_tuple()
{ {
return s_container.m_empty_tuple; empty_tuple* result = s_empty_tuple.load();
if (result == nullptr)
{
auto tmp = new empty_tuple;
if (s_empty_tuple.compare_exchange_weak(result, tmp) == false)
{
delete tmp;
}
else
{
result = tmp;
result->ref();
}
}
return result;
} }
} } // namespace cppa::detail } } // namespace cppa::detail
...@@ -130,13 +130,7 @@ int main(int argc, char** argv) ...@@ -130,13 +130,7 @@ int main(int argc, char** argv)
if (found_key(i, args, "run")) if (found_key(i, args, "run"))
{ {
auto& what = i->second; auto& what = i->second;
/* if (what == "performance_test") if (what == "remote_actor")
{
cout << endl << "run queue performance test ... " << endl;
test__queue_performance();
return 0;
}
else*/ if (what == "remote_actor")
{ {
test__remote_actor(argv[0], true, args); test__remote_actor(argv[0], true, args);
return 0; return 0;
......
...@@ -11,21 +11,20 @@ using std::cout; ...@@ -11,21 +11,20 @@ using std::cout;
using std::endl; using std::endl;
using namespace cppa; using namespace cppa;
void pong(actor_ptr ping_actor); int pongs()
{
return s_pongs;
}
void ping() void ping()
{ {
s_pongs = 0; s_pongs = 0;
actor_ptr pong_actor;
auto response = make_tuple(atom("Ping"), 0);
// invoke rule
receive_loop receive_loop
( (
on<atom("Pong"), std::int32_t>() >> [&](std::int32_t value) on<atom("pong"), int>() >> [](int value)
{ {
++s_pongs; ++s_pongs;
get_ref<1>(response) = value + 1; reply(atom("ping"), value + 1);
last_sender() << response;
} }
); );
} }
...@@ -33,27 +32,19 @@ void ping() ...@@ -33,27 +32,19 @@ void ping()
void pong(actor_ptr ping_actor) void pong(actor_ptr ping_actor)
{ {
link(ping_actor); link(ping_actor);
auto pong_tuple = make_tuple(atom("Pong"), 0);
// kickoff // kickoff
ping_actor << pong_tuple; send(ping_actor, atom("pong"), 0);
// invoke rules
receive_loop receive_loop
( (
on(atom("Ping"), 9) >> []() on(atom("ping"), 9) >> []()
{ {
// terminate with non-normal exit reason // terminate with non-normal exit reason
// to force ping actor to quit // to force ping actor to quit
quit(exit_reason::user_defined); quit(exit_reason::user_defined);
}, },
on<atom("Ping"), int>() >> [&](int value) on<atom("ping"), int>() >> [](int value)
{ {
get_ref<1>(pong_tuple) = value + 1; reply(atom("pong"), value + 1);
last_sender() << pong_tuple;
} }
); );
} }
int pongs()
{
return s_pongs;
}
...@@ -45,15 +45,15 @@ size_t test__local_group() ...@@ -45,15 +45,15 @@ size_t test__local_group()
//receive_until([&result]() { return result == 10; }) //receive_until([&result]() { return result == 10; })
do_receive do_receive
( (
on<int>() >> [&result](int value) on<int>() >> [&](int value)
{ {
CPPA_CHECK(value == 2);
result += value; result += value;
}, },
after(std::chrono::seconds(2)) >> [&]() after(std::chrono::seconds(2)) >> [&]()
{ {
CPPA_CHECK(false); CPPA_CHECK(false);
} }
) )
.until([&result]() { return result == 10; }); .until([&result]() { return result == 10; });
await_all_others_done(); await_all_others_done();
......
...@@ -9,8 +9,116 @@ ...@@ -9,8 +9,116 @@
using namespace cppa; using namespace cppa;
static uniform_type_info const* iinfo = uniform_typeid<int>();
template<typename T>
struct uti_util
{
static inline uniform_type_info const* get() { return uniform_typeid<T>(); }
};
template<>
struct uti_util<anything>
{
static inline uniform_type_info const* get() { return nullptr; }
};
template<typename T>
struct tinfo_util
{
static inline std::type_info const* get() { return &(typeid(T)); }
};
template<>
struct tinfo_util<anything>
{
static inline std::type_info const* get() { return nullptr; }
};
template<bool BuiltinOnlyTypes, typename... T>
struct type_arr_base
{
uniform_type_info const* data[sizeof...(T)];
type_arr_base() : data({uti_util<T>::get()...})
{
}
inline uniform_type_info const* operator[](size_t p) const
{
return data[p];
}
inline size_t size() const { return sizeof...(T); }
};
template<typename... T>
struct type_arr_base<false, T...>
{
std::type_info const* data[sizeof...(T)];
type_arr_base() : data({tinfo_util<T>::get()...})
{
}
inline std::type_info const* operator[](size_t p) const
{
return data[p];
}
inline size_t size() const
{
return sizeof...(T);
}
};
template<typename T>
struct is_builtin
{
static constexpr bool value = std::is_arithmetic<T>::value;
};
template<>
struct is_builtin<anything>
{
static constexpr bool value = true;
};
template<typename... T>
struct type_arr : type_arr_base<util::eval_type_list<util::type_list<T...>,is_builtin>::value, T...>
{
};
void subtest()
{
typedef pattern<int,int,int> i3;
i3 ip;
std::vector<int> vec = {1, 2, 3};
detail::pattern_iterator arg0(i3::size, ip.m_data_ptr, ip.m_utis);
detail::tuple_iterator_arg<std::vector<int>, i3::mapping_vector> arg1(vec);
cout << "match(vec) ( on<int,int,int> =====> " << detail::do_match(arg0, arg1) << endl;
}
template<typename Arr>
void plot(Arr const& arr)
{
cout << "{ ";
for (size_t i = 0; i < arr.size(); ++i)
{
if (i > 0) cout << ", ";
cout << "arr[" << i << "] = " << ((arr[i]) ? arr[i]->name() : "anything");
}
cout << " }" << endl;
}
class foobar { };
static type_arr_base<true,int,anything,float> arr1;
static type_arr_base<false,int,anything,foobar> arr2;
size_t test__pattern() size_t test__pattern()
{ {
cout << "iinfo->name() = " << iinfo->name() << endl;
plot(arr1);
plot(arr2);
//subtest();
CPPA_TEST(test__pattern); CPPA_TEST(test__pattern);
// some pattern objects to play with // some pattern objects to play with
pattern<atom_value, int, std::string> p0{util::wrapped<atom_value>()}; pattern<atom_value, int, std::string> p0{util::wrapped<atom_value>()};
......
...@@ -15,7 +15,7 @@ using namespace cppa; ...@@ -15,7 +15,7 @@ using namespace cppa;
namespace { namespace {
void client_part(const std::map<std::string, std::string>& args) void client_part(std::map<std::string, std::string> const& args)
{ {
auto i = args.find("port"); auto i = args.find("port");
if (i == args.end()) if (i == args.end())
...@@ -26,20 +26,14 @@ void client_part(const std::map<std::string, std::string>& args) ...@@ -26,20 +26,14 @@ void client_part(const std::map<std::string, std::string>& args)
std::uint16_t port; std::uint16_t port;
iss >> port; iss >> port;
auto ping_actor = remote_actor("localhost", port); auto ping_actor = remote_actor("localhost", port);
try spawn(pong, ping_actor);
{
pong(ping_actor);
}
catch (cppa::actor_exited&)
{
}
await_all_others_done(); await_all_others_done();
} }
} // namespace <anonymous> } // namespace <anonymous>
size_t test__remote_actor(const char* app_path, bool is_client, size_t test__remote_actor(char const* app_path, bool is_client,
const std::map<std::string, std::string>& args) std::map<std::string, std::string> const& args)
{ {
if (is_client) if (is_client)
{ {
...@@ -64,16 +58,11 @@ size_t test__remote_actor(const char* app_path, bool is_client, ...@@ -64,16 +58,11 @@ size_t test__remote_actor(const char* app_path, bool is_client,
} }
} }
while (!success); while (!success);
//cout << "port = " << port << endl; std::ostringstream oss;
std::string cmd; oss << app_path << " run=remote_actor port=" << port << " &>/dev/null";
{
std::ostringstream oss;
oss << app_path << " run=remote_actor port=" << port << " &>/dev/null";
cmd = oss.str();
}
// execute client_part() in a separate process, // execute client_part() in a separate process,
// connected via localhost socket // connected via localhost socket
detail::thread child([&cmd]() { system(cmd.c_str()); }); detail::thread child([&oss]() { system(oss.str().c_str()); });
await_all_others_done(); await_all_others_done();
CPPA_CHECK_EQUAL(pongs(), 5); CPPA_CHECK_EQUAL(pongs(), 5);
// wait until separate process (in sep. thread) finished execution // wait until separate process (in sep. thread) finished execution
......
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