Commit dc003fd3 authored by Dominik Charousset's avatar Dominik Charousset

Modularize middleman and networking internals

parent 2c9f4ca9
...@@ -14,6 +14,7 @@ set (LIBCAF_IO_SRCS ...@@ -14,6 +14,7 @@ set (LIBCAF_IO_SRCS
src/doorman.cpp src/doorman.cpp
src/middleman.cpp src/middleman.cpp
src/middleman_actor.cpp src/middleman_actor.cpp
src/middleman_actor_impl.cpp
src/hook.cpp src/hook.cpp
src/interfaces.cpp src/interfaces.cpp
src/manager.cpp src/manager.cpp
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#define CAF_IO_MIDDLEMAN_ACTOR_HPP #define CAF_IO_MIDDLEMAN_ACTOR_HPP
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/atom.hpp"
#include "caf/typed_actor.hpp" #include "caf/typed_actor.hpp"
namespace caf { namespace caf {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO_MIDDLEMAN_ACTOR_IMPL_HPP
#define CAF_IO_MIDDLEMAN_ACTOR_IMPL_HPP
#include "caf/fwd.hpp"
#include "caf/atom.hpp"
#include "caf/typed_actor.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/io/fwd.hpp"
#include "caf/io/middleman_actor.hpp"
namespace caf {
namespace io {
/// Default implementation of the `middleman_actor` interface.
class middleman_actor_impl : public middleman_actor::base {
public:
using put_res = result<uint16_t>;
using mpi_set = std::set<std::string>;
using get_res = delegated<node_id, strong_actor_ptr, mpi_set>;
using del_res = delegated<void>;
using endpoint_data = std::tuple<node_id, strong_actor_ptr, mpi_set>;
using endpoint = std::pair<std::string, uint16_t>;
middleman_actor_impl(actor_config& cfg, actor default_broker);
void on_exit() override;
const char* name() const override;
behavior_type make_behavior() override;
protected:
/// Tries to connect to given `host` and `port`. The default implementation
/// calls `system().middleman().backend().new_tcp_scribe(host, port)`.
virtual expected<scribe_ptr> connect(const std::string& host, uint16_t port);
/// Tries to open a local port. The default implementation calls
/// `system().middleman().backend().new_tcp_doorman(port, addr, reuse)`.
virtual expected<doorman_ptr> open(uint16_t port, const char* addr,
bool reuse);
private:
result<uint16_t> put(uint16_t port, strong_actor_ptr& whom, mpi_set& sigs,
const char* in = nullptr, bool reuse_addr = false);
optional<endpoint_data&> cached(const endpoint& ep);
optional<std::vector<response_promise>&> pending(const endpoint& ep);
actor broker_;
std::map<endpoint, endpoint_data> cached_;
std::map<endpoint, std::vector<response_promise>> pending_;
};
} // namespace io
} // namespace caf
#endif // CAF_IO_MIDDLEMAN_ACTOR_IMPL_HPP
...@@ -176,6 +176,18 @@ bool write_some(size_t& result, native_socket fd, const void* buf, size_t len); ...@@ -176,6 +176,18 @@ bool write_some(size_t& result, native_socket fd, const void* buf, size_t len);
/// as long as /// as long as
bool try_accept(native_socket& result, native_socket fd); bool try_accept(native_socket& result, native_socket fd);
/// Returns the locally assigned port of `fd`.
expected<uint16_t> local_port_of_fd(native_socket fd);
/// Returns the locally assigned address of `fd`.
expected<std::string> local_addr_of_fd(native_socket fd);
/// Returns the port used by the remote host of `fd`.
expected<uint16_t> remote_port_of_fd(native_socket fd);
/// Returns the remote host address of `fd`.
expected<std::string> remote_addr_of_fd(native_socket fd);
class default_multiplexer; class default_multiplexer;
/// A socket I/O event handler. /// A socket I/O event handler.
...@@ -419,9 +431,69 @@ public: ...@@ -419,9 +431,69 @@ public:
void removed_from_loop(operation op) override; void removed_from_loop(operation op) override;
void handle_event(operation op) override; protected:
template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) {
CAF_LOG_TRACE(CAF_ARG(op));
auto mcr = max_consecutive_reads();
switch (op) {
case io::network::operation::read: {
// Loop until an error occurs or we have nothing more to read
// or until we have handled `mcr` reads.
size_t rb;
for (size_t i = 0; i < mcr; ++i) {
if (!policy.read_some(rb, fd(), rd_buf_.data() + collected_,
rd_buf_.size() - collected_)) {
reader_->io_failure(&backend(), operation::read);
passivate();
return;
}
if (rb == 0)
return;
collected_ += rb;
if (collected_ >= read_threshold_) {
auto res = reader_->consume(&backend(), rd_buf_.data(), collected_);
prepare_next_read();
if (!res) {
passivate();
return;
}
}
}
break;
}
case io::network::operation::write: {
size_t wb; // written bytes
if (!policy.write_some(wb, fd(), wr_buf_.data() + written_,
wr_buf_.size() - written_)) {
writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, fd(), this);
} else if (wb > 0) {
written_ += wb;
CAF_ASSERT(written_ <= wr_buf_.size());
auto remaining = wr_buf_.size() - written_;
if (ack_writes_)
writer_->data_transferred(&backend(), wb,
remaining + wr_offline_buf_.size());
// prepare next send (or stop sending)
if (remaining == 0)
prepare_next_write();
}
break;
}
case operation::propagate_error:
if (reader_)
reader_->io_failure(&backend(), operation::read);
if (writer_)
writer_->io_failure(&backend(), operation::write);
// backend will delete this handler anyway,
// no need to call backend().del() here
}
}
private: private:
size_t max_consecutive_reads();
void prepare_next_read(); void prepare_next_read();
void prepare_next_write(); void prepare_next_write();
...@@ -443,6 +515,26 @@ private: ...@@ -443,6 +515,26 @@ private:
buffer_type wr_offline_buf_; buffer_type wr_offline_buf_;
}; };
/// A concrete stream with a technology-dependent policy for sending and
/// receiving data from a socket.
template <class ProtocolPolicy>
class stream_impl : public stream {
public:
template <class... Ts>
stream_impl(default_multiplexer& mpx, native_socket sockfd, Ts&&... xs)
: stream(mpx, sockfd),
policy_(std::forward<Ts>(xs)...) {
// nop
}
void handle_event(io::network::operation op) override {
this->handle_event_impl(op, policy_);
}
private:
ProtocolPolicy policy_;
};
/// An acceptor is responsible for accepting incoming connections. /// An acceptor is responsible for accepting incoming connections.
class acceptor : public event_handler { class acceptor : public event_handler {
public: public:
...@@ -471,15 +563,47 @@ public: ...@@ -471,15 +563,47 @@ public:
/// Closes the network connection and removes this handler from its parent. /// Closes the network connection and removes this handler from its parent.
void stop_reading(); void stop_reading();
void handle_event(operation op) override;
void removed_from_loop(operation op) override; void removed_from_loop(operation op) override;
protected:
template <class Policy>
void handle_event_impl(io::network::operation op, Policy& policy) {
CAF_LOG_TRACE(CAF_ARG(fd()) << CAF_ARG(op));
if (mgr_ && op == operation::read) {
native_socket sockfd = invalid_native_socket;
if (policy.try_accept(sockfd, fd())) {
if (sockfd != invalid_native_socket) {
sock_ = sockfd;
mgr_->new_connection();
}
}
}
}
private: private:
manager_ptr mgr_; manager_ptr mgr_;
native_socket sock_; native_socket sock_;
}; };
/// A concrete acceptor with a technology-dependent policy.
template <class ProtocolPolicy>
class acceptor_impl : public acceptor {
public:
template <class... Ts>
acceptor_impl(default_multiplexer& mpx, native_socket sockfd, Ts&&... xs)
: acceptor(mpx, sockfd),
policy_(std::forward<Ts>(xs)...) {
// nop
}
void handle_event(io::network::operation op) override {
this->handle_event_impl(op, policy_);
}
private:
ProtocolPolicy policy_;
};
expected<native_socket> new_tcp_connection(const std::string& host, expected<native_socket> new_tcp_connection(const std::string& host,
uint16_t port, uint16_t port,
optional<protocol> preferred = none); optional<protocol> preferred = none);
......
...@@ -52,6 +52,8 @@ ...@@ -52,6 +52,8 @@
using std::string; using std::string;
// -- Utiliy functions for converting errno into CAF errors --------------------
namespace { namespace {
#if defined(CAF_MACOS) || defined(CAF_IOS) #if defined(CAF_MACOS) || defined(CAF_IOS)
...@@ -110,15 +112,7 @@ namespace caf { ...@@ -110,15 +112,7 @@ namespace caf {
namespace io { namespace io {
namespace network { namespace network {
// helper function // -- OS-specific functions for sockets and pipes ------------------------------
expected<std::string> local_addr_of_fd(native_socket fd);
expected<uint16_t> local_port_of_fd(native_socket fd);
expected<std::string> remote_addr_of_fd(native_socket fd);
expected<uint16_t> remote_port_of_fd(native_socket fd);
/******************************************************************************
* platform-dependent implementations *
******************************************************************************/
#ifndef CAF_WINDOWS #ifndef CAF_WINDOWS
...@@ -274,7 +268,8 @@ expected<uint16_t> remote_port_of_fd(native_socket fd); ...@@ -274,7 +268,8 @@ expected<uint16_t> remote_port_of_fd(native_socket fd);
connect(read_fd, &a.addr, connect(read_fd, &a.addr,
static_cast<int>(sizeof(a.inaddr)))); static_cast<int>(sizeof(a.inaddr))));
// get write-only end of the pipe // get write-only end of the pipe
CALL_CRITICAL_CFUN(write_fd, cc_valid_socket, "accept", accept(listener, nullptr, nullptr)); CALL_CRITICAL_CFUN(write_fd, cc_valid_socket, "accept",
accept(listener, nullptr, nullptr));
closesocket(listener); closesocket(listener);
guard.disable(); guard.disable();
return std::make_pair(read_fd, write_fd); return std::make_pair(read_fd, write_fd);
...@@ -282,9 +277,7 @@ expected<uint16_t> remote_port_of_fd(native_socket fd); ...@@ -282,9 +277,7 @@ expected<uint16_t> remote_port_of_fd(native_socket fd);
#endif #endif
/****************************************************************************** // -- Platform-dependent abstraction over epoll() or poll() --------------------
* epoll() vs. poll() *
******************************************************************************/
#ifdef CAF_EPOLL_MULTIPLEXER #ifdef CAF_EPOLL_MULTIPLEXER
...@@ -577,6 +570,8 @@ expected<uint16_t> remote_port_of_fd(native_socket fd); ...@@ -577,6 +570,8 @@ expected<uint16_t> remote_port_of_fd(native_socket fd);
#endif // CAF_EPOLL_MULTIPLEXER #endif // CAF_EPOLL_MULTIPLEXER
// -- Helper functions for defining bitmasks of event handlers -----------------
int add_flag(operation op, int bf) { int add_flag(operation op, int bf) {
switch (op) { switch (op) {
case operation::read: case operation::read:
...@@ -605,6 +600,91 @@ int del_flag(operation op, int bf) { ...@@ -605,6 +600,91 @@ int del_flag(operation op, int bf) {
return 0; return 0;
} }
// -- Platform-independent free functions --------------------------------------
expected<void> tcp_nodelay(native_socket fd, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(new_value));
int flag = new_value ? 1 : 0;
CALL_CFUN(res, cc_zero, "setsockopt",
setsockopt(fd, IPPROTO_TCP, TCP_NODELAY,
reinterpret_cast<setsockopt_ptr>(&flag),
static_cast<socklen_t>(sizeof(flag))));
return unit;
}
bool is_error(ssize_t res, bool is_nonblock) {
if (res < 0) {
auto err = last_socket_error();
if (!is_nonblock || !would_block_or_temporarily_unavailable(err)) {
return true;
}
// don't report an error in case of
// spurious wakeup or something similar
}
return false;
}
bool read_some(size_t& result, native_socket fd, void* buf, size_t len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
auto sres = ::recv(fd, reinterpret_cast<socket_recv_ptr>(buf), len, 0);
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(fd) << CAF_ARG(sres));
if (is_error(sres, true) || sres == 0) {
// recv returns 0 when the peer has performed an orderly shutdown
return false;
}
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return true;
}
bool write_some(size_t& result, native_socket fd, const void* buf, size_t len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
auto sres = ::send(fd, reinterpret_cast<socket_send_ptr>(buf),
len, no_sigpipe_flag);
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(fd) << CAF_ARG(sres));
if (is_error(sres, true))
return false;
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return true;
}
bool try_accept(native_socket& result, native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
sockaddr_storage addr;
memset(&addr, 0, sizeof(addr));
socklen_t addrlen = sizeof(addr);
result = ::accept(fd, reinterpret_cast<sockaddr*>(&addr), &addrlen);
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(result));
if (result == invalid_native_socket) {
auto err = last_socket_error();
if (!would_block_or_temporarily_unavailable(err)) {
return false;
}
}
return true;
}
// -- Policy class for TCP wrapping above free functions -----------------------
namespace {
using read_some_fun = decltype(read_some)*;
using write_some_fun = decltype(write_some)*;
using try_accept_fun = decltype(try_accept)*;
struct tcp_policy {
static read_some_fun read_some;
static write_some_fun write_some;
static try_accept_fun try_accept;
};
read_some_fun tcp_policy::read_some = network::read_some;
write_some_fun tcp_policy::write_some = network::write_some;
try_accept_fun tcp_policy::try_accept = network::try_accept;
} // namespace <anonymous>
// -- Platform-independent parts of the default_multiplexer --------------------
void default_multiplexer::add(operation op, native_socket fd, void default_multiplexer::add(operation op, native_socket fd,
event_handler* ptr) { event_handler* ptr) {
CAF_ASSERT(fd != invalid_native_socket); CAF_ASSERT(fd != invalid_native_socket);
...@@ -792,7 +872,7 @@ scribe_ptr default_multiplexer::new_scribe(native_socket fd) { ...@@ -792,7 +872,7 @@ scribe_ptr default_multiplexer::new_scribe(native_socket fd) {
} }
private: private:
bool launched_; bool launched_;
stream stream_; stream_impl<tcp_policy> stream_;
}; };
return make_counted<impl>(*this, fd); return make_counted<impl>(*this, fd);
} }
...@@ -857,7 +937,7 @@ doorman_ptr default_multiplexer::new_doorman(native_socket fd) { ...@@ -857,7 +937,7 @@ doorman_ptr default_multiplexer::new_doorman(native_socket fd) {
acceptor_.passivate(); acceptor_.passivate();
} }
private: private:
network::acceptor acceptor_; acceptor_impl<tcp_policy> acceptor_;
}; };
return make_counted<impl>(*this, fd); return make_counted<impl>(*this, fd);
} }
...@@ -871,70 +951,6 @@ expected<doorman_ptr> default_multiplexer::new_tcp_doorman(uint16_t port, ...@@ -871,70 +951,6 @@ expected<doorman_ptr> default_multiplexer::new_tcp_doorman(uint16_t port,
return std::move(fd.error()); return std::move(fd.error());
} }
/******************************************************************************
* platform-independent implementations (finally) *
******************************************************************************/
expected<void> tcp_nodelay(native_socket fd, bool new_value) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(new_value));
int flag = new_value ? 1 : 0;
CALL_CFUN(res, cc_zero, "setsockopt",
setsockopt(fd, IPPROTO_TCP, TCP_NODELAY,
reinterpret_cast<setsockopt_ptr>(&flag),
static_cast<socklen_t>(sizeof(flag))));
return unit;
}
bool is_error(ssize_t res, bool is_nonblock) {
if (res < 0) {
auto err = last_socket_error();
if (!is_nonblock || !would_block_or_temporarily_unavailable(err)) {
return true;
}
// don't report an error in case of
// spurious wakeup or something similar
}
return false;
}
bool read_some(size_t& result, native_socket fd, void* buf, size_t len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
auto sres = ::recv(fd, reinterpret_cast<socket_recv_ptr>(buf), len, 0);
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(fd) << CAF_ARG(sres));
if (is_error(sres, true) || sres == 0) {
// recv returns 0 when the peer has performed an orderly shutdown
return false;
}
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return true;
}
bool write_some(size_t& result, native_socket fd, const void* buf, size_t len) {
CAF_LOG_TRACE(CAF_ARG(fd) << CAF_ARG(len));
auto sres = ::send(fd, reinterpret_cast<socket_send_ptr>(buf),
len, no_sigpipe_flag);
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(fd) << CAF_ARG(sres));
if (is_error(sres, true))
return false;
result = (sres > 0) ? static_cast<size_t>(sres) : 0;
return true;
}
bool try_accept(native_socket& result, native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd));
sockaddr_storage addr;
memset(&addr, 0, sizeof(addr));
socklen_t addrlen = sizeof(addr);
result = ::accept(fd, reinterpret_cast<sockaddr*>(&addr), &addrlen);
CAF_LOG_DEBUG(CAF_ARG(fd) << CAF_ARG(result));
if (result == invalid_native_socket) {
auto err = last_socket_error();
if (!would_block_or_temporarily_unavailable(err)) {
return false;
}
}
return true;
}
event_handler::event_handler(default_multiplexer& dm, native_socket sockfd) event_handler::event_handler(default_multiplexer& dm, native_socket sockfd)
: eventbf_(0), : eventbf_(0),
...@@ -1089,65 +1105,8 @@ void stream::removed_from_loop(operation op) { ...@@ -1089,65 +1105,8 @@ void stream::removed_from_loop(operation op) {
} }
} }
void stream::handle_event(operation op) { size_t stream::max_consecutive_reads() {
CAF_LOG_TRACE(CAF_ARG(op)); return backend().system().config().middleman_max_consecutive_reads;
auto mcr = backend().system().config().middleman_max_consecutive_reads;
switch (op) {
case operation::read: {
// loop until an error occurs or we have nothing more to read
// or until we have handled 50 reads
size_t rb;
for (size_t i = 0; i < mcr; ++i) {
if (!read_some(rb, fd(),
rd_buf_.data() + collected_,
rd_buf_.size() - collected_)) {
reader_->io_failure(&backend(), operation::read);
passivate();
return;
}
if (rb == 0)
return;
collected_ += rb;
if (collected_ >= read_threshold_) {
auto res = reader_->consume(&backend(), rd_buf_.data(), collected_);
prepare_next_read();
if (!res) {
passivate();
return;
}
}
}
break;
}
case operation::write: {
size_t wb; // written bytes
if (!write_some(wb, fd(),
wr_buf_.data() + written_,
wr_buf_.size() - written_)) {
writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, fd(), this);
} else if (wb > 0) {
written_ += wb;
CAF_ASSERT(written_ <= wr_buf_.size());
auto remaining = wr_buf_.size() - written_;
if (ack_writes_)
writer_->data_transferred(&backend(), wb,
remaining + wr_offline_buf_.size());
// prepare next send (or stop sending)
if (remaining == 0)
prepare_next_write();
}
break;
}
case operation::propagate_error:
if (reader_)
reader_->io_failure(&backend(), operation::read);
if (writer_)
writer_->io_failure(&backend(), operation::write);
// backend will delete this handler anyway,
// no need to call backend().del() here
break;
}
} }
void stream::prepare_next_read() { void stream::prepare_next_read() {
...@@ -1211,19 +1170,6 @@ void acceptor::stop_reading() { ...@@ -1211,19 +1170,6 @@ void acceptor::stop_reading() {
passivate(); passivate();
} }
void acceptor::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(fd()) << CAF_ARG(op));
if (mgr_ && op == operation::read) {
native_socket sockfd = invalid_native_socket;
if (try_accept(sockfd, fd())) {
if (sockfd != invalid_native_socket) {
sock_ = sockfd;
mgr_->new_connection();
}
}
}
}
void acceptor::removed_from_loop(operation op) { void acceptor::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG(fd()) << CAF_ARG(op)); CAF_LOG_TRACE(CAF_ARG(fd()) << CAF_ARG(op));
if (op == operation::read) if (op == operation::read)
......
...@@ -23,206 +23,15 @@ ...@@ -23,206 +23,15 @@
#include <stdexcept> #include <stdexcept>
#include <utility> #include <utility>
#include "caf/sec.hpp" #include "caf/actor_system.hpp"
#include "caf/send.hpp" #include "caf/spawn_options.hpp"
#include "caf/actor.hpp"
#include "caf/logger.hpp"
#include "caf/node_id.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/io/basp_broker.hpp" #include "caf/io/middleman_actor_impl.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
namespace {
class middleman_actor_impl : public middleman_actor::base {
public:
middleman_actor_impl(actor_config& cfg, actor default_broker)
: middleman_actor::base(cfg),
broker_(std::move(default_broker)) {
set_down_handler([=](down_msg& dm) {
auto i = cached_.begin();
auto e = cached_.end();
while (i != e) {
if (get<1>(i->second) == dm.source)
i = cached_.erase(i);
else
++i;
}
});
set_exit_handler([=](exit_msg&) {
// ignored, the MM links group nameservers
// to this actor for proper shutdown ordering
});
}
void on_exit() override {
CAF_LOG_TRACE("");
destroy(broker_);
}
const char* name() const override {
return "middleman_actor";
}
using put_res = result<uint16_t>;
using mpi_set = std::set<std::string>;
using get_res = delegated<node_id, strong_actor_ptr, mpi_set>;
using del_res = delegated<void>;
using endpoint_data = std::tuple<node_id, strong_actor_ptr, mpi_set>;
using endpoint = std::pair<std::string, uint16_t>;
behavior_type make_behavior() override {
CAF_LOG_TRACE("");
return {
[=](publish_atom, uint16_t port, strong_actor_ptr& whom,
mpi_set& sigs, std::string& addr, bool reuse) -> put_res {
CAF_LOG_TRACE("");
return put(port, whom, sigs, addr.c_str(), reuse);
},
[=](open_atom, uint16_t port, std::string& addr, bool reuse) -> put_res {
CAF_LOG_TRACE("");
strong_actor_ptr whom;
mpi_set sigs;
return put(port, whom, sigs, addr.c_str(), reuse);
},
[=](connect_atom, std::string& hostname, uint16_t port) -> get_res {
CAF_LOG_TRACE(CAF_ARG(hostname) << CAF_ARG(port));
auto rp = make_response_promise();
endpoint key{std::move(hostname), port};
// respond immediately if endpoint is cached
auto x = cached(key);
if (x) {
CAF_LOG_DEBUG("found cached entry" << CAF_ARG(*x));
rp.deliver(get<0>(*x), get<1>(*x), get<2>(*x));
return {};
}
// attach this promise to a pending request if possible
auto rps = pending(key);
if (rps) {
CAF_LOG_DEBUG("attach to pending request");
rps->emplace_back(std::move(rp));
return {};
}
// connect to endpoint and initiate handhsake etc.
auto r = system().middleman().backend().new_tcp_scribe(key.first, port);
if (!r) {
rp.deliver(std::move(r.error()));
return {};
}
auto& ptr = *r;
std::vector<response_promise> tmp{std::move(rp)};
pending_.emplace(key, std::move(tmp));
request(broker_, infinite, connect_atom::value, std::move(ptr), port)
.then(
[=](node_id& nid, strong_actor_ptr& addr, mpi_set& sigs) {
auto i = pending_.find(key);
if (i == pending_.end())
return;
if (nid && addr) {
monitor(addr);
cached_.emplace(key, std::make_tuple(nid, addr, sigs));
}
auto res = make_message(std::move(nid), std::move(addr),
std::move(sigs));
for (auto& promise : i->second)
promise.deliver(res);
pending_.erase(i);
},
[=](error& err) {
auto i = pending_.find(key);
if (i == pending_.end())
return;
for (auto& promise : i->second)
promise.deliver(err);
pending_.erase(i);
}
);
return {};
},
[=](unpublish_atom atm, actor_addr addr, uint16_t p) -> del_res {
CAF_LOG_TRACE("");
delegate(broker_, atm, std::move(addr), p);
return {};
},
[=](close_atom atm, uint16_t p) -> del_res {
CAF_LOG_TRACE("");
delegate(broker_, atm, p);
return {};
},
[=](spawn_atom atm, node_id& nid, std::string& str,
message& msg, std::set<std::string>& ifs)
-> delegated<strong_actor_ptr> {
CAF_LOG_TRACE("");
delegate(broker_, forward_atom::value, nid, atom("SpawnServ"),
make_message(atm, std::move(str),
std::move(msg), std::move(ifs)));
return {};
},
[=](get_atom atm, node_id nid)
-> delegated<node_id, std::string, uint16_t> {
CAF_LOG_TRACE("");
delegate(broker_, atm, std::move(nid));
return {};
}
};
}
private:
put_res put(uint16_t port, strong_actor_ptr& whom,
mpi_set& sigs, const char* in = nullptr,
bool reuse_addr = false) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs)
<< CAF_ARG(in) << CAF_ARG(reuse_addr));
uint16_t actual_port;
// treat empty strings like nullptr
if (in != nullptr && in[0] == '\0')
in = nullptr;
auto res = system().middleman().backend().new_tcp_doorman(port, in,
reuse_addr);
if (!res)
return std::move(res.error());
auto& ptr = *res;
actual_port = ptr->port();
anon_send(broker_, publish_atom::value, std::move(ptr), actual_port,
std::move(whom), std::move(sigs));
return actual_port;
}
optional<endpoint_data&> cached(const endpoint& ep) {
auto i = cached_.find(ep);
if (i != cached_.end())
return i->second;
return none;
}
optional<std::vector<response_promise>&> pending(const endpoint& ep) {
auto i = pending_.find(ep);
if (i != pending_.end())
return i->second;
return none;
}
actor broker_;
std::map<endpoint, endpoint_data> cached_;
std::map<endpoint, std::vector<response_promise>> pending_;
};
} // namespace <anonymous>
middleman_actor make_middleman_actor(actor_system& sys, actor db) { middleman_actor make_middleman_actor(actor_system& sys, actor db) {
return sys.config().middleman_detach_utility_actors return sys.config().middleman_detach_utility_actors
? sys.spawn<middleman_actor_impl, detached + hidden>(std::move(db)) ? sys.spawn<middleman_actor_impl, detached + hidden>(std::move(db))
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2017 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/middleman_actor_impl.hpp"
#include <tuple>
#include <stdexcept>
#include <utility>
#include "caf/sec.hpp"
#include "caf/send.hpp"
#include "caf/actor.hpp"
#include "caf/logger.hpp"
#include "caf/node_id.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/typed_event_based_actor.hpp"
#include "caf/io/basp_broker.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/default_multiplexer.hpp"
namespace caf {
namespace io {
middleman_actor_impl::middleman_actor_impl(actor_config& cfg,
actor default_broker)
: middleman_actor::base(cfg),
broker_(std::move(default_broker)) {
set_down_handler([=](down_msg& dm) {
auto i = cached_.begin();
auto e = cached_.end();
while (i != e) {
if (get<1>(i->second) == dm.source)
i = cached_.erase(i);
else
++i;
}
});
set_exit_handler([=](exit_msg&) {
// ignored, the MM links group nameservers
// to this actor for proper shutdown ordering
});
}
void middleman_actor_impl::on_exit() {
CAF_LOG_TRACE("");
broker_ = nullptr;
}
const char* middleman_actor_impl::name() const {
return "middleman_actor";
}
auto middleman_actor_impl::make_behavior() -> behavior_type {
CAF_LOG_TRACE("");
return {
[=](publish_atom, uint16_t port, strong_actor_ptr& whom, mpi_set& sigs,
std::string& addr, bool reuse) -> put_res {
CAF_LOG_TRACE("");
return put(port, whom, sigs, addr.c_str(), reuse);
},
[=](open_atom, uint16_t port, std::string& addr, bool reuse) -> put_res {
CAF_LOG_TRACE("");
strong_actor_ptr whom;
mpi_set sigs;
return put(port, whom, sigs, addr.c_str(), reuse);
},
[=](connect_atom, std::string& hostname, uint16_t port) -> get_res {
CAF_LOG_TRACE(CAF_ARG(hostname) << CAF_ARG(port));
auto rp = make_response_promise();
endpoint key{std::move(hostname), port};
// respond immediately if endpoint is cached
auto x = cached(key);
if (x) {
CAF_LOG_DEBUG("found cached entry" << CAF_ARG(*x));
rp.deliver(get<0>(*x), get<1>(*x), get<2>(*x));
return {};
}
// attach this promise to a pending request if possible
auto rps = pending(key);
if (rps) {
CAF_LOG_DEBUG("attach to pending request");
rps->emplace_back(std::move(rp));
return {};
}
// connect to endpoint and initiate handhsake etc.
auto r = connect(key.first, port);
if (!r) {
rp.deliver(std::move(r.error()));
return {};
}
auto& ptr = *r;
std::vector<response_promise> tmp{std::move(rp)};
pending_.emplace(key, std::move(tmp));
request(broker_, infinite, connect_atom::value, std::move(ptr), port)
.then(
[=](node_id& nid, strong_actor_ptr& addr, mpi_set& sigs) {
auto i = pending_.find(key);
if (i == pending_.end())
return;
if (nid && addr) {
monitor(addr);
cached_.emplace(key, std::make_tuple(nid, addr, sigs));
}
auto res = make_message(std::move(nid), std::move(addr),
std::move(sigs));
for (auto& promise : i->second)
promise.deliver(res);
pending_.erase(i);
},
[=](error& err) {
auto i = pending_.find(key);
if (i == pending_.end())
return;
for (auto& promise : i->second)
promise.deliver(err);
pending_.erase(i);
});
return {};
},
[=](unpublish_atom atm, actor_addr addr, uint16_t p) -> del_res {
CAF_LOG_TRACE("");
delegate(broker_, atm, std::move(addr), p);
return {};
},
[=](close_atom atm, uint16_t p) -> del_res {
CAF_LOG_TRACE("");
delegate(broker_, atm, p);
return {};
},
[=](spawn_atom atm, node_id& nid, std::string& str, message& msg,
std::set<std::string>& ifs) -> delegated<strong_actor_ptr> {
CAF_LOG_TRACE("");
delegate(
broker_, forward_atom::value, nid, atom("SpawnServ"),
make_message(atm, std::move(str), std::move(msg), std::move(ifs)));
return {};
},
[=](get_atom atm,
node_id nid) -> delegated<node_id, std::string, uint16_t> {
CAF_LOG_TRACE("");
delegate(broker_, atm, std::move(nid));
return {};
}
};
}
middleman_actor_impl::put_res
middleman_actor_impl::put(uint16_t port, strong_actor_ptr& whom, mpi_set& sigs,
const char* in, bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << CAF_ARG(whom) << CAF_ARG(sigs)
<< CAF_ARG(in) << CAF_ARG(reuse_addr));
uint16_t actual_port;
// treat empty strings like nullptr
if (in != nullptr && in[0] == '\0')
in = nullptr;
auto res = open(port, in, reuse_addr);
if (!res)
return std::move(res.error());
auto& ptr = *res;
actual_port = ptr->port();
anon_send(broker_, publish_atom::value, std::move(ptr), actual_port,
std::move(whom), std::move(sigs));
return actual_port;
}
optional<middleman_actor_impl::endpoint_data&>
middleman_actor_impl::cached(const endpoint& ep) {
auto i = cached_.find(ep);
if (i != cached_.end())
return i->second;
return none;
}
optional<std::vector<response_promise>&>
middleman_actor_impl::pending(const endpoint& ep) {
auto i = pending_.find(ep);
if (i != pending_.end())
return i->second;
return none;
}
expected<scribe_ptr> middleman_actor_impl::connect(const std::string& host,
uint16_t port) {
return system().middleman().backend().new_tcp_scribe(host, port);
}
expected<doorman_ptr>
middleman_actor_impl::open(uint16_t port, const char* addr, bool reuse) {
return system().middleman().backend().new_tcp_doorman(port, addr, reuse);
}
} // namespace io
} // namespace caf
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