Commit 178334ac authored by neverlord's avatar neverlord

implemented trap_exit() and quit(), fixed test__spawn() and started unicast_network.cpp

parent b96539c5
......@@ -135,7 +135,7 @@
</data>
<data>
<variable>ProjectExplorer.Project.Updater.EnvironmentId</variable>
<value type="QString">{23902c37-f07e-47cd-bb19-c366b9f708db}</value>
<value type="QString">{07fcd197-092d-45a0-8500-3be614e6ae31}</value>
</data>
<data>
<variable>ProjectExplorer.Project.Updater.FileVersion</variable>
......
......@@ -161,3 +161,6 @@ cppa/process_information.hpp
src/process_information.cpp
cppa/util/ripemd_160.hpp
src/ripemd_160.cpp
src/unicast_network.cpp
cppa/actor_exited.hpp
src/actor_exited.cpp
......@@ -5,6 +5,7 @@
#include "cppa/group.hpp"
#include "cppa/channel.hpp"
#include "cppa/process_information.hpp"
namespace cppa {
......@@ -59,6 +60,11 @@ class actor : public channel
*/
virtual bool establish_backlink(const intrusive_ptr<actor>& to) = 0;
/**
* @brief Gets the {@link process_information} of the parent process.
*/
virtual const process_information& parent_process() const;
/**
* @brief
* @return
......
#ifndef ACTOR_EXITED_HPP
#define ACTOR_EXITED_HPP
#include <string>
#include <cstdint>
#include <exception>
namespace cppa {
class actor_exited : public std::exception
{
std::uint32_t m_reason;
std::string m_what;
public:
~actor_exited() throw();
actor_exited(std::uint32_t exit_reason);
std::uint32_t reason() const;
virtual const char* what() const throw();
};
} // namespace cppa
#endif // ACTOR_EXITED_HPP
......@@ -17,7 +17,7 @@ class binary_serializer : public serializer
char* m_end;
char* m_wr_pos;
// make that it's safe to write num_bytes bytes to m_wr_pos
// make sure that it's safe to write num_bytes bytes to m_wr_pos
void acquire(size_t num_bytes);
public:
......@@ -54,6 +54,11 @@ class binary_serializer : public serializer
*/
const char* data() const;
/**
* @brief Resets the internal buffer.
*/
void reset();
};
} // namespace cppa
......
......@@ -43,6 +43,8 @@ class context : public actor
*/
virtual void enqueue /*[[override]]*/ (const message& msg);
void trap_exit(bool new_value);
};
/**
......
......@@ -30,6 +30,7 @@
#define CPPA_HPP
#include <tuple>
#include <cstdint>
#include <type_traits>
#include "cppa/tuple.hpp"
......@@ -52,16 +53,30 @@
namespace cppa {
/**
* @brief Links the calling actor to @p other.
*/
inline void link(actor_ptr& other)
{
self()->link(other);
}
/**
* @brief Links the calling actor to @p other.
*/
inline void link(actor_ptr&& other)
{
self()->link(other);
}
inline void trap_exit(bool new_value)
{
self()->trap_exit(new_value);
}
/**
* @brief Spawns a new actor that executes @p what with given arguments.
*/
template<scheduling_hint Hint, typename F, typename... Args>
actor_ptr spawn(F&& what, const Args&... args)
{
......@@ -71,27 +86,43 @@ actor_ptr spawn(F&& what, const Args&... args)
return get_scheduler()->spawn(ptr, Hint);
}
/**
* @brief Alias for <tt>spawn<scheduled>(what, args...)</tt>.
*/
template<typename F, typename... Args>
actor_ptr spawn(F&& what, const Args&... args)
{
return spawn<scheduled>(std::forward<F>(what), args...);
}
/**
* @brief Quits execution of the calling actor.
*/
inline void quit(std::uint32_t reason)
{
self()->quit(reason);
}
/**
* @brief Quits execution of the calling actor.
*/
inline void quit(exit_reason reason)
{
self()->quit(reason);
}
/**
* @brief Dequeues the next message from the mailbox.
*/
inline const message& receive()
{
return self()->mailbox().dequeue();
}
/**
* @brief Dequeues the next message from the mailbox that's matched
* by @p rules and executes the corresponding callback.
*/
inline void receive(invoke_rules& rules)
{
self()->mailbox().dequeue(rules);
......@@ -112,6 +143,10 @@ inline bool try_receive(invoke_rules& rules)
return self()->mailbox().try_dequeue(rules);
}
/**
* @brief Reads the last dequeued message from the mailbox.
* @return The last dequeued message from the mailbox.
*/
inline const message& last_received()
{
return self()->mailbox().last_dequeued();
......@@ -182,6 +217,16 @@ inline void await_all_others_done()
get_scheduler()->await_others_done();
}
/**
* @brief Publishes @p whom at given @p port.
*/
void publish(const actor_ptr& whom, std::uint16_t port);
/**
* @brief Establish a new connection to the actor at @p host on given @p port.
*/
actor_ptr remote_actor(const char* host, std::uint16_t port);
} // namespace cppa
#endif // CPPA_HPP
......@@ -18,11 +18,16 @@ class blocking_message_queue : public message_queue
queue_node(const message& from) : next(0), msg(from) { }
};
bool m_trap_exit;
message m_last_dequeued;
util::single_reader_queue<queue_node> m_queue;
public:
blocking_message_queue();
virtual void trap_exit(bool new_value);
virtual void enqueue /*[[override]]*/ (const message& msg);
virtual const message& dequeue /*[[override]]*/ ();
......
......@@ -11,13 +11,20 @@ enum class exit_reason : std::uint32_t
/**
* @brief Indicates that an actor finished execution.
*/
normal = 0x00,
normal = 0x00000,
/**
* @brief Indicates that an actor finished execution
* because of an unhandled exception.
*/
unhandled_exception = 0x01,
unhandled_exception = 0x00001,
/**
* @brief Indicates that an actor finishied execution
* because a connection to a remote link was
* closed unexpectedly.
*/
remote_link_unreachable = 0x00101,
/**
* @brief Any user defined exit reason should have a
......
......@@ -13,6 +13,7 @@ class message_queue : public ref_counted
public:
virtual void trap_exit(bool) = 0;
virtual void enqueue(const message&) = 0;
virtual const message& dequeue() = 0;
virtual void dequeue(invoke_rules&) = 0;
......
......@@ -4,6 +4,7 @@
#include <string>
#include <cstdint>
#include "cppa/ref_counted.hpp"
#include "cppa/util/comparable.hpp"
namespace cppa {
......@@ -11,9 +12,20 @@ namespace cppa {
/**
* @brief Identifies a process.
*/
struct process_information : util::comparable<process_information>
class process_information : public ref_counted,
util::comparable<process_information>
{
public:
static constexpr size_t node_id_size = 20;
process_information();
process_information(const process_information& other);
process_information& operator=(const process_information& other);
/**
* @brief Identifies the running process.
*/
......@@ -25,7 +37,7 @@ struct process_information : util::comparable<process_information>
* A hash build from the MAC address of the first network device
* and the serial number from the root HD (mounted in "/" or "C:").
*/
std::uint8_t node_id[20];
std::uint8_t node_id[node_id_size];
/**
* @brief Converts {@link node_id} to an hexadecimal string.
......
......@@ -41,4 +41,11 @@ intrusive_ptr<actor> actor::by_id(std::uint32_t actor_id)
return nullptr;
}
const process_information& actor::parent_process() const
{
// default implementation assumes that the actor belongs to
// the running process
return process_information::get();
}
} // namespace cppa
#include <sstream>
#include "cppa/actor_exited.hpp"
namespace cppa {
actor_exited::~actor_exited() throw()
{
}
actor_exited::actor_exited(std::uint32_t exit_reason) : m_reason(exit_reason)
{
std::ostringstream oss;
oss << "actor exited with reason " << exit_reason;
m_what = oss.str();
}
std::uint32_t actor_exited::reason() const
{
return m_reason;
}
const char* actor_exited::what() const throw()
{
return m_what.c_str();
}
} // namespace cppa
......@@ -102,7 +102,7 @@ binary_serializer::~binary_serializer()
void binary_serializer::acquire(size_t num_bytes)
{
if (m_begin == nullptr)
if (!m_begin)
{
size_t new_size = chunk_size;
while (new_size <= num_bytes)
......@@ -192,4 +192,9 @@ const char* binary_serializer::data() const
return m_begin;
}
void binary_serializer::reset()
{
m_wr_pos = m_begin;
}
} // namespace cppa
#include <vector>
#include "cppa/match.hpp"
#include "cppa/context.hpp"
#include "cppa/exit_signal.hpp"
#include "cppa/invoke_rules.hpp"
#include "cppa/util/singly_linked_list.hpp"
......@@ -5,8 +10,48 @@
#include "cppa/detail/intermediate.hpp"
#include "cppa/detail/blocking_message_queue.hpp"
namespace {
enum throw_on_exit_result
{
not_an_exit_signal,
normal_exit_signal
};
throw_on_exit_result throw_on_exit(const cppa::message& msg)
{
std::vector<size_t> mappings;
if (cppa::match<cppa::exit_signal>(msg.content(), mappings))
{
cppa::tuple_view<cppa::exit_signal> tview(msg.content().vals(),
std::move(mappings));
auto reason = cppa::get<0>(tview).reason();
if (reason != static_cast<std::uint32_t>(cppa::exit_reason::normal))
{
// throws
cppa::self()->quit(reason);
}
else
{
return normal_exit_signal;
}
}
return not_an_exit_signal;
}
} // namespace <anonymous>
namespace cppa { namespace detail {
blocking_message_queue::blocking_message_queue() : m_trap_exit(false)
{
}
void blocking_message_queue::trap_exit(bool new_value)
{
m_trap_exit = new_value;
}
void blocking_message_queue::enqueue(const message& msg)
{
m_queue.push_back(new queue_node(msg));
......@@ -17,19 +62,50 @@ const message& blocking_message_queue::dequeue()
queue_node* msg = m_queue.pop();
m_last_dequeued = std::move(msg->msg);
delete msg;
if (!m_trap_exit)
{
if (throw_on_exit(m_last_dequeued) == normal_exit_signal)
{
// exit_reason::normal is ignored by default,
// dequeue next message
return dequeue();
}
}
return m_last_dequeued;
}
void blocking_message_queue::dequeue(invoke_rules& rules)
{
queue_node* amsg = m_queue.pop();
queue_node* amsg = nullptr;//= m_queue.pop();
util::singly_linked_list<queue_node> buffer;
intrusive_ptr<detail::intermediate> imd;
while (!(imd = rules.get_intermediate(amsg->msg.content())))
bool done = false;
do
{
buffer.push_back(amsg);
amsg = m_queue.pop();
if (!m_trap_exit)
{
if (throw_on_exit(amsg->msg) == normal_exit_signal)
{
// ignored by default
delete amsg;
amsg = nullptr;
}
}
if (amsg)
{
imd = rules.get_intermediate(amsg->msg.content());
if (imd)
{
done = true;
}
else
{
buffer.push_back(amsg);
}
}
}
while (!done);
m_last_dequeued = amsg->msg;
if (!buffer.empty()) m_queue.push_front(std::move(buffer));
imd->invoke();
......
#include <boost/thread.hpp>
#include "cppa/context.hpp"
#include "cppa/message.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/detail/converted_thread_context.hpp"
#include <boost/thread.hpp>
#include "cppa/scheduler.hpp"
namespace {
void cleanup_fun(cppa::context* what)
......@@ -28,6 +27,11 @@ void context::enqueue(const message& msg)
mailbox().enqueue(msg);
}
void context::trap_exit(bool new_value)
{
mailbox().trap_exit(new_value);
}
context* unchecked_self()
{
return m_this_context.get();
......
#include <algorithm>
#include "cppa/exit_signal.hpp"
#include "cppa/actor_exited.hpp"
#include "cppa/detail/converted_thread_context.hpp"
namespace {
......@@ -64,6 +65,7 @@ void converted_thread_context::quit(std::uint32_t reason)
aptr->enqueue(message(mself, aptr, exit_signal(reason)));
}
throw actor_exited(reason);
}
void converted_thread_context::link(intrusive_ptr<actor>& other)
......
......@@ -47,14 +47,10 @@ namespace cppa { namespace detail {
actor_ptr mock_scheduler::spawn(actor_behavior* ab, scheduling_hint)
{
if (ab)
{
++m_running_actors;
intrusive_ptr<context> ctx(new detail::converted_thread_context);
boost::thread(run_actor, ctx, ab).detach();
return ctx;
}
return nullptr;
}
void mock_scheduler::register_converted_context(context*)
......
......@@ -49,7 +49,7 @@ cppa::process_information compute_proc_info()
pclose(cmd);
erase_trailing_newline(first_mac_addr);
auto tmp = cppa::util::ripemd_160(first_mac_addr + hd_serial);
memcpy(result.node_id, tmp.data(), 20);
memcpy(result.node_id, tmp.data(), cppa::process_information::node_id_size);
return result;
}
......@@ -57,12 +57,36 @@ cppa::process_information compute_proc_info()
namespace cppa {
process_information::process_information() : process_id(0)
{
memset(node_id, 0, node_id_size);
}
process_information::process_information(const process_information& other)
: ref_counted(), process_id(other.process_id)
{
memcpy(node_id, other.node_id, node_id_size);
}
process_information&
process_information::operator=(const process_information& other)
{
// prevent self-assignment
if (this != &other)
{
process_id = other.process_id;
memcpy(node_id, other.node_id, node_id_size);
}
return *this;
}
std::string process_information::node_id_as_string() const
{
std::ostringstream oss;
oss << std::hex;
oss.fill('0');
for (size_t i = 0; i < 20; ++i)
for (size_t i = 0; i < node_id_size; ++i)
{
oss.width(2);
oss << static_cast<std::uint32_t>(node_id[i]);
......@@ -79,7 +103,8 @@ const process_information& process_information::get()
int process_information::compare(const process_information& other) const
{
int tmp = strncmp(reinterpret_cast<const char*>(node_id),
reinterpret_cast<const char*>(other.node_id), 20);
reinterpret_cast<const char*>(other.node_id),
node_id_size);
if (tmp == 0)
{
if (process_id < other.process_id) return -1;
......
#include "cppa/config.hpp"
#include <ios> // ios_base::failure
#include <list>
#include <memory>
#include <stdexcept>
#include <netdb.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <boost/thread.hpp>
#include "cppa/cppa.hpp"
#include "cppa/match.hpp"
#include "cppa/binary_serializer.hpp"
#include "cppa/binary_deserializer.hpp"
#include "cppa/util/single_reader_queue.hpp"
namespace cppa {
namespace {
#ifdef ACEDIA_WINDOWS
typedef SOCKET native_socket_t;
typedef const char* socket_send_ptr;
typedef char* socket_recv_ptr;
#else
typedef int native_socket_t;
typedef const void* socket_send_ptr;
typedef void* socket_recv_ptr;
inline void closesocket(native_socket_t s) { close(s); }
#endif
typedef intrusive_ptr<process_information> pinfo_ptr;
class actor_proxy : public actor
{
pinfo_ptr m_parent;
public:
actor_proxy(const pinfo_ptr& parent) : m_parent(parent)
{
if (!m_parent) throw std::runtime_error("parent == nullptr");
}
actor_proxy(pinfo_ptr&& parent) : m_parent(std::move(parent))
{
if (!m_parent) throw std::runtime_error("parent == nullptr");
}
void enqueue(const message& msg);
void join(group_ptr& what);
void leave(const group_ptr& what);
void link(intrusive_ptr<actor>& other);
void unlink(intrusive_ptr<actor>& other);
bool remove_backlink(const intrusive_ptr<actor>& to);
bool establish_backlink(const intrusive_ptr<actor>& to);
const process_information& parent_process() const
{
return *m_parent;
}
};
typedef intrusive_ptr<actor_proxy> actor_proxy_ptr;
struct mailman_job
{
mailman_job* next;
actor_proxy_ptr client;
message original_message;
mailman_job(actor_proxy_ptr apptr, message omsg)
: next(0), client(apptr), original_message(omsg)
{
}
};
util::single_reader_queue<mailman_job> s_mailman_queue;
void actor_proxy::enqueue(const message& msg)
{
s_mailman_queue.push_back(new mailman_job(this, msg));
}
void actor_proxy::join(group_ptr& what)
{
}
void actor_proxy::leave(const group_ptr& what)
{
}
void actor_proxy::link(intrusive_ptr<actor>& other)
{
}
void actor_proxy::unlink(intrusive_ptr<actor>& other)
{
}
bool actor_proxy::remove_backlink(const intrusive_ptr<actor>& to)
{
}
bool actor_proxy::establish_backlink(const intrusive_ptr<actor>& to)
{
}
// a map that manages links between local actors and remote actors (proxies)
typedef std::map<actor_ptr, std::list<actor_proxy_ptr> > link_map;
void fake_exits_from_disconnected_links(link_map& links)
{
for (auto& element : links)
{
auto local_actor = element.first;
auto& remote_actors = element.second;
for (auto& rem_actor : remote_actors)
{
message msg(rem_actor, local_actor,
exit_signal(exit_reason::remote_link_unreachable));
local_actor->enqueue(msg);
}
}
}
void add_link(link_map& links,
const actor_ptr& local_actor,
const actor_proxy_ptr& remote_actor)
{
if (local_actor && remote_actor)
{
links[local_actor].push_back(remote_actor);
}
}
void remove_link(link_map& links,
const actor_ptr& local_actor,
const actor_proxy_ptr& remote_actor)
{
if (local_actor && remote_actor)
{
auto& link_list = links[local_actor];
link_list.remove(remote_actor);
if (link_list.empty()) links.erase(local_actor);
}
}
// handles *all* outgoing messages
void mailman_loop()
{
link_map links;
int send_flags = 0;
binary_serializer bs;
mailman_job* job = nullptr;
const auto& pself = process_information::get();
std::map<process_information, native_socket_t> peers;
for (;;)
{
job = s_mailman_queue.pop();
// keep track about link states of local actors
// (remove link states between local and remote actors if needed)
if (match<exit_signal>(job->original_message.content()))
{
auto sender = job->original_message.sender();
if (pself == sender->parent_process())
{
// an exit message from a local actor to a remote actor
job->client->unlink(sender);
}
else
{
// an exit message from a remote actor to another remote actor
}
}
// forward message to receiver peer
auto peer_element = peers.find(job->client->parent_process());
if (peer_element != peers.end())
{
auto peer = peer_element->second;
try
{
bs << job->original_message;
auto size32 = static_cast<std::uint32_t>(bs.size());
auto sent = ::send(peer, sizeof(size32), &size32, send_flags);
if (sent != -1)
{
sent = ::send(peer, bs.data(), bs.size(), send_flags);
}
if (sent == -1)
{
// peer unreachable
peers.erase(job->client->parent_process());
}
}
catch (...)
{
// TODO: some kind of error handling?
}
bs.reset();
}
}
}
void read_from_socket(native_socket_t sfd, void* buf, size_t buf_size)
{
char* cbuf = reinterpret_cast<char*>(buf);
size_t read_bytes = 0;
size_t left = buf_size;
int rres = 0;
size_t urres = 0;
do
{
rres = ::recv(sfd, cbuf + read_bytes, left, 0);
if (rres <= 0)
{
throw std::ios_base::failure("cannot read from closed socket");
}
urres = static_cast<size_t>(rres);
read_bytes += urres;
left -= urres;
}
while (urres < left);
}
// handles *one* socket
void post_office_loop(native_socket_t socket_fd) // socket file descriptor
{
auto meta_msg = uniform_typeid<message>();
message msg;
int read_flags = 0;
std::uint32_t rsize;
char* buf = nullptr;
size_t buf_size = 0;
size_t buf_allocated = 0;
try
{
for (;;)
{
read_from_socket(socket_fd, &rsize, sizeof(rsize));
if (buf_allocated < rsize)
{
// release old memory
delete[] buf;
// always allocate 1KB chunks
buf_allocated = 1024;
while (buf_allocated <= rsize)
{
buf_allocated += 1024;
}
buf = new char[buf_allocated];
}
buf_size = rsize;
read_from_socket(socket_fd, buf, buf_size);
binary_deserializer bd(buf, buf_size);
uniform_type_info* meta_msg;
meta_msg->deserialize(&msg, &bd);
auto r = msg.receiver();
if (r) r->enqueue(msg);
}
}
catch (std::ios_base::failure& e)
{
}
catch (std::exception& e)
{
}
}
void middle_man_loop(native_socket_t server_socket_fd)
{
sockaddr addr;
socklen_t addrlen;
std::vector<boost::thread> children;
try
{
for (;;)
{
auto sockfd = accept(server_socket_fd, &addr, &addrlen);
if (sockfd < 0)
{
throw std::ios_base::failure("invalid socket accepted");
}
// todo: check if connected peer is compatible
children.push_back(boost::thread(post_office_loop, sockfd));
}
}
catch (...)
{
}
for (auto& child : children)
{
child.join();
}
}
} // namespace <anonmyous>
void publish(const actor_ptr& whom, std::uint16_t port)
{
native_socket_t sockfd;
struct sockaddr_in serv_addr;
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0)
{
throw std::ios_base::failure("could not create server socket");
}
memset((char*) &serv_addr, 0, sizeof(serv_addr));
serv_addr.sin_family = AF_INET;
serv_addr.sin_addr.s_addr = INADDR_ANY;
serv_addr.sin_port = htons(port);
if (bind(sockfd, (struct sockaddr*) &serv_addr, sizeof(serv_addr)) < 0)
{
throw std::ios_base::failure("could not bind socket to port");
}
if (listen(sockfd, 10) != 0)
{
throw std::ios_base::failure("listen() failed");
}
boost::thread(middle_man_loop, sockfd).detach();
}
actor_ptr remote_actor(const char* host, std::uint16_t port)
{
}
} // namespace cppa
#define CPPA_VERBOSE_CHECK
#include <iostream>
#include <functional>
......@@ -28,8 +26,8 @@ void pong(actor_ptr ping_actor)
on<std::int32_t>() >> [](int v) { reply(v+1); });
// loop
while (!done) receive(rules);
// terminate with non-normal exit reason to
// force ping actor to quit
// terminate with non-normal exit reason
// to force ping actor to quit
quit(user_defined_exit_reason);
}
......
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