Commit 9e34fdf9 authored by neverlord's avatar neverlord

bugfixing

parent b61316b9
......@@ -66,9 +66,10 @@
# error Plattform and/or compiler not supported
#endif
#ifdef CPPA_DEBUG
#include <cstdio>
#include <cstdlib>
#ifdef CPPA_DEBUG
#include <execinfo.h>
#define CPPA_REQUIRE__(stmt, file, line) \
......@@ -83,18 +84,16 @@
if ((stmt) == false) { \
CPPA_REQUIRE__(#stmt, __FILE__, __LINE__); \
}((void) 0)
#define CPPA_CRITICAL__(error, file, line) \
printf("%s:%u: critical error: '%s'\n", file, line, error); \
{ \
void *array[10]; \
size_t size = backtrace(array, 10); \
backtrace_symbols_fd(array, size, 2); \
} \
exit(7)
#define CPPA_CRITICAL(error) CPPA_CRITICAL__(error, __FILE__, __LINE__)
#else // CPPA_DEBUG
#define CPPA_REQUIRE(unused) ((void) 0)
#define CPPA_CRITICAL(unused) exit(7)
#endif // CPPA_DEBUG
#define CPPA_CRITICAL__(error, file, line) \
{ \
printf("%s:%u: critical error: '%s'\n", file, line, error); \
exit(7); \
} ((void) 0)
#define CPPA_CRITICAL(error) CPPA_CRITICAL__(error, __FILE__, __LINE__)
#endif // CPPA_CONFIG_HPP
......@@ -45,13 +45,7 @@ class addressed_message
public:
addressed_message(actor_ptr const& from,
channel_ptr const& to,
any_tuple const& ut);
addressed_message(actor_ptr const& from,
channel_ptr const& to,
any_tuple&& ut);
addressed_message(actor_ptr from, channel_ptr to, any_tuple ut);
addressed_message() = default;
addressed_message(addressed_message&&) = default;
......
......@@ -31,7 +31,9 @@
#ifndef THREAD_HPP
#define THREAD_HPP
#ifdef __APPLE__
//#ifdef __APPLE__
#if 1
#include <boost/thread.hpp>
#include "cppa/util/duration.hpp"
......
......@@ -49,8 +49,9 @@ void actor_proxy::forward_message(process_information_ptr const& piptr,
actor* sender,
any_tuple&& msg)
{
detail::addressed_message amsg{sender, this, std::move(msg)};
detail::singleton_manager::get_network_manager()
->send_to_mailman(make_any_tuple(piptr, actor_ptr{sender}, std::move(msg)));
->send_to_mailman(make_any_tuple(piptr, std::move(amsg)));
}
void actor_proxy::enqueue(actor* sender, any_tuple msg)
......
......@@ -33,17 +33,8 @@
namespace cppa { namespace detail {
addressed_message::addressed_message(const actor_ptr& from,
const channel_ptr& to,
const any_tuple& ut)
: m_sender(from), m_receiver(to), m_content(ut)
{
}
addressed_message::addressed_message(const actor_ptr& from,
const channel_ptr& to,
any_tuple&& ut)
: m_sender(from), m_receiver(to), m_content(std::move(ut))
addressed_message::addressed_message(actor_ptr from, channel_ptr to, any_tuple ut)
: m_sender(std::move(from)), m_receiver(std::move(to)), m_content(std::move(ut))
{
}
......
......@@ -99,11 +99,12 @@ void mailman_loop()
}
else
{
DEBUG("message to an unknown peer");
DEBUG("message to an unknown peer: " << to_string(msg));
}
},
on_arg_match >> [&](native_socket_type sockfd, process_information_ptr pinfo)
{
DEBUG("mailman: add_peer");
auto i = peers.find(*pinfo);
if (i == peers.end())
{
......
......@@ -167,41 +167,29 @@ class po_peer
// @returns false if an error occured; otherwise true
void operator()()
{
static constexpr size_t wfp_size = sizeof(std::uint32_t)
+ process_information::node_id_size;
auto nm = singleton_manager::get_network_manager();
auto meta_msg = uniform_typeid<addressed_message>();
buffer<s_chunk_size, s_max_buffer_size> rdbuf;
auto guard = make_scope_guard([&]()
{
DEBUG("po_peer loop done");
send2po(atom("RM_PEER"), m_socket);
});
DEBUG("po_peer loop started");
std::uint32_t msg_size;
std::vector<char> msg_buf;
for (;;)
{
switch (m_state)
{
case wait_for_process_info:
{
if (rdbuf.final_size() != wfp_size)
{
rdbuf.reset(wfp_size);
}
if (rdbuf.append_from(m_socket) == false)
{
DEBUG("m_rdbuf.append_from() failed");
return;
}
CPPA_REQUIRE(rdbuf.ready());
std::uint32_t process_id;
memcpy(&process_id, rdbuf.data(), sizeof(std::uint32_t));
process_information::node_id_type node_id;
memcpy(node_id.data(),
rdbuf.data() + sizeof(std::uint32_t),
process_information::node_id_size);
recv(m_socket, &process_id, sizeof(std::uint32_t), 0);
recv(m_socket, node_id.data(), node_id.size(), 0);
m_peer.reset(new process_information(process_id, node_id));
// inform mailman about new peer
nm->send_to_mailman(make_any_tuple(m_socket, m_peer));
rdbuf.reset();
m_state = wait_for_msg_size;
DEBUG("pinfo read: "
<< m_peer->process_id()
......@@ -211,33 +199,24 @@ class po_peer
}
case wait_for_msg_size:
{
if (rdbuf.final_size() != sizeof(std::uint32_t))
recv(m_socket, &msg_size, sizeof(std::uint32_t), 0);
if (msg_size > (16 * 1024 * 1024))
{
rdbuf.reset(sizeof(std::uint32_t));
// maximum of 16MB
return;
}
if (!rdbuf.append_from(m_socket))
if (msg_buf.size() < msg_size)
{
DEBUG("m_rdbuf.append_from() failed");
return;
msg_buf.resize(msg_size);
}
CPPA_REQUIRE(rdbuf.ready());
// read and set message size
std::uint32_t msg_size;
memcpy(&msg_size, rdbuf.data(), sizeof(std::uint32_t));
rdbuf.reset(msg_size);
m_state = read_message;
// fall through and try to read more from socket
}
case read_message:
{
if (!rdbuf.append_from(m_socket))
{
DEBUG("m_rdbuf.append_from() failed");
return;
}
CPPA_REQUIRE(rdbuf.ready());
recv(m_socket, msg_buf.data(), msg_size, 0);
addressed_message msg;
binary_deserializer bd(rdbuf.data(), rdbuf.size());
binary_deserializer bd(msg_buf.data(), msg_size);
try
{
meta_msg->deserialize(&msg, &bd);
......@@ -307,7 +286,6 @@ class po_peer
}
}
);
rdbuf.reset();
m_state = wait_for_msg_size;
break;
}
......@@ -325,27 +303,42 @@ class po_peer
class po_doorman
{
// server socket
native_socket_type m_socket;
actor_ptr published_actor;
// caches process_information::get()
process_information_ptr m_pself;
int m_pipe_rd;
int m_pipe_wr;
thread m_thread;
struct socket_aid_pair
{
native_socket_type fd;
actor_id aid;
};
public:
po_doorman(native_socket_type fd, actor_ptr mactor)
: m_socket(fd)
, published_actor(std::move(mactor))
, m_pself(process_information::get())
po_doorman() : m_pself(process_information::get())
{
int mpipe[2];
if (pipe(mpipe) < 0)
{
CPPA_CRITICAL("cannot open pipe");
}
m_pipe_rd = mpipe[0];
m_pipe_wr = mpipe[1];
}
~po_doorman()
{
if (m_socket != -1) closesocket(m_socket);
DEBUG(__PRETTY_FUNCTION__);
socket_aid_pair msg{-1, 0};
write(m_pipe_wr, &msg, sizeof(socket_aid_pair));
m_thread.join();
close(m_pipe_rd);
close(m_pipe_wr);
}
void start()
......@@ -353,27 +346,135 @@ class po_doorman
m_thread = thread{std::bind(&po_doorman::operator(), this)};
}
void add(native_socket_type fd, actor_id aid)
{
DEBUG("add, aid = " << aid);
CPPA_REQUIRE(fd != -1);
CPPA_REQUIRE(aid != 0);
socket_aid_pair msg{fd, aid};
if (write(m_pipe_wr, &msg, sizeof(socket_aid_pair)) != sizeof(socket_aid_pair))
{
CPPA_CRITICAL("cannot write to pipe");
}
}
void rm(actor_id aid)
{
CPPA_REQUIRE(aid != 0);
socket_aid_pair msg{-1, aid};
if (write(m_pipe_wr, &msg, sizeof(socket_aid_pair)) != sizeof(socket_aid_pair))
{
CPPA_CRITICAL("cannot write to pipe");
}
}
void operator()()
{
sockaddr addr;
socklen_t addrlen;
std::vector<socket_aid_pair> pvec;
int maxfd = 0;
fd_set readset;
auto guard = make_scope_guard([&]()
{
DEBUG(__PRETTY_FUNCTION__);
for (auto& pm : pvec)
{
closesocket(pm.fd);
}
});
for (;;)
{
FD_ZERO(&readset);
FD_SET(m_pipe_rd, &readset);
maxfd = m_pipe_rd;
for (auto i = pvec.begin(); i != pvec.end(); ++i)
{
maxfd = std::max(maxfd, i->fd);
FD_SET(i->fd, &readset);
}
if (select(maxfd + 1, &readset, nullptr, nullptr, nullptr) < 0)
{
// must not happen
perror("select()");
exit(3);
}
// iterate over sockets
{
auto i = pvec.begin();
while (i != pvec.end())
{
if (FD_ISSET(i->fd, &readset))
{
memset(&addr, 0, sizeof(addr));
memset(&addrlen, 0, sizeof(addrlen));
auto sfd = ::accept(m_socket, &addr, &addrlen);
auto sfd = ::accept(i->fd, &addr, &addrlen);
if (sfd < 0)
{
DEBUG("accept failed (actor unpublished?)");
return;
switch (errno)
{
case EAGAIN:
# if EAGAIN != EWOULDBLOCK
case EWOULDBLOCK:
# endif
// just try again
++i;
break;
default:
// remove socket on failure
i = pvec.erase(i);
break;
}
}
auto id = published_actor->id();
else
{
auto id = i->aid;
std::uint32_t process_id = m_pself->process_id();
::send(sfd, &id, sizeof(std::uint32_t), 0);
::send(sfd, &process_id, sizeof(std::uint32_t), 0);
::send(sfd, m_pself->node_id().data(), m_pself->node_id().size(), 0);
send2po(atom("ADD_PEER"), sfd, process_information_ptr{});
DEBUG("socket accepted; published actor: " << id);
++i;
}
}
}
}
if (FD_ISSET(m_pipe_rd, &readset))
{
DEBUG("po_doorman: read from pipe");
socket_aid_pair msg;
if (read(m_pipe_rd, &msg, sizeof(socket_aid_pair)) != sizeof(socket_aid_pair))
{
CPPA_CRITICAL("cannot read from pipe");
}
if (msg.fd == -1)
{
if (msg.aid == 0)
{
DEBUG("fd == -1 && aid == 0 [done]");
return;
}
else
{
auto i = std::find_if(pvec.begin(), pvec.end(), [&](socket_aid_pair const& m)
{
return m.aid == msg.aid;
});
if (i != pvec.end())
{
DEBUG("removed socket of actor" << i->aid);
closesocket(i->fd);
pvec.erase(i);
}
}
}
else
{
DEBUG("added socket for actor" << msg.aid);
pvec.push_back(msg);
}
}
}
}
......@@ -381,9 +482,9 @@ class po_doorman
void post_office_loop()
{
po_doorman doorman;
doorman.start();
bool done = false;
// map of all published actors
std::map<actor_id, std::list<po_doorman> > doormen;
// list of all peers to which we established a connection via remote_actor()
std::list<po_peer> peers;
do_receive
......@@ -391,13 +492,13 @@ void post_office_loop()
on(atom("ADD_PEER"), arg_match) >> [&](native_socket_type fd,
process_information_ptr piptr)
{
DEBUG("add_peer_msg");
DEBUG("post_office: add_peer");
peers.emplace_back(fd, std::move(piptr));
peers.back().start();
},
on(atom("RM_PEER"), arg_match) >> [&](native_socket_type fd)
{
DEBUG("rm_peer_msg");
DEBUG("post_office: rm_peer");
auto i = std::find_if(peers.begin(), peers.end(), [fd](po_peer& pp)
{
return pp.get_socket() == fd;
......@@ -406,40 +507,25 @@ void post_office_loop()
},
on(atom("ADD_PROXY"), arg_match) >> [&](actor_proxy_ptr)
{
DEBUG("add_proxy_msg");
DEBUG("post_office: add_proxy");
},
on(atom("RM_PROXY"), arg_match) >> [&](actor_proxy_ptr pptr)
{
DEBUG("rm_proxy_msg");
DEBUG("post_office: rm_proxy");
CPPA_REQUIRE(pptr.get() != nullptr);
get_actor_proxy_cache().erase(pptr);
},
on(atom("PUBLISH"), arg_match) >> [&](native_socket_type sockfd,
actor_ptr whom)
{
DEBUG("unpublish_actor_event");
DEBUG("post_office: publish_actor");
CPPA_REQUIRE(whom.get() != nullptr);
auto aid = whom->id();
auto callback = [aid](std::uint32_t)
{
send2po(atom("UNPUBLISH"), aid);
};
if (whom->attach_functor(std::move(callback)))
{
auto& ls = doormen[aid];
ls.emplace_back(sockfd, whom);
ls.back().start();
DEBUG("new doorman");
}
else
{
closesocket(sockfd);
}
doorman.add(sockfd, whom->id());
},
on(atom("UNPUBLISH"), arg_match) >> [&](actor_id whom)
{
DEBUG("unpublish_actor_event");
doormen.erase(whom);
DEBUG("post_office: unpublish_actor");
doorman.rm(whom);
},
on(atom("DONE")) >> [&]()
{
......@@ -461,8 +547,9 @@ void post_office_loop()
******************************************************************************/
void post_office_add_peer(native_socket_type a0,
const process_information_ptr& a1)
process_information_ptr const& a1)
{
DEBUG("post_office_add_peer(" << a0 << ", " << to_string(a1) << ")");
send2po(atom("ADD_PEER"), a0, a1);
}
......
......@@ -100,7 +100,8 @@ struct thread_pool_scheduler::worker
{
return result;
}
# ifdef __APPLE__
# if 1
//# ifdef __APPLE__
auto timeout = boost::get_system_time();
timeout += boost::posix_time::milliseconds(1);
boost::this_thread::sleep(timeout);
......@@ -121,7 +122,8 @@ struct thread_pool_scheduler::worker
{
return result;
}
# ifdef __APPLE__
# if 1
//# ifdef __APPLE__
auto timeout = boost::get_system_time();
timeout += boost::posix_time::milliseconds(10);
boost::this_thread::sleep(timeout);
......@@ -154,6 +156,9 @@ struct thread_pool_scheduler::worker
handler h;
for (;;)
{
h.job = aggressive_polling();
while (!h.job) h.job = less_aggressive_polling();
/*
h.job = aggressive_polling();
if (!h.job)
{
......@@ -163,6 +168,7 @@ struct thread_pool_scheduler::worker
h.job = relaxed_polling();
}
}
*/
if (h.job == m_dummy)
{
// dummy of doom received ...
......
......@@ -125,7 +125,8 @@ std::string to_uniform_name_impl(Iterator begin, Iterator end,
{ demangled<cppa::actor_ptr>(), "@actor" },
{ demangled<cppa::group_ptr>(), "@group" },
{ demangled<cppa::channel_ptr>(), "@channel" },
{ demangled<cppa::detail::addressed_message>(), "@msg" }
{ demangled<cppa::detail::addressed_message>(), "@msg" },
{ demangled< cppa::intrusive_ptr<cppa::process_information> >(), "@process_info" }
};
// check if we could find the whole string in our lookup map
......
......@@ -133,6 +133,10 @@ void publish(actor_ptr& whom, std::uint16_t port)
{
throw network_error("unable to get socket flags");
}
if (fcntl(sockfd, F_SETFL, flags | O_NONBLOCK) < 0)
{
throw network_error("unable to set socket to nonblock");
}
if (bind(sockfd, (struct sockaddr*) &serv_addr, sizeof(serv_addr)) < 0)
{
throw bind_failure(errno);
......@@ -172,7 +176,7 @@ actor_ptr remote_actor(const char* host, std::uint16_t port)
serv_addr.sin_family = AF_INET;
memmove(&serv_addr.sin_addr.s_addr, server->h_addr, server->h_length);
serv_addr.sin_port = htons(port);
if (connect(sockfd, (const sockaddr*) &serv_addr, sizeof(serv_addr)) < 0)
if (connect(sockfd, (const sockaddr*) &serv_addr, sizeof(serv_addr)) != 0)
{
throw network_error("could not connect to host");
}
......@@ -189,12 +193,11 @@ actor_ptr remote_actor(const char* host, std::uint16_t port)
auto peer_pinf = new process_information(peer_pid, peer_node_id);
process_information_ptr pinfptr(peer_pinf);
auto key = std::make_tuple(remote_actor_id, pinfptr->process_id(), pinfptr->node_id());
auto result = detail::get_actor_proxy_cache().get(key);
detail::singleton_manager::get_network_manager()
->send_to_mailman(make_any_tuple(sockfd, pinfptr));
detail::post_office_add_peer(sockfd, pinfptr);
return detail::get_actor_proxy_cache().get(key);
//auto ptr = get_scheduler()->register_hidden_context();
return result;
}
} // namespace cppa
......@@ -158,7 +158,7 @@ class void_type_tinfo : public util::abstract_uniform_type_info<void_type>
protected:
void serialize(const void*, serializer* sink) const
void serialize(void const*, serializer* sink) const
{
serialize_nullptr(sink);
}
......@@ -251,7 +251,7 @@ class actor_ptr_tinfo : public util::abstract_uniform_type_info<actor_ptr>
protected:
void serialize(const void* ptr, serializer* sink) const
void serialize(void const* ptr, serializer* sink) const
{
s_serialize(*reinterpret_cast<const actor_ptr*>(ptr),
sink,
......@@ -317,7 +317,7 @@ class group_ptr_tinfo : public util::abstract_uniform_type_info<group_ptr>
protected:
void serialize(const void* ptr, serializer* sink) const
void serialize(void const* ptr, serializer* sink) const
{
s_serialize(*reinterpret_cast<const group_ptr*>(ptr),
sink,
......@@ -411,7 +411,7 @@ class channel_ptr_tinfo : public util::abstract_uniform_type_info<channel_ptr>
protected:
void serialize(const void* instance, serializer* sink) const
void serialize(void const* instance, serializer* sink) const
{
s_serialize(*reinterpret_cast<const channel_ptr*>(instance),
sink,
......@@ -479,7 +479,7 @@ class any_tuple_tinfo : public util::abstract_uniform_type_info<any_tuple>
protected:
void serialize(const void* instance, serializer* sink) const
void serialize(void const* instance, serializer* sink) const
{
s_serialize(*reinterpret_cast<const any_tuple*>(instance),sink,name());
}
......@@ -501,7 +501,7 @@ class addr_msg_tinfo : public util::abstract_uniform_type_info<addressed_message
public:
virtual void serialize(const void* instance, serializer* sink) const
virtual void serialize(void const* instance, serializer* sink) const
{
const addressed_message& msg = *reinterpret_cast<const addressed_message*>(instance);
const any_tuple& data = msg.content();
......@@ -547,12 +547,68 @@ class addr_msg_tinfo : public util::abstract_uniform_type_info<addressed_message
};
class process_info_ptr_tinfo : public util::abstract_uniform_type_info<process_information_ptr>
{
typedef process_information_ptr ptr_type;
public:
virtual void serialize(void const* instance, serializer* sink) const
{
auto& ptr = *reinterpret_cast<ptr_type const*>(instance);
if (ptr == nullptr)
{
serialize_nullptr(sink);
}
else
{
primitive_variant ptup[2];
ptup[0] = ptr->process_id();
ptup[1] = to_string(ptr->node_id());
sink->begin_object(name());
sink->write_tuple(2, ptup);
sink->end_object();
}
}
virtual void deserialize(void* instance, deserializer* source) const
{
auto& ptrref = *reinterpret_cast<ptr_type*>(instance);
std::string cname = source->seek_object();
if (cname != name())
{
if (cname == nullptr_type_name)
{
deserialize_nullptr(source);
ptrref.reset();
}
else
{
throw std::logic_error("wrong type name found");
}
}
else
{
primitive_variant ptup[2];
primitive_type ptypes[] = { pt_uint32, pt_u8string };
source->begin_object(cname);
source->read_tuple(2, ptypes, ptup);
source->end_object();
process_information::node_id_type nid;
node_id_from_string(get<std::string>(ptup[1]), nid);
ptrref.reset(new process_information{get<std::uint32_t>(ptup[0]), nid});
}
}
};
class atom_value_tinfo : public util::abstract_uniform_type_info<atom_value>
{
public:
virtual void serialize(const void* instance, serializer* sink) const
virtual void serialize(void const* instance, serializer* sink) const
{
auto val = reinterpret_cast<const atom_value*>(instance);
sink->begin_object(name());
......@@ -576,7 +632,7 @@ class atom_value_tinfo : public util::abstract_uniform_type_info<atom_value>
class duration_tinfo : public util::abstract_uniform_type_info<util::duration>
{
virtual void serialize(const void* instance, serializer* sink) const
virtual void serialize(void const* instance, serializer* sink) const
{
auto val = reinterpret_cast<const util::duration*>(instance);
sink->begin_object(name());
......@@ -694,6 +750,7 @@ class uniform_type_info_map_helper
insert(d, new atom_value_tinfo, { raw_name<atom_value>() });
insert(d, new addr_msg_tinfo, {raw_name<detail::addressed_message>() });
insert(d, new void_type_tinfo, { raw_name<void_type>() });
insert(d, new process_info_ptr_tinfo, {raw_name<process_information_ptr>()});
insert<float>(d);
insert<double>(d);
/*
......
......@@ -175,8 +175,8 @@ int main(int argc, char** argv)
},
on("run_ping", arg_match) >> [&](std::string const& num_pings)
{
//auto ping_actor = spawn<detached>(ping, std::stoi(num_pings));
auto ping_actor = spawn_event_based_ping(std::stoi(num_pings));
auto ping_actor = spawn<detached>(ping, std::stoi(num_pings));
//auto ping_actor = spawn_event_based_ping(std::stoi(num_pings));
std::uint16_t port = 4242;
bool success = false;
do
......
......@@ -26,7 +26,8 @@ void client_part(std::vector<string_pair> const& args)
}
auto port = static_cast<std::uint16_t>(std::stoi(i->second));
auto ping_actor = remote_actor("localhost", port);
spawn_event_based_pong(ping_actor);
//spawn_event_based_pong(ping_actor);
spawn<detached>(pong, ping_actor);
await_all_others_done();
}
......
......@@ -87,6 +87,7 @@ size_t test__uniform_type()
"@actor", // cppa::actor_ptr
"@group", // cppa::group_ptr
"@channel", // cppa::channel_ptr
"@process_info", // cppa::intrusive_ptr<cppa::process_information>
"cppa::util::duration"
};
// holds the type names we see at runtime
......
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