Commit c75ccdd8 authored by Dominik Charousset's avatar Dominik Charousset

Synchronize endpoint managers only via their queue

Using both a queue plus atomic bitmask for read and write flags
introduces race conditions between the two. As a result, the endpoint
manager ends up in corrupt states. This redesign synchronizes the
endpoint manager in the same way CAF synchronizes actors: via a single
mailbox that indicates whether the manager is blocked (currently waiting
for new events or messages) or not.
parent d2765ac2
...@@ -20,6 +20,7 @@ set(LIBCAF_NET_SRCS ...@@ -20,6 +20,7 @@ set(LIBCAF_NET_SRCS
src/message_type.cpp src/message_type.cpp
src/multiplexer.cpp src/multiplexer.cpp
src/net/backend/test.cpp src/net/backend/test.cpp
src/net/endpoint_manager_queue.cpp
src/net/middleman.cpp src/net/middleman.cpp
src/net/middleman_backend.cpp src/net/middleman_backend.cpp
src/network_socket.cpp src/network_socket.cpp
......
/****************************************************************************** /******************************************************************************_opt
* ____ _ _____ * * ____ _ _____ *
* / ___| / \ | ___| C++ * * / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor * * | | / _ \ | |_ Actor *
......
...@@ -88,7 +88,7 @@ public: ...@@ -88,7 +88,7 @@ public:
template <class Parent> template <class Parent>
error write_message(Parent& parent, error write_message(Parent& parent,
std::unique_ptr<endpoint_manager::message> ptr) { std::unique_ptr<endpoint_manager_queue::message> ptr) {
auto write_packet = make_callback([&](byte_span hdr, byte_span payload) { auto write_packet = make_callback([&](byte_span hdr, byte_span payload) {
parent.write_packet(hdr, payload); parent.write_packet(hdr, payload);
return none; return none;
...@@ -170,7 +170,7 @@ private: ...@@ -170,7 +170,7 @@ private:
// -- handling of outgoing messages ------------------------------------------ // -- handling of outgoing messages ------------------------------------------
error write(write_packet_callback& write_packet, error write(write_packet_callback& write_packet,
std::unique_ptr<endpoint_manager::message> ptr); std::unique_ptr<endpoint_manager_queue::message> ptr);
// -- handling of incoming messages ------------------------------------------ // -- handling of incoming messages ------------------------------------------
......
...@@ -30,6 +30,7 @@ ...@@ -30,6 +30,7 @@
#include "caf/intrusive/fifo_inbox.hpp" #include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp" #include "caf/intrusive/singly_linked.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/net/endpoint_manager_queue.hpp"
#include "caf/net/socket_manager.hpp" #include "caf/net/socket_manager.hpp"
#include "caf/variant.hpp" #include "caf/variant.hpp"
...@@ -50,91 +51,6 @@ public: ...@@ -50,91 +51,6 @@ public:
using serialize_fun_type = maybe_buffer (*)(actor_system&, using serialize_fun_type = maybe_buffer (*)(actor_system&,
const type_erased_tuple&); const type_erased_tuple&);
struct event : intrusive::singly_linked<event> {
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);
/// Holds the event data.
variant<resolve_request, new_proxy, local_actor_down, 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 at least 1 if the payload is empty.
return x.payload.size() + static_cast<task_size_type>(x.payload.empty());
}
};
using message_queue_type = intrusive::fifo_inbox<message_policy>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
endpoint_manager(socket handle, const multiplexer_ptr& parent, endpoint_manager(socket handle, const multiplexer_ptr& parent,
...@@ -148,7 +64,7 @@ public: ...@@ -148,7 +64,7 @@ public:
return sys_; return sys_;
} }
std::unique_ptr<message> next_message(); endpoint_manager_queue::message_ptr next_message();
// -- event management ------------------------------------------------------- // -- event management -------------------------------------------------------
...@@ -162,7 +78,7 @@ public: ...@@ -162,7 +78,7 @@ public:
/// Enqueues an event to the endpoint. /// Enqueues an event to the endpoint.
template <class... Ts> template <class... Ts>
void enqueue_event(Ts&&... xs) { void enqueue_event(Ts&&... xs) {
enqueue(new event(std::forward<Ts>(xs)...)); enqueue(new endpoint_manager_queue::event(std::forward<Ts>(xs)...));
} }
// -- pure virtual member functions ------------------------------------------ // -- pure virtual member functions ------------------------------------------
...@@ -174,16 +90,13 @@ public: ...@@ -174,16 +90,13 @@ public:
virtual serialize_fun_type serialize_fun() const noexcept = 0; virtual serialize_fun_type serialize_fun() const noexcept = 0;
protected: protected:
void enqueue(event* ptr); bool enqueue(endpoint_manager_queue::element* ptr);
/// Points to the hosting actor system. /// Points to the hosting actor system.
actor_system& sys_; actor_system& sys_;
/// Stores control events. /// Stores control events and outbound messages.
event_queue_type events_; endpoint_manager_queue::type queue_;
/// Stores outbound messages.
message_queue_type messages_;
/// Stores a proxy for interacting with the actor clock. /// Stores a proxy for interacting with the actor clock.
actor timeout_proxy_; actor timeout_proxy_;
......
...@@ -80,6 +80,7 @@ public: ...@@ -80,6 +80,7 @@ public:
// -- interface functions ---------------------------------------------------- // -- interface functions ----------------------------------------------------
error init() override { error init() override {
this->register_reading();
return transport_.init(*this); return transport_.init(*this);
} }
...@@ -88,31 +89,36 @@ public: ...@@ -88,31 +89,36 @@ public:
} }
bool handle_write_event() override { 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 { do {
this->events_.fetch_more();
auto& q = this->events_.queue();
q.inc_deficit(q.total_task_size()); q.inc_deficit(q.total_task_size());
for (auto ptr = q.next(); ptr != nullptr; ptr = q.next()) { for (auto ptr = q.next(); ptr != nullptr; ptr = q.next()) {
auto f = detail::make_overload( auto f = detail::make_overload(
[&](endpoint_manager::event::resolve_request& x) { [&](endpoint_manager_queue::event::resolve_request& x) {
transport_.resolve(*this, x.locator, x.listener); transport_.resolve(*this, x.locator, x.listener);
}, },
[&](endpoint_manager::event::new_proxy& x) { [&](endpoint_manager_queue::event::new_proxy& x) {
transport_.new_proxy(*this, x.peer, x.id); transport_.new_proxy(*this, x.peer, x.id);
}, },
[&](endpoint_manager::event::local_actor_down& x) { [&](endpoint_manager_queue::event::local_actor_down& x) {
transport_.local_actor_down(*this, x.observing_peer, x.id, transport_.local_actor_down(*this, x.observing_peer, x.id,
std::move(x.reason)); std::move(x.reason));
}, },
[&](endpoint_manager::event::timeout& x) { [&](endpoint_manager_queue::event::timeout& x) {
transport_.timeout(*this, x.type, x.id); transport_.timeout(*this, x.type, x.id);
}); });
visit(f, ptr->value); 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 { 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
...@@ -64,9 +64,9 @@ public: ...@@ -64,9 +64,9 @@ public:
// -- thread-safe signaling -------------------------------------------------- // -- thread-safe signaling --------------------------------------------------
/// Causes the multiplexer to update its event bitmask for `mgr`. /// Registers `mgr` for write events.
/// @thread-safe /// @thread-safe
void update(const socket_manager_ptr& mgr); void register_writing(const socket_manager_ptr& mgr);
/// Closes the pipe for signaling updates to the multiplexer. After closing /// Closes the pipe for signaling updates to the multiplexer. After closing
/// the pipe, calls to `update` no longer have any effect. /// the pipe, calls to `update` no longer have any effect.
...@@ -75,10 +75,17 @@ public: ...@@ -75,10 +75,17 @@ public:
// -- control flow ----------------------------------------------------------- // -- control flow -----------------------------------------------------------
/// Registers `mgr` for read events.
/// @pre `std::this_thread::id() == tid_`
void register_reading(const socket_manager_ptr& mgr);
/// Polls I/O activity once and runs all socket event handlers that become /// Polls I/O activity once and runs all socket event handlers that become
/// ready as a result. /// ready as a result.
bool poll_once(bool blocking); 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. /// Polls until no socket event handler remains.
void run(); void run();
......
...@@ -45,5 +45,13 @@ constexpr operation operator~(operation x) { ...@@ -45,5 +45,13 @@ constexpr operation operator~(operation x) {
return static_cast<operation>(~static_cast<int>(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 net
} // namespace caf } // namespace caf
...@@ -18,8 +18,6 @@ ...@@ -18,8 +18,6 @@
#pragma once #pragma once
#include <atomic>
#include "caf/error.hpp" #include "caf/error.hpp"
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/net/fwd.hpp" #include "caf/net/fwd.hpp"
...@@ -53,21 +51,26 @@ public: ...@@ -53,21 +51,26 @@ public:
} }
/// Returns registered operations (read, write, or both). /// 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. /// @returns `false` if `mask() | flag == mask()`, `true` otherwise.
/// @pre `has_parent()`
/// @pre `flag != operation::none` /// @pre `flag != operation::none`
bool mask_add(operation flag) noexcept; bool mask_add(operation flag) noexcept;
/// Deletes given flag(s) from the event mask and updates the parent on /// Tries to clear given flag(s) from the event mask.
/// success.
/// @returns `false` if `mask() & ~flag == mask()`, `true` otherwise. /// @returns `false` if `mask() & ~flag == mask()`, `true` otherwise.
/// @pre `has_parent()`
/// @pre `flag != operation::none` /// @pre `flag != operation::none`
bool mask_del(operation flag) noexcept; bool mask_del(operation flag) noexcept;
// -- event loop management --------------------------------------------------
void register_reading();
void register_writing();
// -- pure virtual member functions ------------------------------------------ // -- pure virtual member functions ------------------------------------------
/// Called whenever the socket received new data. /// Called whenever the socket received new data.
...@@ -85,7 +88,7 @@ protected: ...@@ -85,7 +88,7 @@ protected:
socket handle_; socket handle_;
std::atomic<operation> mask_; operation mask_;
weak_multiplexer_ptr parent_; weak_multiplexer_ptr parent_;
}; };
......
...@@ -89,7 +89,6 @@ public: ...@@ -89,7 +89,6 @@ public:
manager_ = &parent; manager_ = &parent;
if (auto err = worker_.init(*this)) if (auto err = worker_.init(*this))
return err; return err;
parent.mask_add(operation::read);
return none; return none;
} }
...@@ -200,7 +199,7 @@ public: ...@@ -200,7 +199,7 @@ public:
void write_packet(span<const byte> header, span<const byte> payload, void write_packet(span<const byte> header, span<const byte> payload,
typename worker_type::id_type) { typename worker_type::id_type) {
if (write_buf_.empty()) 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(), header.begin(), header.end());
write_buf_.insert(write_buf_.end(), payload.begin(), payload.end()); write_buf_.insert(write_buf_.end(), payload.begin(), payload.end());
} }
......
...@@ -76,7 +76,7 @@ public: ...@@ -76,7 +76,7 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& 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); auto decorator = make_write_packet_decorator(*this, parent);
application_.write_message(decorator, std::move(msg)); application_.write_message(decorator, std::move(msg));
} }
......
...@@ -77,7 +77,7 @@ public: ...@@ -77,7 +77,7 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& 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; auto sender = msg->msg->sender;
if (!sender) if (!sender)
return; return;
......
...@@ -38,6 +38,7 @@ actor_proxy_impl::~actor_proxy_impl() { ...@@ -38,6 +38,7 @@ actor_proxy_impl::~actor_proxy_impl() {
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);
CAF_LOG_SEND_EVENT(what);
if (auto payload = sf_(home_system(), what->content())) if (auto payload = sf_(home_system(), what->content()))
dst_->enqueue(std::move(what), ctrl(), std::move(*payload)); dst_->enqueue(std::move(what), ctrl(), std::move(*payload));
else else
......
...@@ -100,9 +100,10 @@ void application::resolve_remote_path(write_packet_callback& write_packet, ...@@ -100,9 +100,10 @@ void application::resolve_remote_path(write_packet_callback& write_packet,
} }
error application::write(write_packet_callback& write_packet, error application::write(write_packet_callback& write_packet,
std::unique_ptr<endpoint_manager::message> ptr) { std::unique_ptr<endpoint_manager_queue::message> ptr) {
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->msg != nullptr); CAF_ASSERT(ptr->msg != nullptr);
CAF_LOG_TRACE(CAF_ARG2("content", ptr->msg->content()));
buf_.clear(); buf_.clear();
serializer_impl<buffer_type> sink{system(), buf_}; serializer_impl<buffer_type> sink{system(), buf_};
const auto& src = ptr->msg->sender; const auto& src = ptr->msg->sender;
......
...@@ -20,98 +20,71 @@ ...@@ -20,98 +20,71 @@
#include "caf/byte.hpp" #include "caf/byte.hpp"
#include "caf/intrusive/inbox_result.hpp" #include "caf/intrusive/inbox_result.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
namespace caf { namespace caf {
namespace net { namespace net {
endpoint_manager::event::event(uri locator, actor listener)
: value(resolve_request{std::move(locator), std::move(listener)}) {
// nop
}
endpoint_manager::event::event(node_id peer, actor_id proxy_id)
: value(new_proxy{peer, proxy_id}) {
// nop
}
endpoint_manager::event::event(node_id observing_peer, actor_id local_actor_id,
error reason)
: value(local_actor_down{observing_peer, local_actor_id, std::move(reason)}) {
// 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, endpoint_manager::endpoint_manager(socket handle, const multiplexer_ptr& parent,
actor_system& sys) actor_system& sys)
: super(handle, parent), : super(handle, parent), sys_(sys), queue_(unit, unit, unit) {
sys_(sys), queue_.try_block();
events_(event_policy{}),
messages_(message_policy{}) {
events_.try_block();
messages_.try_block();
} }
endpoint_manager::~endpoint_manager() { endpoint_manager::~endpoint_manager() {
// nop // nop
} }
std::unique_ptr<endpoint_manager::message> endpoint_manager::next_message() { endpoint_manager_queue::message_ptr endpoint_manager::next_message() {
if (messages_.blocked()) if (queue_.blocked())
return nullptr; return nullptr;
messages_.fetch_more(); queue_.fetch_more();
auto& q = messages_.queue(); auto& q = std::get<1>(queue_.queue().queues());
auto ts = q.next_task_size(); auto ts = q.next_task_size();
if (ts == 0) if (ts == 0)
return nullptr; return nullptr;
q.inc_deficit(ts); q.inc_deficit(ts);
auto result = q.next(); auto result = q.next();
if (q.empty()) if (queue_.empty())
messages_.try_block(); queue_.try_block();
return result; return result;
} }
void endpoint_manager::resolve(uri locator, actor listener) { void endpoint_manager::resolve(uri locator, actor listener) {
using intrusive::inbox_result; using intrusive::inbox_result;
auto ptr = new event(std::move(locator), std::move(listener)); using event_type = endpoint_manager_queue::event;
switch (events_.push_back(ptr)) { auto ptr = new event_type(std::move(locator), std::move(listener));
default: if (!enqueue(ptr))
break;
case inbox_result::unblocked_reader:
mask_add(operation::write);
break;
case inbox_result::queue_closed:
anon_send(listener, resolve_atom::value, anon_send(listener, resolve_atom::value,
make_error(sec::request_receiver_down)); make_error(sec::request_receiver_down));
}
} }
void endpoint_manager::enqueue(mailbox_element_ptr msg, void endpoint_manager::enqueue(mailbox_element_ptr msg,
strong_actor_ptr receiver, strong_actor_ptr receiver,
std::vector<byte> payload) { std::vector<byte> payload) {
auto ptr = new message(std::move(msg), std::move(receiver), using message_type = endpoint_manager_queue::message;
auto ptr = new message_type(std::move(msg), std::move(receiver),
std::move(payload)); std::move(payload));
if (messages_.push_back(ptr) == intrusive::inbox_result::unblocked_reader) enqueue(ptr);
mask_add(operation::write);
} }
void endpoint_manager::enqueue(event* ptr) { bool endpoint_manager::enqueue(endpoint_manager_queue::element* ptr) {
if (events_.push_back(ptr) == intrusive::inbox_result::unblocked_reader) switch (queue_.push_back(ptr)) {
mask_add(operation::write); 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 } // namespace net
......
...@@ -90,7 +90,6 @@ error multiplexer::init() { ...@@ -90,7 +90,6 @@ error multiplexer::init() {
auto pipe_handles = make_pipe(); auto pipe_handles = make_pipe();
if (!pipe_handles) if (!pipe_handles)
return std::move(pipe_handles.error()); return std::move(pipe_handles.error());
tid_ = std::this_thread::get_id();
add(make_counted<pollset_updater>(pipe_handles->first, shared_from_this())); add(make_counted<pollset_updater>(pipe_handles->first, shared_from_this()));
write_handle_ = pipe_handles->second; write_handle_ = pipe_handles->second;
return none; return none;
...@@ -107,11 +106,13 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) { ...@@ -107,11 +106,13 @@ ptrdiff_t multiplexer::index_of(const socket_manager_ptr& mgr) {
return i == last ? -1 : std::distance(first, i); return i == last ? -1 : std::distance(first, i);
} }
void multiplexer::update(const socket_manager_ptr& mgr) { void multiplexer::register_writing(const socket_manager_ptr& mgr) {
if (std::this_thread::get_id() == tid_) { if (std::this_thread::get_id() == tid_) {
if (mgr->mask_add(operation::write)) {
auto i = std::find(dirty_managers_.begin(), dirty_managers_.end(), mgr); auto i = std::find(dirty_managers_.begin(), dirty_managers_.end(), mgr);
if (i == dirty_managers_.end()) if (i == dirty_managers_.end())
dirty_managers_.emplace_back(mgr); dirty_managers_.emplace_back(mgr);
}
} else { } else {
mgr->ref(); mgr->ref();
auto value = reinterpret_cast<intptr_t>(mgr.get()); auto value = reinterpret_cast<intptr_t>(mgr.get());
...@@ -136,6 +137,15 @@ void multiplexer::close_pipe() { ...@@ -136,6 +137,15 @@ void multiplexer::close_pipe() {
} }
} }
void multiplexer::register_reading(const socket_manager_ptr& mgr) {
CAF_ASSERT(std::this_thread::get_id() == tid_);
if (mgr->mask_add(operation::read)) {
auto i = std::find(dirty_managers_.begin(), dirty_managers_.end(), mgr);
if (i == dirty_managers_.end())
dirty_managers_.emplace_back(mgr);
}
}
bool multiplexer::poll_once(bool blocking) { bool multiplexer::poll_once(bool blocking) {
if (pollset_.empty()) if (pollset_.empty())
return false; return false;
...@@ -194,7 +204,12 @@ bool multiplexer::poll_once(bool blocking) { ...@@ -194,7 +204,12 @@ bool multiplexer::poll_once(bool blocking) {
} }
} }
void multiplexer::set_thread_id() {
tid_ = std::this_thread::get_id();
}
void multiplexer::run() { void multiplexer::run() {
CAF_LOG_TRACE("");
while (!pollset_.empty()) while (!pollset_.empty())
poll_once(true); poll_once(true);
} }
...@@ -223,13 +238,13 @@ void multiplexer::handle(const socket_manager_ptr& mgr, int mask) { ...@@ -223,13 +238,13 @@ void multiplexer::handle(const socket_manager_ptr& mgr, int mask) {
bool checkerror = true; bool checkerror = true;
if ((mask & input_mask) != 0) { if ((mask & input_mask) != 0) {
checkerror = false; checkerror = false;
if (!mgr->handle_read_event()) if (!mgr->handle_read_event() && mgr->mask_del(operation::read))
mgr->mask_del(operation::read); dirty_managers_.emplace_back(mgr);
} }
if ((mask & output_mask) != 0) { if ((mask & output_mask) != 0) {
checkerror = false; checkerror = false;
if (!mgr->handle_write_event()) if (!mgr->handle_write_event() && mgr->mask_del(operation::write))
mgr->mask_del(operation::write); dirty_managers_.emplace_back(mgr);
} }
if (checkerror && ((mask & error_mask) != 0)) { if (checkerror && ((mask & error_mask) != 0)) {
if (mask & POLLNVAL) if (mask & POLLNVAL)
...@@ -239,6 +254,7 @@ void multiplexer::handle(const socket_manager_ptr& mgr, int mask) { ...@@ -239,6 +254,7 @@ void multiplexer::handle(const socket_manager_ptr& mgr, int mask) {
else else
mgr->handle_error(sec::socket_operation_failed); mgr->handle_error(sec::socket_operation_failed);
mgr->mask_del(operation::read_write); mgr->mask_del(operation::read_write);
dirty_managers_.emplace_back(mgr);
} }
} }
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
#include "caf/net/make_endpoint_manager.hpp" #include "caf/net/make_endpoint_manager.hpp"
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
#include "caf/net/multiplexer.hpp"
#include "caf/net/stream_transport.hpp" #include "caf/net/stream_transport.hpp"
#include "caf/raise_error.hpp" #include "caf/raise_error.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
...@@ -40,8 +41,7 @@ test::test(middleman& mm) ...@@ -40,8 +41,7 @@ test::test(middleman& mm)
} }
test::~test() { test::~test() {
for (auto& kvp : peers_) // nop
close(kvp.second.first);
} }
error test::init() { error test::init() {
...@@ -75,7 +75,6 @@ void test::set_last_hop(node_id*) { ...@@ -75,7 +75,6 @@ void test::set_last_hop(node_id*) {
test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first, test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first,
stream_socket second) { stream_socket second) {
using transport_type = stream_transport<basp::application>; using transport_type = stream_transport<basp::application>;
nonblocking(first, true);
nonblocking(second, true); nonblocking(second, true);
auto mpx = mm_.mpx(); auto mpx = mm_.mpx();
auto mgr = make_endpoint_manager(mpx, mm_.system(), auto mgr = make_endpoint_manager(mpx, mm_.system(),
...@@ -83,6 +82,7 @@ test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first, ...@@ -83,6 +82,7 @@ test::peer_entry& test::emplace(const node_id& peer_id, stream_socket first,
basp::application{proxies_}}); basp::application{proxies_}});
if (auto err = mgr->init()) if (auto err = mgr->init())
CAF_RAISE_ERROR("mgr->init() failed"); CAF_RAISE_ERROR("mgr->init() failed");
mpx->register_reading(mgr);
auto& result = peers_[peer_id]; auto& result = peers_[peer_id];
result = std::make_pair(first, std::move(mgr)); result = std::make_pair(first, std::move(mgr));
return result; return result;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#include "caf/net/middleman.hpp" #include "caf/net/middleman.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/detail/set_thread_name.hpp"
#include "caf/net/basp/ec.hpp" #include "caf/net/basp/ec.hpp"
#include "caf/net/middleman_backend.hpp" #include "caf/net/middleman_backend.hpp"
#include "caf/net/multiplexer.hpp" #include "caf/net/multiplexer.hpp"
...@@ -40,7 +41,15 @@ middleman::~middleman() { ...@@ -40,7 +41,15 @@ middleman::~middleman() {
void middleman::start() { void middleman::start() {
if (!get_or(config(), "middleman.manual-multiplexing", false)) { if (!get_or(config(), "middleman.manual-multiplexing", false)) {
auto mpx = mpx_; auto mpx = mpx_;
mpx_thread_ = std::thread{[mpx] { mpx->run(); }}; 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();
}};
} }
} }
......
...@@ -48,7 +48,7 @@ bool pollset_updater::handle_read_event() { ...@@ -48,7 +48,7 @@ bool pollset_updater::handle_read_event() {
auto value = *reinterpret_cast<intptr_t*>(buf_.data()); auto value = *reinterpret_cast<intptr_t*>(buf_.data());
socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false}; socket_manager_ptr mgr{reinterpret_cast<socket_manager*>(value), false};
if (auto ptr = parent_.lock()) if (auto ptr = parent_.lock())
ptr->update(mgr); ptr->register_writing(mgr);
} }
} else { } else {
return get<sec>(res) == sec::unavailable_or_would_block; return get<sec>(res) == sec::unavailable_or_would_block;
......
...@@ -34,32 +34,38 @@ socket_manager::~socket_manager() { ...@@ -34,32 +34,38 @@ socket_manager::~socket_manager() {
close(handle_); close(handle_);
} }
operation socket_manager::mask() const noexcept {
return mask_.load();
}
bool socket_manager::mask_add(operation flag) noexcept { bool socket_manager::mask_add(operation flag) noexcept {
CAF_ASSERT(flag != operation::none); CAF_ASSERT(flag != operation::none);
auto x = mask(); auto x = mask();
while ((x & flag) != flag) if ((x & flag) == flag)
if (mask_.compare_exchange_strong(x, x | flag)) {
if (auto ptr = parent_.lock())
ptr->update(this);
return true;
}
return false; return false;
mask_ = x | flag;
return true;
} }
bool socket_manager::mask_del(operation flag) noexcept { bool socket_manager::mask_del(operation flag) noexcept {
CAF_ASSERT(flag != operation::none); CAF_ASSERT(flag != operation::none);
auto x = mask(); auto x = mask();
while ((x & flag) != operation::none) if ((x & flag) == operation::none)
if (mask_.compare_exchange_strong(x, x & ~flag)) {
if (auto ptr = parent_.lock())
ptr->update(this);
return true;
}
return false; 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 } // namespace net
......
...@@ -46,8 +46,11 @@ string_view hello_test{"hello test!"}; ...@@ -46,8 +46,11 @@ string_view hello_test{"hello test!"};
struct fixture : test_coordinator_fixture<>, host_fixture { struct fixture : test_coordinator_fixture<>, host_fixture {
fixture() { fixture() {
mpx = std::make_shared<multiplexer>(); mpx = std::make_shared<multiplexer>();
mpx->set_thread_id();
if (auto err = mpx->init()) if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err)); 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 { bool handle_io_event() override {
...@@ -87,7 +90,7 @@ public: ...@@ -87,7 +90,7 @@ public:
error init(Manager& manager) { error init(Manager& manager) {
auto test_bytes = as_bytes(make_span(hello_test)); auto test_bytes = as_bytes(make_span(hello_test));
buf_.insert(buf_.end(), test_bytes.begin(), test_bytes.end()); buf_.insert(buf_.end(), test_bytes.begin(), test_bytes.end());
CAF_CHECK(manager.mask_add(operation::read_write)); manager.register_writing();
return none; return none;
} }
...@@ -164,7 +167,6 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture) ...@@ -164,7 +167,6 @@ CAF_TEST_FIXTURE_SCOPE(endpoint_manager_tests, fixture)
CAF_TEST(send and receive) { CAF_TEST(send and receive) {
std::vector<byte> read_buf(1024); std::vector<byte> read_buf(1024);
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 1u);
auto buf = std::make_shared<std::vector<byte>>(); auto buf = std::make_shared<std::vector<byte>>();
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
nonblocking(sockets.second, true); nonblocking(sockets.second, true);
...@@ -173,7 +175,9 @@ CAF_TEST(send and receive) { ...@@ -173,7 +175,9 @@ CAF_TEST(send and receive) {
auto guard = detail::make_scope_guard([&] { close(sockets.second); }); auto guard = detail::make_scope_guard([&] { close(sockets.second); });
auto mgr = make_endpoint_manager(mpx, sys, auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf}); dummy_transport{sockets.first, buf});
CAF_CHECK_EQUAL(mgr->mask(), operation::none);
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
CAF_CHECK_EQUAL(mgr->mask(), operation::read_write);
mpx->handle_updates(); mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 2u);
CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))), CAF_CHECK_EQUAL(write(sockets.second, as_bytes(make_span(hello_manager))),
...@@ -197,6 +201,7 @@ CAF_TEST(resolve and proxy communication) { ...@@ -197,6 +201,7 @@ CAF_TEST(resolve and proxy communication) {
auto mgr = make_endpoint_manager(mpx, sys, auto mgr = make_endpoint_manager(mpx, sys,
dummy_transport{sockets.first, buf}); dummy_transport{sockets.first, buf});
CAF_CHECK_EQUAL(mgr->init(), none); CAF_CHECK_EQUAL(mgr->init(), none);
CAF_CHECK_EQUAL(mgr->mask(), operation::read_write);
mpx->handle_updates(); mpx->handle_updates();
run(); run();
CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)), hello_test.size()); CAF_CHECK_EQUAL(read(sockets.second, make_span(read_buf)), hello_test.size());
......
...@@ -123,7 +123,7 @@ using dummy_manager_ptr = intrusive_ptr<dummy_manager>; ...@@ -123,7 +123,7 @@ using dummy_manager_ptr = intrusive_ptr<dummy_manager>;
struct fixture : host_fixture { struct fixture : host_fixture {
fixture() : manager_count(0), mpx(std::make_shared<multiplexer>()) { fixture() : manager_count(0), mpx(std::make_shared<multiplexer>()) {
// nop mpx->set_thread_id();
} }
~fixture() { ~fixture() {
...@@ -165,12 +165,12 @@ CAF_TEST(send and receive) { ...@@ -165,12 +165,12 @@ CAF_TEST(send and receive) {
auto sockets = unbox(make_stream_socket_pair()); auto sockets = unbox(make_stream_socket_pair());
auto alice = make_counted<dummy_manager>(manager_count, sockets.first, mpx); auto alice = make_counted<dummy_manager>(manager_count, sockets.first, mpx);
auto bob = make_counted<dummy_manager>(manager_count, sockets.second, mpx); auto bob = make_counted<dummy_manager>(manager_count, sockets.second, mpx);
alice->mask_add(operation::read); alice->register_reading();
bob->mask_add(operation::read); bob->register_reading();
mpx->handle_updates(); mpx->handle_updates();
CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u); CAF_CHECK_EQUAL(mpx->num_socket_managers(), 3u);
alice->send("hello bob"); alice->send("hello bob");
alice->mask_add(operation::write); alice->register_writing();
mpx->handle_updates(); mpx->handle_updates();
exhaust(); exhaust();
CAF_CHECK_EQUAL(bob->receive(), "hello bob"); CAF_CHECK_EQUAL(bob->receive(), "hello bob");
......
...@@ -71,7 +71,7 @@ class planet : public test_coordinator_fixture<config<Node>> { ...@@ -71,7 +71,7 @@ class planet : public test_coordinator_fixture<config<Node>> {
public: public:
planet(planet_driver& driver) planet(planet_driver& driver)
: mpx(*this->sys.network_manager().mpx()), driver_(driver) { : mpx(*this->sys.network_manager().mpx()), driver_(driver) {
// nop mpx.set_thread_id();
} }
net::backend::test& backend() { net::backend::test& backend() {
......
...@@ -71,6 +71,7 @@ struct config : actor_system_config { ...@@ -71,6 +71,7 @@ struct config : actor_system_config {
struct fixture : test_coordinator_fixture<config>, host_fixture { struct fixture : test_coordinator_fixture<config>, host_fixture {
fixture() : mars(make_node_id(unbox(make_uri("test:mars")))) { fixture() : mars(make_node_id(unbox(make_uri("test:mars")))) {
auto& mm = sys.network_manager(); auto& mm = sys.network_manager();
mm.mpx()->set_thread_id();
auto backend = dynamic_cast<backend::test*>(mm.backend("test")); auto backend = dynamic_cast<backend::test*>(mm.backend("test"));
auto mgr = backend->peer(mars); auto mgr = backend->peer(mars);
auto& dref = dynamic_cast<endpoint_manager_impl<transport_type>&>(*mgr); auto& dref = dynamic_cast<endpoint_manager_impl<transport_type>&>(*mgr);
......
...@@ -48,6 +48,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -48,6 +48,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
mpx = std::make_shared<multiplexer>(); mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init()) if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err)); CAF_FAIL("mpx->init failed: " << sys.render(err));
mpx->set_thread_id();
} }
bool handle_io_event() override { bool handle_io_event() override {
...@@ -74,7 +75,7 @@ public: ...@@ -74,7 +75,7 @@ public:
template <class Transport> template <class Transport>
void write_message(Transport& 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); transport.write_packet(span<byte>{}, msg->payload);
} }
......
...@@ -49,6 +49,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture { ...@@ -49,6 +49,7 @@ struct fixture : test_coordinator_fixture<>, host_fixture {
mpx = std::make_shared<multiplexer>(); mpx = std::make_shared<multiplexer>();
if (auto err = mpx->init()) if (auto err = mpx->init())
CAF_FAIL("mpx->init failed: " << sys.render(err)); CAF_FAIL("mpx->init failed: " << sys.render(err));
mpx->set_thread_id();
} }
bool handle_io_event() override { bool handle_io_event() override {
...@@ -99,7 +100,7 @@ public: ...@@ -99,7 +100,7 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& 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. // Ignore proxy announcement messages.
if (msg->msg == nullptr) if (msg->msg == nullptr)
return; return;
......
...@@ -72,7 +72,7 @@ public: ...@@ -72,7 +72,7 @@ public:
template <class Parent> template <class Parent>
void write_message(Parent& 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); parent.write_packet(span<const byte>{}, msg->payload);
} }
...@@ -192,8 +192,8 @@ CAF_TEST(write_message) { ...@@ -192,8 +192,8 @@ CAF_TEST(write_message) {
const_cast<char*>(hello_test.data())), const_cast<char*>(hello_test.data())),
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), using message_type = endpoint_manager_queue::message;
nullptr, auto message = detail::make_unique<message_type>(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;
......
...@@ -64,7 +64,7 @@ public: ...@@ -64,7 +64,7 @@ public:
template <class Transport> template <class Transport>
void write_message(Transport& 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_)); rec_buf_->push_back(static_cast<byte>(id_));
transport.write_packet(span<byte>{}, make_span(msg->payload)); transport.write_packet(span<byte>{}, make_span(msg->payload));
} }
...@@ -175,7 +175,7 @@ struct fixture : host_fixture { ...@@ -175,7 +175,7 @@ struct fixture : host_fixture {
add_new_workers(); add_new_workers();
} }
std::unique_ptr<net::endpoint_manager::message> std::unique_ptr<net::endpoint_manager_queue::message>
make_dummy_message(node_id nid) { make_dummy_message(node_id nid) {
actor_id aid = 42; actor_id aid = 42;
actor_config cfg; actor_config cfg;
...@@ -187,7 +187,7 @@ struct fixture : host_fixture { ...@@ -187,7 +187,7 @@ struct fixture : host_fixture {
auto elem = make_mailbox_element(std::move(strong_actor), auto elem = make_mailbox_element(std::move(strong_actor),
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_queue::message>(std::move(elem),
strong_actor, 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