Commit 5d3cf75b authored by Dominik Charousset's avatar Dominik Charousset

refactored mem mgmnt in MM

this patch changes the memory management in the middleman, i.e.,
´continuable´ no longer reference counted but provides a `dispose`
member function; using reference counting to manage object lifetimes
in the MM is an overkill, because it is quite simple to decide when
an IO worker is done; this change allowed us to remove `broker_backend`,
because `broker` can now inherit both `local_actor` and `continuable`
without having two reference counts
parent 458f9b29
......@@ -108,7 +108,6 @@ set(LIBCPPA_SRC
src/binary_deserializer.cpp
src/binary_serializer.cpp
src/broker.cpp
src/broker_backend.cpp
src/buffer.cpp
src/buffered_writer.cpp
src/channel.cpp
......
......@@ -81,7 +81,6 @@ cppa/intrusive/single_reader_queue.hpp
cppa/intrusive_ptr.hpp
cppa/io/acceptor.hpp
cppa/io/broker.hpp
cppa/io/broker_backend.hpp
cppa/io/buffered_writer.hpp
cppa/io/continuable.hpp
cppa/io/default_actor_addressing.hpp
......@@ -218,7 +217,6 @@ src/behavior_stack.cpp
src/binary_deserializer.cpp
src/binary_serializer.cpp
src/broker.cpp
src/broker_backend.cpp
src/buffer.cpp
src/buffered_writer.cpp
src/channel.cpp
......
......@@ -73,7 +73,6 @@
#include "cppa/io/io_handle.hpp"
#include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp"
#include "cppa/io/broker_backend.hpp"
#include "cppa/detail/memory.hpp"
#include "cppa/detail/get_behavior.hpp"
......@@ -625,10 +624,12 @@ actor_ptr spawn_io(io::input_stream_ptr in,
Ts&&... args) {
using namespace io;
auto mm = get_middleman();
auto ptr = make_counted<Impl>(std::forward<Ts>(args)...);
auto backend = make_counted<broker_backend>(std::move(in), std::move(out), ptr);
backend->init();
mm->run_later([=] { mm->continue_reader(backend); });
auto ptr = make_counted<Impl>(std::move(in), std::move(out), std::forward<Ts>(args)...);
{
scoped_self_setter sss{ptr.get()};
ptr->init();
}
if (ptr->has_behavior()) mm->run_later([=] { mm->continue_reader(ptr.get()); });
return eval_sopts(Options, std::move(ptr));
}
......@@ -647,10 +648,12 @@ actor_ptr spawn_io(F fun,
Ts&&... args) {
using namespace io;
auto mm = get_middleman();
auto ptr = broker::from(std::move(fun), std::forward<Ts>(args)...);
auto backend = make_counted<broker_backend>(std::move(in), std::move(out), ptr);
backend->init();
mm->run_later([=] { mm->continue_reader(backend); });
auto ptr = broker::from(std::move(fun), std::move(in), std::move(out), std::forward<Ts>(args)...);
{
scoped_self_setter sss{ptr.get()};
ptr->init();
}
// default_broker_impl will call continue_reader()
return eval_sopts(Options, std::move(ptr));
}
......
......@@ -104,6 +104,10 @@ class receive_policy {
return false;
}
inline void add_to_cache(pointer node_ptr) {
m_cache.emplace_back(std::move(node_ptr));
}
template<class Client, class Fun>
bool invoke(Client* client,
pointer node_ptr,
......
......@@ -28,8 +28,8 @@
\******************************************************************************/
#ifndef IO_ACTOR_HPP
#define IO_ACTOR_HPP
#ifndef CPPA_BROKER_HPP
#define CPPA_BROKER_HPP
#include <functional>
......@@ -38,51 +38,81 @@
#include "cppa/local_actor.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/io/io_handle.hpp"
#include "cppa/io/buffered_writer.hpp"
#include "cppa/detail/fwd.hpp"
namespace cppa { namespace io {
class broker_backend;
class broker_continuation;
class broker : public extend<local_actor>::with<threadless, stackless> {
class broker : public extend<local_actor>::with<threadless, stackless>
, public buffered_writer {
typedef combined_type super;
typedef combined_type super1;
typedef buffered_writer super2;
friend class broker_backend;
friend class broker_continuation;
public:
enum policy_flag { at_least, at_most, exactly };
broker(input_stream_ptr in, output_stream_ptr out);
void io_failed() override;
void dispose() override;
void receive_policy(policy_flag policy, size_t buffer_size);
continue_reading_result continue_reading() override;
void write(size_t num_bytes, const void* data);
void enqueue(const message_header& hdr, any_tuple msg);
bool initialized() const;
void quit(std::uint32_t reason);
static intrusive_ptr<broker> from(std::function<void (io_handle*)> fun);
static intrusive_ptr<broker> from(std::function<void (broker*)> fun,
input_stream_ptr in,
output_stream_ptr out);
template<typename F, typename T0, typename... Ts>
static intrusive_ptr<broker> from(F fun, T0&& arg0, Ts&&... args) {
static intrusive_ptr<broker> from(F fun,
input_stream_ptr in,
output_stream_ptr out,
T0&& arg0,
Ts&&... args) {
return from(std::bind(std::move(fun),
std::placeholders::_1,
detail::fwd<T0>(arg0),
detail::fwd<Ts>(args)...));
detail::fwd<Ts>(args)...),
std::move(in),
std::move(out));
}
protected:
io_handle& io();
void cleanup(std::uint32_t reason);
private:
void invoke_message(mailbox_element* elem);
void invoke_message(const message_header& hdr, any_tuple msg);
void disconnect();
void invoke_message(any_tuple msg);
static constexpr size_t default_max_buffer_size = 65535;
intrusive_ptr<broker_backend> m_parent;
bool m_is_continue_reading;
bool m_disconnected;
bool m_dirty;
policy_flag m_policy;
size_t m_policy_buffer_size;
input_stream_ptr m_in;
cow_tuple<atom_value, uint32_t, util::buffer> m_read;
};
......@@ -90,4 +120,4 @@ typedef intrusive_ptr<broker> broker_ptr;
} } // namespace cppa::network
#endif // IO_ACTOR_HPP
#endif // CPPA_BROKER_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \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 IO_ACTOR_BACKEND_HPP
#define IO_ACTOR_BACKEND_HPP
#include <cstdint>
#include "cppa/io/broker.hpp"
#include "cppa/io/io_handle.hpp"
#include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp"
#include "cppa/io/buffered_writer.hpp"
namespace cppa { namespace io {
class broker_backend : public buffered_writer, public io_handle {
typedef buffered_writer super; // io_service is merely an interface
// 65k is the maximum TCP package size
static constexpr size_t default_max_buffer_size = 65535;
public:
broker_backend(input_stream_ptr in, output_stream_ptr out, broker_ptr ptr);
void init();
void handle_disconnect();
void io_failed() override;
void receive_policy(policy_flag policy, size_t buffer_size) override;
continue_reading_result continue_reading() override;
void close() override;
void write(size_t num_bytes, const void* data) override;
private:
bool m_dirty;
policy_flag m_policy;
size_t m_policy_buffer_size;
input_stream_ptr m_in;
intrusive_ptr<broker> m_self;
cow_tuple<atom_value, uint32_t, util::buffer> m_read;
};
} } // namespace cppa::network
#endif // IO_ACTOR_BACKEND_HPP
......@@ -61,21 +61,34 @@ enum continue_writing_result {
/**
* @brief An object performing asynchronous input and output.
*/
class continuable : public ref_counted {
class continuable {
continuable(const continuable&) = delete;
continuable& operator=(const continuable&) = delete;
public:
virtual ~continuable();
/**
* @brief Disposes this instance. This member function is invoked by
* the middleman once it has neither pending reads nor pending
* writes for this instance.
*
* This member function is expected to perform cleanup code
* and to release memory, e.g., by calling <tt>delete this</tt>.
*/
virtual void dispose() = 0;
/**
* @brief Returns the file descriptor for incoming data.
*/
inline native_socket_type read_handle() const { return m_rd; }
inline native_socket_type read_handle() const;
/**
* @brief Returns the file descriptor for outgoing data.
*/
inline native_socket_type write_handle() const {
return m_wr;
}
inline native_socket_type write_handle() const;
/**
* @brief Reads from {@link read_handle()} if valid.
......@@ -96,7 +109,7 @@ class continuable : public ref_counted {
protected:
continuable(native_socket_type read_fd,
native_socket_type write_fd = invalid_socket);
native_socket_type write_fd = invalid_socket);
private:
......@@ -105,7 +118,21 @@ class continuable : public ref_counted {
};
typedef intrusive_ptr<continuable> continuable_ptr;
/******************************************************************************
* inline and template member function implementations *
******************************************************************************/
inline native_socket_type continuable::read_handle() const {
return m_rd;
}
/**
* @brief Returns the file descriptor for outgoing data.
*/
inline native_socket_type continuable::write_handle() const {
return m_wr;
}
} } // namespace cppa::network
......
......@@ -67,6 +67,8 @@ class default_peer : public buffered_writer {
continue_writing_result continue_writing() override;
void dispose() override;
void io_failed() override;
void enqueue(const message_header& hdr, const any_tuple& msg);
......@@ -135,8 +137,6 @@ class default_peer : public buffered_writer {
};
typedef intrusive_ptr<default_peer> default_peer_ptr;
} } // namespace cppa::network
#endif // CPPA_DEFAULT_PEER_IMPL_HPP
......@@ -54,6 +54,8 @@ class default_peer_acceptor : public continuable {
inline const actor_ptr& published_actor() const { return m_pa; }
void dispose() override;
void io_failed();
private:
......@@ -64,8 +66,6 @@ class default_peer_acceptor : public continuable {
};
typedef intrusive_ptr<default_peer_acceptor> default_peer_acceptor_ptr;
} } // namespace cppa::detail
#endif // IPV4_PEER_ACCEPTOR_HPP
......@@ -50,6 +50,9 @@ class default_protocol : public protocol {
typedef protocol super;
friend class default_peer;
friend class default_peer_acceptor;
public:
default_protocol(middleman* multiplexer);
......@@ -70,7 +73,7 @@ class default_protocol : public protocol {
void register_peer(const process_information& node, default_peer* ptr);
default_peer_ptr get_peer(const process_information& node);
default_peer* get_peer(const process_information& node);
void enqueue(const process_information& node,
const message_header& hdr,
......@@ -80,22 +83,26 @@ class default_protocol : public protocol {
const output_stream_ptr& out,
const process_information_ptr& node = nullptr);
void last_proxy_exited(const default_peer_ptr& pptr);
void last_proxy_exited(default_peer* pptr);
void continue_writer(const default_peer_ptr& pptr);
void continue_writer(default_peer* pptr);
// covariant return type
default_actor_addressing* addressing();
private:
void del_acceptor(default_peer_acceptor* ptr);
void del_peer(default_peer* ptr);
struct peer_entry {
default_peer_ptr impl;
default_peer* impl;
default_message_queue_ptr queue;
};
default_actor_addressing m_addressing;
std::map<actor_ptr, std::vector<default_peer_acceptor_ptr> > m_acceptors;
std::map<actor_ptr, std::vector<default_peer_acceptor*>> m_acceptors;
std::map<process_information, peer_entry> m_peers;
};
......
......@@ -70,25 +70,37 @@ class middleman {
* @brief Removes @p ptr from the list of active writers.
* @warning This member function is not thread-safe.
*/
void stop_writer(const continuable_ptr& ptr);
void stop_writer(continuable* ptr);
/**
* @brief Adds @p ptr to the list of active writers.
* @warning This member function is not thread-safe.
*/
void continue_writer(const continuable_ptr& ptr);
void continue_writer(continuable* ptr);
/**
* @brief Checks wheter @p ptr is an active writer.
* @warning This member function is not thread-safe.
*/
bool has_writer(continuable* ptr);
/**
* @brief Removes @p ptr from the list of active readers.
* @warning This member function is not thread-safe.
*/
void stop_reader(const continuable_ptr& ptr);
void stop_reader(continuable* ptr);
/**
* @brief Adds @p ptr to the list of active readers.
* @warning This member function is not thread-safe.
*/
void continue_reader(const continuable_ptr& ptr);
void continue_reader(continuable* ptr);
/**
* @brief Checks wheter @p ptr is an active reader.
* @warning This member function is not thread-safe.
*/
bool has_reader(continuable* ptr);
protected:
......
......@@ -65,10 +65,10 @@ inline event_bitmask from_int_bitmask(unsigned mask) {
struct fd_meta_info {
native_socket_type fd;
continuable_ptr ptr;
continuable* ptr;
event_bitmask mask;
fd_meta_info(native_socket_type a0,
const continuable_ptr& a1,
continuable* a1,
event_bitmask a2)
: fd(a0), ptr(a1), mask(a2) { }
};
......@@ -84,12 +84,12 @@ class middleman_event_handler {
/**
* @brief Enqueues an add operation.
*/
void add_later(const continuable_ptr& ptr, event_bitmask e);
void add_later(continuable* ptr, event_bitmask e);
/**
* @brief Enqueues an erase operation.
*/
void erase_later(const continuable_ptr& ptr, event_bitmask e);
void erase_later(continuable* ptr, event_bitmask e);
/**
* @brief Poll all events.
......@@ -117,6 +117,12 @@ class middleman_event_handler {
*/
void update();
std::vector<continuable*> readers();
bool has_reader(continuable* ptr);
bool has_writer(continuable* ptr);
protected:
std::vector<fd_meta_info> m_meta; // this vector is *always* sorted
......@@ -125,6 +131,8 @@ class middleman_event_handler {
std::vector<std::pair<event_bitmask, continuable*>> m_events;
std::vector<continuable*> m_dispose_list;
middleman_event_handler();
// fills the event vector
......@@ -138,7 +146,7 @@ class middleman_event_handler {
private:
void alteration(const continuable_ptr& ptr, event_bitmask e, fd_meta_event etype);
void alteration(continuable* ptr, event_bitmask e, fd_meta_event etype);
event_bitmask next_bitmask(event_bitmask old, event_bitmask arg, fd_meta_event op) const;
......
......@@ -89,25 +89,25 @@ class protocol {
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void stop_writer(const continuable_ptr& ptr);
inline void stop_writer(continuable* ptr);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void continue_writer(const continuable_ptr& ptr);
inline void continue_writer(continuable* ptr);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void stop_reader(const continuable_ptr& ptr);
inline void stop_reader(continuable* ptr);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void continue_reader(const continuable_ptr& ptr);
inline void continue_reader(continuable* ptr);
/**
* @brief Returns the parent of this protocol instance.
......@@ -135,19 +135,19 @@ inline void protocol::run_later(T&& what) {
m_parent->run_later(std::forward<T>(what));
}
inline void protocol::stop_writer(const continuable_ptr& ptr) {
inline void protocol::stop_writer(continuable* ptr) {
m_parent->stop_writer(ptr);
}
inline void protocol::continue_writer(const continuable_ptr& ptr) {
inline void protocol::continue_writer(continuable* ptr) {
m_parent->continue_writer(ptr);
}
inline void protocol::stop_reader(const continuable_ptr& ptr) {
inline void protocol::stop_reader(continuable* ptr) {
m_parent->stop_reader(ptr);
}
inline void protocol::continue_reader(const continuable_ptr& ptr) {
inline void protocol::continue_reader(continuable* ptr) {
m_parent->continue_reader(ptr);
}
......
......@@ -36,21 +36,18 @@
#include "cppa/cppa.hpp"
#include "cppa/logging.hpp"
#include "cppa/singletons.hpp"
#include "cppa/io/broker.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/ipv4_acceptor.hpp"
#include "cppa/io/ipv4_io_stream.hpp"
#include "pingpong.pb.h"
#include "cppa/singletons.hpp"
#include "cppa/io/broker.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/io_handle.hpp"
#include "cppa/io/broker_backend.hpp"
using namespace std;
using namespace cppa;
using namespace cppa::network;
using namespace cppa::io;
void ping(size_t num_pings) {
auto count = make_shared<size_t>(0);
......@@ -77,7 +74,7 @@ void pong() {
);
}
void protobuf_io(io_handle* ios, const actor_ptr& buddy) {
void protobuf_io(broker* ios, const actor_ptr& buddy) {
self->monitor(buddy);
auto write = [=](const org::libcppa::PingOrPong& p) {
string buf = p.SerializeAsString();
......@@ -125,7 +122,7 @@ void protobuf_io(io_handle* ios, const actor_ptr& buddy) {
cerr << "neither Pong nor Ping!" << endl;
}
// receive next length prefix
ios->receive_policy(io_handle::exactly, 4);
ios->receive_policy(broker::exactly, 4);
unbecome();
},
default_bhvr
......@@ -140,13 +137,13 @@ void protobuf_io(io_handle* ios, const actor_ptr& buddy) {
return;
}
// receive protobuf data
ios->receive_policy(io_handle::exactly, (size_t) num_bytes);
ios->receive_policy(broker::exactly, (size_t) num_bytes);
become(keep_behavior, await_protobuf_data);
},
default_bhvr
};
// initial setup
ios->receive_policy(io_handle::exactly, 4);
ios->receive_policy(broker::exactly, 4);
become(await_length_prefix);
}
......
......@@ -28,33 +28,41 @@
\******************************************************************************/
#include <iostream>
#include "cppa/cppa.hpp"
#include "cppa/singletons.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/util/scope_guard.hpp"
#include "cppa/io/broker.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/broker_backend.hpp"
#include "cppa/io/broker.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/sync_request_bouncer.hpp"
using std::cout;
using std::endl;
namespace cppa { namespace io {
class broker_continuation {
public:
broker_continuation(broker_ptr ptr, mailbox_element* elem)
: m_self(std::move(ptr)), m_elem(elem) { }
broker_continuation(broker_ptr ptr, const message_header& hdr, any_tuple&& msg)
: m_self(std::move(ptr)), m_hdr(hdr), m_data(std::move(msg)) { }
inline void operator()() const {
m_self->invoke_message(m_elem);
inline void operator()() {
m_self->invoke_message(m_hdr, std::move(m_data));
}
private:
broker_ptr m_self;
mailbox_element* m_elem;
message_header m_hdr;
any_tuple m_data;
};
......@@ -62,15 +70,28 @@ class default_broker_impl : public broker {
public:
typedef std::function<void (io_handle*)> function_type;
typedef std::function<void (broker*)> function_type;
default_broker_impl(function_type&& fun) : m_fun(std::move(fun)) { }
default_broker_impl(function_type&& fun,
input_stream_ptr in,
output_stream_ptr out)
: broker(in, out), m_fun(std::move(fun)) { }
void init() override {
enqueue(nullptr, make_any_tuple(atom("INITMSG")));
become(
on(atom("INITMSG")) >> [=] {
m_fun(&io());
unbecome();
m_fun(this);
auto mm = get_middleman();
if (has_behavior()) mm->continue_reader(this);
else if (not mm->has_writer(this)) {
CPPA_LOG_WARNING("spawn_io: no behavior was set and "
"no data was written");
// middleman knows nothing about this broker;
// release its implicit reference count
deref();
}
}
);
}
......@@ -81,44 +102,44 @@ class default_broker_impl : public broker {
};
void broker::invoke_message(mailbox_element* elem) {
if (exit_reason() != exit_reason::not_exited) {
if (elem->mid.valid()) {
void broker::invoke_message(const message_header& hdr, any_tuple msg) {
if (exit_reason() != exit_reason::not_exited || m_bhvr_stack.empty()) {
if (hdr.id.valid()) {
detail::sync_request_bouncer srb{exit_reason()};
srb(*elem);
detail::disposer d;
d(elem);
srb(hdr.sender, hdr.id);
}
return;
}
// prepare actor for invocation of message handler
m_dummy_node.sender = hdr.sender;
m_dummy_node.msg = std::move(msg);
m_dummy_node.mid = hdr.id;
try {
scoped_self_setter sss{this};
m_bhvr_stack.invoke(m_recv_policy, this, elem);
auto& bhvr = m_bhvr_stack.back();
if (not bhvr(m_dummy_node.msg)) {
auto e = mailbox_element::create(hdr, std::move(m_dummy_node.msg));
m_recv_policy.add_to_cache(e);
}
}
catch (std::exception& e) {
CPPA_LOG_ERROR("IO actor killed due to an unhandled exception: "
CPPA_LOG_ERROR("broker killed due to an unhandled exception: "
<< to_verbose_string(e));
// keep compiler happy in non-debug mode
static_cast<void>(e);
quit(exit_reason::unhandled_exception);
}
catch (...) {
CPPA_LOG_ERROR("IO actor killed due to an unhandled exception");
CPPA_LOG_ERROR("broker killed due to an unhandled exception");
quit(exit_reason::unhandled_exception);
}
if (exit_reason() != exit_reason::not_exited) {
get_actor_registry()->dec_running();
m_parent.reset();
}
}
void broker::invoke_message(any_tuple msg) {
invoke_message(mailbox_element::create(this, std::move(msg)));
// restore dummy node
m_dummy_node.sender.reset();
m_dummy_node.msg.reset();
}
void broker::enqueue(const message_header& hdr, any_tuple msg) {
auto e = mailbox_element::create(hdr, std::move(msg));
get_middleman()->run_later(broker_continuation{this, e});
get_middleman()->run_later(broker_continuation{this, hdr, std::move(msg)});
}
bool broker::initialized() const {
......@@ -127,15 +148,106 @@ bool broker::initialized() const {
void broker::quit(std::uint32_t reason) {
cleanup(reason);
m_parent->handle_disconnect();
}
io_handle& broker::io() {
return *m_parent;
intrusive_ptr<broker> broker::from(std::function<void (broker*)> fun,
input_stream_ptr in,
output_stream_ptr out) {
return make_counted<default_broker_impl>(std::move(fun), in, out);
}
broker::broker(input_stream_ptr in, output_stream_ptr out)
: super1(), super2(get_middleman(), in->read_handle(), std::move(out))
, m_is_continue_reading(false), m_disconnected(false), m_dirty(false)
, m_policy(at_least), m_policy_buffer_size(0), m_in(in)
, m_read(atom("IO_read"), static_cast<uint32_t>(in->read_handle())) {
get_ref<2>(m_read).final_size(default_max_buffer_size);
// acquire implicit reference count held by the middleman
ref();
// actor is running now
get_actor_registry()->inc_running();
}
void broker::disconnect() {
if (not m_disconnected) {
m_disconnected = true;
if (exit_reason() == exit_reason::not_exited) {
auto msg = make_any_tuple(atom("IO_closed"),
static_cast<uint32_t>(m_in->read_handle()));
invoke_message(nullptr, std::move(msg));
}
}
}
void broker::cleanup(std::uint32_t reason) {
if (not m_is_continue_reading) {
// if we are currently in the continue_reading handler, we don't have
// tell the MM to stop this actor, because the member function will
// return stop_reading
get_middleman()->stop_reader(this);
}
super1::cleanup(reason);
get_actor_registry()->dec_running();
}
void broker::io_failed() {
disconnect();
}
void broker::dispose() {
deref(); // release implicit reference count of the middleman
}
void broker::receive_policy(policy_flag policy, size_t buffer_size) {
CPPA_LOG_TRACE(CPPA_ARG(policy) << ", " << CPPA_ARG(buffer_size));
if (not m_disconnected) {
m_dirty = true;
m_policy = policy;
m_policy_buffer_size = buffer_size;
}
}
continue_reading_result broker::continue_reading() {
CPPA_LOG_TRACE("");
m_is_continue_reading = true;
auto sg = util::make_scope_guard([=] {
m_is_continue_reading = false;
});
for (;;) {
// stop reading if actor finished execution
if (exit_reason() != exit_reason::not_exited) return read_closed;
auto& buf = get_ref<2>(m_read);
if (m_dirty) {
m_dirty = false;
if (m_policy == at_most || m_policy == exactly) {
buf.final_size(m_policy_buffer_size);
}
else buf.final_size(default_max_buffer_size);
}
auto before = buf.size();
try { buf.append_from(m_in.get()); }
catch (std::ios_base::failure&) {
disconnect();
return read_failure;
}
CPPA_LOG_DEBUG("received " << (buf.size() - before) << " bytes");
if ( before == buf.size()
|| (m_policy == exactly && buf.size() != m_policy_buffer_size)) {
return read_continue_later;
}
if ( (m_policy == at_least && buf.size() >= m_policy_buffer_size)
|| m_policy == exactly
|| m_policy == at_most) {
CPPA_LOG_DEBUG("invoke io actor");
invoke_message(nullptr, m_read);
CPPA_LOG_INFO_IF(!m_read.vals()->unique(), "buffer became detached");
get_ref<2>(m_read).clear();
}
}
}
intrusive_ptr<broker> broker::from(std::function<void (io_handle*)> fun) {
return make_counted<default_broker_impl>(std::move(fun));
void broker::write(size_t num_bytes, const void* data) {
super2::write(num_bytes, data);
}
} } // 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/singletons.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/broker_backend.hpp"
#include "cppa/detail/actor_registry.hpp"
namespace cppa { namespace io {
broker_backend::broker_backend(input_stream_ptr in,
output_stream_ptr out,
broker_ptr ptr)
: super(get_middleman(), in->read_handle(), std::move(out))
, m_dirty(false), m_policy(at_least)
, m_policy_buffer_size(0)
, m_in(in), m_self(ptr)
, m_read(atom("IO_read"), static_cast<uint32_t>(in->read_handle())) {
m_self->m_parent = this;
get_ref<2>(m_read).final_size(default_max_buffer_size);
}
void broker_backend::init() {
get_actor_registry()->inc_running();
auto selfptr = m_self.get();
scoped_self_setter sss{selfptr};
selfptr->init();
}
void broker_backend::handle_disconnect() {
if (m_self == nullptr) return;
auto ms = m_self;
m_self.reset();
CPPA_LOG_DEBUG("became disconnected");
if (ms->exit_reason() == exit_reason::not_exited) {
auto msg = make_any_tuple(atom("IO_closed"),
static_cast<uint32_t>(m_in->read_handle()));
ms->invoke_message(std::move(msg));
}
get_middleman()->stop_reader(this);
}
void broker_backend::io_failed() {
handle_disconnect();
}
void broker_backend::receive_policy(policy_flag policy, size_t buffer_size) {
CPPA_LOG_TRACE(CPPA_ARG(policy) << ", " << CPPA_ARG(buffer_size));
m_dirty = true;
m_policy = policy;
m_policy_buffer_size = buffer_size;
}
continue_reading_result broker_backend::continue_reading() {
CPPA_LOG_TRACE("");
for (;;) {
auto& buf = get_ref<2>(m_read);
if (m_dirty) {
m_dirty = false;
if (m_policy == at_most || m_policy == exactly) {
buf.final_size(m_policy_buffer_size);
}
else buf.final_size(default_max_buffer_size);
}
auto before = buf.size();
try { buf.append_from(m_in.get()); }
catch (std::ios_base::failure&) {
handle_disconnect();
return read_failure;
}
CPPA_LOG_DEBUG("received " << (buf.size() - before) << " bytes");
if ( before == buf.size()
|| (m_policy == exactly && buf.size() != m_policy_buffer_size)) {
return read_continue_later;
}
if ( (m_policy == at_least && buf.size() >= m_policy_buffer_size)
|| m_policy == exactly
|| m_policy == at_most) {
CPPA_LOG_DEBUG("invoke io actor");
m_self->invoke_message(m_read);
CPPA_LOG_INFO_IF(!m_read.vals()->unique(), "buffer became detached");
if (m_self == nullptr) {
// broker::quit() calls handle_disconnect, which sets
// m_self to nullptr
return read_closed;
}
get_ref<2>(m_read).clear();
}
}
}
void broker_backend::close() {
CPPA_LOG_DEBUG("");
get_middleman()->stop_reader(this);
}
void broker_backend::write(size_t num_bytes, const void* data) {
super::write(num_bytes, data);
}
} } // namespace cppa::network
......@@ -28,10 +28,14 @@
\******************************************************************************/
#include "cppa/detail/demangle.hpp"
#include "cppa/io/continuable.hpp"
namespace cppa { namespace io {
continuable::~continuable() { }
continuable::continuable(native_socket_type rd, native_socket_type wr)
: m_rd(rd), m_wr(wr) { }
......
......@@ -340,4 +340,9 @@ void default_peer::enqueue(const message_header& hdr, const any_tuple& msg) {
register_for_writing();
}
void default_peer::dispose() {
m_parent->del_peer(this);
delete this;
}
} } // namespace cppa::network
......@@ -88,4 +88,9 @@ void default_peer_acceptor::io_failed() {
<< this << " due to an IO failure");
}
void default_peer_acceptor::dispose() {
m_parent->del_acceptor(this);
delete this;
}
} } // namespace cppa::network
......@@ -91,12 +91,12 @@ void default_protocol::publish(const actor_ptr& whom,
static_cast<void>(args); // keep compiler happy
get_actor_registry()->put(whom->id(), whom);
default_protocol* proto = this;
auto impl = make_counted<default_peer_acceptor>(this, move(ptr), whom);
auto impl = new default_peer_acceptor(this, move(ptr), whom);
run_later([=] {
CPPA_LOGC_TRACE("cppa::io::default_protocol",
"publish$add_acceptor", "");
proto->m_acceptors[whom].push_back(impl);
proto->continue_reader(impl.get());
proto->continue_reader(impl);
});
}
......@@ -107,7 +107,7 @@ void default_protocol::unpublish(const actor_ptr& whom) {
CPPA_LOGC_TRACE("cppa::io::default_protocol",
"unpublish$remove_acceptors", "");
auto& acceptors = m_acceptors[whom];
for (auto& ptr : acceptors) proto->stop_reader(ptr.get());
for (auto& ptr : acceptors) proto->stop_reader(ptr);
m_acceptors.erase(whom);
});
}
......@@ -119,7 +119,7 @@ void default_protocol::register_peer(const process_information& node,
if (entry.impl == nullptr) {
if (entry.queue == nullptr) entry.queue.emplace();
ptr->set_queue(entry.queue);
entry.impl.reset(ptr);
entry.impl = ptr;
if (!entry.queue->empty()) {
auto tmp = entry.queue->pop();
ptr->enqueue(tmp.first, tmp.second);
......@@ -132,7 +132,7 @@ void default_protocol::register_peer(const process_information& node,
}
}
default_peer_ptr default_protocol::get_peer(const process_information& n) {
default_peer* default_protocol::get_peer(const process_information& n) {
CPPA_LOG_TRACE("n = " << to_string(n));
auto i = m_peers.find(n);
if (i != m_peers.end()) {
......@@ -143,6 +143,23 @@ default_peer_ptr default_protocol::get_peer(const process_information& n) {
return nullptr;
}
void default_protocol::del_peer(default_peer* ptr) {
m_peers.erase(ptr->node());
}
void default_protocol::del_acceptor(default_peer_acceptor* ptr) {
auto i = m_acceptors.begin();
auto e = m_acceptors.end();
while (i != e) {
auto& vec = i->second;
auto last = vec.end();
auto iter = std::find(vec.begin(), last, ptr);
if (iter != last) vec.erase(iter);
if (not vec.empty()) ++i;
else i = m_acceptors.erase(i);
}
}
void default_protocol::enqueue(const process_information& node,
const message_header& hdr,
any_tuple msg) {
......@@ -215,12 +232,12 @@ actor_ptr default_protocol::remote_actor(stream_ptr_pair io,
return result->value;
}
void default_protocol::last_proxy_exited(const default_peer_ptr& pptr) {
void default_protocol::last_proxy_exited(default_peer* pptr) {
CPPA_REQUIRE(pptr != nullptr);
CPPA_LOG_TRACE("pptr = " << pptr.get()
<< ", pptr->node() = " << to_string(pptr->node()));
if (pptr->erase_on_last_proxy_exited() && pptr->queue().empty()) {
stop_reader(pptr.get());
stop_reader(pptr);
auto i = m_peers.find(pptr->node());
if (i != m_peers.end()) {
CPPA_LOG_DEBUG_IF(i->second.impl != pptr,
......@@ -237,14 +254,14 @@ void default_protocol::new_peer(const input_stream_ptr& in,
const output_stream_ptr& out,
const process_information_ptr& node) {
CPPA_LOG_TRACE("");
auto ptr = make_counted<default_peer>(this, in, out, node);
continue_reader(ptr.get());
if (node) register_peer(*node, ptr.get());
auto ptr = new default_peer(this, in, out, node);
continue_reader(ptr);
if (node) register_peer(*node, ptr);
}
void default_protocol::continue_writer(const default_peer_ptr& pptr) {
void default_protocol::continue_writer(default_peer* pptr) {
CPPA_LOG_TRACE(CPPA_MARG(pptr, get));
super::continue_writer(pptr.get());
super::continue_writer(pptr);
}
default_actor_addressing* default_protocol::addressing() {
......
......@@ -111,30 +111,32 @@ class middleman_impl {
static_cast<void>(write(m_pipe_write, &dummy, sizeof(dummy)));
}
void continue_writer(const continuable_ptr& ptr) {
void continue_writer(continuable* ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_handler->add_later(ptr, event::write);
}
void stop_writer(const continuable_ptr& ptr) {
void stop_writer(continuable* ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_handler->erase_later(ptr, event::write);
}
void continue_reader(const continuable_ptr& ptr) {
inline bool has_writer(continuable* ptr) {
return m_handler->has_writer(ptr);
}
void continue_reader(continuable* ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_readers.push_back(ptr);
m_handler->add_later(ptr, event::read);
}
void stop_reader(const continuable_ptr& ptr) {
void stop_reader(continuable* 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_ptr& lhs) {
return lhs == ptr;
});
if (i != last) m_readers.erase(i);
}
inline bool has_reader(continuable* ptr) {
return m_handler->has_reader(ptr);
}
protected:
......@@ -166,7 +168,6 @@ class middleman_impl {
inline bool done() const { return m_done; }
bool m_done;
std::vector<continuable_ptr> m_readers;
middleman_event_handler& handler();
......@@ -199,6 +200,10 @@ class middleman_overseer : public continuable {
middleman_overseer(int pipe_fd, middleman_queue& q)
: super(pipe_fd), m_queue(q) { }
void dispose() override {
delete this;
}
continue_reading_result continue_reading() {
CPPA_LOG_TRACE("");
static constexpr size_t num_dummies = 64;
......@@ -254,22 +259,31 @@ void middleman::run_later(std::function<void()> fun) {
m_impl->run_later(std::move(fun));
}
void middleman::continue_writer(const continuable_ptr& ptr) {
void middleman::continue_writer(continuable* ptr) {
m_impl->continue_writer(ptr);
}
void middleman::stop_writer(const continuable_ptr& ptr) {
void middleman::stop_writer(continuable* ptr) {
m_impl->stop_writer(ptr);
}
void middleman::continue_reader(const continuable_ptr& ptr) {
bool middleman::has_writer(continuable* ptr) {
return m_impl->has_writer(ptr);
}
void middleman::continue_reader(continuable* ptr) {
m_impl->continue_reader(ptr);
}
void middleman::stop_reader(const continuable_ptr& ptr) {
void middleman::stop_reader(continuable* ptr) {
m_impl->stop_reader(ptr);
}
bool middleman::has_reader(continuable* ptr) {
return m_impl->has_reader(ptr);
}
void middleman_loop(middleman_impl* impl) {
# ifdef CPPA_LOG_LEVEL
auto mself = make_counted<thread_mapped_actor>();
......@@ -281,7 +295,7 @@ void middleman_loop(middleman_impl* impl) {
CPPA_LOGF_INFO("middleman runs at "
<< to_string(*process_information::get()));
handler->init();
impl->continue_reader(make_counted<middleman_overseer>(impl->m_pipe_read, impl->m_queue));
impl->continue_reader(new middleman_overseer(impl->m_pipe_read, impl->m_queue));
handler->update();
while (!impl->done()) {
handler->poll([&](event_bitmask mask, continuable* io) {
......@@ -329,7 +343,8 @@ void middleman_loop(middleman_impl* impl) {
}
CPPA_LOGF_DEBUG("event loop done, erase all readers");
// make sure to write everything before shutting down
for (auto ptr : impl->m_readers) { handler->erase_later(ptr, event::read); }
auto readers = handler->readers();
for (auto reader : readers) { handler->erase_later(reader, event::read); }
handler->update();
CPPA_LOGF_DEBUG("flush outgoing messages");
CPPA_LOGF_DEBUG_IF(handler->num_sockets() == 0,
......@@ -359,9 +374,6 @@ void middleman_loop(middleman_impl* impl) {
}
});
}
CPPA_LOGF_DEBUG("clear all containers");
//impl->m_peers.clear();
impl->m_readers.clear();
CPPA_LOGF_DEBUG("middleman loop done");
}
......
......@@ -47,7 +47,7 @@ middleman_event_handler::middleman_event_handler() { }
middleman_event_handler::~middleman_event_handler() { }
void middleman_event_handler::alteration(const continuable_ptr& ptr,
void middleman_event_handler::alteration(continuable* ptr,
event_bitmask e,
fd_meta_event etype) {
native_socket_type fd;
......@@ -63,7 +63,7 @@ void middleman_event_handler::alteration(const continuable_ptr& ptr,
fd = ptr->read_handle();
auto wrfd = ptr->write_handle();
if (fd != wrfd) {
CPPA_LOGMF(CPPA_DEBUG, self, "read_handle != write_handle, split "
CPPA_LOG_DEBUG("read_handle != write_handle, split "
"into two function calls");
// split into two function calls
e = event::read;
......@@ -72,18 +72,18 @@ void middleman_event_handler::alteration(const continuable_ptr& ptr,
break;
}
default:
CPPA_LOGMF(CPPA_ERROR, self, "invalid bitmask");
CPPA_CRITICAL("invalid bitmask");
return;
}
m_alterations.emplace_back(fd_meta_info(fd, ptr, e), etype);
}
void middleman_event_handler::add_later(const continuable_ptr& ptr, event_bitmask e) {
void middleman_event_handler::add_later(continuable* ptr, event_bitmask e) {
CPPA_LOG_TRACE("ptr = " << ptr.get() << ", e = " << eb2str(e));
alteration(ptr, e, fd_meta_event::add);
}
void middleman_event_handler::erase_later(const continuable_ptr& ptr, event_bitmask e) {
void middleman_event_handler::erase_later(continuable* ptr, event_bitmask e) {
CPPA_LOG_TRACE("ptr = " << ptr.get() << ", e = " << eb2str(e));
alteration(ptr, e, fd_meta_event::erase);
}
......@@ -105,13 +105,12 @@ void middleman_event_handler::update() {
auto iter = std::lower_bound(m_meta.begin(), last, elem.fd, mless);
if (iter != last) old = iter->mask;
auto mask = next_bitmask(old, elem.mask, elem_pair.second);
auto ptr = elem.ptr.get();
CPPA_LOGMF(CPPA_DEBUG, self, "new bitmask for "
auto ptr = elem.ptr;
CPPA_LOG_DEBUG("new bitmask for "
<< elem.ptr.get() << ": " << eb2str(mask));
if (iter == last || iter->fd != elem.fd) {
CPPA_LOG_INFO_IF(mask == event::none,
"cannot erase " << ptr
<< " (not found in m_meta)");
CPPA_LOG_ERROR_IF(mask == event::none,
"cannot erase " << ptr << " (no such element)");
if (mask != event::none) {
m_meta.insert(iter, elem);
handle_event(fd_meta_event::add, elem.fd,
......@@ -121,6 +120,9 @@ void middleman_event_handler::update() {
else if (iter->fd == elem.fd) {
CPPA_REQUIRE(iter->ptr == elem.ptr);
if (mask == event::none) {
// note: we cannot decide whether it's safe to dispose `ptr`,
// because we didn't parse all alterations yet
m_dispose_list.emplace_back(ptr);
m_meta.erase(iter);
handle_event(fd_meta_event::erase, elem.fd, old, mask, ptr);
}
......@@ -131,6 +133,45 @@ void middleman_event_handler::update() {
}
}
m_alterations.clear();
// m_meta won't be touched inside loop
auto first = m_meta.begin();
auto last = m_meta.end();
auto is_alive = [&](native_socket_type fd) -> bool {
auto iter = std::lower_bound(first, last, fd, mless);
return iter != last && iter->fd == fd;
};
// check whether elements in dispose list can be safely deleted
for (auto& elem : m_dispose_list) {
auto rd = elem->read_handle();
auto wr = elem->write_handle();
if ( (rd == wr && !is_alive(rd))
|| (rd != wr && !is_alive(rd) && !is_alive(wr))) {
std::cout << "SAFE TO DISPOSE ELEMENT: " << elem << std::endl;
elem->dispose();
}
else std::cout << "GOTTA KEEP ELEMENT: " << elem << std::endl;
}
m_dispose_list.clear();
}
std::vector<continuable*> middleman_event_handler::readers() {
std::vector<continuable*> result;
for (auto& meta : m_meta) {
if (meta.mask & event::read) result.push_back(meta.ptr);
}
return result;
}
bool middleman_event_handler::has_reader(continuable* ptr) {
return std::any_of(m_meta.begin(), m_meta.end(), [=](fd_meta_info& meta) {
return meta.ptr == ptr && (meta.mask & event::read);
});
}
bool middleman_event_handler::has_writer(continuable* ptr) {
return std::any_of(m_meta.begin(), m_meta.end(), [=](fd_meta_info& meta) {
return meta.ptr == ptr && (meta.mask & event::write);
});
}
} } // namespace cppa::network
......@@ -98,7 +98,7 @@ class middleman_event_handler_impl : public middleman_event_handler {
output_event,
error_event>(mask);
m_pollset[i].revents = 0;
if (eb != event::none) m_events.emplace_back(eb, m_meta[i].ptr.get());
if (eb != event::none) m_events.emplace_back(eb, m_meta[i].ptr);
}
}
......
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