Unverified Commit 7fb9bd97 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #24

Add scaffold for new BASP implementation
parents c08a983d 2c996f4b
...@@ -8,10 +8,16 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp") ...@@ -8,10 +8,16 @@ file(GLOB_RECURSE LIBCAF_NET_HDRS "caf/*.hpp")
# list cpp files excluding platform-dependent files # list cpp files excluding platform-dependent files
set(LIBCAF_NET_SRCS set(LIBCAF_NET_SRCS
src/actor_proxy_impl.cpp src/actor_proxy_impl.cpp
src/application.cpp
src/connection_state.cpp
src/convert_ip_endpoint.cpp
src/datagram_socket.cpp src/datagram_socket.cpp
src/ec.cpp
src/endpoint_manager.cpp src/endpoint_manager.cpp
src/header.cpp
src/host.cpp src/host.cpp
src/ip.cpp src/ip.cpp
src/message_type.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/network_socket.cpp src/network_socket.cpp
src/pipe_socket.cpp src/pipe_socket.cpp
...@@ -19,10 +25,10 @@ set(LIBCAF_NET_SRCS ...@@ -19,10 +25,10 @@ set(LIBCAF_NET_SRCS
src/socket.cpp src/socket.cpp
src/socket_manager.cpp src/socket_manager.cpp
src/stream_socket.cpp src/stream_socket.cpp
src/convert_ip_endpoint.cpp
src/udp_datagram_socket.cpp
src/tcp_stream_socket.cpp
src/tcp_accept_socket.cpp src/tcp_accept_socket.cpp
src/tcp_stream_socket.cpp
src/tcp_stream_socket.cpp
src/udp_datagram_socket.cpp
) )
add_custom_target(libcaf_net) add_custom_target(libcaf_net)
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 <cstdint>
#include <memory>
#include <unordered_map>
#include <unordered_set>
#include <vector>
#include "caf/actor_addr.hpp"
#include "caf/byte.hpp"
#include "caf/callback.hpp"
#include "caf/error.hpp"
#include "caf/net/basp/connection_state.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/header.hpp"
#include "caf/net/basp/message_type.hpp"
#include "caf/net/endpoint_manager.hpp"
#include "caf/net/receive_policy.hpp"
#include "caf/node_id.hpp"
#include "caf/proxy_registry.hpp"
#include "caf/response_promise.hpp"
#include "caf/serializer_impl.hpp"
#include "caf/span.hpp"
#include "caf/unit.hpp"
namespace caf {
namespace net {
namespace basp {
/// An implementation of BASP as an application layer protocol.
class application {
public:
// -- member types -----------------------------------------------------------
using buffer_type = std::vector<byte>;
using byte_span = span<const byte>;
using write_packet_callback = callback<byte_span, byte_span>;
using proxy_registry_ptr = std::shared_ptr<proxy_registry>;
// -- constructors, destructors, and assignment operators --------------------
explicit application(proxy_registry_ptr proxies);
// -- interface functions ----------------------------------------------------
template <class Parent>
error init(Parent& parent) {
// Initialize member variables.
system_ = &parent.system();
// Write handshake.
if (auto err = generate_handshake())
return err;
auto hdr = to_bytes(header{message_type::handshake,
static_cast<uint32_t>(buf_.size()), version});
parent.write_packet(hdr, buf_);
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;
}
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;
});
size_t next_read_size = header_size;
if (auto err = handle(next_read_size, write_packet, 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);
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
// nop
}
void handle_error(sec) {
// nop
}
static expected<std::vector<byte>> 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 {
return state_;
}
actor_system& system() const noexcept {
return *system_;
}
private:
// -- message handling -------------------------------------------------------
error handle(size_t& next_read_size, write_packet_callback& write_packet,
byte_span bytes);
error handle(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_handshake(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_actor_message(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_resolve_request(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_resolve_response(write_packet_callback& write_packet, header hdr,
byte_span payload);
/// Writes the handshake payload to `buf_`.
error generate_handshake();
// -- member variables -------------------------------------------------------
/// Stores a pointer to the parent actor system.
actor_system* system_ = nullptr;
/// Stores the expected type of the next incoming message.
connection_state state_ = connection_state::await_handshake_header;
/// 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_;
/// Caches actor handles obtained via `resolve`.
std::unordered_map<uint64_t, response_promise> pending_resolves_;
/// Ascending ID generator for requests to our peer.
uint64_t next_request_id_ = 1;
/// Points to the factory object for generating proxies.
proxy_registry_ptr proxies_;
};
} // namespace basp
} // 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>
namespace caf {
namespace net {
namespace basp {
/// @addtogroup BASP
/// Stores the state of a connection in a `basp::application`.
enum class connection_state {
/// Initial state for any connection to wait for the peer's handshake.
await_handshake_header,
/// Indicates that the header for the peer's handshake arrived and BASP
/// requires the payload next.
await_handshake_payload,
/// Indicates that a connection is established and this node is waiting for
/// the next BASP header.
await_header,
/// Indicates that this node has received a header with non-zero payload and
/// is waiting for the data.
await_payload,
/// Indicates that the connection is about to shut down.
shutdown,
};
/// @relates connection_state
std::string to_string(connection_state x);
/// @}
} // namespace basp
} // 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 <cstddef>
#include <cstdint>
namespace caf {
namespace net {
namespace basp {
/// @addtogroup BASP
/// The current BASP version.
/// @note BASP is not backwards compatible.
constexpr uint64_t version = 1;
/// Size of a BASP header in serialized form.
constexpr size_t header_size = 13;
/// @}
} // namespace basp
} // 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 <cstdint>
#include <string>
#include "caf/fwd.hpp"
namespace caf {
namespace net {
namespace basp {
/// BASP-specific error codes.
enum class ec : uint8_t {
invalid_magic_number = 1,
unexpected_number_of_bytes,
unexpected_payload,
missing_payload,
illegal_state,
invalid_handshake,
missing_handshake,
unexpected_handshake,
version_mismatch,
unimplemented = 10,
app_identifiers_mismatch,
invalid_payload,
};
/// @relates ec
std::string to_string(ec x);
/// @relates ec
error make_error(ec x);
} // namespace basp
} // 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 <array>
#include <cstdint>
#include "caf/detail/comparable.hpp"
#include "caf/fwd.hpp"
#include "caf/meta/hex_formatted.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/net/basp/constants.hpp"
#include "caf/net/basp/message_type.hpp"
namespace caf {
namespace net {
namespace basp {
/// @addtogroup BASP
/// The header of a Binary Actor System Protocol (BASP) message.
struct header : detail::comparable<header> {
// -- constructors, destructors, and assignment operators --------------------
constexpr header() noexcept
: type(message_type::handshake), payload_len(0), operation_data(0) {
// nop
}
constexpr header(message_type type, uint32_t payload_len,
uint64_t operation_data) noexcept
: type(type), payload_len(payload_len), operation_data(operation_data) {
// nop
}
header(const header&) noexcept = default;
header& operator=(const header&) noexcept = default;
// -- factory functions ------------------------------------------------------
/// @pre `bytes.size() == header_size`
static header from_bytes(span<const byte> bytes);
// -- comparison -------------------------------------------------------------
int compare(header other) const noexcept;
// -- member variables -------------------------------------------------------
/// Denotes the BASP operation and how `operation_data` gets interpreted.
message_type type;
/// Stores the size in bytes for the payload that follows this header.
uint32_t payload_len;
/// Stores type-specific information such as the BASP version in handshakes.
uint64_t operation_data;
};
/// Serializes a header to a byte representation.
/// @relates header
std::array<byte, header_size> to_bytes(header x);
/// @relates header
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, header& x) {
return f(meta::type_name("basp::header"), x.type, x.payload_len,
x.operation_data);
}
/// @}
} // namespace basp
} // 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 <cstdint>
#include <string>
namespace caf {
namespace net {
namespace basp {
/// @addtogroup BASP
/// Describes the first header field of a BASP message and determines the
/// interpretation of the other header fields.
enum class message_type : uint8_t {
/// Sends supported BASP version and node information to the server.
///
/// ![](client_handshake.png)
handshake = 0,
/// Transmits an actor-to-actor messages.
///
/// ![](direct_message.png)
actor_message = 1,
/// Tries to resolve a path on the receiving node.
///
/// ![](resolve_request.png)
resolve_request = 2,
/// Transmits the result of a path lookup.
///
/// ![](resolve_response.png)
resolve_response = 3,
/// Informs the receiving node that the sending node has created a proxy
/// instance for one of its actors. Causes the receiving node to attach a
/// functor to the actor that triggers a down_message on termination.
///
/// ![](monitor_message.png)
monitor_message = 4,
/// Informs the receiving node that it has a proxy for an actor that has been
/// terminated.
///
/// ![](down_message.png)
down_message = 5,
/// Used to generate periodic traffic between two nodes in order to detect
/// disconnects.
///
/// ![](heartbeat.png)
heartbeat = 6,
};
/// @relates message_type
std::string to_string(message_type);
/// @}
} // namespace basp
} // namespace net
} // namespace caf
...@@ -91,10 +91,14 @@ public: ...@@ -91,10 +91,14 @@ public:
/// Original message to a remote actor. /// Original message to a remote actor.
mailbox_element_ptr msg; mailbox_element_ptr msg;
/// ID of the receiving actor.
strong_actor_ptr receiver;
/// Serialized representation of of `msg->content()`. /// Serialized representation of of `msg->content()`.
std::vector<byte> payload; std::vector<byte> payload;
message(mailbox_element_ptr msg, std::vector<byte> payload); message(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
}; };
struct message_policy { struct message_policy {
...@@ -136,7 +140,8 @@ public: ...@@ -136,7 +140,8 @@ public:
void resolve(std::string path, actor listener); void resolve(std::string path, actor listener);
/// Enqueues a message to the endpoint. /// Enqueues a message to the endpoint.
void enqueue(mailbox_element_ptr msg, std::vector<byte> payload); void enqueue(mailbox_element_ptr msg, strong_actor_ptr receiver,
std::vector<byte> payload);
/// Enqueues a timeout to the endpoint. /// Enqueues a timeout to the endpoint.
void enqueue(timeout_msg msg); void enqueue(timeout_msg msg);
......
...@@ -18,6 +18,10 @@ ...@@ -18,6 +18,10 @@
#pragma once #pragma once
#include "caf/abstract_actor.hpp"
#include "caf/actor_cast.hpp"
#include "caf/actor_system.hpp"
#include "caf/atom.hpp"
#include "caf/net/endpoint_manager.hpp" #include "caf/net/endpoint_manager.hpp"
namespace caf { namespace caf {
...@@ -48,10 +52,18 @@ public: ...@@ -48,10 +52,18 @@ public:
// -- properties ------------------------------------------------------------- // -- properties -------------------------------------------------------------
application_type& application() {
return transport_.application();
}
transport_type& transport() { transport_type& transport() {
return transport_; return transport_;
} }
endpoint_manager_impl& manager() {
return *this;
}
// -- timeout management ----------------------------------------------------- // -- timeout management -----------------------------------------------------
template <class... Ts> template <class... Ts>
......
...@@ -41,6 +41,10 @@ public: ...@@ -41,6 +41,10 @@ public:
using application_type = Application; using application_type = Application;
using transport_type = stream_transport;
using worker_type = transport_worker<application_type>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
stream_transport(stream_socket handle, application_type application) stream_transport(stream_socket handle, application_type application)
...@@ -51,7 +55,8 @@ public: ...@@ -51,7 +55,8 @@ public:
collected_(0), collected_(0),
max_(1024), max_(1024),
rd_flag_(net::receive_policy_flag::exactly), rd_flag_(net::receive_policy_flag::exactly),
written_(0) { written_(0),
manager_(nullptr) {
// nop // nop
} }
...@@ -61,18 +66,35 @@ public: ...@@ -61,18 +66,35 @@ public:
return handle_; return handle_;
} }
actor_system& system() {
return manager().system();
}
application_type& application() {
return worker_.application();
}
transport_type& transport() {
return *this;
}
endpoint_manager& manager() {
return *manager_;
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
if (auto err = worker_.init(parent)) manager_ = &parent;
if (auto err = worker_.init(*this))
return err; return err;
parent.mask_add(operation::read); parent.mask_add(operation::read);
return none; return none;
} }
template <class Parent> template <class Parent>
bool handle_read_event(Parent& parent) { bool handle_read_event(Parent&) {
auto buf = read_buf_.data() + collected_; auto buf = read_buf_.data() + collected_;
size_t len = read_threshold_ - collected_; size_t len = read_threshold_ - collected_;
CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len)); CAF_LOG_TRACE(CAF_ARG(handle_.id) << CAF_ARG(len));
...@@ -82,8 +104,7 @@ public: ...@@ -82,8 +104,7 @@ public:
CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes)); CAF_LOG_DEBUG(CAF_ARG(len) << CAF_ARG(handle_.id) << CAF_ARG(*num_bytes));
collected_ += *num_bytes; collected_ += *num_bytes;
if (collected_ >= read_threshold_) { if (collected_ >= read_threshold_) {
auto decorator = make_write_packet_decorator(*this, parent); worker_.handle_data(*this, read_buf_);
worker_.handle_data(decorator, read_buf_);
prepare_next_read(); prepare_next_read();
} }
return true; return true;
...@@ -103,16 +124,15 @@ public: ...@@ -103,16 +124,15 @@ public:
// TODO: dont read all messages at once - get one by one. // TODO: dont read all messages at once - get one by one.
for (auto msg = parent.next_message(); msg != nullptr; for (auto msg = parent.next_message(); msg != nullptr;
msg = parent.next_message()) { msg = parent.next_message()) {
auto decorator = make_write_packet_decorator(*this, parent); worker_.write_message(*this, std::move(msg));
worker_.write_message(decorator, std::move(msg));
} }
// Write prepared data. // Write prepared data.
return write_some(); return write_some();
} }
template <class Parent> template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) { void resolve(Parent&, const std::string& path, actor listener) {
worker_.resolve(parent, path, listener); worker_.resolve(*this, path, listener);
} }
template <class... Ts> template <class... Ts>
...@@ -121,8 +141,8 @@ public: ...@@ -121,8 +141,8 @@ public:
} }
template <class Parent> template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) { void timeout(Parent&, atom_value value, uint64_t id) {
worker_.timeout(parent, value, id); worker_.timeout(*this, value, id);
} }
void handle_error(sec code) { void handle_error(sec code) {
...@@ -163,9 +183,10 @@ public: ...@@ -163,9 +183,10 @@ public:
prepare_next_read(); prepare_next_read();
} }
template <class Parent> void write_packet(span<const byte> header, span<const byte> payload,
void write_packet(Parent&, span<const byte> header, span<const byte> payload, typename worker_type::id_type) {
unit_t) { 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(), header.begin(), header.end());
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end()); write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
} }
...@@ -198,7 +219,7 @@ private: ...@@ -198,7 +219,7 @@ private:
return true; return true;
} }
transport_worker<application_type> worker_; worker_type worker_;
stream_socket handle_; stream_socket handle_;
std::vector<byte> read_buf_; std::vector<byte> read_buf_;
...@@ -212,6 +233,8 @@ private: ...@@ -212,6 +233,8 @@ private:
receive_policy_flag rd_flag_; receive_policy_flag rd_flag_;
size_t written_; size_t written_;
endpoint_manager* manager_;
}; };
} // namespace net } // namespace net
......
...@@ -46,16 +46,32 @@ public: ...@@ -46,16 +46,32 @@ public:
// nop // nop
} }
// -- properties -------------------------------------------------------------
application_type& application() noexcept {
return application_;
}
const application_type& application() const noexcept {
return application_;
}
const id_type& id() const noexcept {
return id_;
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class Parent> template <class Parent>
error init(Parent& parent) { error init(Parent& parent) {
return application_.init(parent); auto decorator = make_write_packet_decorator(*this, parent);
return application_.init(decorator);
} }
template <class Parent> template <class Parent>
void handle_data(Parent& parent, span<const byte> data) { void handle_data(Parent& parent, span<const byte> data) {
application_.handle_data(parent, data); auto decorator = make_write_packet_decorator(*this, parent);
application_.handle_data(decorator, data);
} }
template <class Parent> template <class Parent>
...@@ -65,20 +81,16 @@ public: ...@@ -65,20 +81,16 @@ public:
application_.write_message(decorator, std::move(msg)); application_.write_message(decorator, std::move(msg));
} }
template <class Parent>
void write_packet(Parent& parent, span<const byte> header,
span<const byte> payload) {
parent.write_packet(header, payload, id_);
}
template <class Parent> template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) { void resolve(Parent& parent, const std::string& path, actor listener) {
application_.resolve(parent, path, listener); auto decorator = make_write_packet_decorator(*this, parent);
application_.resolve(decorator, path, listener);
} }
template <class Parent> template <class Parent>
void timeout(Parent& parent, atom_value value, uint64_t id) { void timeout(Parent& parent, atom_value value, uint64_t id) {
application_.timeout(parent, value, id); auto decorator = make_write_packet_decorator(*this, parent);
application_.timeout(decorator, value, id);
} }
void handle_error(sec error) { void handle_error(sec error) {
......
...@@ -52,12 +52,17 @@ public: ...@@ -52,12 +52,17 @@ public:
return parent_.transport(); return parent_.transport();
} }
endpoint_manager& manager() {
return parent_.manager();
}
// -- member functions ------------------------------------------------------- // -- member functions -------------------------------------------------------
template <class... Ts> template <class... Ts>
void write_packet(span<const byte> header, span<const byte> payload, void write_packet(span<const byte> header, span<const byte> payload,
Ts&&... xs) { Ts&&... xs) {
object_.write_packet(parent_, header, payload, std::forward<Ts>(xs)...); parent_.write_packet(header, payload, std::forward<Ts>(xs)...,
object_.id());
} }
void cancel_timeout(atom_value type, uint64_t id) { void cancel_timeout(atom_value type, uint64_t id) {
......
...@@ -27,18 +27,18 @@ namespace net { ...@@ -27,18 +27,18 @@ namespace net {
actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst) actor_proxy_impl::actor_proxy_impl(actor_config& cfg, endpoint_manager_ptr dst)
: super(cfg), sf_(dst->serialize_fun()), dst_(std::move(dst)) { : super(cfg), sf_(dst->serialize_fun()), dst_(std::move(dst)) {
// anon_send(broker_, monitor_atom::value, ctrl()); // nop
} }
actor_proxy_impl::~actor_proxy_impl() { actor_proxy_impl::~actor_proxy_impl() {
// anon_send(broker_, make_message(delete_atom::value, node(), id())); // nop
} }
void actor_proxy_impl::enqueue(mailbox_element_ptr what, execution_unit*) { void actor_proxy_impl::enqueue(mailbox_element_ptr what, execution_unit*) {
CAF_PUSH_AID(0); CAF_PUSH_AID(0);
CAF_ASSERT(what != nullptr); CAF_ASSERT(what != nullptr);
if (auto payload = sf_(home_system(), what->content())) if (auto payload = sf_(home_system(), what->content()))
dst_->enqueue(std::move(what), std::move(*payload)); dst_->enqueue(std::move(what), ctrl(), std::move(*payload));
else else
CAF_LOG_ERROR( CAF_LOG_ERROR(
"unable to serialize payload: " << home_system().render(payload.error())); "unable to serialize payload: " << home_system().render(payload.error()));
......
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/basp/connection_state.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
std::string to_string(connection_state x) {
return connection_state_names[static_cast<uint8_t>(x)];
}
} // namespace basp
} // 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/basp/ec.hpp"
#include "caf/atom.hpp"
#include "caf/error.hpp"
#include "caf/string_view.hpp"
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")};
}
} // namespace basp
} // namespace net
} // namespace caf
...@@ -37,8 +37,11 @@ endpoint_manager::event::event(atom_value type, uint64_t id) ...@@ -37,8 +37,11 @@ endpoint_manager::event::event(atom_value type, uint64_t id)
} }
endpoint_manager::message::message(mailbox_element_ptr msg, endpoint_manager::message::message(mailbox_element_ptr msg,
strong_actor_ptr receiver,
std::vector<byte> payload) std::vector<byte> payload)
: msg(std::move(msg)), payload(std::move(payload)) { : msg(std::move(msg)),
receiver(std::move(receiver)),
payload(std::move(payload)) {
// nop // nop
} }
...@@ -87,8 +90,10 @@ void endpoint_manager::resolve(std::string path, actor listener) { ...@@ -87,8 +90,10 @@ void endpoint_manager::resolve(std::string path, actor listener) {
} }
void endpoint_manager::enqueue(mailbox_element_ptr msg, void endpoint_manager::enqueue(mailbox_element_ptr msg,
strong_actor_ptr receiver,
std::vector<byte> payload) { std::vector<byte> payload) {
auto ptr = new message(std::move(msg), std::move(payload)); auto ptr = new message(std::move(msg), std::move(receiver),
std::move(payload));
if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader) if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader)
mask_add(operation::write); mask_add(operation::write);
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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/basp/header.hpp"
#include <cstring>
#include "caf/byte.hpp"
#include "caf/detail/network_order.hpp"
#include "caf/span.hpp"
namespace caf {
namespace net {
namespace basp {
int header::compare(header other) const noexcept {
auto x = to_bytes(*this);
auto y = to_bytes(other);
return memcmp(x.data(), y.data(), header_size);
}
header header::from_bytes(span<const byte> bytes) {
CAF_ASSERT(bytes.size() >= header_size);
header result;
auto ptr = bytes.data();
result.type = *reinterpret_cast<const message_type*>(ptr);
auto payload_len = *reinterpret_cast<const uint32_t*>(ptr + 1);
result.payload_len = detail::from_network_order(payload_len);
auto operation_data = *reinterpret_cast<const uint64_t*>(ptr + 5);
result.operation_data = detail::from_network_order(operation_data);
return result;
}
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));
return result;
}
} // namespace basp
} // 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/basp/message_type.hpp"
#include "caf/string_view.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()};
}
} // namespace basp
} // namespace net
} // namespace caf
This diff is collapsed.
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 basp.header
#include "caf/net/basp/header.hpp"
#include "caf/test/dsl.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/serializer_impl.hpp"
using namespace caf;
using namespace caf::net;
CAF_TEST(serialization) {
basp::header x{basp::message_type::handshake, 42, 4};
std::vector<byte> buf;
{
serializer_impl<std::vector<byte>> sink{nullptr, buf};
CAF_CHECK_EQUAL(sink(x), none);
}
CAF_CHECK_EQUAL(buf.size(), basp::header_size);
auto buf2 = to_bytes(x);
CAF_REQUIRE_EQUAL(buf.size(), buf2.size());
CAF_CHECK(std::equal(buf.begin(), buf.end(), buf2.begin()));
basp::header y;
{
binary_deserializer source{nullptr, buf};
CAF_CHECK_EQUAL(source(y), none);
}
CAF_CHECK_EQUAL(x, y);
auto z = basp::header::from_bytes(buf);
CAF_CHECK_EQUAL(x, z);
CAF_CHECK_EQUAL(y, z);
}
CAF_TEST(to_string) {
basp::header x{basp::message_type::handshake, 42, 4};
CAF_CHECK_EQUAL(to_string(x), "basp::header(handshake, 42, 4)");
}
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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 basp.stream_application
#include "caf/net/basp/application.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/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/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;
using namespace caf::net;
#define REQUIRE_OK(statement) \
if (auto err = statement) \
CAF_FAIL(sys.render(err));
namespace {
using buffer_type = std::vector<byte>;
using transport_type = stream_transport<basp::application>;
size_t fetch_size(variant<size_t, sec> x) {
if (holds_alternative<sec>(x))
CAF_FAIL("read/write failed: " << to_string(get<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();
}
~fixture() {
close(sock);
}
bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false);
}
template <class... Ts>
buffer_type to_buf(const Ts&... xs) {
buffer_type buf;
serializer_impl<buffer_type> sink{system(), buf};
REQUIRE_OK(sink(xs...));
return buf;
}
template <class... Ts>
void mock(const Ts&... xs) {
auto buf = to_buf(xs...);
if (fetch_size(write(sock, make_span(buf))) != buf.size())
CAF_FAIL("unable to write " << buf.size() << " bytes");
run();
}
void handle_handshake() {
CAF_CHECK_EQUAL(app->state(),
basp::connection_state::await_handshake_header);
auto payload = to_buf(mars, defaults::middleman::app_identifiers);
mock(basp::header{basp::message_type::handshake,
static_cast<uint32_t>(payload.size()), basp::version});
CAF_CHECK_EQUAL(app->state(),
basp::connection_state::await_handshake_payload);
write(sock, make_span(payload));
run();
}
void consume_handshake() {
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);
if (hdr.type != basp::message_type::handshake || hdr.payload_len == 0
|| hdr.operation_data != basp::version)
CAF_FAIL("invalid handshake header");
buf.resize(hdr.payload_len);
if (fetch_size(read(sock, make_span(buf))) != hdr.payload_len)
CAF_FAIL("unable to read " << hdr.payload_len << " bytes");
node_id nid;
std::vector<std::string> app_ids;
binary_deserializer source{sys, buf};
if (auto err = source(nid, app_ids))
CAF_FAIL("unable to deserialize payload: " << sys.render(err));
if (source.remaining() > 0)
CAF_FAIL("trailing bytes after reading payload");
}
actor_system& system() {
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;
};
} // 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)); \
run(); \
} while (false)
#define RECEIVE(msg_type, op_data, ...) \
do { \
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)); \
} while (false)
CAF_TEST_FIXTURE_SCOPE(application_tests, fixture)
CAF_TEST(actor message) {
handle_handshake();
consume_handshake();
sys.registry().put(self->id(), self);
CAF_REQUIRE_EQUAL(self->mailbox().size(), 0u);
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!"));
}
CAF_TEST(resolve request without result) {
handle_handshake();
consume_handshake();
CAF_CHECK_EQUAL(app->state(), basp::connection_state::await_header);
MOCK(basp::message_type::resolve_request, 42, std::string{"foo/bar"});
CAF_CHECK_EQUAL(app->state(), basp::connection_state::await_header);
actor_id aid;
std::set<std::string> ifs;
RECEIVE(basp::message_type::resolve_response, 42u, aid, ifs);
CAF_CHECK_EQUAL(aid, 0u);
CAF_CHECK(ifs.empty());
}
CAF_TEST(resolve request on id with result) {
handle_handshake();
consume_handshake();
sys.registry().put(self->id(), self);
auto path = "id/" + std::to_string(self->id());
CAF_CHECK_EQUAL(app->state(), basp::connection_state::await_header);
MOCK(basp::message_type::resolve_request, 42, path);
CAF_CHECK_EQUAL(app->state(), basp::connection_state::await_header);
actor_id aid;
std::set<std::string> ifs;
RECEIVE(basp::message_type::resolve_response, 42u, aid, ifs);
CAF_CHECK_EQUAL(aid, self->id());
CAF_CHECK(ifs.empty());
}
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -84,15 +84,17 @@ public: ...@@ -84,15 +84,17 @@ public:
rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end()); rec_buf_->insert(rec_buf_->begin(), data.begin(), data.end());
} }
template <class Manager> template <class Parent>
void resolve(Manager& manager, const std::string& path, actor listener) { void resolve(Parent& parent, const std::string& path, actor listener) {
actor_id aid = 42; actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899"; auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(42, hid)); auto nid = unbox(make_node_id(42, hid));
actor_config cfg; actor_config cfg;
endpoint_manager_ptr ptr{&parent.manager()};
auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid, auto p = make_actor<actor_proxy_impl, strong_actor_ptr>(aid, nid,
&manager.system(), &parent.system(),
cfg, &manager); cfg,
std::move(ptr));
anon_send(listener, resolve_atom::value, std::move(path), p); anon_send(listener, resolve_atom::value, std::move(path), p);
} }
......
...@@ -155,16 +155,16 @@ public: ...@@ -155,16 +155,16 @@ public:
} }
} }
template <class Manager> template <class Parent>
void resolve(Manager& manager, const std::string& path, actor listener) { void resolve(Parent& parent, const std::string& path, actor listener) {
actor_id aid = 42; actor_id aid = 42;
auto hid = "0011223344556677889900112233445566778899"; auto hid = "0011223344556677889900112233445566778899";
auto nid = unbox(make_node_id(aid, hid)); auto nid = unbox(make_node_id(aid, hid));
actor_config cfg; actor_config cfg;
auto p = make_actor<net::actor_proxy_impl, strong_actor_ptr>(aid, nid, auto sys = &parent.system();
&manager auto mgr = &parent.manager();
.system(), auto p = make_actor<net::actor_proxy_impl, strong_actor_ptr>(aid, nid, sys,
cfg, &manager); cfg, mgr);
anon_send(listener, resolve_atom::value, std::move(path), p); anon_send(listener, resolve_atom::value, std::move(path), p);
} }
......
...@@ -70,10 +70,10 @@ public: ...@@ -70,10 +70,10 @@ public:
return none; return none;
} }
template <class Transport> template <class Parent>
void write_message(Transport& transport, void write_message(Parent& parent,
std::unique_ptr<endpoint_manager::message> msg) { std::unique_ptr<endpoint_manager::message> msg) {
transport.write_packet(span<byte>{}, msg->payload); parent.write_packet(span<const byte>{}, msg->payload);
} }
template <class Parent> template <class Parent>
...@@ -192,6 +192,7 @@ CAF_TEST(write_message) { ...@@ -192,6 +192,7 @@ CAF_TEST(write_message) {
hello_test.size()); hello_test.size());
std::vector<byte> payload(test_span.begin(), test_span.end()); std::vector<byte> payload(test_span.begin(), test_span.end());
auto message = detail::make_unique<endpoint_manager::message>(std::move(elem), auto message = detail::make_unique<endpoint_manager::message>(std::move(elem),
nullptr,
payload); payload);
worker.write_message(transport, std::move(message)); worker.write_message(transport, std::move(message));
auto& buf = transport_results->packet_buffer; auto& buf = transport_results->packet_buffer;
......
...@@ -187,6 +187,7 @@ struct fixture : host_fixture { ...@@ -187,6 +187,7 @@ struct fixture : host_fixture {
make_message_id(12345), std::move(stack), make_message_id(12345), std::move(stack),
make_message()); make_message());
return detail::make_unique<endpoint_manager::message>(std::move(elem), return detail::make_unique<endpoint_manager::message>(std::move(elem),
strong_actor,
payload); payload);
} }
......
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