Unverified Commit f4de15a9 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #28

Finish BASP implementation
parents 38e6ccdf 75ffd95b
......@@ -19,6 +19,10 @@ set(LIBCAF_NET_SRCS
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/network_socket.cpp
src/pipe_socket.cpp
src/pollset_updater.cpp
......@@ -27,7 +31,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>
......@@ -37,6 +38,7 @@
#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"
......@@ -56,11 +58,11 @@ public:
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,6 +70,12 @@ 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())
return err;
......@@ -80,29 +88,12 @@ public:
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?
std::unique_ptr<endpoint_manager_queue::message> ptr) {
auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(hdr, payload);
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;
});
return write(write_packet, std::move(ptr));
}
template <class Parent>
......@@ -127,8 +118,27 @@ public:
resolve_remote_path(write_packet, path, listener);
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
template <class Parent>
void new_proxy(Parent& parent, actor_id id) {
header hdr{message_type::monitor_message, 0, static_cast<uint64_t>(id)};
auto bytes = to_bytes(hdr);
parent.write_packet(make_span(bytes), span<const byte>{});
}
template <class Parent>
void local_actor_down(Parent& parent, actor_id id, error reason) {
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
if (auto err = sink(reason))
CAF_RAISE_ERROR("unable to serialize an error");
header hdr{message_type::down_message, static_cast<uint32_t>(buf_.size()),
static_cast<uint64_t>(id)};
auto bytes = to_bytes(hdr);
parent.write_packet(make_span(bytes), make_span(buf_));
}
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
// nop
}
......@@ -157,7 +167,12 @@ public:
}
private:
// -- message handling -------------------------------------------------------
// -- handling of outgoing messages ------------------------------------------
error write(write_packet_callback& write_packet,
std::unique_ptr<endpoint_manager_queue::message> ptr);
// -- handling of incoming messages ------------------------------------------
error handle(size_t& next_read_size, write_packet_callback& write_packet,
byte_span bytes);
......@@ -177,6 +192,12 @@ private:
error handle_resolve_response(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_monitor_message(write_packet_callback& write_packet, header hdr,
byte_span payload);
error handle_down_message(write_packet_callback& write_packet, header hdr,
byte_span payload);
/// Writes the handshake payload to `buf_`.
error generate_handshake();
......@@ -210,7 +231,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
......
......@@ -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
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_;
......
......@@ -45,5 +45,13 @@ constexpr operation operator~(operation x) {
return static_cast<operation>(~static_cast<int>(x));
}
constexpr const char* to_string(operation x) {
return x == operation::none
? "none"
: (x == operation::read
? "read"
: (x == operation::write ? "write" : "read_write"));
}
} // namespace net
} // namespace caf
......@@ -20,6 +20,7 @@
#include <array>
#include <cstdint>
#include <mutex>
#include "caf/byte.hpp"
#include "caf/net/pipe_socket.hpp"
......@@ -34,6 +35,8 @@ 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);
......@@ -56,8 +59,9 @@ public:
void handle_error(sec code) override;
private:
std::array<byte, sizeof(intptr_t)> buf_;
msg_buf buf_;
size_t buf_size_;
std::mutex write_lock_;
};
} // namespace net
......
......@@ -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_;
};
......
......@@ -89,7 +89,6 @@ public:
manager_ = &parent;
if (auto err = worker_.init(*this))
return err;
parent.mask_add(operation::read);
return none;
}
......@@ -104,16 +103,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 +135,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 +155,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);
}
......@@ -186,7 +201,7 @@ public:
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);
manager().register_writing();
write_buf_.insert(write_buf_.end(), header.begin(), header.end());
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
}
......@@ -212,9 +227,11 @@ private:
}
} else {
auto err = get<sec>(ret);
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
worker_.handle_error(err);
return false;
if (err != sec::unavailable_or_would_block) {
CAF_LOG_DEBUG("send failed" << CAF_ARG(err));
worker_.handle_error(err);
return false;
}
}
return true;
}
......
......@@ -69,24 +69,37 @@ public:
}
template <class Parent>
void handle_data(Parent& parent, span<const byte> data) {
error handle_data(Parent& parent, span<const byte> data) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.handle_data(decorator, data);
return application_.handle_data(decorator, 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 decorator = make_write_packet_decorator(*this, parent);
application_.write_message(decorator, std::move(msg));
}
template <class Parent>
void resolve(Parent& parent, const std::string& path, actor listener) {
void resolve(Parent& parent, string_view path, const actor& listener) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.resolve(decorator, path, listener);
}
template <class Parent>
void new_proxy(Parent& parent, const node_id&, actor_id id) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.new_proxy(decorator, id);
}
template <class Parent>
void local_actor_down(Parent& parent, const node_id&, actor_id id,
error reason) {
auto decorator = make_write_packet_decorator(*this, parent);
application_.local_actor_down(decorator, 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);
......
......@@ -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,6 +29,7 @@
#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/no_stages.hpp"
......@@ -42,8 +43,7 @@ namespace caf {
namespace net {
namespace basp {
application::application(proxy_registry_ptr proxies)
: proxies_(std::move(proxies)) {
application::application(proxy_registry& proxies) : proxies_(proxies) {
// nop
}
......@@ -57,6 +57,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)) {
......@@ -79,8 +80,9 @@ strong_actor_ptr application::resolve_local_path(string_view path) {
void application::resolve_remote_path(write_packet_callback& write_packet,
string_view path, actor listener) {
CAF_LOG_TRACE(CAF_ARG(path) << CAF_ARG(listener));
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
serializer_impl<buffer_type> sink{&executor_, buf_};
if (auto err = sink(path)) {
CAF_LOG_ERROR("unable to serialize path");
return;
......@@ -97,9 +99,40 @@ void application::resolve_remote_path(write_packet_callback& write_packet,
pending_resolves_.emplace(req_id, std::move(rp));
}
error application::write(write_packet_callback& write_packet,
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()));
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) {
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;
}
// TODO: avoid extra copy of the payload
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);
return write_packet(make_span(bytes), make_span(buf_));
}
error application::handle(size_t& next_read_size,
write_packet_callback& write_packet,
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)
......@@ -144,6 +177,7 @@ error application::handle(size_t& next_read_size,
error application::handle(write_packet_callback& write_packet, 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;
......@@ -153,6 +187,10 @@ error application::handle(write_packet_callback& write_packet, header hdr,
return handle_resolve_request(write_packet, hdr, payload);
case message_type::resolve_response:
return handle_resolve_response(write_packet, hdr, payload);
case message_type::monitor_message:
return handle_monitor_message(write_packet, hdr, payload);
case message_type::down_message:
return handle_down_message(write_packet, hdr, payload);
case message_type::heartbeat:
return none;
default:
......@@ -162,13 +200,14 @@ error application::handle(write_packet_callback& write_packet, header hdr,
error application::handle_handshake(write_packet_callback&, 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())
......@@ -187,13 +226,14 @@ error application::handle_handshake(write_packet_callback&, header hdr,
error application::handle_actor_message(write_packet_callback&, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
// Deserialize payload.
actor_id src_id;
node_id src_node;
actor_id dst_id;
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 +248,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),
......@@ -219,9 +259,10 @@ error application::handle_actor_message(write_packet_callback&, header hdr,
error application::handle_resolve_request(write_packet_callback& write_packet,
header hdr, byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
CAF_ASSERT(hdr.type == message_type::resolve_request);
size_t path_size = 0;
binary_deserializer source{system(), payload};
binary_deserializer source{&executor_, payload};
if (auto err = source.begin_sequence(path_size))
return err;
// We expect the payload to consist only of the path.
......@@ -233,10 +274,16 @@ error application::handle_resolve_request(write_packet_callback& write_packet,
// 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_};
serializer_impl<buffer_type> sink{&executor_, buf_};
if (auto err = sink(aid, ifs))
return err;
auto out_hdr = to_bytes(header{message_type::resolve_response,
......@@ -247,6 +294,7 @@ error application::handle_resolve_request(write_packet_callback& write_packet,
error application::handle_resolve_response(write_packet_callback&, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
CAF_ASSERT(hdr.type == message_type::resolve_response);
auto i = pending_resolves_.find(hdr.operation_data);
if (i == pending_resolves_.end()) {
......@@ -260,20 +308,58 @@ 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_, payload};
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(write_packet_callback& write_packet,
header hdr, byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
if (!payload.empty())
return ec::unexpected_payload;
auto aid = static_cast<actor_id>(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;
buf_.clear();
serializer_impl<buffer_type> sink{&executor_, buf_};
if (auto err = sink(reason))
return err;
auto out_hdr = to_bytes(header{message_type::down_message,
static_cast<uint32_t>(buf_.size()),
hdr.operation_data});
return write_packet(out_hdr, buf_);
}
return none;
}
error application::handle_down_message(write_packet_callback&, header hdr,
byte_span payload) {
CAF_LOG_TRACE(CAF_ARG(hdr) << CAF_ARG2("payload.size", payload.size()));
error reason;
binary_deserializer source{&executor_, payload};
if (auto err = source(reason))
return err;
proxies_.erase(peer_id_, hdr.operation_data, std::move(reason));
return none;
}
error application::generate_handshake() {
buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_};
serializer_impl<buffer_type> sink{&executor_, buf_};
return sink(system().node(),
get_or(system().config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers));
......
......@@ -42,6 +42,7 @@ string_view ec_names[] = {
"unimplemented",
"app_identifiers_mismatch",
"invalid_payload",
"invalid_scheme",
};
} // namespace
......
......@@ -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), std::move(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
......
......@@ -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 {
using transport_type = stream_transport<basp::application>;
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_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_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_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_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
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_backend.hpp"
namespace caf {
namespace net {
middleman_backend::middleman_backend(std::string id) : id_(std::move(id)) {
// nop
}
middleman_backend::~middleman_backend() {
// nop
}
} // namespace net
} // namespace caf
......@@ -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
......
......@@ -41,10 +41,12 @@ 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 {
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));
......@@ -108,6 +110,10 @@ struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
return *this;
}
endpoint_manager& manager() {
CAF_FAIL("unexpected function call");
}
template <class... Ts>
void configure_read(Ts...) {
// nop
......@@ -130,6 +136,8 @@ struct fixture : test_coordinator_fixture<>, proxy_registry::backend {
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"
......
......@@ -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,10 +20,9 @@
#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"
......
......@@ -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,18 +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/actor_proxy_impl.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/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_socket.hpp"
#include "caf/net/stream_transport.hpp"
......@@ -57,37 +60,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 +135,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 +193,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,10 +20,9 @@
#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"
......
......@@ -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);
}
......@@ -74,18 +73,19 @@ public:
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager::message> msg) {
std::unique_ptr<endpoint_manager_queue::message> msg) {
transport.write_packet(span<byte>{}, msg->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, 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 Transport>
void timeout(Transport&, atom_value, uint64_t) {
template <class Parent>
void timeout(Parent&, atom_value, uint64_t) {
// nop
}
template <class Parent>
void new_proxy(Parent&, actor_id) {
// nop
}
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"
......@@ -49,10 +48,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);
}
......@@ -99,7 +98,10 @@ public:
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<net::endpoint_manager::message> msg) {
std::unique_ptr<endpoint_manager_queue::message> msg) {
// Ignore proxy announcement messages.
if (msg->msg == nullptr)
return;
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));
......@@ -138,7 +140,7 @@ 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;
......@@ -153,10 +155,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,11 +168,22 @@ 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 Parent>
void timeout(Parent&, atom_value, uint64_t) {
// nop
}
template <class Transport>
void timeout(Transport&, atom_value, uint64_t) {
template <class Parent>
void new_proxy(Parent&, actor_id) {
// nop
}
template <class Parent>
void local_actor_down(Parent&, actor_id, error) {
// nop
}
......@@ -202,16 +216,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"
......@@ -72,21 +71,22 @@ public:
template <class Parent>
void write_message(Parent& parent,
std::unique_ptr<endpoint_manager::message> msg) {
std::unique_ptr<endpoint_manager_queue::message> msg) {
parent.write_packet(span<const byte>{}, 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>
......@@ -151,7 +151,6 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
}
bool handle_io_event() override {
mpx->handle_updates();
return mpx->poll_once(false);
}
......@@ -191,9 +190,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"
......@@ -64,14 +63,15 @@ public:
template <class Transport>
void write_message(Transport& transport,
std::unique_ptr<endpoint_manager::message> msg) {
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));
}
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>
......@@ -174,7 +174,7 @@ struct fixture : host_fixture {
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;
......@@ -186,9 +186,9 @@ struct fixture : host_fixture {
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) {
......
......@@ -20,10 +20,9 @@
#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"
......
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