Commit e2675619 authored by Dominik Charousset's avatar Dominik Charousset

revised middleman layer

this patch removes the possibility to add more than one protocol to a
middleman, because this would most likely break any distributed actor system;
instead, this patch prepares libcppa for allowing user-defined middlemans;
furthermore, this patch establishes a more straightforward parent-child relation
between the middleman and its protocol and added the buffered_writer class
to ease development of asynchronous IO classes
parent e57a183c
......@@ -108,6 +108,7 @@ set(LIBCPPA_SRC
src/binary_deserializer.cpp
src/binary_serializer.cpp
src/buffer.cpp
src/buffered_writer.cpp
src/channel.cpp
src/context_switching_actor.cpp
src/continuable_io.cpp
......
......@@ -91,6 +91,7 @@ cppa/message_header.hpp
cppa/message_id.hpp
cppa/message_priority.hpp
cppa/network/acceptor.hpp
cppa/network/buffered_writer.hpp
cppa/network/continuable_io.hpp
cppa/network/default_actor_addressing.hpp
cppa/network/default_actor_proxy.hpp
......@@ -211,6 +212,7 @@ src/behavior_stack.cpp
src/binary_deserializer.cpp
src/binary_serializer.cpp
src/buffer.cpp
src/buffered_writer.cpp
src/channel.cpp
src/context_switching_actor.cpp
src/continuable_io.cpp
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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 BUFFERED_WRITER_HPP
#define BUFFERED_WRITER_HPP
#include "cppa/util/buffer.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/network/continuable_io.hpp"
namespace cppa { namespace network {
class middleman;
class buffered_writer : public continuable_io {
typedef continuable_io super;
public:
buffered_writer(middleman* parent,
native_socket_type read_fd,
output_stream_ptr out);
continue_writing_result continue_writing() override;
inline bool has_unwritten_data() const {
return m_has_unwritten_data;
}
protected:
void write(size_t num_bytes, const void* data);
void register_for_writing();
inline util::buffer& write_buffer() {
return m_buf;
}
private:
middleman* m_middleman;
output_stream_ptr m_out;
bool m_has_unwritten_data;
util::buffer m_buf;
};
} } // namespace cppa::network
#endif // BUFFERED_WRITER_HPP
......@@ -74,7 +74,7 @@ class default_actor_proxy : public actor_proxy {
default_actor_proxy(actor_id mid,
const process_information_ptr& pinfo,
const default_protocol_ptr& parent);
default_protocol* parent);
void enqueue(const message_header& hdr, any_tuple msg) override;
......@@ -104,7 +104,7 @@ class default_actor_proxy : public actor_proxy {
void forward_msg(const message_header& hdr, any_tuple msg);
default_protocol_ptr m_proto;
default_protocol* m_parent;
process_information_ptr m_pinf;
intrusive::single_reader_queue<sync_request_info, detail::disposer> m_pending_requests;
......
......@@ -43,16 +43,16 @@
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/network/continuable_io.hpp"
#include "cppa/network/buffered_writer.hpp"
#include "cppa/network/default_message_queue.hpp"
namespace cppa { namespace network {
class default_protocol;
class default_peer : public continuable_io {
class default_peer : public buffered_writer {
typedef continuable_io super;
typedef buffered_writer super;
friend class default_protocol;
......@@ -67,7 +67,7 @@ class default_peer : public continuable_io {
continue_writing_result continue_writing() override;
void io_failed();
void io_failed() override;
void enqueue(const message_header& hdr, const any_tuple& msg);
......@@ -79,18 +79,8 @@ class default_peer : public continuable_io {
return *m_node;
}
inline bool has_unwritten_data() const {
return m_has_unwritten_data;
}
protected:
~default_peer();
private:
void disconnected();
enum read_state {
// connection just established; waiting for process information
wait_for_process_info,
......@@ -102,10 +92,8 @@ class default_peer : public continuable_io {
default_protocol* m_parent;
input_stream_ptr m_in;
output_stream_ptr m_out;
read_state m_state;
process_information_ptr m_node;
bool m_has_unwritten_data;
const uniform_type_info* m_meta_hdr;
const uniform_type_info* m_meta_msg;
......@@ -143,6 +131,8 @@ class default_peer : public continuable_io {
enqueue({nullptr, nullptr}, msg);
}
void enqueue_impl(const message_header& hdr, const any_tuple& msg);
};
typedef intrusive_ptr<default_peer> default_peer_ptr;
......
......@@ -38,6 +38,7 @@
#include "cppa/process_information.hpp"
#include "cppa/network/protocol.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/default_peer.hpp"
#include "cppa/network/default_peer_acceptor.hpp"
#include "cppa/network/default_message_queue.hpp"
......@@ -51,7 +52,7 @@ class default_protocol : public protocol {
public:
default_protocol(abstract_middleman* parent);
default_protocol(middleman* multiplexer);
atom_value identifier() const;
......@@ -99,8 +100,6 @@ class default_protocol : public protocol {
};
typedef intrusive_ptr<default_protocol> default_protocol_ptr;
} } // namespace cppa::network
#endif // DEFAULT_PROTOCOL_HPP
......@@ -43,13 +43,12 @@ namespace cppa { namespace detail { class singleton_manager; } }
namespace cppa { namespace network {
class protocol;
typedef intrusive_ptr<protocol> protocol_ptr;
class middleman_impl;
/**
* @brief Multiplexes asynchronous IO.
*/
class middleman : public ref_counted {
class middleman {
friend class detail::singleton_manager;
......@@ -58,55 +57,60 @@ class middleman : public ref_counted {
virtual ~middleman();
/**
* @brief Add a new communication protocol to the middleman.
* @brief Returns the networking protocol in use.
*/
virtual void add_protocol(const protocol_ptr& impl) = 0;
protocol* get_protocol();
/**
* @brief Returns the protocol associated with @p id.
* @brief Runs @p fun in the middleman's event loop.
*/
virtual protocol_ptr protocol(atom_value id) = 0;
void run_later(std::function<void()> fun);
/**
* @brief Runs @p fun in the middleman's event loop.
* @brief Removes @p ptr from the list of active writers.
* @warning This member function is not thread-safe.
*/
virtual void run_later(std::function<void()> fun) = 0;
protected:
virtual void destroy() = 0;
virtual void initialize() = 0;
private:
static middleman* create_singleton();
inline void dispose() { delete this; }
void stop_writer(const continuable_io_ptr& ptr);
};
/**
* @brief Adds @p ptr to the list of active writers.
* @warning This member function is not thread-safe.
*/
void continue_writer(const continuable_io_ptr& ptr);
class middleman_event_handler;
/**
* @brief Removes @p ptr from the list of active readers.
* @warning This member function is not thread-safe.
*/
void stop_reader(const continuable_io_ptr& ptr);
class abstract_middleman : public middleman {
/**
* @brief Adds @p ptr to the list of active readers.
* @warning This member function is not thread-safe.
*/
void continue_reader(const continuable_io_ptr& ptr);
public:
protected:
inline abstract_middleman() : m_done(false) { }
// destroys singleton
void destroy();
void stop_writer(const continuable_io_ptr& ptr);
void continue_writer(const continuable_io_ptr& ptr);
// initializes singletons
void initialize();
void stop_reader(const continuable_io_ptr& what);
void continue_reader(const continuable_io_ptr& what);
private:
protected:
// sets m_impl and binds implementation to given protocol
void set_pimpl(std::unique_ptr<protocol>&&);
inline void quit() { m_done = true; }
inline bool done() const { return m_done; }
// creates a middleman using network::default_protocol
static middleman* create_singleton();
bool m_done;
std::vector<continuable_io_ptr> m_readers;
// destroys uninitialized instances
inline void dispose() { delete this; }
middleman_event_handler& handler();
// pointer to implementation
std::unique_ptr<middleman_impl> m_impl;
};
......
......@@ -53,15 +53,13 @@ class abstract_middleman;
/**
* @brief Implements a communication protocol.
*/
class protocol : public ref_counted {
typedef ref_counted super;
class protocol {
public:
typedef std::initializer_list<primitive_variant> variant_args;
protocol(abstract_middleman* parent);
protocol(middleman* parent);
virtual atom_value identifier() const = 0;
......@@ -80,33 +78,78 @@ class protocol : public ref_counted {
virtual actor_addressing* addressing() = 0;
void run_later(std::function<void()> fun);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
template<typename T>
inline void run_later(T&& what);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void stop_writer(const continuable_io_ptr& ptr);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void continue_writer(const continuable_io_ptr& ptr);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void stop_reader(const continuable_io_ptr& ptr);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void continue_reader(const continuable_io_ptr& ptr);
/**
* @brief Returns the parent of this protocol instance.
*/
inline middleman* parent();
protected:
private:
// note: not thread-safe; call only in run_later functor!
void continue_reader(continuable_io* what);
middleman* m_parent;
// note: not thread-safe; call only in run_later functor!
void continue_writer(continuable_io* what);
};
// note: not thread-safe; call only in run_later functor!
void stop_reader(continuable_io* what);
typedef intrusive_ptr<protocol> protocol_ptr;
// note: not thread-safe; call only in run_later functor!
void stop_writer(continuable_io* what);
/******************************************************************************
* inline and template member function implementations *
******************************************************************************/
inline abstract_middleman* parent() { return m_parent.get(); }
inline middleman* protocol::parent() {
return m_parent;
}
inline const abstract_middleman* parent() const { return m_parent.get(); }
template<typename T>
inline void protocol::run_later(T&& what) {
m_parent->run_later(std::forward<T>(what));
}
private:
inline void protocol::stop_writer(const continuable_io_ptr& ptr) {
m_parent->stop_writer(ptr);
}
intrusive_ptr<abstract_middleman> m_parent;
inline void protocol::continue_writer(const continuable_io_ptr& ptr) {
m_parent->continue_writer(ptr);
}
};
inline void protocol::stop_reader(const continuable_io_ptr& ptr) {
m_parent->stop_reader(ptr);
}
typedef intrusive_ptr<protocol> protocol_ptr;
inline void protocol::continue_reader(const continuable_io_ptr& ptr) {
m_parent->continue_reader(ptr);
}
} } // namespace cppa::network
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* 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 2.1 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 "cppa/logging.hpp"
#include "cppa/to_string.hpp"
#include "cppa/singletons.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/buffered_writer.hpp"
namespace cppa { namespace network {
buffered_writer::buffered_writer(middleman* pptr, native_socket_type rfd, output_stream_ptr out)
: super(rfd, out->write_handle()), m_middleman(pptr)
, m_out(out), m_has_unwritten_data(false) { }
continue_writing_result buffered_writer::continue_writing() {
CPPA_LOG_TRACE("");
CPPA_LOG_DEBUG_IF(!m_has_unwritten_data, "nothing to write (done)");
while (m_has_unwritten_data) {
size_t written;
try { written = m_out->write_some(m_buf.data(), m_buf.size()); }
catch (std::exception& e) {
CPPA_LOG_ERROR(to_verbose_string(e));
static_cast<void>(e); // keep compiler happy
return write_failure;
}
if (written != m_buf.size()) {
CPPA_LOGMF(CPPA_DEBUG, self, "tried to write " << m_buf.size() << "bytes, "
<< "only " << written << " bytes written");
m_buf.erase_leading(written);
return write_continue_later;
}
else {
m_buf.clear();
m_has_unwritten_data = false;
CPPA_LOGMF(CPPA_DEBUG, self, "write done, " << written << "bytes written");
}
}
return write_done;
}
void buffered_writer::write(size_t num_bytes, const void* data) {
m_buf.write(num_bytes, data);
register_for_writing();
}
void buffered_writer::register_for_writing() {
if (!m_has_unwritten_data) {
CPPA_LOGMF(CPPA_DEBUG, self, "register for writing");
m_has_unwritten_data = true;
m_middleman->continue_writer(this);
}
}
} } // namespace cppa::network
......@@ -51,8 +51,8 @@ sync_request_info::sync_request_info(actor_ptr sptr, message_id id)
default_actor_proxy::default_actor_proxy(actor_id mid,
const process_information_ptr& pinfo,
const default_protocol_ptr& parent)
: super(mid), m_proto(parent), m_pinf(pinfo) {
default_protocol* parent)
: super(mid), m_parent(parent), m_pinf(pinfo) {
CPPA_REQUIRE(parent != nullptr);
CPPA_LOG_INFO(CPPA_ARG(mid) << ", " << CPPA_TARG(pinfo, to_string)
<< "protocol = " << detail::demangle(typeid(*parent)));
......@@ -61,9 +61,9 @@ default_actor_proxy::default_actor_proxy(actor_id mid,
default_actor_proxy::~default_actor_proxy() {
auto aid = m_id;
auto node = m_pinf;
auto proto = m_proto;
auto proto = m_parent;
CPPA_LOG_INFO(CPPA_ARG(m_id) << ", " << CPPA_TSARG(m_pinf)
<< ", protocol = " << detail::demangle(typeid(*m_proto)));
<< ", protocol = " << detail::demangle(typeid(*m_parent)));
proto->run_later([aid, node, proto] {
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
"~default_actor_proxy$run_later",
......@@ -106,7 +106,7 @@ void default_actor_proxy::forward_msg(const message_header& hdr, any_tuple msg)
switch (m_pending_requests.enqueue(new_req_info(hdr.sender, hdr.id))) {
case intrusive::queue_closed: {
auto rsn = exit_reason();
m_proto->run_later([rsn, hdr] {
m_parent->run_later([rsn, hdr] {
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
"forward_msg$bouncer",
"bounce message for reason " << rsn);
......@@ -119,8 +119,8 @@ void default_actor_proxy::forward_msg(const message_header& hdr, any_tuple msg)
}
}
auto node = m_pinf;
auto proto = m_proto;
m_proto->run_later([hdr, msg, node, proto] {
auto proto = m_parent;
m_parent->run_later([hdr, msg, node, proto] {
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
"forward_msg$forwarder",
"");
......@@ -138,7 +138,7 @@ void default_actor_proxy::enqueue(const message_header& hdr, any_tuple msg) {
CPPA_LOG_DEBUG("received KILL_PROXY message");
intrusive_ptr<default_actor_proxy> _this{this};
auto reason = msg.get_as<uint32_t>(1);
m_proto->run_later([_this, reason] {
m_parent->run_later([_this, reason] {
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
"enqueue$kill_proxy_helper",
"KILL_PROXY with exit reason " << reason);
......
......@@ -60,14 +60,13 @@ default_peer::default_peer(default_protocol* parent,
const input_stream_ptr& in,
const output_stream_ptr& out,
process_information_ptr peer_ptr)
: super(in->read_handle(), out->write_handle())
, m_parent(parent), m_in(in), m_out(out)
: super(parent->parent(), in->read_handle(), out)
, m_parent(parent), m_in(in)
, m_state((peer_ptr) ? wait_for_msg_size : wait_for_process_info)
, m_node(peer_ptr)
, m_has_unwritten_data(false) {
m_rd_buf.reset(m_state == wait_for_process_info
? sizeof(uint32_t) + process_information::node_id_size
: sizeof(uint32_t));
, m_node(peer_ptr) {
m_rd_buf.final_size( m_state == wait_for_process_info
? sizeof(uint32_t) + process_information::node_id_size
: sizeof(uint32_t));
// state == wait_for_msg_size iff peer was created using remote_peer()
// in this case, this peer must be erased if no proxy of it remains
m_erase_on_last_proxy_exited = m_state == wait_for_msg_size;
......@@ -75,11 +74,7 @@ default_peer::default_peer(default_protocol* parent,
m_meta_msg = uniform_typeid<any_tuple>();
}
default_peer::~default_peer() {
disconnected();
}
void default_peer::disconnected() {
void default_peer::io_failed() {
CPPA_LOG_TRACE("node = " << (m_node ? to_string(*m_node) : "nullptr"));
if (m_node) {
// kill all proxies
......@@ -94,17 +89,11 @@ void default_peer::disconnected() {
}
}
void default_peer::io_failed() {
CPPA_LOG_TRACE("");
disconnected();
}
continue_reading_result default_peer::continue_reading() {
CPPA_LOG_TRACE("");
for (;;) {
try { m_rd_buf.append_from(m_in.get()); }
catch (exception&) {
disconnected();
return read_failure;
}
if (!m_rd_buf.full()) return read_continue_later; // try again later
......@@ -115,7 +104,7 @@ continue_reading_result default_peer::continue_reading() {
uint32_t process_id;
process_information::node_id_type node_id;
memcpy(&process_id, m_rd_buf.data(), sizeof(uint32_t));
memcpy(node_id.data(), m_rd_buf.data() + sizeof(uint32_t),
memcpy(node_id.data(), m_rd_buf.offset_data(sizeof(uint32_t)),
process_information::node_id_size);
m_node.reset(new process_information(process_id, node_id));
if (*process_information::get() == *m_node) {
......@@ -128,14 +117,16 @@ continue_reading_result default_peer::continue_reading() {
m_parent->register_peer(*m_node, this);
// initialization done
m_state = wait_for_msg_size;
m_rd_buf.reset(sizeof(uint32_t));
m_rd_buf.clear();
m_rd_buf.final_size(sizeof(uint32_t));
break;
}
case wait_for_msg_size: {
//DEBUG("peer_connection::continue_reading: wait_for_msg_size");
uint32_t msg_size;
memcpy(&msg_size, m_rd_buf.data(), sizeof(uint32_t));
m_rd_buf.reset(msg_size);
m_rd_buf.clear();
m_rd_buf.final_size(msg_size);
m_state = read_message;
break;
}
......@@ -177,7 +168,8 @@ continue_reading_result default_peer::continue_reading() {
deliver(hdr, move(msg));
}
);
m_rd_buf.reset(sizeof(uint32_t));
m_rd_buf.clear();
m_rd_buf.final_size(sizeof(uint32_t));
m_state = wait_for_msg_size;
break;
}
......@@ -220,7 +212,7 @@ void default_peer::monitor(const actor_ptr&,
}
else {
CPPA_LOGMF(CPPA_DEBUG, self, "attach functor to " << entry.first.get());
default_protocol_ptr proto = m_parent;
default_protocol* proto = m_parent;
entry.first->attach_functor([=](uint32_t reason) {
proto->run_later([=] {
CPPA_LOGC_TRACE("cppa::network::default_peer",
......@@ -308,47 +300,27 @@ void default_peer::unlink(const actor_ptr& sender, const actor_ptr& ptr) {
continue_writing_result default_peer::continue_writing() {
CPPA_LOG_TRACE("");
CPPA_LOG_DEBUG_IF(!m_has_unwritten_data, "nothing to write (done)");
while (m_has_unwritten_data) {
size_t written;
try { written = m_out->write_some(m_wr_buf.data(), m_wr_buf.size()); }
catch (exception& e) {
CPPA_LOGMF(CPPA_ERROR, self, to_verbose_string(e));
static_cast<void>(e); // keep compiler happy
disconnected();
return write_failure;
}
if (written != m_wr_buf.size()) {
CPPA_LOGMF(CPPA_DEBUG, self, "tried to write " << m_wr_buf.size() << "bytes, "
<< "only " << written << " bytes written");
m_wr_buf.erase_leading(written);
return write_continue_later;
}
else {
m_wr_buf.reset();
m_has_unwritten_data = false;
CPPA_LOGMF(CPPA_DEBUG, self, "write done, " << written << "bytes written");
}
// try to write next message in queue
while (!m_has_unwritten_data && !queue().empty()) {
auto tmp = queue().pop();
enqueue(tmp.first, tmp.second);
}
auto result = super::continue_writing();
while (result == write_done && !queue().empty()) {
auto tmp = queue().pop();
enqueue(tmp.first, tmp.second);
result = super::continue_writing();
}
if (erase_on_last_proxy_exited() && !has_unwritten_data()) {
if (result == write_done && erase_on_last_proxy_exited() && !has_unwritten_data()) {
if (m_parent->addressing()->count_proxies(*m_node) == 0) {
m_parent->last_proxy_exited(this);
}
}
return write_done;
return result;
}
void default_peer::enqueue(const message_header& hdr, const any_tuple& msg) {
void default_peer::enqueue_impl(const message_header& hdr, const any_tuple& msg) {
CPPA_LOG_TRACE("");
binary_serializer bs(&m_wr_buf, m_parent->addressing());
uint32_t size = 0;
auto before = m_wr_buf.size();
m_wr_buf.write(sizeof(uint32_t), &size, util::grow_if_needed);
auto& wbuf = write_buffer();
auto before = wbuf.size();
binary_serializer bs(&wbuf, m_parent->addressing());
wbuf.write(sizeof(uint32_t), &size);
try { bs << hdr << msg; }
catch (exception& e) {
CPPA_LOGMF(CPPA_ERROR, self, to_verbose_string(e));
......@@ -358,15 +330,14 @@ void default_peer::enqueue(const message_header& hdr, const any_tuple& msg) {
return;
}
CPPA_LOGMF(CPPA_DEBUG, self, "serialized: " << to_string(hdr) << " " << to_string(msg));
size = (m_wr_buf.size() - before) - sizeof(std::uint32_t);
size = (wbuf.size() - before) - sizeof(std::uint32_t);
// update size in buffer
memcpy(m_wr_buf.data() + before, &size, sizeof(std::uint32_t));
CPPA_LOG_DEBUG_IF(m_has_unwritten_data, "still registered for writing");
if (!m_has_unwritten_data) {
CPPA_LOGMF(CPPA_DEBUG, self, "register for writing");
m_has_unwritten_data = true;
m_parent->continue_writer(this);
}
memcpy(wbuf.offset_data(before), &size, sizeof(std::uint32_t));
}
void default_peer::enqueue(const message_header& hdr, const any_tuple& msg) {
enqueue_impl(hdr, msg);
register_for_writing();
}
} } // namespace cppa::network
......@@ -53,7 +53,7 @@ using namespace cppa::detail;
namespace cppa { namespace network {
default_protocol::default_protocol(abstract_middleman* parent)
default_protocol::default_protocol(middleman* parent)
: super(parent), m_addressing(this) { }
atom_value default_protocol::identifier() const {
......@@ -90,7 +90,7 @@ void default_protocol::publish(const actor_ptr& whom,
CPPA_REQUIRE(args.size() == 0);
static_cast<void>(args); // keep compiler happy
get_actor_registry()->put(whom->id(), whom);
default_protocol_ptr proto = this;
default_protocol* proto = this;
auto impl = make_counted<default_peer_acceptor>(this, move(ptr), whom);
run_later([=] {
CPPA_LOGC_TRACE("cppa::network::default_protocol",
......@@ -102,7 +102,7 @@ void default_protocol::publish(const actor_ptr& whom,
void default_protocol::unpublish(const actor_ptr& whom) {
CPPA_LOG_TRACE("whom = " << to_string(whom));
default_protocol_ptr proto = this;
default_protocol* proto = this;
run_later([=] {
CPPA_LOGC_TRACE("cppa::network::default_protocol",
"unpublish$remove_acceptors", "");
......@@ -199,7 +199,7 @@ actor_ptr default_protocol::remote_actor(io_stream_ptr_pair io,
# endif
return get_actor_registry()->get(remote_aid);
}
default_protocol_ptr proto = this;
default_protocol* proto = this;
intrusive::blocking_single_reader_queue<remote_actor_result> q;
run_later([proto, io, pinfptr, remote_aid, &q] {
CPPA_LOGC_TRACE("cppa::network::default_protocol",
......
......@@ -51,6 +51,7 @@
#include "cppa/util/buffer.hpp"
#include "cppa/network/protocol.hpp"
#include "cppa/network/acceptor.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/input_stream.hpp"
......@@ -87,32 +88,19 @@ class middleman_event {
typedef intrusive::single_reader_queue<middleman_event> middleman_queue;
class default_middleman_impl : public abstract_middleman {
class middleman_impl {
friend class abstract_middleman;
friend void middleman_loop(default_middleman_impl*);
friend class middleman;
friend void middleman_loop(middleman_impl*);
public:
default_middleman_impl() : m_handler(middleman_event_handler::create()) {
m_protocols.insert(make_pair(atom("DEFAULT"),
new network::default_protocol(this)));
}
void add_protocol(const protocol_ptr& impl) {
if (!impl) {
CPPA_LOGMF(CPPA_ERROR, self, "impl == nullptr");
throw std::invalid_argument("impl == nullptr");
}
CPPA_LOG_TRACE("identifier = " << to_string(impl->identifier()));
std::lock_guard<util::shared_spinlock> guard(m_protocols_lock);
m_protocols.insert(make_pair(impl->identifier(), impl));
}
middleman_impl(std::unique_ptr<protocol>&& proto)
: m_handler(middleman_event_handler::create())
, m_protocol(std::move(proto)) { }
protocol_ptr protocol(atom_value id) {
util::shared_lock_guard<util::shared_spinlock> guard(m_protocols_lock);
auto i = m_protocols.find(id);
return (i != m_protocols.end()) ? i->second : nullptr;
protocol* get_protocol() {
return m_protocol.get();
}
void run_later(function<void()> fun) {
......@@ -123,6 +111,32 @@ class default_middleman_impl : public abstract_middleman {
static_cast<void>(write(m_pipe_write, &dummy, sizeof(dummy)));
}
void continue_writer(const continuable_io_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_handler->add_later(ptr, event::write);
}
void stop_writer(const continuable_io_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_handler->erase_later(ptr, event::write);
}
void continue_reader(const continuable_io_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_readers.push_back(ptr);
m_handler->add_later(ptr, event::read);
}
void stop_reader(const continuable_io_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_handler->erase_later(ptr, event::read);
auto last = m_readers.end();
auto i = find_if(m_readers.begin(), last, [&](const continuable_io_ptr& lhs) {
return lhs == ptr;
});
if (i != last) m_readers.erase(i);
}
protected:
void initialize() {
......@@ -133,8 +147,6 @@ class default_middleman_impl : public abstract_middleman {
detail::fd_util::nonblocking(m_pipe_read, true);
// start threads
m_thread = thread([this] { middleman_loop(this); });
// increase reference count for singleton manager
ref();
}
void destroy() {
......@@ -145,26 +157,37 @@ class default_middleman_impl : public abstract_middleman {
m_thread.join();
close(m_pipe_read);
close(m_pipe_write);
// decrease reference count for singleton manager
deref();
//delete this;
}
private:
inline void quit() { m_done = true; }
inline bool done() const { return m_done; }
bool m_done;
std::vector<continuable_io_ptr> m_readers;
middleman_event_handler& handler();
thread m_thread;
native_socket_type m_pipe_read;
native_socket_type m_pipe_write;
middleman_queue m_queue;
std::unique_ptr<middleman_event_handler> m_handler;
util::shared_spinlock m_protocols_lock;
map<atom_value, protocol_ptr> m_protocols;
std::unique_ptr<protocol> m_protocol;
};
void middleman::set_pimpl(std::unique_ptr<protocol>&& proto) {
m_impl.reset(new middleman_impl(std::move(proto)));
}
middleman* middleman::create_singleton() {
return new default_middleman_impl;
auto ptr = new middleman;
ptr->set_pimpl(std::unique_ptr<protocol>{new default_protocol(ptr)});
return ptr;
}
class middleman_overseer : public continuable_io {
......@@ -215,38 +238,39 @@ class middleman_overseer : public continuable_io {
middleman::~middleman() { }
middleman_event_handler& abstract_middleman::handler() {
return *(static_cast<default_middleman_impl*>(this)->m_handler);
void middleman::destroy() {
m_impl->destroy();
}
void middleman::initialize() {
m_impl->initialize();
}
void abstract_middleman::continue_writer(const continuable_io_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
handler().add_later(ptr, event::write);
protocol* middleman::get_protocol() {
return m_impl->get_protocol();
}
void abstract_middleman::stop_writer(const continuable_io_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
handler().erase_later(ptr, event::write);
void middleman::run_later(std::function<void()> fun) {
m_impl->run_later(std::move(fun));
}
void abstract_middleman::continue_reader(const continuable_io_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_readers.push_back(ptr);
handler().add_later(ptr, event::read);
void middleman::continue_writer(const continuable_io_ptr& ptr) {
m_impl->continue_writer(ptr);
}
void abstract_middleman::stop_reader(const continuable_io_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
handler().erase_later(ptr, event::read);
void middleman::stop_writer(const continuable_io_ptr& ptr) {
m_impl->stop_writer(ptr);
}
void middleman::continue_reader(const continuable_io_ptr& ptr) {
m_impl->continue_reader(ptr);
}
auto last = end(m_readers);
auto i = find_if(begin(m_readers), last, [ptr](const continuable_io_ptr& value) {
return value == ptr;
});
if (i != last) m_readers.erase(i);
void middleman::stop_reader(const continuable_io_ptr& ptr) {
m_impl->stop_reader(ptr);
}
void middleman_loop(default_middleman_impl* impl) {
void middleman_loop(middleman_impl* impl) {
# ifdef CPPA_LOG_LEVEL
auto mself = make_counted<thread_mapped_actor>();
scoped_self_setter sss(mself.get());
......
......@@ -34,33 +34,8 @@
namespace cppa { namespace network {
protocol::protocol(abstract_middleman* parent) : m_parent(parent) {
protocol::protocol(middleman* parent) : m_parent(parent) {
CPPA_REQUIRE(parent != nullptr);
}
void protocol::run_later(std::function<void()> fun) {
CPPA_REQUIRE(m_parent != nullptr);
m_parent->run_later(std::move(fun));
}
void protocol::continue_reader(continuable_io* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_parent->continue_reader(ptr);
}
void protocol::continue_writer(continuable_io* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_parent->continue_writer(ptr);
}
void protocol::stop_reader(continuable_io* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_parent->stop_reader(ptr);
}
void protocol::stop_writer(continuable_io* ptr) {
CPPA_LOG_TRACE(CPPA_ARG(ptr));
m_parent->stop_writer(ptr);
}
} } // namespace cppa::network
......@@ -61,8 +61,8 @@ namespace cppa {
using namespace detail;
using namespace network;
namespace { protocol_ptr proto() {
return get_middleman()->protocol(atom("DEFAULT")).get();
namespace { protocol* proto() {
return get_middleman()->get_protocol();
} }
void publish(actor_ptr whom, std::unique_ptr<acceptor> aptr) {
......
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