Commit d7208d90 authored by Jakob Otto's avatar Jakob Otto

Merge branch 'master' into topic/datagram-transport

parents 564489f8 493ec8b2
......@@ -91,6 +91,16 @@ function(pretty_yes var)
endif()
endfunction(pretty_yes)
add_executable(caf-generate-enum-strings cmake/caf-generate-enum-strings.cpp)
function(enum_to_string relative_input_file relative_output_file)
set(input "${CMAKE_CURRENT_SOURCE_DIR}/${relative_input_file}")
set(output "${CMAKE_CURRENT_BINARY_DIR}/${relative_output_file}")
add_custom_command(OUTPUT "${output}"
COMMAND caf-generate-enum-strings "${input}" "${output}"
DEPENDS caf-generate-enum-strings "${input}")
endfunction()
# -- binary and library path setup ---------------------------------------------
# Prohibit in-source builds.
......
#include <algorithm>
#include <cstring>
#include <fstream>
#include <iostream>
#include <string>
#include <vector>
using std::cerr;
using std::find;
using std::find_if;
using std::string;
using std::vector;
void trim(string& str) {
auto not_space = [](char c) { return isspace(c) == 0; };
str.erase(str.begin(), find_if(str.begin(), str.end(), not_space));
str.erase(find_if(str.rbegin(), str.rend(), not_space).base(), str.end());
}
template <size_t N>
bool starts_with(const string& str, const char (&prefix)[N]) {
return str.compare(0, N - 1, prefix) == 0;
}
template <size_t N>
void drop_prefix(string& str, const char (&prefix)[N]) {
if (str.compare(0, N - 1, prefix) == 0)
str.erase(str.begin(), str.begin() + (N - 1));
}
void keep_alnum(string& str) {
auto not_alnum = [](char c) { return isalnum(c) == 0 && c != '_'; };
str.erase(find_if(str.begin(), str.end(), not_alnum), str.end());
}
int main(int argc, char** argv) {
if (argc != 3) {
cerr << "wrong number of arguments.\n"
<< "usage: " << argv[0] << "input-file output-file\n";
return EXIT_FAILURE;
}
std::ifstream in{argv[1]};
if (!in) {
cerr << "unable to open input file: " << argv[1] << '\n';
return EXIT_FAILURE;
}
vector<string> namespaces;
string enum_name;
string line;
bool is_enum_class = false;
// Locate the beginning of the enum.
for (;;) {
if (!getline(in, line)) {
cerr << "unable to locate enum in file: " << argv[1] << '\n';
return EXIT_FAILURE;
}
trim(line);
if (starts_with(line, "enum ")) {
drop_prefix(line, "enum ");
if (starts_with(line, "class ")) {
is_enum_class = true;
drop_prefix(line, "class ");
}
trim(line);
keep_alnum(line);
enum_name = line;
break;
}
if (starts_with(line, "namespace ")) {
if (line.back() == '{')
line.pop_back();
line.erase(line.begin(), find(line.begin(), line.end(), ' '));
trim(line);
namespaces.emplace_back(line);
}
}
// Sanity checking.
if (namespaces.empty()) {
cerr << "enum found outside of a namespace\n";
return EXIT_FAILURE;
}
if (enum_name.empty()) {
cerr << "empty enum name found\n";
return EXIT_FAILURE;
}
std::ofstream out{argv[2]};
if (!out) {
cerr << "unable to open output file: " << argv[1] << '\n';
return EXIT_FAILURE;
}
// Print file header.
out << "#include \"" << namespaces[0];
for (size_t i = 1; i < namespaces.size(); ++i)
out << '/' << namespaces[i];
out << '/' << enum_name << ".hpp\"\n\n"
<< "#include <string>\n\n"
<< "namespace " << namespaces[0] << " {\n";
for (size_t i = 1; i < namespaces.size(); ++i)
out << "namespace " << namespaces[i] << " {\n";
out << "\nstd::string to_string(" << enum_name << " x) {\n"
<< " switch(x) {\n"
<< " default:\n"
<< " return \"???\";\n";
// Read until hitting the closing '}'.
std::string case_label_prefix;
if (is_enum_class)
case_label_prefix = enum_name + "::";
for (;;) {
if (!getline(in, line)) {
cerr << "unable to read enum values\n";
return EXIT_FAILURE;
}
trim(line);
if (line.empty())
continue;
if (line[0] == '}')
break;
if (line[0] != '/') {
keep_alnum(line);
out << " case " << case_label_prefix << line << ":\n"
<< " return \"" << line << "\";\n";
}
}
// Done. Print file footer and exit.
out << " };\n"
<< "}\n\n";
for (auto i = namespaces.rbegin(); i != namespaces.rend(); ++i)
out << "} // namespace " << *i << '\n';
}
......@@ -5,20 +5,33 @@ project(caf_net C CXX)
# e.g., for creating proper Xcode projects
file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
enum_to_string("caf/net/basp/connection_state.hpp" "basp_conn_strings.cpp")
enum_to_string("caf/net/basp/ec.hpp" "basp_ec_strings.cpp")
enum_to_string("caf/net/basp/message_type.hpp" "basp_message_type_strings.cpp")
enum_to_string("caf/net/operation.hpp" "operation_strings.cpp")
# list cpp files excluding platform-dependent files
set(LIBCAF_NET_SRCS
"${CMAKE_CURRENT_BINARY_DIR}/basp_conn_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/basp_ec_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/basp_message_type_strings.cpp"
"${CMAKE_CURRENT_BINARY_DIR}/operation_strings.cpp"
src/actor_proxy_impl.cpp
src/application.cpp
src/connection_state.cpp
src/convert_ip_endpoint.cpp
src/datagram_socket.cpp
src/defaults.cpp
src/ec.cpp
src/endpoint_manager.cpp
src/header.cpp
src/host.cpp
src/ip.cpp
src/message_type.cpp
src/multiplexer.cpp
src/net/backend/test.cpp
src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp
src/net/middleman_backend.cpp
src/net/packet_writer.cpp
src/network_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
......@@ -27,7 +40,6 @@ set(LIBCAF_NET_SRCS
src/stream_socket.cpp
src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp
)
......
......@@ -35,10 +35,6 @@ public:
void enqueue(mailbox_element_ptr what, execution_unit* context) override;
bool add_backlink(abstract_actor* x) override;
bool remove_backlink(abstract_actor* x) override;
void kill_proxy(execution_unit* ctx, error rsn) override;
private:
......
/******************************************************************************_opt
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <map>
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/node_id.hpp"
namespace caf {
namespace net {
namespace backend {
/// Minimal backend for unit testing.
/// @warning this backend is *not* thread safe.
class test : public middleman_backend {
public:
// -- member types -----------------------------------------------------------
using peer_entry = std::pair<stream_socket, endpoint_manager_ptr>;
// -- constructors, destructors, and assignment operators --------------------
test(middleman& mm);
~test() override;
// -- interface functions ----------------------------------------------------
error init() override;
endpoint_manager_ptr peer(const node_id& id) override;
void resolve(const uri& locator, const actor& listener) override;
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override;
void set_last_hop(node_id*) override;
// -- properties -------------------------------------------------------------
stream_socket socket(const node_id& peer_id) {
return get_peer(peer_id).first;
}
peer_entry& emplace(const node_id& peer_id, stream_socket first,
stream_socket second);
private:
peer_entry& get_peer(const node_id& id);
middleman& mm_;
std::map<node_id, peer_entry> peers_;
proxy_registry proxies_;
};
} // namespace backend
} // namespace net
} // namespace caf
......@@ -20,6 +20,7 @@
#include <cstdint>
#include <memory>
#include <type_traits>
#include <unordered_map>
#include <unordered_set>
#include <vector>
......@@ -33,10 +34,12 @@
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/message_type.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/packet_writer.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/node_id.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/response_promise.hpp"
#include "caf/scoped_execution_unit.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
......@@ -54,13 +57,11 @@ public:
using byte_span = span<const byte>;
using write_packet_callback = callback<byte_span, byte_span>;
using proxy_registry_ptr = std::shared_ptr<proxy_registry>;
struct test_tag {};
// -- constructors, destructors, and assignment operators --------------------
explicit application(proxy_registry_ptr proxies);
explicit application(proxy_registry& proxies);
// -- interface functions ----------------------------------------------------
......@@ -68,67 +69,47 @@ public:
error init(Parent& parent) {
// Initialize member variables.
system_ = &parent.system();
executor_.system_ptr(system_);
executor_.proxy_registry_ptr(&proxies_);
// TODO: use `if constexpr` when switching to C++17.
// Allow unit tests to run the application without endpoint manager.
if (!std::is_base_of<test_tag, Parent>::value)
manager_ = &parent.manager();
// Write handshake.
if (auto err = generate_handshake())
auto hdr = parent.next_header_buffer();
auto payload = parent.next_payload_buffer();
if (auto err = generate_handshake(payload))
return err;
auto hdr = to_bytes(header{message_type::handshake,
static_cast<uint32_t>(buf_.size()), version});
parent.write_packet(hdr, buf_);
to_bytes(header{message_type::handshake,
static_cast<uint32_t>(payload.size()), version},
hdr);
parent.write_packet(hdr, payload);
parent.transport().configure_read(receive_policy::exactly(header_size));
return none;
}
template <class Parent>
error write_message(Parent& parent,
std::unique_ptr<endpoint_manager::message> ptr) {
// TODO: avoid extra copy of the payload
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver;
if (dst == nullptr) {
// TODO: valid?
return none;
}
if (src != nullptr) {
if (auto err = sink(src->node(), src->id(), dst->id(), ptr->msg->stages))
return err;
} else {
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages))
return err;
}
buf_.insert(buf_.end(), ptr->payload.begin(), ptr->payload.end());
header hdr{message_type::actor_message, static_cast<uint32_t>(buf_.size()),
ptr->msg->mid.integer_value()};
auto bytes = to_bytes(hdr);
parent.write_packet(make_span(bytes), make_span(buf_));
return none;
}
error write_message(packet_writer& writer,
std::unique_ptr<endpoint_manager_queue::message> ptr);
template <class Parent>
error handle_data(Parent& parent, byte_span bytes) {
auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(hdr, payload);
return none;
});
static_assert(std::is_base_of<packet_writer, Parent>::value,
"parent must implement packet_writer");
size_t next_read_size = header_size;
if (auto err = handle(next_read_size, write_packet, bytes))
if (auto err = handle(next_read_size, parent, bytes))
return err;
parent.transport().configure_read(receive_policy::exactly(next_read_size));
return none;
}
template <class Parent>
void resolve(Parent& parent, string_view path, actor listener) {
auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(hdr, payload);
return none;
});
resolve_remote_path(write_packet, path, listener);
}
void resolve(packet_writer& writer, string_view path, const actor& listener);
static void new_proxy(packet_writer& writer, actor_id id);
void local_actor_down(packet_writer& writer, actor_id id, error reason);
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
// nop
}
......@@ -136,16 +117,13 @@ public:
// nop
}
static expected<std::vector<byte>> serialize(actor_system& sys,
const type_erased_tuple& x);
static expected<buffer_type> serialize(actor_system& sys,
const type_erased_tuple& x);
// -- utility functions ------------------------------------------------------
strong_actor_ptr resolve_local_path(string_view path);
void resolve_remote_path(write_packet_callback& write_packet,
string_view path, actor listener);
// -- properties -------------------------------------------------------------
connection_state state() const noexcept {
......@@ -157,28 +135,31 @@ public:
}
private:
// -- message handling -------------------------------------------------------
// -- handling of incoming messages ------------------------------------------
error handle(size_t& next_read_size, write_packet_callback& write_packet,
byte_span bytes);
error handle(size_t& next_read_size, packet_writer& writer, byte_span bytes);
error handle(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle(packet_writer& writer, header hdr, byte_span payload);
error handle_handshake(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_handshake(packet_writer& writer, header hdr, byte_span payload);
error handle_actor_message(write_packet_callback& write_packet, header hdr,
error handle_actor_message(packet_writer& writer, header hdr,
byte_span payload);
error handle_resolve_request(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_resolve_request(packet_writer& writer, header rec_hdr,
byte_span received);
error handle_resolve_response(packet_writer& writer, header received_hdr,
byte_span received);
error handle_monitor_message(packet_writer& writer, header received_hdr,
byte_span received);
error handle_resolve_response(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_down_message(packet_writer& writer, header received_hdr,
byte_span received);
/// Writes the handshake payload to `buf_`.
error generate_handshake();
error generate_handshake(buffer_type& buf);
// -- member variables -------------------------------------------------------
......@@ -191,17 +172,11 @@ private:
/// Caches the last header while waiting for the matching payload.
header hdr_;
/// Re-usable buffer for storing payloads.
buffer_type buf_;
/// Stores our own ID.
node_id id_;
/// Stores the ID of our peer.
node_id peer_id_;
/// Tracks which local actors our peer monitors.
std::unordered_set<actor_addr> monitored_actors_;
std::unordered_set<actor_addr> monitored_actors_; // TODO: this is unused
/// Caches actor handles obtained via `resolve`.
std::unordered_map<uint64_t, response_promise> pending_resolves_;
......@@ -210,7 +185,14 @@ private:
uint64_t next_request_id_ = 1;
/// Points to the factory object for generating proxies.
proxy_registry_ptr proxies_;
proxy_registry& proxies_;
/// Points to the endpoint manager that owns this applications.
endpoint_manager* manager_ = nullptr;
/// Provides pointers to the actor system as well as the registry,
/// serializers and deserializer.
scoped_execution_unit executor_;
};
} // namespace basp
......
......@@ -41,6 +41,8 @@ enum class ec : uint8_t {
unimplemented = 10,
app_identifiers_mismatch,
invalid_payload,
invalid_scheme,
invalid_locator,
};
/// @relates ec
......
......@@ -78,6 +78,10 @@ struct header : detail::comparable<header> {
/// @relates header
std::array<byte, header_size> to_bytes(header x);
/// Serializes a header to a byte representation.
/// @relates header
void to_bytes(header x, std::vector<byte>& buf);
/// @relates header
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, header& x) {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <cstddef>
// -- hard-coded default values for various CAF options ------------------------
namespace caf {
namespace defaults {
namespace middleman {
/// Maximum number of cached buffers for sending payloads.
extern const size_t max_payload_buffers;
/// Maximum number of cached buffers for sending headers.
extern const size_t max_header_buffers;
} // namespace middleman
} // namespace defaults
} // namespace caf
......@@ -30,6 +30,7 @@
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/socket_manager.hpp"
#include "caf/variant.hpp"
......@@ -50,75 +51,6 @@ public:
using serialize_fun_type = maybe_buffer (*)(actor_system&,
const type_erased_tuple&);
struct event : intrusive::singly_linked<event> {
struct resolve_request {
std::string path;
actor listener;
};
struct timeout {
atom_value type;
uint64_t id;
};
event(std::string path, actor listener);
event(atom_value type, uint64_t id);
/// Either contains a string for `resolve` requests or an `atom_value`
variant<resolve_request, timeout> value;
};
struct event_policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = event;
using unique_pointer = std::unique_ptr<event>;
using queue_type = intrusive::drr_queue<event_policy>;
task_size_type task_size(const event&) const noexcept {
return 1;
}
};
using event_queue_type = intrusive::fifo_inbox<event_policy>;
struct message : intrusive::singly_linked<message> {
/// Original message to a remote actor.
mailbox_element_ptr msg;
/// ID of the receiving actor.
strong_actor_ptr receiver;
/// Serialized representation of of `msg->content()`.
std::vector<byte> payload;
message(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
};
struct message_policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = message;
using unique_pointer = std::unique_ptr<message>;
using queue_type = intrusive::drr_queue<message_policy>;
task_size_type task_size(const message& x) const noexcept {
return x.payload.size();
}
};
using message_queue_type = intrusive::fifo_inbox<message_policy>;
// -- constructors, destructors, and assignment operators --------------------
endpoint_manager(socket handle, const multiplexer_ptr& parent,
......@@ -132,19 +64,22 @@ public:
return sys_;
}
std::unique_ptr<message> next_message();
endpoint_manager_queue::message_ptr next_message();
// -- event management -------------------------------------------------------
/// Resolves a path to a remote actor.
void resolve(std::string path, actor listener);
void resolve(uri locator, actor listener);
/// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
/// Enqueues a timeout to the endpoint.
void enqueue(timeout_msg msg);
/// Enqueues an event to the endpoint.
template <class... Ts>
void enqueue_event(Ts&&... xs) {
enqueue(new endpoint_manager_queue::event(std::forward<Ts>(xs)...));
}
// -- pure virtual member functions ------------------------------------------
......@@ -155,14 +90,13 @@ public:
virtual serialize_fun_type serialize_fun() const noexcept = 0;
protected:
bool enqueue(endpoint_manager_queue::element* ptr);
/// Points to the hosting actor system.
actor_system& sys_;
/// Stores control events.
event_queue_type events_;
/// Stores outbound messages.
message_queue_type messages_;
/// Stores control events and outbound messages.
endpoint_manager_queue::type queue_;
/// Stores a proxy for interacting with the actor clock.
actor timeout_proxy_;
......
......@@ -22,6 +22,7 @@
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/atom.hpp"
#include "caf/detail/overload.hpp"
#include "caf/net/endpoint_manager.hpp"
namespace caf {
......@@ -79,6 +80,7 @@ public:
// -- interface functions ----------------------------------------------------
error init() override {
this->register_reading();
return transport_.init(*this);
}
......@@ -87,25 +89,36 @@ public:
}
bool handle_write_event() override {
if (!this->events_.blocked() && !this->events_.empty()) {
if (!this->queue_.blocked()) {
this->queue_.fetch_more();
auto& q = std::get<0>(this->queue_.queue().queues());
do {
this->events_.fetch_more();
auto& q = this->events_.queue();
q.inc_deficit(q.total_task_size());
for (auto ptr = q.next(); ptr != nullptr; ptr = q.next()) {
using timeout = endpoint_manager::event::timeout;
using resolve_request = endpoint_manager::event::resolve_request;
if (auto rr = get_if<resolve_request>(&ptr->value)) {
transport_.resolve(*this, std::move(rr->path),
std::move(rr->listener));
} else {
auto& t = get<timeout>(ptr->value);
transport_.timeout(*this, t.type, t.id);
}
auto f = detail::make_overload(
[&](endpoint_manager_queue::event::resolve_request& x) {
transport_.resolve(*this, x.locator, x.listener);
},
[&](endpoint_manager_queue::event::new_proxy& x) {
transport_.new_proxy(*this, x.peer, x.id);
},
[&](endpoint_manager_queue::event::local_actor_down& x) {
transport_.local_actor_down(*this, x.observing_peer, x.id,
std::move(x.reason));
},
[&](endpoint_manager_queue::event::timeout& x) {
transport_.timeout(*this, x.type, x.id);
});
visit(f, ptr->value);
}
} while (!this->events_.try_block());
} while (!q.empty());
}
return transport_.handle_write_event(*this);
if (!transport_.handle_write_event(*this)) {
if (this->queue_.blocked())
return false;
return !(this->queue_.empty() && this->queue_.try_block());
}
return true;
}
void handle_error(sec code) override {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include "caf/actor.hpp"
#include "caf/fwd.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp"
#include "caf/intrusive/wdrr_fixed_multiplexed_queue.hpp"
#include "caf/mailbox_element.hpp"
#include "caf/unit.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace net {
class endpoint_manager_queue {
public:
enum class element_type { event, message };
class element : public intrusive::singly_linked<element> {
public:
explicit element(element_type tag) : tag_(tag) {
// nop
}
virtual ~element();
virtual size_t task_size() const noexcept = 0;
element_type tag() const noexcept {
return tag_;
}
private:
element_type tag_;
};
using element_ptr = std::unique_ptr<element>;
class event final : public element {
public:
struct resolve_request {
uri locator;
actor listener;
};
struct new_proxy {
node_id peer;
actor_id id;
};
struct local_actor_down {
node_id observing_peer;
actor_id id;
error reason;
};
struct timeout {
atom_value type;
uint64_t id;
};
event(uri locator, actor listener);
event(node_id peer, actor_id proxy_id);
event(node_id observing_peer, actor_id local_actor_id, error reason);
event(atom_value type, uint64_t id);
~event() override;
size_t task_size() const noexcept override;
/// Holds the event data.
variant<resolve_request, new_proxy, local_actor_down, timeout> value;
};
using event_ptr = std::unique_ptr<event>;
struct event_policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = event;
using unique_pointer = event_ptr;
using queue_type = intrusive::drr_queue<event_policy>;
constexpr event_policy(unit_t) {
// nop
}
task_size_type task_size(const event&) const noexcept {
return 1;
}
};
class message : public element {
public:
/// Original message to a remote actor.
mailbox_element_ptr msg;
/// ID of the receiving actor.
strong_actor_ptr receiver;
/// Serialized representation of of `msg->content()`.
std::vector<byte> payload;
message(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
~message() override;
size_t task_size() const noexcept override;
};
using message_ptr = std::unique_ptr<message>;
struct message_policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = message;
using unique_pointer = message_ptr;
using queue_type = intrusive::drr_queue<message_policy>;
constexpr message_policy(unit_t) {
// nop
}
static task_size_type task_size(const message& x) noexcept {
// Return at least 1 if the payload is empty.
return x.payload.size() + static_cast<task_size_type>(x.payload.empty());
}
};
struct categorized {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = element;
using unique_pointer = element_ptr;
constexpr categorized(unit_t) {
// nop
}
template <class Queue>
deficit_type quantum(const Queue&, deficit_type x) const noexcept {
return x;
}
size_t id_of(const element& x) const noexcept {
return static_cast<size_t>(x.tag());
}
};
struct policy {
using deficit_type = size_t;
using task_size_type = size_t;
using mapped_type = element;
using unique_pointer = std::unique_ptr<element>;
using queue_type = intrusive::wdrr_fixed_multiplexed_queue<
categorized, event_policy::queue_type, message_policy::queue_type>;
task_size_type task_size(const message& x) const noexcept {
return x.task_size();
}
};
using type = intrusive::fifo_inbox<policy>;
};
} // namespace net
} // namespace caf
......@@ -25,24 +25,37 @@
namespace caf {
namespace net {
class multiplexer;
class socket_manager;
// -- templates ----------------------------------------------------------------
template <class Application, class IdType = unit_t>
class transport_worker;
template <class Application, class IdType = unit_t>
class transport_worker_dispatcher;
// -- classes ------------------------------------------------------------------
class endpoint_manager;
class middleman;
class middleman_backend;
class multiplexer;
class socket_manager;
// -- structs ------------------------------------------------------------------
struct network_socket;
struct pipe_socket;
struct socket;
struct stream_socket;
struct tcp_stream_socket;
struct tcp_accept_socket;
struct tcp_stream_socket;
using socket_manager_ptr = intrusive_ptr<socket_manager>;
// -- smart pointers -----------------------------------------------------------
using endpoint_manager_ptr = intrusive_ptr<endpoint_manager>;
using middleman_backend_ptr = std::unique_ptr<middleman_backend>;
using multiplexer_ptr = std::shared_ptr<multiplexer>;
using socket_manager_ptr = intrusive_ptr<socket_manager>;
using weak_multiplexer_ptr = std::weak_ptr<multiplexer>;
} // namespace net
......
......@@ -27,9 +27,19 @@ namespace caf {
namespace net {
namespace ip {
/// Returns all IP addresses of to `host` (if any).
/// Returns all IP addresses of `host` (if any).
std::vector<ip_address> resolve(string_view host);
/// Returns all IP addresses of `host` (if any).
std::vector<ip_address> resolve(ip_address host);
/// Returns the IP addresses for a local endpoint, which is either an address,
/// an interface name, or the string "localhost".
std::vector<ip_address> local_addresses(string_view host);
/// Returns the IP addresses for a local endpoint address.
std::vector<ip_address> local_addresses(ip_address host);
/// Returns the hostname of this device.
std::string hostname();
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <thread>
#include "caf/actor_system.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
namespace caf {
namespace net {
class middleman : public actor_system::module {
public:
// -- member types -----------------------------------------------------------
using module = actor_system::module;
using module_ptr = actor_system::module_ptr;
using middleman_backend_list = std::vector<middleman_backend_ptr>;
// -- constructors, destructors, and assignment operators --------------------
~middleman() override;
// -- interface functions ----------------------------------------------------
void start() override;
void stop() override;
void init(actor_system_config&) override;
id_t id() const override;
void* subtype_ptr() override;
// -- factory functions ------------------------------------------------------
template <class... Ts>
static module* make(actor_system& sys, detail::type_list<Ts...> token) {
std::unique_ptr<middleman> result{new middleman(sys)};
if (sizeof...(Ts) > 0) {
result->backends_.reserve(sizeof...(Ts));
create_backends(*result, token);
}
return result.release();
}
// -- remoting ---------------------------------------------------------------
/// Resolves a path to a remote actor.
void resolve(const uri& locator, const actor& listener);
// -- properties -------------------------------------------------------------
actor_system& system() {
return sys_;
}
const actor_system_config& config() const noexcept {
return sys_.config();
}
const multiplexer_ptr& mpx() const noexcept {
return mpx_;
}
middleman_backend* backend(string_view scheme) const noexcept;
private:
// -- constructors, destructors, and assignment operators --------------------
explicit middleman(actor_system& sys);
// -- utility functions ------------------------------------------------------
static void create_backends(middleman&, detail::type_list<>) {
// End of recursion.
}
template <class T, class... Ts>
static void create_backends(middleman& mm, detail::type_list<T, Ts...>) {
mm.backends_.emplace_back(new T(mm));
create_backends(mm, detail::type_list<Ts...>{});
}
// -- member variables -------------------------------------------------------
/// Points to the parent system.
actor_system& sys_;
/// Stores the global socket I/O multiplexer.
multiplexer_ptr mpx_;
/// Stores all available backends for managing peers.
middleman_backend_list backends_;
/// Runs the multiplexer's event loop
std::thread mpx_thread_;
};
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <string>
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
#include "caf/proxy_registry.hpp"
namespace caf {
namespace net {
/// Technology-specific backend for connecting to and managing peer
/// connections.
/// @relates middleman
class middleman_backend : public proxy_registry::backend {
public:
// -- constructors, destructors, and assignment operators --------------------
middleman_backend(std::string id);
virtual ~middleman_backend();
// -- interface functions ----------------------------------------------------
/// Initializes the backend.
virtual error init() = 0;
/// @returns The endpoint manager for `peer` on success, `nullptr` otherwise.
virtual endpoint_manager_ptr peer(const node_id& id) = 0;
/// Resolves a path to a remote actor.
virtual void resolve(const uri& locator, const actor& listener) = 0;
// -- properties -------------------------------------------------------------
const std::string& id() const noexcept {
return id_;
}
private:
/// Stores the technology-specific identifier.
std::string id_;
};
/// @relates middleman_backend
using middleman_backend_ptr = std::unique_ptr<middleman_backend>;
} // namespace net
} // namespace caf
......@@ -64,9 +64,13 @@ public:
// -- thread-safe signaling --------------------------------------------------
/// Causes the multiplexer to update its event bitmask for `mgr`.
/// Registers `mgr` for read events.
/// @thread-safe
void update(const socket_manager_ptr& mgr);
void register_reading(const socket_manager_ptr& mgr);
/// Registers `mgr` for write events.
/// @thread-safe
void register_writing(const socket_manager_ptr& mgr);
/// Closes the pipe for signaling updates to the multiplexer. After closing
/// the pipe, calls to `update` no longer have any effect.
......@@ -79,21 +83,25 @@ public:
/// ready as a result.
bool poll_once(bool blocking);
/// Sets the thread ID to `std::this_thread::id()`.
void set_thread_id();
/// Polls until no socket event handler remains.
void run();
/// Processes all updates on the socket managers.
void handle_updates();
protected:
// -- utility functions ------------------------------------------------------
/// Handles an I/O event on given manager.
void handle(const socket_manager_ptr& mgr, int mask);
short handle(const socket_manager_ptr& mgr, short events, short revents);
/// Adds a new socket manager to the pollset.
void add(socket_manager_ptr mgr);
/// Writes `opcode` and pointer to `mgr` the the pipe for handling an event
/// later via the pollset updater.
void write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr);
// -- member variables -------------------------------------------------------
/// Bookkeeping data for managed sockets.
......@@ -103,9 +111,6 @@ protected:
/// order as their sockets appear in `pollset_`.
manager_list managers_;
/// Managers that updated their event mask and need updating.
manager_list dirty_managers_;
/// Stores the ID of the thread this multiplexer is running in. Set when
/// calling `init()`.
std::thread::id tid_;
......
......@@ -18,6 +18,8 @@
#pragma once
#include <string>
namespace caf {
namespace net {
......@@ -45,5 +47,7 @@ constexpr operation operator~(operation x) {
return static_cast<operation>(~static_cast<int>(x));
}
std::string to_string(operation x);
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <vector>
#include "caf/byte.hpp"
#include "caf/net/fwd.hpp"
#include "caf/span.hpp"
namespace caf {
namespace net {
/// Implements an interface for packet writing in application-layers.
class packet_writer {
public:
using buffer_type = std::vector<byte>;
virtual ~packet_writer();
/// Returns a buffer for writing header information.
virtual buffer_type next_header_buffer() = 0;
/// Returns a buffer for writing payload content.
virtual buffer_type next_payload_buffer() = 0;
/// Convenience function to write a packet consisting of multiple buffers.
/// @param buffers all buffers for the packet. The first buffer is a header
/// buffer, the other buffers are payload buffer.
/// @warning this function takes ownership of `buffers`.
template <class... Ts>
void write_packet(Ts&... buffers) {
buffer_type* bufs[] = {&buffers...};
write_impl(make_span(bufs, sizeof...(Ts)));
}
protected:
/// Implementing function for `write_packet`.
/// @param buffers a `span` containing all buffers of a packet.
virtual void write_impl(span<buffer_type*> buffers) = 0;
};
} // namespace net
} // namespace caf
......@@ -18,6 +18,8 @@
#pragma once
#include "caf/net/packet_writer.hpp"
#include "caf/byte.hpp"
#include "caf/span.hpp"
......@@ -25,9 +27,9 @@ namespace caf {
namespace net {
/// Implements the interface for transport and application policies and
/// dispatches member functions either to `decorator` or `parent`.
/// dispatches member functions either to `object` or `parent`.
template <class Object, class Parent>
class write_packet_decorator {
class packet_writer_decorator final : public packet_writer {
public:
// -- member types -----------------------------------------------------------
......@@ -37,7 +39,7 @@ public:
// -- constructors, destructors, and assignment operators --------------------
write_packet_decorator(Object& object, Parent& parent)
packet_writer_decorator(Object& object, Parent& parent)
: object_(object), parent_(parent) {
// nop
}
......@@ -56,15 +58,16 @@ public:
return parent_.manager();
}
// -- member functions -------------------------------------------------------
buffer_type next_header_buffer() override {
return transport().next_header_buffer();
}
template <class... Ts>
void write_packet(span<const byte> header, span<const byte> payload,
Ts&&... xs) {
parent_.write_packet(header, payload, std::forward<Ts>(xs)...,
object_.id());
buffer_type next_payload_buffer() override {
return transport().next_payload_buffer();
}
// -- member functions -------------------------------------------------------
void cancel_timeout(atom_value type, uint64_t id) {
parent_.cancel_timeout(type, id);
}
......@@ -74,14 +77,19 @@ public:
return parent_.set_timeout(tout, type, std::forward<Ts>(xs)...);
}
protected:
void write_impl(span<buffer_type*> buffers) override {
parent_.write_packet(object_.id(), buffers);
}
private:
Object& object_;
Parent& parent_;
};
template <class Object, class Parent>
write_packet_decorator<Object, Parent>
make_write_packet_decorator(Object& object, Parent& parent) {
packet_writer_decorator<Object, Parent>
make_packet_writer_decorator(Object& object, Parent& parent) {
return {object, parent};
}
......
......@@ -20,6 +20,7 @@
#include <array>
#include <cstdint>
#include <mutex>
#include "caf/byte.hpp"
#include "caf/net/pipe_socket.hpp"
......@@ -34,9 +35,11 @@ public:
using super = socket_manager;
using msg_buf = std::array<byte, sizeof(intptr_t) + 1>;
// -- constructors, destructors, and assignment operators --------------------
pollset_updater(pipe_socket read_handle, multiplexer_ptr parent);
pollset_updater(pipe_socket read_handle, const multiplexer_ptr& parent);
~pollset_updater() override;
......@@ -56,7 +59,7 @@ public:
void handle_error(sec code) override;
private:
std::array<byte, sizeof(intptr_t)> buf_;
msg_buf buf_;
size_t buf_size_;
};
......
......@@ -18,8 +18,6 @@
#pragma once
#include <atomic>
#include "caf/error.hpp"
#include "caf/fwd.hpp"
#include "caf/net/fwd.hpp"
......@@ -53,21 +51,26 @@ public:
}
/// Returns registered operations (read, write, or both).
operation mask() const noexcept;
operation mask() const noexcept {
return mask_;
}
/// Adds given flag(s) to the event mask and updates the parent on success.
/// Adds given flag(s) to the event mask.
/// @returns `false` if `mask() | flag == mask()`, `true` otherwise.
/// @pre `has_parent()`
/// @pre `flag != operation::none`
bool mask_add(operation flag) noexcept;
/// Deletes given flag(s) from the event mask and updates the parent on
/// success.
/// Tries to clear given flag(s) from the event mask.
/// @returns `false` if `mask() & ~flag == mask()`, `true` otherwise.
/// @pre `has_parent()`
/// @pre `flag != operation::none`
bool mask_del(operation flag) noexcept;
// -- event loop management --------------------------------------------------
void register_reading();
void register_writing();
// -- pure virtual member functions ------------------------------------------
/// Called whenever the socket received new data.
......@@ -85,7 +88,7 @@ protected:
socket handle_;
std::atomic<operation> mask_;
operation mask_;
weak_multiplexer_ptr parent_;
};
......
......@@ -48,21 +48,30 @@ error nodelay(stream_socket x, bool new_value);
/// Receives data from `x`.
/// @param x Connected endpoint.
/// @param buf Points to destination buffer.
/// @param buf_size Specifies the maximum size of the buffer in bytes.
/// @returns The number of received bytes on success, an error code otherwise.
/// @relates pipe_socket
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> read(stream_socket x, span<byte> buf);
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @param buf Points to the message to send.
/// @param buf_size Specifies the size of the buffer in bytes.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates pipe_socket
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
variant<size_t, sec> write(stream_socket x, span<const byte> buf);
/// Transmits data from `x` to its peer.
/// @param x Connected endpoint.
/// @param bufs Points to the message to send, scattered across up to 10
/// buffers.
/// @returns The number of written bytes on success, otherwise an error code.
/// @relates stream_socket
/// @post either the result is a `sec` or a positive (non-zero) integer
/// @pre `bufs.size() < 10`
variant<size_t, sec> write(stream_socket x,
std::initializer_list<span<const byte>> bufs);
/// Converts the result from I/O operation on a ::stream_socket to either an
/// error code or a non-zero positive integer.
/// @relates stream_socket
......
......@@ -18,10 +18,13 @@
#pragma once
#include "caf/actor_system_config.hpp"
#include "caf/byte.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
#include "caf/net/defaults.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/net/stream_socket.hpp"
......@@ -45,6 +48,10 @@ public:
using worker_type = transport_worker<application_type>;
using buffer_type = std::vector<byte>;
using buffer_cache_type = std::vector<buffer_type>;
// -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application)
......@@ -87,9 +94,15 @@ public:
template <class Parent>
error init(Parent& parent) {
manager_ = &parent;
auto& cfg = system().config();
auto max_header_bufs = get_or(cfg, "middleman.max-header-buffers",
defaults::middleman::max_header_buffers);
header_bufs_.reserve(max_header_bufs);
auto max_payload_bufs = get_or(cfg, "middleman.max-payload-buffers",
defaults::middleman::max_payload_buffers);
payload_bufs_.reserve(max_payload_bufs);
if (auto err = worker_.init(*this))
return err;
parent.mask_add(operation::read);
return none;
}
......@@ -104,16 +117,21 @@ public:
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
collected_ += *num_bytes;
if (collected_ >= read_threshold_) {
worker_.handle_data(*this, read_buf_);
if (auto err = worker_.handle_data(*this, read_buf_)) {
CAF_LOG_WARNING("handle_data failed:" << CAF_ARG(err));
return false;
}
prepare_next_read();
}
return true;
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("receive failed" << CAF_ARG(err));
worker_.handle_error(err);
return false;
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("receive failed" << CAF_ARG(err));
worker_.handle_error(err);
return false;
}
}
return true;
}
template <class Parent>
......@@ -131,13 +149,19 @@ public:
}
template <class Parent>
void resolve(Parent&, const std::string& path, actor listener) {
worker_.resolve(*this, path, listener);
void resolve(Parent&, const uri& locator, const actor& listener) {
worker_.resolve(*this, locator.path(), listener);
}
template <class... Ts>
void set_timeout(uint64_t, Ts&&...) {
// nop
template <class Parent>
void new_proxy(Parent&, const node_id& peer, actor_id id) {
worker_.new_proxy(*this, peer, id);
}
template <class Parent>
void local_actor_down(Parent&, const node_id& peer, actor_id id,
error reason) {
worker_.local_actor_down(*this, peer, id, std::move(reason));
}
template <class Parent>
......@@ -145,6 +169,11 @@ public:
worker_.timeout(*this, value, id);
}
template <class... Ts>
void set_timeout(uint64_t, Ts&&...) {
// nop
}
void handle_error(sec code) {
worker_.handle_error(code);
}
......@@ -183,47 +212,96 @@ public:
prepare_next_read();
}
void write_packet(span<const byte> header, span<const byte> payload,
typename worker_type::id_type) {
if (write_buf_.empty())
manager().mask_add(operation::write);
write_buf_.insert(write_buf_.end(), header.begin(), header.end());
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
void write_packet(unit_t, span<buffer_type*> buffers) {
CAF_ASSERT(!buffers.empty());
if (write_queue_.empty())
manager().register_writing();
// By convention, the first buffer is a header buffer. Every other buffer is
// a payload buffer.
auto i = buffers.begin();
write_queue_.emplace_back(true, std::move(*(*i++)));
while (i != buffers.end())
write_queue_.emplace_back(false, std::move(*(*i++)));
}
// -- buffer management ------------------------------------------------------
buffer_type next_header_buffer() {
return next_buffer_impl(header_bufs_);
}
buffer_type next_payload_buffer() {
return next_buffer_impl(payload_bufs_);
}
private:
// -- private member functions -----------------------------------------------
// -- utility functions ------------------------------------------------------
static buffer_type next_buffer_impl(buffer_cache_type cache) {
if (cache.empty()) {
return {};
}
auto buf = std::move(cache.back());
cache.pop_back();
return buf;
}
bool write_some() {
if (write_buf_.empty())
CAF_LOG_TRACE(CAF_ARG(handle_.id));
if (write_queue_.empty())
return false;
auto len = write_buf_.size() - written_;
auto buf = write_buf_.data() + written_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
auto ret = write(handle_, make_span(buf, len));
if (auto num_bytes = get_if<size_t>(&ret)) {
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
// Update state.
written_ += *num_bytes;
if (written_ >= write_buf_.size()) {
written_ = 0;
write_buf_.clear();
return false;
// Helper function to sort empty buffers back into the right caches.
auto recycle = [&]() {
auto& front = write_queue_.front();
auto& is_header = front.first;
auto& buf = front.second;
written_ = 0;
buf.clear();
if (is_header) {
if (header_bufs_.size() < header_bufs_.capacity())
header_bufs_.emplace_back(std::move(buf));
} else if (payload_bufs_.size() < payload_bufs_.capacity()) {
payload_bufs_.emplace_back(std::move(buf));
}
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
worker_.handle_error(err);
return false;
}
return true;
write_queue_.pop_front();
};
// Write buffers from the write_queue_ for as long as possible.
do {
auto& buf = write_queue_.front().second;
CAF_ASSERT(!buf.empty());
auto data = buf.data() + written_;
auto len = buf.size() - written_;
auto write_ret = write(handle_, make_span(data, len));
if (auto num_bytes = get_if<size_t>(&write_ret)) {
CAF_LOG_DEBUG(CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
if (*num_bytes + written_ >= buf.size()) {
recycle();
written_ = 0;
} else {
written_ += *num_bytes;
return false;
}
} else {
auto err = get<sec>(write_ret);
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
worker_.handle_error(err);
return false;
}
return true;
}
} while (!write_queue_.empty());
return false;
}
worker_type worker_;
stream_socket handle_;
std::vector<byte> read_buf_;
std::vector<byte> write_buf_;
buffer_cache_type header_bufs_;
buffer_cache_type payload_bufs_;
buffer_type read_buf_;
std::deque<std::pair<bool, buffer_type>> write_queue_;
// TODO implement retries using this member!
// size_t max_consecutive_reads_;
......
......@@ -22,7 +22,7 @@
#include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/write_packet_decorator.hpp"
#include "caf/net/packet_writer_decorator.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
......@@ -64,33 +64,46 @@ public:
template <class Parent>
error init(Parent& parent) {
auto decorator = make_write_packet_decorator(*this, parent);
return application_.init(decorator);
auto writer = make_packet_writer_decorator(*this, parent);
return application_.init(writer);
}
template <class Parent>
void handle_data(Parent& parent, span<const byte> data) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.handle_data(decorator, data);
error handle_data(Parent& parent, span<const byte> data) {
auto writer = make_packet_writer_decorator(*this, parent);
return application_.handle_data(writer, data);
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.write_message(decorator, std::move(msg));
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.write_message(writer, std::move(msg));
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.resolve(decorator, path, listener);
void resolve(Parent& parent, string_view path, const actor& listener) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.resolve(writer, path, listener);
}
template <class Parent>
void new_proxy(Parent& parent, const node_id&, actor_id id) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.new_proxy(writer, id);
}
template <class Parent>
void local_actor_down(Parent& parent, const node_id&, actor_id id,
error reason) {
auto writer = make_packet_writer_decorator(*this, parent);
application_.local_actor_down(writer, id, std::move(reason));
}
template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.timeout(decorator, value, id);
auto writer = make_packet_writer_decorator(*this, parent);
application_.timeout(writer, value, id);
}
void handle_error(sec error) {
......
......@@ -24,8 +24,8 @@
#include "caf/ip_endpoint.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/fwd.hpp"
#include "caf/net/packet_writer_decorator.hpp"
#include "caf/net/transport_worker.hpp"
#include "caf/net/write_packet_decorator.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
......@@ -64,7 +64,7 @@ public:
}
template <class Parent>
void handle_data(Parent& parent, span<byte> data, id_type id) {
error handle_data(Parent& parent, span<byte> data, id_type id) {
auto it = workers_by_id_.find(id);
if (it == workers_by_id_.end()) {
// TODO: where to get node_id from here?
......@@ -72,12 +72,12 @@ public:
it = workers_by_id_.find(id);
}
auto worker = it->second;
worker->handle_data(parent, data);
return worker->handle_data(parent, data);
}
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) {
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto sender = msg->msg->sender;
if (!sender)
return;
......
......@@ -27,7 +27,8 @@ namespace net {
actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst)
: super(cfg), sf_(dst->serialize_fun()), dst_(std::move(dst)) {
// nop
CAF_ASSERT(dst_ != nullptr);
dst_->enqueue_event(node(), id());
}
actor_proxy_impl::~actor_proxy_impl() {
......@@ -37,6 +38,7 @@ actor_proxy_impl::~actor_proxy_impl() {
void actor_proxy_impl::enqueue(mailbox_element_ptr what, execution_unit*) {
CAF_PUSH_AID(0);
CAF_ASSERT(what != nullptr);
CAF_LOG_SEND_EVENT(what);
if (auto payload = sf_(home_system(), what->content()))
dst_->enqueue(std::move(what), ctrl(), std::move(*payload));
else
......@@ -44,26 +46,6 @@ void actor_proxy_impl::enqueue(mailbox_element_ptr what, execution_unit*) {
"unable to serialize payload: " << home_system().render(payload.error()));
}
bool actor_proxy_impl::add_backlink(abstract_actor* x) {
if (monitorable_actor::add_backlink(x)) {
enqueue(make_mailbox_element(ctrl(), make_message_id(), {},
link_atom::value, x->ctrl()),
nullptr);
return true;
}
return false;
}
bool actor_proxy_impl::remove_backlink(abstract_actor* x) {
if (monitorable_actor::remove_backlink(x)) {
enqueue(make_mailbox_element(ctrl(), make_message_id(), {},
unlink_atom::value, x->ctrl()),
nullptr);
return true;
}
return false;
}
void actor_proxy_impl::kill_proxy(execution_unit* ctx, error rsn) {
cleanup(std::move(rsn), ctx);
}
......
......@@ -29,8 +29,10 @@
#include "caf/detail/parse.hpp"
#include "caf/error.hpp"
#include "caf/expected.hpp"
#include "caf/logger.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/packet_writer.hpp"
#include "caf/no_stages.hpp"
#include "caf/none.hpp"
#include "caf/sec.hpp"
......@@ -42,11 +44,83 @@ namespace caf {
namespace net {
namespace basp {
application::application(proxy_registry_ptr proxies)
: proxies_(std::move(proxies)) {
application::application(proxy_registry& proxies) : proxies_(proxies) {
// nop
}
error application::write_message(
packet_writer& writer, std::unique_ptr<endpoint_manager_queue::message> ptr) {
CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->msg != nullptr);
CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content()));
auto payload_prefix = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{system(), payload_prefix};
const auto& src = ptr->msg->sender;
const auto& dst = ptr->receiver;
if (dst == nullptr) {
// TODO: valid?
return none;
}
if (src != nullptr) {
auto src_id = src->id();
system().registry().put(src_id, src);
if (auto err = sink(src->node(), src_id, dst->id(), ptr->msg->stages))
return err;
} else {
if (auto err = sink(node_id{}, actor_id{0}, dst->id(), ptr->msg->stages))
return err;
}
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::actor_message,
static_cast<uint32_t>(payload_prefix.size()
+ ptr->payload.size()),
ptr->msg->mid.integer_value()},
hdr);
writer.write_packet(hdr, payload_prefix, ptr->payload);
return none;
}
void application::resolve(packet_writer& writer, string_view path,
const actor& listener) {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path" << CAF_ARG(err));
return;
}
auto req_id = next_request_id_++;
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::resolve_request,
static_cast<uint32_t>(payload.size()), req_id},
hdr);
writer.write_packet(hdr, payload);
response_promise rp{nullptr, actor_cast<strong_actor_ptr>(listener),
no_stages, make_message_id()};
pending_resolves_.emplace(req_id, std::move(rp));
}
void application::new_proxy(packet_writer& writer, actor_id id) {
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::monitor_message, 0, static_cast<uint64_t>(id)},
hdr);
writer.write_packet(hdr);
}
void application::local_actor_down(packet_writer& writer, actor_id id,
error reason) {
auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{system(), payload};
if (auto err = sink(reason))
CAF_RAISE_ERROR("unable to serialize an error");
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::down_message,
static_cast<uint32_t>(payload.size()),
static_cast<uint64_t>(id)},
hdr);
writer.write_packet(hdr, payload);
}
expected<std::vector<byte>> application::serialize(actor_system& sys,
const type_erased_tuple& x) {
std::vector<byte> result;
......@@ -57,6 +131,7 @@ expected<std::vector<byte>> application::serialize(actor_system& sys,
}
strong_actor_ptr application::resolve_local_path(string_view path) {
CAF_LOG_TRACE(CAF_ARG(path));
// We currently support two path formats: `id/<actor_id>` and `name/<atom>`.
static constexpr string_view id_prefix = "id/";
if (starts_with(path, id_prefix)) {
......@@ -77,29 +152,9 @@ strong_actor_ptr application::resolve_local_path(string_view path) {
return nullptr;
}
void application::resolve_remote_path(write_packet_callback& write_packet,
string_view path, actor listener) {
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path");
return;
}
auto req_id = next_request_id_++;
auto hdr = to_bytes(header{message_type::resolve_request,
static_cast<uint32_t>(buf_.size()), req_id});
if (auto err = write_packet(hdr, buf_)) {
CAF_LOG_ERROR("unable to write resolve_request header");
return;
}
response_promise rp{nullptr, actor_cast<strong_actor_ptr>(listener),
no_stages, make_message_id()};
pending_resolves_.emplace(req_id, std::move(rp));
}
error application::handle(size_t& next_read_size,
write_packet_callback& write_packet,
error application::handle(size_t& next_read_size, packet_writer& writer,
byte_span bytes) {
CAF_LOG_TRACE(CAF_ARG(state_) << CAF_ARG2("bytes.size", bytes.size()));
switch (state_) {
case connection_state::await_handshake_header: {
if (bytes.size() != header_size)
......@@ -116,7 +171,7 @@ error application::handle(size_t& next_read_size,
return none;
}
case connection_state::await_handshake_payload: {
if (auto err = handle_handshake(write_packet, hdr_, bytes))
if (auto err = handle_handshake(writer, hdr_, bytes))
return err;
state_ = connection_state::await_header;
return none;
......@@ -126,7 +181,7 @@ error application::handle(size_t& next_read_size,
return ec::unexpected_number_of_bytes;
hdr_ = header::from_bytes(bytes);
if (hdr_.payload_len == 0)
return handle(write_packet, hdr_, byte_span{});
return handle(writer, hdr_, byte_span{});
next_read_size = hdr_.payload_len;
state_ = connection_state::await_payload;
return none;
......@@ -135,24 +190,29 @@ error application::handle(size_t& next_read_size,
if (bytes.size() != hdr_.payload_len)
return ec::unexpected_number_of_bytes;
state_ = connection_state::await_header;
return handle(write_packet, hdr_, bytes);
return handle(writer, hdr_, bytes);
}
default:
return ec::illegal_state;
}
}
error application::handle(write_packet_callback& write_packet, header hdr,
error application::handle(packet_writer& writer, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
switch (hdr.type) {
case message_type::handshake:
return ec::unexpected_handshake;
case message_type::actor_message:
return handle_actor_message(write_packet, hdr, payload);
return handle_actor_message(writer, hdr, payload);
case message_type::resolve_request:
return handle_resolve_request(write_packet, hdr, payload);
return handle_resolve_request(writer, hdr, payload);
case message_type::resolve_response:
return handle_resolve_response(write_packet, hdr, payload);
return handle_resolve_response(writer, hdr, payload);
case message_type::monitor_message:
return handle_monitor_message(writer, hdr, payload);
case message_type::down_message:
return handle_down_message(writer, hdr, payload);
case message_type::heartbeat:
return none;
default:
......@@ -160,15 +220,16 @@ error application::handle(write_packet_callback& write_packet, header hdr,
}
}
error application::handle_handshake(write_packet_callback&, header hdr,
error application::handle_handshake(packet_writer&, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
if (hdr.type != message_type::handshake)
return ec::missing_handshake;
if (hdr.operation_data != version)
return ec::version_mismatch;
node_id peer_id;
std::vector<std::string> app_ids;
binary_deserializer source{system(), payload};
binary_deserializer source{&executor_, payload};
if (auto err = source(peer_id, app_ids))
return err;
if (!peer_id || app_ids.empty())
......@@ -185,15 +246,16 @@ error application::handle_handshake(write_packet_callback&, header hdr,
return none;
}
error application::handle_actor_message(write_packet_callback&, header hdr,
error application::handle_actor_message(packet_writer&, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
// Deserialize payload.
actor_id src_id;
actor_id src_id = 0;
node_id src_node;
actor_id dst_id;
actor_id dst_id = 0;
std::vector<strong_actor_ptr> fwd_stack;
message content;
binary_deserializer source{system(), payload};
binary_deserializer source{&executor_, payload};
if (auto err = source(src_node, src_id, dst_id, fwd_stack, content))
return err;
// Sanity checks.
......@@ -208,7 +270,7 @@ error application::handle_actor_message(write_packet_callback&, header hdr,
// Try to fetch the sender.
strong_actor_ptr src_hdl;
if (src_node != none && src_id != 0)
src_hdl = proxies_->get_or_put(src_node, src_id);
src_hdl = proxies_.get_or_put(src_node, src_id);
// Ship the message.
auto ptr = make_mailbox_element(std::move(src_hdl),
make_message_id(hdr.operation_data),
......@@ -217,14 +279,15 @@ error application::handle_actor_message(write_packet_callback&, header hdr,
return none;
}
error application::handle_resolve_request(write_packet_callback& write_packet,
header hdr, byte_span payload) {
CAF_ASSERT(hdr.type == message_type::resolve_request);
error application::handle_resolve_request(packet_writer& writer, header rec_hdr,
byte_span received) {
CAF_LOG_TRACE(CAF_ARG(rec_hdr) << CAF_ARG2("received.size", received.size()));
CAF_ASSERT(rec_hdr.type == message_type::resolve_request);
size_t path_size = 0;
binary_deserializer source{system(), payload};
binary_deserializer source{&executor_, received};
if (auto err = source.begin_sequence(path_size))
return err;
// We expect the payload to consist only of the path.
// We expect the received buffer to contain the path only.
if (path_size != source.remaining())
return ec::invalid_payload;
auto remainder = source.remainder();
......@@ -232,23 +295,34 @@ error application::handle_resolve_request(write_packet_callback& write_packet,
remainder.size()};
// Write result.
auto result = resolve_local_path(path);
buf_.clear();
actor_id aid = result ? result->id() : 0;
actor_id aid;
std::set<std::string> ifs;
if (result) {
aid = result->id();
system().registry().put(aid, result);
} else {
aid = 0;
}
// TODO: figure out how to obtain messaging interface.
serializer_impl<buffer_type> sink{system(), buf_};
auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(aid, ifs))
return err;
auto out_hdr = to_bytes(header{message_type::resolve_response,
static_cast<uint32_t>(buf_.size()),
hdr.operation_data});
return write_packet(out_hdr, buf_);
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::resolve_response,
static_cast<uint32_t>(payload.size()),
rec_hdr.operation_data},
hdr);
writer.write_packet(hdr, payload);
return none;
}
error application::handle_resolve_response(write_packet_callback&, header hdr,
byte_span payload) {
CAF_ASSERT(hdr.type == message_type::resolve_response);
auto i = pending_resolves_.find(hdr.operation_data);
error application::handle_resolve_response(packet_writer&, header received_hdr,
byte_span received) {
CAF_LOG_TRACE(CAF_ARG(received_hdr)
<< CAF_ARG2("received.size", received.size()));
CAF_ASSERT(received_hdr.type == message_type::resolve_response);
auto i = pending_resolves_.find(received_hdr.operation_data);
if (i == pending_resolves_.end()) {
CAF_LOG_ERROR("received unknown ID in resolve_response message");
return none;
......@@ -260,20 +334,62 @@ error application::handle_resolve_response(write_packet_callback&, header hdr,
});
actor_id aid;
std::set<std::string> ifs;
binary_deserializer source{system(), payload};
binary_deserializer source{&executor_, received};
if (auto err = source(aid, ifs))
return err;
if (aid == 0) {
i->second.deliver(strong_actor_ptr{nullptr}, std::move(ifs));
return none;
}
i->second.deliver(proxies_->get_or_put(peer_id_, aid), std::move(ifs));
i->second.deliver(proxies_.get_or_put(peer_id_, aid), std::move(ifs));
return none;
}
error application::handle_monitor_message(packet_writer& writer,
header received_hdr,
byte_span received) {
CAF_LOG_TRACE(CAF_ARG(received_hdr)
<< CAF_ARG2("received.size", received.size()));
if (!received.empty())
return ec::unexpected_payload;
auto aid = static_cast<actor_id>(received_hdr.operation_data);
auto hdl = system().registry().get(aid);
if (hdl != nullptr) {
endpoint_manager_ptr mgr = manager_;
auto nid = peer_id_;
hdl->get()->attach_functor([mgr, nid, aid](error reason) mutable {
mgr->enqueue_event(std::move(nid), aid, std::move(reason));
});
} else {
error reason = exit_reason::unknown;
auto payload = writer.next_payload_buffer();
serializer_impl<buffer_type> sink{&executor_, payload};
if (auto err = sink(reason))
return err;
auto hdr = writer.next_header_buffer();
to_bytes(header{message_type::down_message,
static_cast<uint32_t>(payload.size()),
received_hdr.operation_data},
hdr);
writer.write_packet(hdr, payload);
}
return none;
}
error application::handle_down_message(packet_writer&, header received_hdr,
byte_span received) {
CAF_LOG_TRACE(CAF_ARG(received_hdr)
<< CAF_ARG2("received.size", received.size()));
error reason;
binary_deserializer source{&executor_, received};
if (auto err = source(reason))
return err;
proxies_.erase(peer_id_, received_hdr.operation_data, std::move(reason));
return none;
}
error application::generate_handshake() {
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
error application::generate_handshake(std::vector<byte>& buf) {
serializer_impl<buffer_type> sink{&executor_, buf};
return sink(system().node(),
get_or(system().config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers));
......
......@@ -27,6 +27,7 @@ namespace caf {
namespace detail {
void convert(const ip_endpoint& src, sockaddr_storage& dst) {
memset(&dst, 0, sizeof(sockaddr_storage));
if (src.address().embeds_v4()) {
auto sockaddr4 = reinterpret_cast<sockaddr_in*>(&dst);
sockaddr4->sin_family = AF_INET;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* 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/net/defaults.hpp"
namespace caf {
namespace defaults {
namespace middleman {
const size_t max_payload_buffers = 100;
const size_t max_header_buffers = 10;
} // namespace middleman
} // namespace defaults
} // namespace caf
......@@ -26,31 +26,6 @@ namespace caf {
namespace net {
namespace basp {
namespace {
string_view ec_names[] = {
"none",
"invalid_magic_number",
"unexpected_number_of_bytes",
"unexpected_payload",
"missing_payload",
"illegal_state",
"invalid_handshake",
"missing_handshake",
"unexpected_handshake",
"version_mismatch",
"unimplemented",
"app_identifiers_mismatch",
"invalid_payload",
};
} // namespace
std::string to_string(ec x) {
auto result = ec_names[static_cast<uint8_t>(x)];
return std::string{result.begin(), result.end()};
}
error make_error(ec x) {
return {static_cast<uint8_t>(x), atom("basp")};
}
......
......@@ -20,82 +20,71 @@
#include "caf/byte.hpp"
#include "caf/intrusive/inbox_result.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp"
#include "caf/send.hpp"
namespace caf {
namespace net {
endpoint_manager::event::event(std::string path, actor listener)
: value(resolve_request{std::move(path), std::move(listener)}) {
// nop
}
endpoint_manager::event::event(atom_value type, uint64_t id)
: value(timeout{type, id}) {
// nop
}
endpoint_manager::message::message(mailbox_element_ptr msg,
strong_actor_ptr receiver,
std::vector<byte> payload)
: msg(std::move(msg)),
receiver(std::move(receiver)),
payload(std::move(payload)) {
// nop
}
endpoint_manager::endpoint_manager(socket handle, const multiplexer_ptr& parent,
actor_system& sys)
: super(handle, parent),
sys_(sys),
events_(event_policy{}),
messages_(message_policy{}) {
events_.try_block();
messages_.try_block();
: super(handle, parent), sys_(sys), queue_(unit, unit, unit) {
queue_.try_block();
}
endpoint_manager::~endpoint_manager() {
// nop
}
std::unique_ptr<endpoint_manager::message> endpoint_manager::next_message() {
if (messages_.blocked())
endpoint_manager_queue::message_ptr endpoint_manager::next_message() {
if (queue_.blocked())
return nullptr;
messages_.fetch_more();
auto& q = messages_.queue();
queue_.fetch_more();
auto& q = std::get<1>(queue_.queue().queues());
auto ts = q.next_task_size();
if (ts == 0)
return nullptr;
q.inc_deficit(ts);
auto result = q.next();
if (q.empty())
messages_.try_block();
if (queue_.empty())
queue_.try_block();
return result;
}
void endpoint_manager::resolve(std::string path, actor listener) {
void endpoint_manager::resolve(uri locator, actor listener) {
using intrusive::inbox_result;
auto ptr = new event(std::move(path), std::move(listener));
switch (events_.push_back(ptr)) {
default:
break;
case inbox_result::unblocked_reader:
mask_add(operation::write);
break;
case inbox_result::queue_closed:
anon_send(listener, resolve_atom::value,
make_error(sec::request_receiver_down));
}
using event_type = endpoint_manager_queue::event;
auto ptr = new event_type(std::move(locator), listener);
if (!enqueue(ptr))
anon_send(listener, resolve_atom::value,
make_error(sec::request_receiver_down));
}
void endpoint_manager::enqueue(mailbox_element_ptr msg,
strong_actor_ptr receiver,
std::vector<byte> payload) {
auto ptr = new message(std::move(msg), std::move(receiver),
std::move(payload));
if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader)
mask_add(operation::write);
using message_type = endpoint_manager_queue::message;
auto ptr = new message_type(std::move(msg), std::move(receiver),
std::move(payload));
enqueue(ptr);
}
bool endpoint_manager::enqueue(endpoint_manager_queue::element* ptr) {
switch (queue_.push_back(ptr)) {
case intrusive::inbox_result::success:
return true;
case intrusive::inbox_result::unblocked_reader: {
auto mpx = parent_.lock();
if (mpx) {
mpx->register_writing(this);
return true;
}
return false;
}
default:
return false;
}
}
} // namespace net
......
......@@ -28,6 +28,18 @@ namespace caf {
namespace net {
namespace basp {
namespace {
void to_bytes_impl(const header& x, byte* ptr) {
*ptr = static_cast<byte>(x.type);
auto payload_len = detail::to_network_order(x.payload_len);
memcpy(ptr + 1, &payload_len, sizeof(payload_len));
auto operation_data = detail::to_network_order(x.operation_data);
memcpy(ptr + 5, &operation_data, sizeof(operation_data));
}
} // namespace
int header::compare(header other) const noexcept {
auto x = to_bytes(*this);
auto y = to_bytes(other);
......@@ -47,16 +59,16 @@ header header::from_bytes(span<const byte> bytes) {
}
std::array<byte, header_size> to_bytes(header x) {
std::array<byte, header_size> result;
auto ptr = result.data();
*ptr = static_cast<byte>(x.type);
auto payload_len = detail::to_network_order(x.payload_len);
memcpy(ptr + 1, &payload_len, sizeof(payload_len));
auto operation_data = detail::to_network_order(x.operation_data);
memcpy(ptr + 5, &operation_data, sizeof(operation_data));
std::array<byte, header_size> result{};
to_bytes_impl(x, result.data());
return result;
}
void to_bytes(header x, std::vector<byte>& buf) {
buf.resize(header_size);
to_bytes_impl(x, buf.data());
}
} // namespace basp
} // namespace net
} // namespace caf
......@@ -16,6 +16,8 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/ip.hpp"
#include <cstddef>
#include <string>
#include <utility>
......@@ -25,7 +27,10 @@
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/error.hpp"
#include "caf/ip_address.hpp"
#include "caf/ip_subnet.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/logger.hpp"
#include "caf/string_algorithms.hpp"
#include "caf/string_view.hpp"
#ifdef CAF_WINDOWS
......@@ -56,7 +61,8 @@ namespace ip {
namespace {
// Dummy port to resolve empty string with getaddrinfo.
constexpr auto dummy_port = "42";
constexpr string_view dummy_port = "42";
constexpr string_view localhost = "localhost";
void* fetch_in_addr(int family, sockaddr* addr) {
if (family == AF_INET)
......@@ -75,6 +81,86 @@ int fetch_addr_str(char (&buf)[INET6_ADDRSTRLEN], sockaddr* addr) {
: AF_UNSPEC;
}
#ifdef CAF_WINDOWS
template <class F>
void for_each_adapter(F f, bool is_link_local = false) {
using adapters_ptr = std::unique_ptr<IP_ADAPTER_ADDRESSES, void (*)(void*)>;
ULONG len = 0;
if (GetAdaptersAddresses(AF_UNSPEC, GAA_FLAG_INCLUDE_PREFIX, nullptr, nullptr,
&len)
!= ERROR_BUFFER_OVERFLOW) {
CAF_LOG_ERROR("failed to get adapter addresses buffer length");
return;
}
auto adapters = adapters_ptr{reinterpret_cast<IP_ADAPTER_ADDRESSES*>(
::malloc(len)),
free};
if (!adapters) {
CAF_LOG_ERROR("malloc failed");
return;
}
// TODO: The Microsoft WIN32 API example propopses to try three times, other
// examples online just perform the call once. If we notice the call to be
// unreliable, we might adapt that behavior.
if (GetAdaptersAddresses(AF_UNSPEC, GAA_FLAG_INCLUDE_PREFIX, nullptr,
adapters.get(), &len)
!= ERROR_SUCCESS) {
CAF_LOG_ERROR("failed to get adapter addresses");
return;
}
char ip_buf[INET6_ADDRSTRLEN];
char name_buf[HOST_NAME_MAX];
std::vector<std::pair<std::string, ip_address>> interfaces;
for (auto* it = adapters.get(); it != nullptr; it = it->Next) {
memset(name_buf, 0, HOST_NAME_MAX);
WideCharToMultiByte(CP_ACP, 0, it->FriendlyName, wcslen(it->FriendlyName),
name_buf, HOST_NAME_MAX, nullptr, nullptr);
std::string name{name_buf};
for (auto* addr = it->FirstUnicastAddress; addr != nullptr;
addr = addr->Next) {
memset(ip_buf, 0, INET6_ADDRSTRLEN);
getnameinfo(addr->Address.lpSockaddr, addr->Address.iSockaddrLength,
ip_buf, sizeof(ip_buf), nullptr, 0, NI_NUMERICHOST);
ip_address ip;
if (!is_link_local && starts_with(ip_buf, "fe80:")) {
CAF_LOG_DEBUG("skipping link-local address: " << ip_buf);
continue;
} else if (auto err = parse(ip_buf, ip))
continue;
f(name_buf, ip);
}
}
}
#else // CAF_WINDOWS
template <class F>
void for_each_adapter(F f, bool is_link_local = false) {
ifaddrs* tmp = nullptr;
if (getifaddrs(&tmp) != 0)
return;
std::unique_ptr<ifaddrs, decltype(freeifaddrs)*> addrs{tmp, freeifaddrs};
char buffer[INET6_ADDRSTRLEN];
std::vector<std::pair<std::string, ip_address>> interfaces;
for (auto i = addrs.get(); i != nullptr; i = i->ifa_next) {
auto family = fetch_addr_str(buffer, i->ifa_addr);
if (family != AF_UNSPEC) {
ip_address ip;
if (!is_link_local && starts_with(buffer, "fe80:")) {
CAF_LOG_DEBUG("skipping link-local address: " << buffer);
continue;
} else if (auto err = parse(buffer, ip)) {
CAF_LOG_ERROR("could not parse into ip address " << buffer);
continue;
}
f({i->ifa_name, strlen(i->ifa_name)}, ip);
}
}
}
#endif // CAF_WINDOWS
} // namespace
std::vector<ip_address> resolve(string_view host) {
......@@ -87,7 +173,7 @@ std::vector<ip_address> resolve(string_view host) {
addrinfo* tmp = nullptr;
std::string host_str{host.begin(), host.end()};
if (getaddrinfo(host.empty() ? nullptr : host_str.c_str(),
host.empty() ? dummy_port : nullptr, &hint, &tmp)
host.empty() ? dummy_port.data() : nullptr, &hint, &tmp)
!= 0)
return {};
std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo};
......@@ -111,6 +197,53 @@ std::vector<ip_address> resolve(string_view host) {
return results;
}
std::vector<ip_address> resolve(ip_address host) {
return resolve(to_string(host));
}
std::vector<ip_address> local_addresses(string_view host) {
ip_address host_ip;
std::vector<ip_address> results;
if (host.empty()) {
for_each_adapter(
[&](string_view, ip_address ip) { results.push_back(ip); });
} else if (host == localhost) {
auto v6_local = ip_address{{0}, {0x1}};
auto v4_local = ip_address{make_ipv4_address(127, 0, 0, 1)};
for_each_adapter([&](string_view, ip_address ip) {
if (ip == v4_local || ip == v6_local)
results.push_back(ip);
});
} else if (auto err = parse(host, host_ip)) {
for_each_adapter([&](string_view iface, ip_address ip) {
if (iface == host)
results.push_back(ip);
});
} else {
return local_addresses(host_ip);
}
return results;
}
std::vector<ip_address> local_addresses(ip_address host) {
static auto v6_any = ip_address{{0}, {0}};
static auto v4_any = ip_address{make_ipv4_address(0, 0, 0, 0)};
if (host == v4_any || host == v6_any)
return resolve("");
auto link_local = ip_address({0xfe, 0x8, 0x0, 0x0}, {0x0, 0x0, 0x0, 0x0});
auto ll_prefix = ip_subnet(link_local, 10);
// Unless explicitly specified we are going to skip link-local addresses.
auto is_link_local = ll_prefix.contains(host);
std::vector<ip_address> results;
for_each_adapter(
[&](string_view, ip_address ip) {
if (host == ip)
results.push_back(ip);
},
is_link_local);
return results;
}
std::string hostname() {
char buf[HOST_NAME_MAX + 1];
buf[HOST_NAME_MAX] = '\0';
......
......@@ -90,7 +90,6 @@ error multiplexer::init() {
auto pipe_handles = make_pipe();
if (!pipe_handles)
return std::move(pipe_handles.error());
tid_ = std::this_thread::get_id();
add(make_counted<pollset_updater>(pipe_handles->first, shared_from_this()));
write_handle_ = pipe_handles->second;
return none;
......@@ -107,24 +106,35 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) {
return i == last ? -1 : std::distance(first, i);
}
void multiplexer::update(const socket_manager_ptr& mgr) {
void multiplexer::register_reading(const socket_manager_ptr& mgr) {
if (std::this_thread::get_id() == tid_) {
auto i = std::find(dirty_managers_.begin(), dirty_managers_.end(), mgr);
if (i == dirty_managers_.end())
dirty_managers_.emplace_back(mgr);
if (mgr->mask() != operation::none) {
CAF_ASSERT(index_of(mgr) != -1);
if (mgr->mask_add(operation::read)) {
auto& fd = pollset_[index_of(mgr)];
fd.events |= input_mask;
}
} else if (mgr->mask_add(operation::read)) {
add(mgr);
}
} else {
mgr->ref();
auto value = reinterpret_cast<intptr_t>(mgr.get());
variant<size_t, sec> res;
{ // Lifetime scope of guard.
std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket)
res = write(write_handle_, as_bytes(make_span(&value, 1)));
else
res = sec::socket_invalid;
write_to_pipe(0, mgr);
}
}
void multiplexer::register_writing(const socket_manager_ptr& mgr) {
if (std::this_thread::get_id() == tid_) {
if (mgr->mask() != operation::none) {
CAF_ASSERT(index_of(mgr) != -1);
if (mgr->mask_add(operation::write)) {
auto& fd = pollset_[index_of(mgr)];
fd.events |= output_mask;
}
} else if (mgr->mask_add(operation::write)) {
add(mgr);
}
if (holds_alternative<sec>(res))
mgr->deref();
} else {
write_to_pipe(1, mgr);
}
}
......@@ -182,72 +192,96 @@ bool multiplexer::poll_once(bool blocking) {
return false;
// Scan pollset for events.
CAF_LOG_DEBUG("scan pollset for socket events");
for (size_t i = 0; i < pollset_.size() && presult > 0; ++i) {
auto& x = pollset_[i];
if (x.revents != 0) {
handle(managers_[i], x.revents);
for (size_t i = 0; i < pollset_.size() && presult > 0;) {
auto revents = pollset_[i].revents;
if (revents != 0) {
auto events = pollset_[i].events;
auto mgr = managers_[i];
auto new_events = handle(mgr, events, revents);
--presult;
if (new_events == 0) {
pollset_.erase(pollset_.begin() + i);
managers_.erase(managers_.begin() + i);
continue;
} else if (new_events != events) {
pollset_[i].events = new_events;
}
}
++i;
}
handle_updates();
return true;
}
}
void multiplexer::set_thread_id() {
tid_ = std::this_thread::get_id();
}
void multiplexer::run() {
CAF_LOG_TRACE("");
while (!pollset_.empty())
poll_once(true);
}
void multiplexer::handle_updates() {
for (auto mgr : dirty_managers_) {
auto index = index_of(mgr.get());
if (index == -1) {
add(std::move(mgr));
} else {
// Update or remove an existing manager in the pollset.
if (mgr->mask() == operation::none) {
pollset_.erase(pollset_.begin() + index);
managers_.erase(managers_.begin() + index);
} else {
pollset_[index].events = to_bitmask(mgr->mask());
}
}
}
dirty_managers_.clear();
}
void multiplexer::handle(const socket_manager_ptr& mgr, int mask) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle()) << CAF_ARG(mask));
short multiplexer::handle(const socket_manager_ptr& mgr, short events,
short revents) {
CAF_LOG_TRACE(CAF_ARG2("socket", mgr->handle()));
CAF_ASSERT(mgr != nullptr);
bool checkerror = true;
if ((mask & input_mask) != 0) {
if ((revents & input_mask) != 0) {
checkerror = false;
if (!mgr->handle_read_event())
if (!mgr->handle_read_event()) {
mgr->mask_del(operation::read);
events &= ~input_mask;
}
}
if ((mask & output_mask) != 0) {
if ((revents & output_mask) != 0) {
checkerror = false;
if (!mgr->handle_write_event())
if (!mgr->handle_write_event()) {
mgr->mask_del(operation::write);
events &= ~output_mask;
}
}
if (checkerror && ((mask & error_mask) != 0)) {
if (mask & POLLNVAL)
if (checkerror && ((revents & error_mask) != 0)) {
if (revents & POLLNVAL)
mgr->handle_error(sec::socket_invalid);
else if (mask & POLLHUP)
else if (revents & POLLHUP)
mgr->handle_error(sec::socket_disconnected);
else
mgr->handle_error(sec::socket_operation_failed);
mgr->mask_del(operation::read_write);
events = 0;
}
return events;
}
void multiplexer::add(socket_manager_ptr mgr) {
CAF_ASSERT(index_of(mgr) == -1);
pollfd new_entry{socket_cast<socket_id>(mgr->handle()),
to_bitmask(mgr->mask()), 0};
pollset_.emplace_back(new_entry);
managers_.emplace_back(std::move(mgr));
}
void multiplexer::write_to_pipe(uint8_t opcode, const socket_manager_ptr& mgr) {
CAF_ASSERT(opcode == 0 || opcode == 1);
CAF_ASSERT(mgr != nullptr);
pollset_updater::msg_buf buf;
mgr->ref();
buf[0] = static_cast<byte>(opcode);
auto value = reinterpret_cast<intptr_t>(mgr.get());
memcpy(buf.data() + 1, &value, sizeof(intptr_t));
variant<size_t, sec> res;
{ // Lifetime scope of guard.
std::lock_guard<std::mutex> guard{write_lock_};
if (write_handle_ != invalid_socket)
res = write(write_handle_, make_span(buf));
else
res = sec::socket_invalid;
}
if (holds_alternative<sec>(res))
mgr->deref();
}
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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/net/backend/test.hpp"
#include "caf/expected.hpp"
#include "caf/net/actor_proxy_impl.hpp"
#include "caf/net/basp/application.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/raise_error.hpp"
#include "caf/send.hpp"
namespace caf {
namespace net {
namespace backend {
test::test(middleman& mm)
: middleman_backend("test"), mm_(mm), proxies_(mm.system(), *this) {
// nop
}
test::~test() {
// nop
}
error test::init() {
return none;
}
endpoint_manager_ptr test::peer(const node_id& id) {
return get_peer(id).second;
}
void test::resolve(const uri& locator, const actor& listener) {
auto id = locator.authority_only();
if (id)
peer(make_node_id(*id))->resolve(locator, listener);
else
anon_send(listener, error(basp::ec::invalid_locator));
}
strong_actor_ptr test::make_proxy(node_id nid, actor_id aid) {
using impl_type = actor_proxy_impl;
using hdl_type = strong_actor_ptr;
actor_config cfg;
return make_actor<impl_type, hdl_type>(aid, nid, &mm_.system(), cfg,
peer(nid));
}
void test::set_last_hop(node_id*) {
// nop
}
test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first,
stream_socket second) {
using transport_type = stream_transport<basp::application>;
nonblocking(second, true);
auto mpx = mm_.mpx();
auto mgr = make_endpoint_manager(mpx, mm_.system(),
transport_type{second,
basp::application{proxies_}});
if (auto err = mgr->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << mm_.system().render(err));
CAF_RAISE_ERROR("mgr->init() failed");
}
mpx->register_reading(mgr);
auto& result = peers_[peer_id];
result = std::make_pair(first, std::move(mgr));
return result;
}
test::peer_entry& test::get_peer(const node_id& id) {
auto i = peers_.find(id);
if (i != peers_.end())
return i->second;
auto sockets = make_stream_socket_pair();
if (!sockets) {
CAF_LOG_ERROR("make_stream_socket_pair failed: "
<< mm_.system().render(sockets.error()));
CAF_RAISE_ERROR("make_stream_socket_pair failed");
}
return emplace(id, sockets->first, sockets->second);
}
} // namespace backend
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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/net/endpoint_manager_queue.hpp"
namespace caf {
namespace net {
endpoint_manager_queue::element::~element() {
// nop
}
endpoint_manager_queue::event::event(uri locator, actor listener)
: element(element_type::event),
value(resolve_request{std::move(locator), std::move(listener)}) {
// nop
}
endpoint_manager_queue::event::event(node_id peer, actor_id proxy_id)
: element(element_type::event), value(new_proxy{peer, proxy_id}) {
// nop
}
endpoint_manager_queue::event::event(node_id observing_peer,
actor_id local_actor_id, error reason)
: element(element_type::event),
value(local_actor_down{observing_peer, local_actor_id, std::move(reason)}) {
// nop
}
endpoint_manager_queue::event::event(atom_value type, uint64_t id)
: element(element_type::event), value(timeout{type, id}) {
// nop
}
endpoint_manager_queue::event::~event() {
// nop
}
size_t endpoint_manager_queue::event::task_size() const noexcept {
return 1;
}
endpoint_manager_queue::message::message(mailbox_element_ptr msg,
strong_actor_ptr receiver,
std::vector<byte> payload)
: element(element_type::message),
msg(std::move(msg)),
receiver(std::move(receiver)),
payload(std::move(payload)) {
// nop
}
size_t endpoint_manager_queue::message::task_size() const noexcept {
return message_policy::task_size(*this);
}
endpoint_manager_queue::message::~message() {
// nop
}
} // namespace net
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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/net/middleman.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/raise_error.hpp"
#include "caf/send.hpp"
#include "caf/uri.hpp"
namespace caf {
namespace net {
middleman::middleman(actor_system& sys) : sys_(sys) {
mpx_ = std::make_shared<multiplexer>();
}
middleman::~middleman() {
// nop
}
void middleman::start() {
if (!get_or(config(), "middleman.manual-multiplexing", false)) {
auto mpx = mpx_;
auto sys_ptr = &system();
mpx_thread_ = std::thread{[mpx, sys_ptr] {
CAF_SET_LOGGER_SYS(sys_ptr);
detail::set_thread_name("caf.multiplexer");
sys_ptr->thread_started();
mpx->set_thread_id();
mpx->run();
sys_ptr->thread_terminates();
}};
}
}
void middleman::stop() {
mpx_->close_pipe();
if (mpx_thread_.joinable())
mpx_thread_.join();
}
void middleman::init(actor_system_config& cfg) {
if (auto err = mpx_->init()) {
CAF_LOG_ERROR("mgr->init() failed: " << system().render(err));
CAF_RAISE_ERROR("mpx->init failed");
}
if (auto node_uri = get_if<uri>(&cfg, "middleman.this-node")) {
auto this_node = make_node_id(std::move(*node_uri));
sys_.node_.swap(this_node);
} else {
CAF_RAISE_ERROR("no valid entry for middleman.this-node found");
}
for (auto& backend : backends_)
if (auto err = backend->init()) {
CAF_LOG_ERROR("failed to initialize backend: " << system().render(err));
CAF_RAISE_ERROR("failed to initialize backend");
}
}
middleman::module::id_t middleman::id() const {
return module::network_manager;
}
void* middleman::subtype_ptr() {
return this;
}
void middleman::resolve(const uri& locator, const actor& listener) {
auto ptr = backend(locator.scheme());
if (ptr != nullptr)
ptr->resolve(locator, listener);
else
anon_send(listener, error{basp::ec::invalid_scheme});
}
middleman_backend* middleman::backend(string_view scheme) const noexcept {
auto predicate = [&](const middleman_backend_ptr& ptr) {
return ptr->id() == scheme;
};
auto i = std::find_if(backends_.begin(), backends_.end(), predicate);
if (i != backends_.end())
return i->get();
return nullptr;
}
} // namespace net
} // namespace caf
......@@ -16,28 +16,18 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/middleman_backend.hpp"
namespace caf {
namespace net {
namespace basp {
namespace {
const char* connection_state_names[] = {
"await_handshake_header",
"await_handshake_payload",
"await_header",
"await_payload",
"shutdown",
};
} // namespace
middleman_backend::middleman_backend(std::string id) : id_(std::move(id)) {
// nop
}
std::string to_string(connection_state x) {
return connection_state_names[static_cast<uint8_t>(x)];
middleman_backend::~middleman_backend() {
// nop
}
} // namespace basp
} // namespace net
} // namespace caf
......@@ -16,28 +16,14 @@
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/net/basp/message_type.hpp"
#include "caf/string_view.hpp"
#include "caf/net/packet_writer.hpp"
namespace caf {
namespace net {
namespace basp {
namespace {
string_view message_type_names[] = {
"handshake", "actor_message", "resolve_request", "resolve_response",
"monitor_message", "down_message", "heartbeat",
};
} // namespace
std::string to_string(message_type x) {
auto result = message_type_names[static_cast<uint8_t>(x)];
return std::string{result.begin(), result.end()};
packet_writer::~packet_writer() {
// nop
}
} // namespace basp
} // namespace net
} // namespace caf
......@@ -27,8 +27,8 @@ namespace caf {
namespace net {
pollset_updater::pollset_updater(pipe_socket read_handle,
multiplexer_ptr parent)
: super(read_handle, std::move(parent)), buf_size_(0) {
const multiplexer_ptr& parent)
: super(read_handle, parent), buf_size_(0) {
mask_ = operation::read;
nonblocking(read_handle, true);
}
......@@ -45,10 +45,18 @@ bool pollset_updater::handle_read_event() {
buf_size_ += *num_bytes;
if (buf_.size() == buf_size_) {
buf_size_ = 0;
auto value = *reinterpret_cast<intptr_t*>(buf_.data());
auto opcode = static_cast<uint8_t>(buf_[0]);
intptr_t value;
memcpy(&value, buf_.data() + 1, sizeof(intptr_t));
socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false};
if (auto ptr = parent_.lock())
ptr->update(mgr);
if (auto ptr = parent_.lock()) {
if (opcode == 0) {
ptr->register_reading(mgr);
} else {
CAF_ASSERT(opcode == 1);
ptr->register_writing(mgr);
}
}
}
} else {
return get<sec>(res) == sec::unavailable_or_would_block;
......
......@@ -34,32 +34,38 @@ socket_manager::~socket_manager() {
close(handle_);
}
operation socket_manager::mask() const noexcept {
return mask_.load();
}
bool socket_manager::mask_add(operation flag) noexcept {
CAF_ASSERT(flag != operation::none);
auto x = mask();
while ((x & flag) != flag)
if (mask_.compare_exchange_strong(x, x | flag)) {
if (auto ptr = parent_.lock())
ptr->update(this);
return true;
}
return false;
if ((x & flag) == flag)
return false;
mask_ = x | flag;
return true;
}
bool socket_manager::mask_del(operation flag) noexcept {
CAF_ASSERT(flag != operation::none);
auto x = mask();
while ((x & flag) != operation::none)
if (mask_.compare_exchange_strong(x, x & ~flag)) {
if (auto ptr = parent_.lock())
ptr->update(this);
return true;
}
return false;
if ((x & flag) == operation::none)
return false;
mask_ = x & ~flag;
return true;
}
void socket_manager::register_reading() {
if ((mask() & operation::read) == operation::read)
return;
auto ptr = parent_.lock();
if (ptr != nullptr)
ptr->register_reading(this);
}
void socket_manager::register_writing() {
if ((mask() & operation::write) == operation::write)
return;
auto ptr = parent_.lock();
if (ptr != nullptr)
ptr->register_writing(this);
}
} // namespace net
......
......@@ -27,6 +27,10 @@
#include "caf/span.hpp"
#include "caf/variant.hpp"
#ifdef CAF_POSIX
# include <sys/uio.h>
#endif
namespace caf {
namespace net {
......@@ -175,6 +179,47 @@ variant<size_t, sec> write(stream_socket x, span<const byte> buf) {
return check_stream_socket_io_res(res);
}
#ifdef CAF_WINDOWS
variant<size_t, sec> write(stream_socket x,
std::initializer_list<span<const byte>> bufs) {
CAF_ASSERT(bufs.size() < 10);
WSABUF buf_array[10];
auto convert = [](span<const byte> buf) {
auto data = const_cast<byte*>(buf.data());
return WSABUF{static_cast<ULONG>(buf.size()),
reinterpret_cast<CHAR*>(data)};
};
std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert);
DWORD bytes_sent = 0;
auto res = WSASend(x.id, buf_array, static_cast<DWORD>(bufs.size()),
&bytes_sent, 0, nullptr, nullptr);
if (res != 0) {
auto code = last_socket_error();
if (code == std::errc::operation_would_block
|| code == std::errc::resource_unavailable_try_again)
return sec::unavailable_or_would_block;
return sec::socket_operation_failed;
}
return static_cast<size_t>(bytes_sent);
}
#else // CAF_WINDOWS
variant<size_t, sec> write(stream_socket x,
std::initializer_list<span<const byte>> bufs) {
CAF_ASSERT(bufs.size() < 10);
iovec buf_array[10];
auto convert = [](span<const byte> buf) {
return iovec{const_cast<byte*>(buf.data()), buf.size()};
};
std::transform(bufs.begin(), bufs.end(), std::begin(buf_array), convert);
auto res = writev(x.id, buf_array, static_cast<int>(bufs.size()));
return check_stream_socket_io_res(res);
}
#endif // CAF_WINDOWS
variant<size_t, sec>
check_stream_socket_io_res(std::make_signed<size_t>::type res) {
if (res == 0)
......
......@@ -100,9 +100,10 @@ variant<std::pair<size_t, ip_endpoint>, sec> read(udp_datagram_socket x,
<< CAF_ARG(buf.size()) << " of " << CAF_ARG(num_bytes)
<< " bytes");
ip_endpoint ep;
// TODO: how to properly pass error further?
if (auto err = detail::convert(addr, ep))
return sec::runtime_error;
if (auto err = detail::convert(addr, ep)) {
CAF_ASSERT(err.category() == atom("system"));
return static_cast<sec>(err.code());
}
return std::pair<size_t, ip_endpoint>(*num_bytes, ep);
} else {
return get<sec>(ret);
......
......@@ -29,6 +29,7 @@
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/packet_writer.hpp"
#include "caf/none.hpp"
#include "caf/uri.hpp"
......@@ -41,10 +42,13 @@ using namespace caf::net;
namespace {
struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
struct fixture : test_coordinator_fixture<>,
proxy_registry::backend,
basp::application::test_tag,
public packet_writer {
using buffer_type = std::vector<byte>;
fixture() : app(std::make_shared<proxy_registry>(sys, *this)) {
fixture() : proxies(sys, *this), app(proxies) {
REQUIRE_OK(app.init(*this));
uri mars_uri;
REQUIRE_OK(parse("tcp://mars", mars_uri));
......@@ -64,11 +68,6 @@ struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
input = to_buf(xs...);
}
void write_packet(span<const byte> hdr, span<const byte> payload) {
output.insert(output.end(), hdr.begin(), hdr.end());
output.insert(output.end(), payload.begin(), payload.end());
}
void handle_handshake() {
CAF_CHECK_EQUAL(app.state(),
basp::connection_state::await_handshake_header);
......@@ -108,6 +107,18 @@ struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
return *this;
}
endpoint_manager& manager() {
CAF_FAIL("unexpected function call");
}
buffer_type next_payload_buffer() override {
return {};
}
buffer_type next_header_buffer() override {
return {};
}
template <class... Ts>
void configure_read(Ts...) {
// nop
......@@ -124,12 +135,20 @@ struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
// nop
}
protected:
void write_impl(span<buffer_type*> buffers) override {
for (auto buf : buffers)
output.insert(output.end(), buf->begin(), buf->end());
}
buffer_type input;
buffer_type output;
node_id mars;
proxy_registry proxies;
basp::application app;
};
......
......@@ -20,10 +20,9 @@
#include "caf/detail/convert_ip_endpoint.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <cstring>
#include "caf/detail/socket_sys_includes.hpp"
......@@ -37,22 +36,21 @@ namespace {
struct fixture : host_fixture {
fixture() : host_fixture() {
memset(&sockaddr6_src, 0, sizeof(sockaddr_in6));
memset(&sockaddr6_dst, 0, sizeof(sockaddr_in6));
sockaddr6_src.sin6_family = AF_INET6;
sockaddr6_src.sin6_port = htons(23);
sockaddr6_src.sin6_addr = in6addr_loopback;
memset(&sockaddr4_src, 0, sizeof(sockaddr_in));
memset(&sockaddr4_dst, 0, sizeof(sockaddr_in));
sockaddr4_src.sin_family = AF_INET;
sockaddr4_src.sin_port = htons(23);
sockaddr4_src.sin_addr.s_addr = INADDR_LOOPBACK;
memset(&sockaddr4_src, 0, sizeof(sockaddr_storage));
memset(&sockaddr6_src, 0, sizeof(sockaddr_storage));
auto sockaddr6_ptr = reinterpret_cast<sockaddr_in6*>(&sockaddr6_src);
sockaddr6_ptr->sin6_family = AF_INET6;
sockaddr6_ptr->sin6_port = htons(23);
sockaddr6_ptr->sin6_addr = in6addr_loopback;
auto sockaddr4_ptr = reinterpret_cast<sockaddr_in*>(&sockaddr4_src);
sockaddr4_ptr->sin_family = AF_INET;
sockaddr4_ptr->sin_port = htons(23);
sockaddr4_ptr->sin_addr.s_addr = INADDR_LOOPBACK;
}
sockaddr_in6 sockaddr6_src;
sockaddr_in6 sockaddr6_dst;
sockaddr_in sockaddr4_src;
sockaddr_in sockaddr4_dst;
sockaddr_storage sockaddr6_src;
sockaddr_storage sockaddr4_src;
sockaddr_storage dst;
ip_endpoint ep_src;
ip_endpoint ep_dst;
};
......@@ -62,51 +60,43 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(convert_ip_endpoint_tests, fixture)
CAF_TEST(sockaddr_in6 roundtrip) {
ip_endpoint ep;
ip_endpoint tmp;
CAF_MESSAGE("converting sockaddr_in6 to ip_endpoint");
CAF_CHECK_EQUAL(convert(reinterpret_cast<sockaddr_storage&>(sockaddr6_src),
ep),
none);
CAF_CHECK_EQUAL(convert(sockaddr6_src, tmp), none);
CAF_MESSAGE("converting ip_endpoint to sockaddr_in6");
convert(ep, reinterpret_cast<sockaddr_storage&>(sockaddr6_dst));
CAF_CHECK_EQUAL(memcmp(&sockaddr6_src, &sockaddr6_dst, sizeof(sockaddr_in6)),
0);
convert(tmp, dst);
CAF_CHECK_EQUAL(memcmp(&sockaddr6_src, &dst, sizeof(sockaddr_storage)), 0);
}
CAF_TEST(ipv6_endpoint roundtrip) {
sockaddr_storage addr;
memset(&addr, 0, sizeof(sockaddr_storage));
sockaddr_storage tmp = {};
if (auto err = detail::parse("[::1]:55555", ep_src))
CAF_FAIL("unable to parse input: " << err);
CAF_MESSAGE("converting ip_endpoint to sockaddr_in6");
convert(ep_src, addr);
convert(ep_src, tmp);
CAF_MESSAGE("converting sockaddr_in6 to ip_endpoint");
CAF_CHECK_EQUAL(convert(addr, ep_dst), none);
CAF_CHECK_EQUAL(convert(tmp, ep_dst), none);
CAF_CHECK_EQUAL(ep_src, ep_dst);
}
CAF_TEST(sockaddr_in4 roundtrip) {
ip_endpoint ep;
ip_endpoint tmp;
CAF_MESSAGE("converting sockaddr_in to ip_endpoint");
CAF_CHECK_EQUAL(convert(reinterpret_cast<sockaddr_storage&>(sockaddr4_src),
ep),
none);
CAF_CHECK_EQUAL(convert(sockaddr4_src, tmp), none);
CAF_MESSAGE("converting ip_endpoint to sockaddr_in");
convert(ep, reinterpret_cast<sockaddr_storage&>(sockaddr4_dst));
CAF_CHECK_EQUAL(memcmp(&sockaddr4_src, &sockaddr4_dst, sizeof(sockaddr_in)),
0);
convert(tmp, dst);
CAF_CHECK_EQUAL(memcmp(&sockaddr4_src, &dst, sizeof(sockaddr_storage)), 0);
}
CAF_TEST(ipv4_endpoint roundtrip) {
sockaddr_storage addr;
memset(&addr, 0, sizeof(sockaddr_storage));
sockaddr_storage tmp = {};
if (auto err = detail::parse("127.0.0.1:55555", ep_src))
CAF_FAIL("unable to parse input: " << err);
CAF_MESSAGE("converting ip_endpoint to sockaddr_in");
convert(ep_src, addr);
convert(ep_src, tmp);
CAF_MESSAGE("converting sockaddr_in to ip_endpoint");
CAF_CHECK_EQUAL(convert(addr, ep_dst), none);
CAF_CHECK_EQUAL(convert(tmp, ep_dst), none);
CAF_CHECK_EQUAL(ep_src, ep_dst);
}
CAF_TEST_FIXTURE_SCOPE_END();
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -20,10 +20,9 @@
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
......@@ -46,12 +45,14 @@ string_view hello_test{"hello test!"};
struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() {
mpx = std::make_shared<multiplexer>();
mpx->set_thread_id();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
if (mpx->num_socket_managers() != 1)
CAF_FAIL("mpx->num_socket_managers() != 1");
}
bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false);
}
......@@ -87,7 +88,7 @@ public:
error init(Manager& manager) {
auto test_bytes = as_bytes(make_span(hello_test));
buf_.insert(buf_.end(), test_bytes.begin(), test_bytes.end());
CAF_CHECK(manager.mask_add(operation::read_write));
manager.register_writing();
return none;
}
......@@ -121,7 +122,7 @@ public:
}
template <class Manager>
void resolve(Manager& mgr, std::string path, actor listener) {
void resolve(Manager& mgr, const uri& locator, const actor& listener) {
actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(42, hid));
......@@ -129,6 +130,7 @@ public:
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid,
&mgr.system(), cfg,
&mgr);
std::string path{locator.path().begin(), locator.path().end()};
anon_send(listener, resolve_atom::value, std::move(path), p);
}
......@@ -137,6 +139,16 @@ public:
// nop
}
template <class Parent>
void new_proxy(Parent&, const node_id&, actor_id) {
// nop
}
template <class Parent>
void local_actor_down(Parent&, const node_id&, actor_id, error) {
// nop
}
private:
stream_socket handle_;
......@@ -153,7 +165,6 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(send and receive) {
std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true);
......@@ -162,8 +173,9 @@ CAF_TEST(send and receive) {
auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf});
CAF_CHECK_EQUAL(mgr->mask(), operation::none);
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mgr->mask(), operation::read_write);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))),
hello_manager.size());
......@@ -186,10 +198,10 @@ CAF_TEST(resolve and proxy communication) {
auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf});
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mgr->mask(), operation::read_write);
run();
CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)), hello_test.size());
mgr->resolve("/id/42", self);
mgr->resolve(unbox(make_uri("test:id/42")), self);
run();
self->receive(
[&](resolve_atom, const std::string&, const strong_actor_ptr& p) {
......
......@@ -20,38 +20,63 @@
#include "caf/net/ip.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/ip_address.hpp"
#include "caf/ipv4_address.hpp"
using namespace caf;
using namespace caf::net;
CAF_TEST_FIXTURE_SCOPE(ip_tests, host_fixture)
namespace {
struct fixture : host_fixture {
fixture() : v6_local{{0}, {0x1}} {
v4_local = ip_address{make_ipv4_address(127, 0, 0, 1)};
v4_any_addr = ip_address{make_ipv4_address(0, 0, 0, 0)};
}
bool contains(ip_address x) {
return std::count(addrs.begin(), addrs.end(), x) > 0;
}
ip_address v4_any_addr;
ip_address v6_any_addr;
ip_address v4_local;
ip_address v6_local;
std::vector<ip_address> addrs;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(ip_tests, fixture)
CAF_TEST(resolve localhost) {
ip_address v4_local{make_ipv4_address(127, 0, 0, 1)};
ip_address v6_local{{0}, {0x1}};
auto addrs = ip::resolve("localhost");
addrs = ip::resolve("localhost");
CAF_CHECK(!addrs.empty());
auto contains = [&](ip_address x) {
return std::count(addrs.begin(), addrs.end(), x) > 0;
};
CAF_CHECK(contains(v4_local) || contains(v6_local));
}
CAF_TEST(resolve any) {
ip_address v4_any{make_ipv4_address(0, 0, 0, 0)};
ip_address v6_any{{0}, {0}};
auto addrs = ip::resolve("");
addrs = ip::resolve("");
CAF_CHECK(!addrs.empty());
auto contains = [&](ip_address x) {
return std::count(addrs.begin(), addrs.end(), x) > 0;
};
CAF_CHECK(contains(v4_any) || contains(v6_any));
CAF_CHECK(contains(v4_any_addr) || contains(v6_any_addr));
}
CAF_TEST(local addresses localhost) {
addrs = ip::local_addresses("localhost");
CAF_CHECK(!addrs.empty());
CAF_CHECK(contains(v4_local) || contains(v6_local));
}
CAF_TEST(local addresses any) {
addrs = ip::local_addresses("0.0.0.0");
auto tmp = ip::local_addresses("::");
addrs.insert(addrs.end(), tmp.begin(), tmp.end());
CAF_CHECK(!addrs.empty());
CAF_CHECK(contains(v4_any_addr) || contains(v6_any_addr));
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -20,10 +20,9 @@
#include "caf/net/multiplexer.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <new>
#include <tuple>
#include <vector>
......@@ -123,7 +122,7 @@ using dummy_manager_ptr = intrusive_ptr<dummy_manager>;
struct fixture : host_fixture {
fixture() : manager_count(0), mpx(std::make_shared<multiplexer>()) {
// nop
mpx->set_thread_id();
}
~fixture() {
......@@ -165,13 +164,11 @@ CAF_TEST(send and receive) {
auto sockets = unbox(make_stream_socket_pair());
auto alice = make_counted<dummy_manager>(manager_count, sockets.first, mpx);
auto bob = make_counted<dummy_manager>(manager_count, sockets.second, mpx);
alice->mask_add(operation::read);
bob->mask_add(operation::read);
mpx->handle_updates();
alice->register_reading();
bob->register_reading();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
alice->send("hello bob");
alice->mask_add(operation::write);
mpx->handle_updates();
alice->register_writing();
exhaust();
CAF_CHECK_EQUAL(bob->receive(), "hello bob");
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#define CAF_SUITE net.basp.ping_pong
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "caf/net/backend/test.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/multiplexer.hpp"
using namespace caf;
using namespace caf::net;
namespace {
using ping_atom = caf::atom_constant<caf::atom("ping")>;
using pong_atom = caf::atom_constant<caf::atom("pong")>;
struct earth_node {
uri operator()() {
return unbox(make_uri("test://earth"));
}
};
struct mars_node {
uri operator()() {
return unbox(make_uri("test://mars"));
}
};
template <class Node>
struct config : actor_system_config {
config() {
Node this_node;
put(content, "middleman.this-node", this_node());
load<middleman, backend::test>();
}
};
class planet_driver {
public:
virtual ~planet_driver() {
// nop
}
virtual bool consume_message() = 0;
virtual bool handle_io_event() = 0;
virtual bool trigger_timeout() = 0;
};
template <class Node>
class planet : public test_coordinator_fixture<config<Node>> {
public:
planet(planet_driver& driver)
: mpx(*this->sys.network_manager().mpx()), driver_(driver) {
mpx.set_thread_id();
}
net::backend::test& backend() {
auto& mm = this->sys.network_manager();
return *dynamic_cast<net::backend::test*>(mm.backend("test"));
}
node_id id() const {
return this->sys.node();
}
bool consume_message() override {
return driver_.consume_message();
}
bool handle_io_event() override {
return driver_.handle_io_event();
}
bool trigger_timeout() override {
return driver_.trigger_timeout();
}
actor resolve(string_view locator) {
auto hdl = actor_cast<actor>(this->self);
this->sys.network_manager().resolve(unbox(make_uri(locator)), hdl);
this->run();
actor result;
this->self->receive(
[&](strong_actor_ptr& ptr, const std::set<std::string>&) {
CAF_MESSAGE("resolved " << locator);
result = actor_cast<actor>(std::move(ptr));
});
return result;
}
multiplexer& mpx;
private:
planet_driver& driver_;
};
behavior ping_actor(event_based_actor* self, actor pong, size_t num_pings,
std::shared_ptr<size_t> count) {
CAF_MESSAGE("num_pings: " << num_pings);
self->send(pong, ping_atom::value, 1);
return {
[=](pong_atom, int value) -> std::tuple<atom_value, int> {
CAF_MESSAGE("received `pong_atom`");
if (++*count >= num_pings) {
CAF_MESSAGE("received " << num_pings << " pings, call self->quit");
self->quit();
}
return std::make_tuple(ping_atom::value, value + 1);
},
};
}
behavior pong_actor(event_based_actor* self) {
CAF_MESSAGE("pong actor started");
self->set_down_handler([=](down_msg& dm) {
CAF_MESSAGE("received down_msg{" << to_string(dm.reason) << "}");
self->quit(dm.reason);
});
return {
[=](ping_atom, int value) -> std::tuple<atom_value, int> {
CAF_MESSAGE("received `ping_atom` from " << self->current_sender());
if (self->current_sender() == self->ctrl())
abort();
self->monitor(self->current_sender());
// set next behavior
self->become([](ping_atom, int val) {
return std::make_tuple(pong_atom::value, val);
});
// reply to 'ping'
return std::make_tuple(pong_atom::value, value);
},
};
}
struct fixture : host_fixture, planet_driver {
fixture() : earth(*this), mars(*this) {
auto sockets = unbox(make_stream_socket_pair());
earth.backend().emplace(mars.id(), sockets.first, sockets.second);
mars.backend().emplace(earth.id(), sockets.second, sockets.first);
run();
}
bool consume_message() override {
return earth.sched.try_run_once() || mars.sched.try_run_once();
}
bool handle_io_event() override {
return earth.mpx.poll_once(false) || mars.mpx.poll_once(false);
}
bool trigger_timeout() override {
return earth.sched.trigger_timeout() || mars.sched.trigger_timeout();
}
void run() {
earth.run();
}
planet<earth_node> earth;
planet<mars_node> mars;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(ping_pong_tests, fixture)
CAF_TEST(full setup) {
auto pong = earth.sys.spawn(pong_actor);
run();
earth.sys.registry().put(atom("pong"), pong);
auto remote_pong = mars.resolve("test://earth/name/pong");
auto count = std::make_shared<size_t>(0);
auto ping = mars.sys.spawn(ping_actor, remote_pong, 10, count);
run();
anon_send_exit(pong, exit_reason::kill);
anon_send_exit(ping, exit_reason::kill);
CAF_CHECK_EQUAL(*count, 10u);
run();
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -20,10 +20,9 @@
#include "caf/net/network_socket.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
using namespace caf;
using namespace caf::net;
......
......@@ -20,10 +20,9 @@
#include "caf/net/pipe_socket.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <vector>
#include "caf/byte.hpp"
......
......@@ -20,10 +20,9 @@
#include "caf/net/socket.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
using namespace caf;
using namespace caf::net;
......
......@@ -20,22 +20,21 @@
#include "caf/net/basp/application.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <vector>
#include "caf/byte.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/net/backend/test.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/ec.hpp"
#include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
#include "caf/none.hpp"
#include "caf/uri.hpp"
using namespace caf;
......@@ -57,37 +56,26 @@ size_t fetch_size(variant<size_t, sec> x) {
return get<size_t>(x);
}
struct fixture : test_coordinator_fixture<>,
host_fixture,
proxy_registry::backend {
fixture() {
uri mars_uri;
REQUIRE_OK(parse("tcp://mars", mars_uri));
mars = make_node_id(mars_uri);
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
auto proxies = std::make_shared<proxy_registry>(sys, *this);
auto sockets = unbox(make_stream_socket_pair());
sock = sockets.first;
nonblocking(sockets.first, true);
nonblocking(sockets.second, true);
mgr = make_endpoint_manager(mpx, sys,
transport_type{sockets.second,
basp::application{proxies}});
REQUIRE_OK(mgr->init());
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
auto& dref = dynamic_cast<endpoint_manager_impl<transport_type>&>(*mgr);
app = &dref.application();
struct config : actor_system_config {
config() {
put(content, "middleman.this-node", unbox(make_uri("test:earth")));
load<middleman, backend::test>();
}
};
~fixture() {
close(sock);
struct fixture : host_fixture, test_coordinator_fixture<config> {
fixture() : mars(make_node_id(unbox(make_uri("test:mars")))) {
auto& mm = sys.network_manager();
mm.mpx()->set_thread_id();
auto backend = dynamic_cast<backend::test*>(mm.backend("test"));
auto mgr = backend->peer(mars);
auto& dref = dynamic_cast<endpoint_manager_impl<transport_type>&>(*mgr);
app = &dref.application();
sock = backend->socket(mars);
}
bool handle_io_event() override {
mpx->handle_updates();
auto mpx = sys.network_manager().mpx();
return mpx->poll_once(false);
}
......@@ -143,57 +131,57 @@ struct fixture : test_coordinator_fixture<>,
return sys;
}
strong_actor_ptr make_proxy(node_id nid, actor_id aid) override {
using impl_type = forwarding_actor_proxy;
using hdl_type = strong_actor_ptr;
actor_config cfg;
return make_actor<impl_type, hdl_type>(aid, nid, &sys, cfg, self);
}
void set_last_hop(node_id*) override {
// nop
}
node_id mars;
multiplexer_ptr mpx;
endpoint_manager_ptr mgr;
stream_socket sock;
basp::application* app;
unit_t no_payload;
};
} // namespace
#define MOCK(kind, op, ...) \
do { \
auto payload = to_buf(__VA_ARGS__); \
mock(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \
write(sock, make_span(payload)); \
CAF_MESSAGE("mock " << kind); \
if (!std::is_same<decltype(std::make_tuple(__VA_ARGS__)), \
std::tuple<unit_t>>::value) { \
auto payload = to_buf(__VA_ARGS__); \
mock(basp::header{kind, static_cast<uint32_t>(payload.size()), op}); \
write(sock, make_span(payload)); \
} else { \
mock(basp::header{kind, 0, op}); \
} \
run(); \
} while (false)
#define RECEIVE(msg_type, op_data, ...) \
do { \
CAF_MESSAGE("receive " << msg_type); \
buffer_type buf(basp::header_size); \
if (fetch_size(read(sock, make_span(buf))) != basp::header_size) \
CAF_FAIL("unable to read " << basp::header_size << " bytes"); \
auto hdr = basp::header::from_bytes(buf); \
CAF_CHECK_EQUAL(hdr.type, msg_type); \
CAF_CHECK_EQUAL(hdr.operation_data, op_data); \
buf.resize(hdr.payload_len); \
if (fetch_size(read(sock, make_span(buf))) != size_t{hdr.payload_len}) \
CAF_FAIL("unable to read " << hdr.payload_len << " bytes"); \
binary_deserializer source{sys, buf}; \
if (auto err = source(__VA_ARGS__)) \
CAF_FAIL("failed to receive data: " << sys.render(err)); \
if (!std::is_same<decltype(std::make_tuple(__VA_ARGS__)), \
std::tuple<unit_t>>::value) { \
buf.resize(hdr.payload_len); \
if (fetch_size(read(sock, make_span(buf))) != size_t{hdr.payload_len}) \
CAF_FAIL("unable to read " << hdr.payload_len << " bytes"); \
binary_deserializer source{sys, buf}; \
if (auto err = source(__VA_ARGS__)) \
CAF_FAIL("failed to receive data: " << sys.render(err)); \
} else { \
if (hdr.payload_len != 0) \
CAF_FAIL("unexpected payload"); \
} \
} while (false)
CAF_TEST_FIXTURE_SCOPE(application_tests, fixture)
CAF_TEST(actor message) {
CAF_TEST(actor message and down message) {
handle_handshake();
consume_handshake();
sys.registry().put(self->id(), self);
......@@ -201,9 +189,19 @@ CAF_TEST(actor message) {
MOCK(basp::message_type::actor_message, make_message_id().integer_value(),
mars, actor_id{42}, self->id(), std::vector<strong_actor_ptr>{},
make_message("hello world!"));
allow((atom_value, strong_actor_ptr),
from(_).to(self).with(atom("monitor"), _));
expect((std::string), from(_).to(self).with("hello world!"));
MOCK(basp::message_type::monitor_message, 42u, no_payload);
strong_actor_ptr proxy;
self->receive([&](const std::string& str) {
CAF_CHECK_EQUAL(str, "hello world!");
proxy = self->current_sender();
CAF_REQUIRE_NOT_EQUAL(proxy, nullptr);
self->monitor(proxy);
});
MOCK(basp::message_type::down_message, 42u,
error{exit_reason::user_shutdown});
expect((down_msg),
from(_).to(self).with(down_msg{actor_cast<actor_addr>(proxy),
exit_reason::user_shutdown}));
}
CAF_TEST(resolve request without result) {
......
......@@ -20,17 +20,24 @@
#include "caf/net/stream_socket.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/byte.hpp"
#include "caf/span.hpp"
using namespace caf;
using namespace caf::net;
CAF_TEST_FIXTURE_SCOPE(network_socket_tests, host_fixture)
namespace {
byte operator"" _b(unsigned long long x) {
return static_cast<byte>(x);
}
} // namespace
CAF_TEST_FIXTURE_SCOPE(network_socket_simple_tests, host_fixture)
CAF_TEST(invalid socket) {
stream_socket x;
......@@ -39,36 +46,72 @@ CAF_TEST(invalid socket) {
CAF_CHECK_EQUAL(allow_sigpipe(x, true), sec::network_syscall_failed);
}
CAF_TEST(connected socket pair) {
std::vector<byte> wr_buf{byte(1), byte(2), byte(4), byte(8),
byte(16), byte(32), byte(64)};
std::vector<byte> rd_buf(124);
CAF_MESSAGE("create sockets and configure nonblocking I/O");
auto x = unbox(make_stream_socket_pair());
CAF_CHECK_EQUAL(nonblocking(x.first, true), caf::none);
CAF_CHECK_EQUAL(nonblocking(x.second, true), caf::none);
CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.first)), 0u);
CAF_CHECK_NOT_EQUAL(unbox(send_buffer_size(x.second)), 0u);
CAF_MESSAGE("verify nonblocking communication");
CAF_CHECK_EQUAL(read(x.first, make_span(rd_buf)),
CAF_TEST_FIXTURE_SCOPE_END()
namespace {
struct fixture : host_fixture {
fixture() : rd_buf(124) {
std::tie(first, second) = unbox(make_stream_socket_pair());
CAF_REQUIRE_EQUAL(nonblocking(first, true), caf::none);
CAF_REQUIRE_EQUAL(nonblocking(second, true), caf::none);
CAF_REQUIRE_NOT_EQUAL(unbox(send_buffer_size(first)), 0u);
CAF_REQUIRE_NOT_EQUAL(unbox(send_buffer_size(second)), 0u);
}
~fixture() {
close(first);
close(second);
}
stream_socket first;
stream_socket second;
std::vector<byte> rd_buf;
};
} // namespace
CAF_TEST_FIXTURE_SCOPE(network_socket_tests, fixture)
CAF_TEST(read on empty sockets) {
CAF_CHECK_EQUAL(read(first, make_span(rd_buf)),
sec::unavailable_or_would_block);
CAF_CHECK_EQUAL(read(x.second, make_span(rd_buf)),
CAF_CHECK_EQUAL(read(second, make_span(rd_buf)),
sec::unavailable_or_would_block);
}
CAF_TEST(transfer data from first to second socket) {
std::vector<byte> wr_buf{1_b, 2_b, 4_b, 8_b, 16_b, 32_b, 64_b};
CAF_MESSAGE("transfer data from first to second socket");
CAF_CHECK_EQUAL(write(x.first, make_span(wr_buf)), wr_buf.size());
CAF_CHECK_EQUAL(read(x.second, make_span(rd_buf)), wr_buf.size());
CAF_CHECK_EQUAL(write(first, make_span(wr_buf)), wr_buf.size());
CAF_CHECK_EQUAL(read(second, make_span(rd_buf)), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), byte(0));
CAF_MESSAGE("transfer data from second to first socket");
CAF_CHECK_EQUAL(write(x.second, make_span(wr_buf)), wr_buf.size());
CAF_CHECK_EQUAL(read(x.first, make_span(rd_buf)), wr_buf.size());
}
CAF_TEST(transfer data from second to first socket) {
std::vector<byte> wr_buf{1_b, 2_b, 4_b, 8_b, 16_b, 32_b, 64_b};
CAF_CHECK_EQUAL(write(second, make_span(wr_buf)), wr_buf.size());
CAF_CHECK_EQUAL(read(first, make_span(rd_buf)), wr_buf.size());
CAF_CHECK(std::equal(wr_buf.begin(), wr_buf.end(), rd_buf.begin()));
rd_buf.assign(rd_buf.size(), byte(0));
CAF_MESSAGE("shut down first socket and observe shutdown on the second one");
close(x.first);
CAF_CHECK_EQUAL(read(x.second, make_span(rd_buf)), sec::socket_disconnected);
CAF_MESSAGE("done (cleanup)");
close(x.second);
}
CAF_TEST(shut down first socket and observe shutdown on the second one) {
close(first);
CAF_CHECK_EQUAL(read(second, make_span(rd_buf)), sec::socket_disconnected);
first.id = invalid_socket_id;
}
CAF_TEST(transfer data using multiple buffers) {
std::vector<byte> wr_buf_1{1_b, 2_b, 4_b};
std::vector<byte> wr_buf_2{8_b, 16_b, 32_b, 64_b};
std::vector<byte> full_buf;
full_buf.insert(full_buf.end(), wr_buf_1.begin(), wr_buf_1.end());
full_buf.insert(full_buf.end(), wr_buf_2.begin(), wr_buf_2.end());
CAF_CHECK_EQUAL(write(second, {make_span(wr_buf_1), make_span(wr_buf_2)}),
full_buf.size());
CAF_CHECK_EQUAL(read(first, make_span(rd_buf)), full_buf.size());
CAF_CHECK(std::equal(full_buf.begin(), full_buf.end(), rd_buf.begin()));
}
CAF_TEST_FIXTURE_SCOPE_END()
......@@ -20,10 +20,9 @@
#include "caf/net/stream_transport.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp"
......@@ -48,10 +47,10 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
mpx->set_thread_id();
}
bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false);
}
......@@ -72,20 +71,21 @@ public:
return none;
}
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager::message> msg) {
transport.write_packet(span<byte>{}, msg->payload);
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> ptr) {
parent.write_packet(ptr->payload);
}
template <class Parent>
void handle_data(Parent&, span<const byte> data) {
error handle_data(Parent&, span<const byte> data) {
rec_buf_->clear();
rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end());
return none;
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
void resolve(Parent& parent, string_view path, const actor& listener) {
actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(42, hid));
......@@ -95,11 +95,22 @@ public:
&parent.system(),
cfg,
std::move(ptr));
anon_send(listener, resolve_atom::value, std::move(path), p);
anon_send(listener, resolve_atom::value,
std::string{path.begin(), path.end()}, p);
}
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
// nop
}
template <class Parent>
void new_proxy(Parent&, actor_id) {
// nop
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
template <class Parent>
void local_actor_down(Parent&, actor_id, error) {
// nop
}
......@@ -139,7 +150,6 @@ CAF_TEST(receive) {
transport.configure_read(net::receive_policy::exactly(hello_manager.size()));
auto mgr = make_endpoint_manager(mpx, sys, transport);
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))),
hello_manager.size());
......@@ -161,9 +171,8 @@ CAF_TEST(resolve and proxy communication) {
transport_type{sockets.first,
dummy_application{buf}});
CAF_CHECK_EQUAL(mgr->init(), none);
mpx->handle_updates();
run();
mgr->resolve("/id/42", self);
mgr->resolve(unbox(make_uri("test:/id/42")), self);
run();
self->receive(
[&](resolve_atom, const std::string&, const strong_actor_ptr& p) {
......
......@@ -20,10 +20,9 @@
#include "caf/net/stream_transport.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include <vector>
#include "caf/binary_deserializer.hpp"
......@@ -44,15 +43,17 @@ using namespace caf::policy;
namespace {
using buffer_type = std::vector<byte>;
struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() {
mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err));
mpx->set_thread_id();
}
bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false);
}
......@@ -99,10 +100,15 @@ public:
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) {
header_type header{static_cast<uint32_t>(msg->payload.size())};
std::vector<byte> payload(msg->payload.begin(), msg->payload.end());
parent.write_packet(as_bytes(make_span(&header, 1)), make_span(payload));
std::unique_ptr<endpoint_manager_queue::message> ptr) {
// Ignore proxy announcement messages.
if (ptr->msg == nullptr)
return;
auto header_buf = parent.next_header_buffer();
serializer_impl<buffer_type> sink{sys_, header_buf};
header_type header{static_cast<uint32_t>(ptr->payload.size())};
sink(header);
parent.write_packet(header_buf, ptr->payload);
}
static expected<std::vector<byte>> serialize(actor_system& sys,
......@@ -138,14 +144,15 @@ public:
}
template <class Parent>
void handle_data(Parent& parent, span<const byte> data) {
error handle_data(Parent& parent, span<const byte> data) {
if (await_payload_) {
Base::handle_packet(parent, header_, data);
await_payload_ = false;
} else {
if (data.size() != sizeof(header_type))
CAF_FAIL("");
memcpy(&header_, data.data(), sizeof(header_type));
binary_deserializer source{nullptr, data};
source(header_);
if (header_.payload == 0)
Base::handle_packet(parent, header_, span<const byte>{});
else
......@@ -153,10 +160,11 @@ public:
net::receive_policy::exactly(header_.payload));
await_payload_ = true;
}
return none;
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
void resolve(Parent& parent, string_view path, actor listener) {
actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(aid, hid));
......@@ -165,18 +173,29 @@ public:
auto mgr = &parent.manager();
auto p = make_actor<net::actor_proxy_impl, strong_actor_ptr>(aid, nid, sys,
cfg, mgr);
anon_send(listener, resolve_atom::value, std::move(path), p);
anon_send(listener, resolve_atom::value,
std::string{path.begin(), path.end()}, p);
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
// nop
}
void handle_error(sec) {
template <class Parent>
void new_proxy(Parent&, actor_id) {
// nop
}
template <class Parent>
void local_actor_down(Parent&, actor_id, error) {
// nop
}
void handle_error(sec sec) {
CAF_FAIL("handle_error called: " << CAF_ARG(sec));
}
private:
header_type header_;
bool await_payload_;
......@@ -202,16 +221,14 @@ CAF_TEST(receive) {
transport_type{sockets.first,
application_type{sys, buf}});
CAF_CHECK_EQUAL(mgr1->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
auto mgr2 = make_endpoint_manager(mpx, sys,
transport_type{sockets.second,
application_type{sys, buf}});
CAF_CHECK_EQUAL(mgr2->init(), none);
mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
CAF_MESSAGE("resolve actor-proxy");
mgr1->resolve("/id/42", self);
mgr1->resolve(unbox(make_uri("test:/id/42")), self);
run();
self->receive(
[&](resolve_atom, const std::string&, const strong_actor_ptr& p) {
......
......@@ -21,10 +21,9 @@
#include "caf/net/tcp_accept_socket.hpp"
#include "caf/net/tcp_stream_socket.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/net/socket_guard.hpp"
using namespace caf;
......
......@@ -20,10 +20,9 @@
#include "caf/net/transport_worker.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/byte.hpp"
#include "caf/detail/scope_guard.hpp"
#include "caf/make_actor.hpp"
......@@ -38,6 +37,8 @@ using namespace caf::net;
namespace {
using buffer_type = std::vector<byte>;
constexpr string_view hello_test = "hello test!";
struct application_result {
......@@ -72,21 +73,23 @@ public:
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager::message> msg) {
parent.write_packet(span<const byte>{}, msg->payload);
std::unique_ptr<endpoint_manager_queue::message> msg) {
auto header_buffer = parent.next_header_buffer();
parent.write_packet(header_buffer, msg->payload);
}
template <class Parent>
void handle_data(Parent&, span<const byte> data) {
error handle_data(Parent&, span<const byte> data) {
auto& buf = res_->data_buffer;
buf.clear();
buf.insert(buf.begin(), data.begin(), data.end());
return none;
}
template <class Parent>
void resolve(Parent&, const std::string& path, actor listener) {
res_->resolve_path = path;
res_->resolve_listener = std::move(listener);
void resolve(Parent&, string_view path, const actor& listener) {
res_->resolve_path.assign(path.begin(), path.end());
res_->resolve_listener = listener;
}
template <class Parent>
......@@ -118,16 +121,29 @@ public:
using application_type = dummy_application;
dummy_transport(std::shared_ptr<transport_result> res) : res_(res) {
dummy_transport(std::shared_ptr<transport_result> res)
: res_(std::move(res)) {
// nop
}
void write_packet(span<const byte> header, span<const byte> payload,
ip_endpoint ep) {
auto& buf = res_->packet_buffer;
buf.insert(buf.begin(), header.begin(), header.end());
buf.insert(buf.begin(), payload.begin(), payload.end());
void write_packet(ip_endpoint ep, span<buffer_type*> buffers) {
res_->ep = ep;
auto& packet_buf = res_->packet_buffer;
packet_buf.clear();
for (auto buf : buffers)
packet_buf.insert(packet_buf.end(), buf->begin(), buf->end());
}
transport_type& transport() {
return *this;
}
std::vector<byte> next_header_buffer() {
return {};
}
std::vector<byte> next_payload_buffer() {
return {};
}
private:
......@@ -151,7 +167,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
}
bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false);
}
......@@ -191,9 +206,9 @@ CAF_TEST(write_message) {
const_cast<char*>(hello_test.data())),
hello_test.size());
std::vector<byte> payload(test_span.begin(), test_span.end());
auto message = detail::make_unique<endpoint_manager::message>(std::move(elem),
nullptr,
payload);
using message_type = endpoint_manager_queue::message;
auto message = detail::make_unique<message_type>(std::move(elem), nullptr,
payload);
worker.write_message(transport, std::move(message));
auto& buf = transport_results->packet_buffer;
string_view result{reinterpret_cast<char*>(buf.data()), buf.size()};
......
......@@ -20,10 +20,9 @@
#include "caf/net/transport_worker_dispatcher.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/make_actor.hpp"
#include "caf/monitorable_actor.hpp"
#include "caf/node_id.hpp"
......@@ -34,6 +33,8 @@ using namespace caf::net;
namespace {
using buffer_type = std::vector<byte>;
constexpr string_view hello_test = "hello_test";
struct dummy_actor : public monitorable_actor {
......@@ -48,7 +49,7 @@ struct dummy_actor : public monitorable_actor {
class dummy_application {
public:
dummy_application(std::shared_ptr<std::vector<byte>> rec_buf, uint8_t id)
dummy_application(std::shared_ptr<buffer_type> rec_buf, uint8_t id)
: rec_buf_(std::move(rec_buf)),
id_(id){
// nop
......@@ -62,16 +63,18 @@ public:
return none;
}
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager::message> msg) {
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager_queue::message> msg) {
rec_buf_->push_back(static_cast<byte>(id_));
transport.write_packet(span<byte>{}, make_span(msg->payload));
auto header_buf = parent.next_header_buffer();
parent.write_packet(header_buf, msg->payload);
}
template <class Parent>
void handle_data(Parent&, span<const byte>) {
error handle_data(Parent&, span<const byte>) {
rec_buf_->push_back(static_cast<byte>(id_));
return none;
}
template <class Manager>
......@@ -88,13 +91,13 @@ public:
rec_buf_->push_back(static_cast<byte>(id_));
}
static expected<std::vector<byte>> serialize(actor_system&,
const type_erased_tuple&) {
return std::vector<byte>{};
static expected<buffer_type> serialize(actor_system&,
const type_erased_tuple&) {
return buffer_type{};
}
private:
std::shared_ptr<std::vector<byte>> rec_buf_;
std::shared_ptr<buffer_type> rec_buf_;
uint8_t id_;
};
......@@ -102,8 +105,8 @@ struct dummy_application_factory {
public:
using application_type = dummy_application;
dummy_application_factory(std::shared_ptr<std::vector<byte>> buf)
: buf_(buf), application_cnt_(0) {
dummy_application_factory(std::shared_ptr<buffer_type> buf)
: buf_(std::move(buf)), application_cnt_(0) {
// nop
}
......@@ -112,7 +115,7 @@ public:
}
private:
std::shared_ptr<std::vector<byte>> buf_;
std::shared_ptr<buffer_type> buf_;
uint8_t application_cnt_;
};
......@@ -121,22 +124,35 @@ struct dummy_transport {
using application_type = dummy_application;
dummy_transport(std::shared_ptr<std::vector<byte>> buf) : buf_(buf) {
dummy_transport(std::shared_ptr<buffer_type> buf) : buf_(std::move(buf)) {
// nop
}
template <class IdType>
void write_packet(span<const byte> header, span<const byte> payload, IdType) {
buf_->insert(buf_->end(), header.begin(), header.end());
buf_->insert(buf_->end(), payload.begin(), payload.end());
void write_packet(IdType, span<buffer_type*> buffers) {
for (auto buf : buffers)
buf_->insert(buf_->end(), buf->begin(), buf->end());
}
transport_type& transport() {
return *this;
}
buffer_type next_header_buffer() {
return {};
}
buffer_type next_payload_buffer() {
return {};
}
private:
std::shared_ptr<std::vector<byte>> buf_;
std::shared_ptr<buffer_type> buf_;
};
struct testdata {
testdata(uint8_t worker_id, node_id id, ip_endpoint ep)
: worker_id(worker_id), nid(id), ep(ep) {
: worker_id(worker_id), nid(std::move(id)), ep(ep) {
// nop
}
......@@ -168,27 +184,27 @@ struct fixture : host_fixture {
ip_endpoint>;
fixture()
: buf{std::make_shared<std::vector<byte>>()},
: buf{std::make_shared<buffer_type>()},
dispatcher{dummy_application_factory{buf}},
dummy{buf} {
add_new_workers();
}
std::unique_ptr<net::endpoint_manager::message>
std::unique_ptr<net::endpoint_manager_queue::message>
make_dummy_message(node_id nid) {
actor_id aid = 42;
actor_config cfg;
auto p = make_actor<dummy_actor, strong_actor_ptr>(aid, nid, &sys, cfg);
auto test_span = as_bytes(make_span(hello_test));
std::vector<byte> payload(test_span.begin(), test_span.end());
buffer_type payload(test_span.begin(), test_span.end());
auto strong_actor = actor_cast<strong_actor_ptr>(p);
mailbox_element::forwarding_stack stack;
auto elem = make_mailbox_element(std::move(strong_actor),
make_message_id(12345), std::move(stack),
make_message());
return detail::make_unique<endpoint_manager::message>(std::move(elem),
strong_actor,
payload);
return detail::make_unique<endpoint_manager_queue::message>(std::move(elem),
strong_actor,
payload);
}
bool contains(byte x) {
......@@ -213,7 +229,7 @@ struct fixture : host_fixture {
actor_system_config cfg{};
actor_system sys{cfg};
std::shared_ptr<std::vector<byte>> buf;
std::shared_ptr<buffer_type> buf;
dispatcher_type dispatcher;
dummy_transport dummy;
......
......@@ -20,79 +20,47 @@
#include "caf/net/udp_datagram_socket.hpp"
#include "caf/net/test/host_fixture.hpp"
#include "caf/test/dsl.hpp"
#include "host_fixture.hpp"
#include "caf/detail/net_syscall.hpp"
#include "caf/detail/socket_sys_includes.hpp"
#include "caf/ip_address.hpp"
#include "caf/ip_endpoint.hpp"
#include "caf/ipv4_address.hpp"
#include "caf/net/ip.hpp"
#include "caf/net/socket_guard.hpp"
using namespace caf;
using namespace caf::net;
using namespace caf::net::ip;
namespace {
constexpr string_view hello_test = "Hello test!";
struct fixture : host_fixture {
fixture()
: host_fixture(),
v4_local{make_ipv4_address(127, 0, 0, 1)},
v6_local{{0}, {0x1}},
// TODO: use local_addresses() when merged
addrs{net::ip::resolve("localhost")} {
}
bool contains(ip_address x) {
return std::count(addrs.begin(), addrs.end(), x) > 0;
}
void test_send_receive(ip_address addr) {
std::vector<byte> buf(1024);
ip_endpoint ep(addr, 0);
fixture() : host_fixture() {
addresses = local_addresses("localhost");
CAF_CHECK(!addresses.empty());
ep = ip_endpoint(*addresses.begin(), 0);
auto send_pair = unbox(make_udp_datagram_socket(ep));
auto send_socket = send_pair.first;
send_socket = send_pair.first;
auto receive_pair = unbox(make_udp_datagram_socket(ep));
auto receive_socket = receive_pair.first;
receive_socket = receive_pair.first;
ep.port(ntohs(receive_pair.second));
CAF_MESSAGE("sending data to: " << to_string(ep));
auto send_guard = make_socket_guard(send_socket);
auto receive_guard = make_socket_guard(receive_socket);
if (auto err = nonblocking(socket_cast<net::socket>(receive_socket), true))
CAF_FAIL("nonblocking returned an error" << err);
auto test_read_res = read(receive_socket, make_span(buf));
if (auto err = get_if<sec>(&test_read_res))
CAF_CHECK_EQUAL(*err, sec::unavailable_or_would_block);
else
CAF_FAIL("read should have failed");
auto write_ret = write(send_socket, as_bytes(make_span(hello_test)), ep);
if (auto num_bytes = get_if<size_t>(&write_ret))
CAF_CHECK_EQUAL(*num_bytes, hello_test.size());
else
CAF_FAIL("write returned an error: " << sys.render(get<sec>(write_ret)));
}
auto read_ret = read(receive_socket, make_span(buf));
if (auto read_res = get_if<std::pair<size_t, ip_endpoint>>(&read_ret)) {
CAF_CHECK_EQUAL(read_res->first, hello_test.size());
buf.resize(read_res->first);
} else {
CAF_FAIL("write returned an error: " << sys.render(get<sec>(read_ret)));
}
string_view buf_view{reinterpret_cast<const char*>(buf.data()), buf.size()};
CAF_CHECK_EQUAL(buf_view, hello_test);
~fixture() {
close(send_socket);
close(receive_socket);
}
std::vector<ip_address> addresses;
actor_system_config cfg;
actor_system sys{cfg};
ip_address v4_local;
ip_address v6_local;
std::vector<ip_address> addrs;
ip_endpoint ep;
udp_datagram_socket send_socket;
udp_datagram_socket receive_socket;
};
} // namespace
......@@ -100,15 +68,29 @@ struct fixture : host_fixture {
CAF_TEST_FIXTURE_SCOPE(udp_datagram_socket_test, fixture)
CAF_TEST(send and receive) {
// TODO: check which versions exist and test existing versions accordingly
// -> local_addresses()
if (contains(v4_local)) {
test_send_receive(v4_local);
} else if (contains(v6_local)) {
test_send_receive(v6_local);
} else {
CAF_FAIL("could not resolve 'localhost'");
}
std::vector<byte> buf(1024);
if (auto err = nonblocking(socket_cast<net::socket>(receive_socket), true))
CAF_FAIL("setting socket to nonblocking failed: " << err);
CAF_CHECK_EQUAL(read(receive_socket, make_span(buf)),
sec::unavailable_or_would_block);
CAF_MESSAGE("sending data to " << to_string(ep));
CAF_CHECK_EQUAL(write(send_socket, as_bytes(make_span(hello_test)), ep),
hello_test.size());
int receive_attempts = 0;
variant<std::pair<size_t, ip_endpoint>, sec> read_ret;
do {
read_ret = read(receive_socket, make_span(buf));
if (auto read_res = get_if<std::pair<size_t, ip_endpoint>>(&read_ret)) {
CAF_CHECK_EQUAL(read_res->first, hello_test.size());
buf.resize(read_res->first);
} else if (get<sec>(read_ret) != sec::unavailable_or_would_block) {
CAF_FAIL("read returned an error: " << sys.render(get<sec>(read_ret)));
}
if (++receive_attempts > 100)
CAF_FAIL("unable to receive data from the UDP socket");
} while (read_ret.index() != 0);
string_view received{reinterpret_cast<const char*>(buf.data()), buf.size()};
CAF_CHECK_EQUAL(received, hello_test);
}
CAF_TEST_FIXTURE_SCOPE_END()
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