Commit d10fbef5 authored by Dominik Charousset's avatar Dominik Charousset

Store stream managers by their local slot only

parent 130edc98
...@@ -135,6 +135,9 @@ public: ...@@ -135,6 +135,9 @@ public:
/// Base type. /// Base type.
using super = local_actor; using super = local_actor;
/// Maps slot IDs to stream managers.
using stream_manager_map = std::map<stream_slot, stream_manager_ptr>;
/// Stores asynchronous messages with default priority. /// Stores asynchronous messages with default priority.
using default_queue = intrusive::drr_cached_queue<policy::normal_messages>; using default_queue = intrusive::drr_cached_queue<policy::normal_messages>;
...@@ -445,12 +448,11 @@ public: ...@@ -445,12 +448,11 @@ public:
} }
template <class Driver, class Input, class... Ts> template <class Driver, class Input, class... Ts>
stream_result<typename Driver::output_type> make_sink(const stream<Input>& in, stream_result<typename Driver::output_type> make_sink(const stream<Input>&,
Ts&&... xs) { Ts&&... xs) {
auto slot = next_slot(); auto slot = next_slot();
stream_slots id{in.slot(), slot};
auto ptr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...); auto ptr = detail::make_stream_sink<Driver>(this, std::forward<Ts>(xs)...);
if (!add_stream_manager(id, ptr)) { if (!add_stream_manager(slot, ptr)) {
CAF_LOG_WARNING("unable to add a stream manager for a sink"); CAF_LOG_WARNING("unable to add a stream manager for a sink");
return {0, nullptr}; return {0, nullptr};
} }
...@@ -470,13 +472,12 @@ public: ...@@ -470,13 +472,12 @@ public:
template <class Driver, template <class Driver,
class Scatterer = broadcast_scatterer<typename Driver::output_type>, class Scatterer = broadcast_scatterer<typename Driver::output_type>,
class Input = int, class... Ts> class Input = int, class... Ts>
typename Driver::output_stream_type make_stage(const stream<Input>& in, typename Driver::output_stream_type make_stage(const stream<Input>&,
Ts&&... xs) { Ts&&... xs) {
auto slot = next_slot(); auto slot = next_slot();
stream_slots id{in.slot(), slot};
auto ptr = detail::make_stream_stage<Driver, Scatterer>( auto ptr = detail::make_stream_stage<Driver, Scatterer>(
this, std::forward<Ts>(xs)...); this, std::forward<Ts>(xs)...);
if (!add_stream_manager(id, ptr)) { if (!add_stream_manager(slot, ptr)) {
CAF_LOG_WARNING("unable to add a stream manager for a stage"); CAF_LOG_WARNING("unable to add a stream manager for a stage");
return {0, nullptr}; return {0, nullptr};
} }
...@@ -1034,7 +1035,7 @@ public: ...@@ -1034,7 +1035,7 @@ public:
template <class T> template <class T>
void handle_upstream_msg(stream_slots slots, actor_addr&, T& x) { void handle_upstream_msg(stream_slots slots, actor_addr&, T& x) {
auto i = stream_managers_.find(slots); auto i = stream_managers_.find(slots.receiver);
if (i == stream_managers_.end()) { if (i == stream_managers_.end()) {
auto j = pending_stream_managers_.find(slots.receiver); auto j = pending_stream_managers_.find(slots.receiver);
if (j != pending_stream_managers_.end()) { if (j != pending_stream_managers_.end()) {
...@@ -1108,10 +1109,10 @@ public: ...@@ -1108,10 +1109,10 @@ public:
/// Adds a new stream manager to the actor and starts cycle management if /// Adds a new stream manager to the actor and starts cycle management if
/// needed. /// needed.
bool add_stream_manager(stream_slots id, stream_manager_ptr ptr); bool add_stream_manager(stream_slot id, stream_manager_ptr ptr);
/// Removes a new stream manager. /// Removes a stream manager.
void erase_stream_manager(stream_slots id); void erase_stream_manager(stream_slot id);
/// @pre `x.content().match_elements<open_stream_msg>()` /// @pre `x.content().match_elements<open_stream_msg>()`
invoke_message_result handle_open_stream_msg(mailbox_element& x); invoke_message_result handle_open_stream_msg(mailbox_element& x);
...@@ -1151,11 +1152,11 @@ protected: ...@@ -1151,11 +1152,11 @@ protected:
exit_handler exit_handler_; exit_handler exit_handler_;
/// Stores stream managers for established streams. /// Stores stream managers for established streams.
std::map<stream_slots, stream_manager_ptr> stream_managers_; stream_manager_map stream_managers_;
/// Stores stream managers for pending streams, i.e., streams that have not /// Stores stream managers for pending streams, i.e., streams that have not
/// yet received an ACK. /// yet received an ACK.
std::map<stream_slot, stream_manager_ptr> pending_stream_managers_; stream_manager_map pending_stream_managers_;
/// Controls batch and credit timeouts. /// Controls batch and credit timeouts.
detail::tick_emitter stream_ticks_; detail::tick_emitter stream_ticks_;
......
...@@ -287,12 +287,13 @@ struct downstream_msg_visitor { ...@@ -287,12 +287,13 @@ struct downstream_msg_visitor {
// Do *not* store a reference here since we potentially reset `inptr`. // Do *not* store a reference here since we potentially reset `inptr`.
auto mgr = inptr->mgr; auto mgr = inptr->mgr;
inptr->handle(x); inptr->handle(x);
CAF_ASSERT(inptr->slots == dm.slots);
// TODO: replace with `if constexpr` when switching to C++17 // TODO: replace with `if constexpr` when switching to C++17
if (std::is_same<T, downstream_msg::close>::value if (std::is_same<T, downstream_msg::close>::value
|| std::is_same<T, downstream_msg::forced_close>::value) { || std::is_same<T, downstream_msg::forced_close>::value) {
inptr.reset(); inptr.reset();
qs_ref.erase_later(dm.slots.receiver); qs_ref.erase_later(dm.slots.receiver);
selfptr->erase_stream_manager(dm.slots); selfptr->erase_stream_manager(dm.slots.receiver);
if (mgr->done()) if (mgr->done())
mgr->stop(); mgr->stop();
return intrusive::task_result::stop; return intrusive::task_result::stop;
...@@ -300,7 +301,7 @@ struct downstream_msg_visitor { ...@@ -300,7 +301,7 @@ struct downstream_msg_visitor {
else if (mgr->done()) { else if (mgr->done()) {
CAF_LOG_DEBUG("path is done receiving and closes its manager"); CAF_LOG_DEBUG("path is done receiving and closes its manager");
mgr->stop(); mgr->stop();
selfptr->erase_stream_manager(dm.slots); selfptr->erase_stream_manager(dm.slots.receiver);
return intrusive::task_result::stop; return intrusive::task_result::stop;
} }
return intrusive::task_result::resume; return intrusive::task_result::resume;
...@@ -889,7 +890,7 @@ void scheduled_actor::handle_upstream_msg(stream_slots slots, ...@@ -889,7 +890,7 @@ void scheduled_actor::handle_upstream_msg(stream_slots slots,
} }
auto ptr = std::move(i->second); auto ptr = std::move(i->second);
pending_stream_managers_.erase(i); pending_stream_managers_.erase(i);
if (!add_stream_manager(slots, ptr)) { if (!add_stream_manager(slots.receiver, ptr)) {
CAF_LOG_WARNING("unable to add stream manager after receiving ack_open"); CAF_LOG_WARNING("unable to add stream manager after receiving ack_open");
return; return;
} }
...@@ -907,17 +908,17 @@ uint64_t scheduled_actor::set_timeout(atom_value type, ...@@ -907,17 +908,17 @@ uint64_t scheduled_actor::set_timeout(atom_value type,
stream_slot scheduled_actor::next_slot() { stream_slot scheduled_actor::next_slot() {
stream_slot result = 0; stream_slot result = 0;
auto nslot = [](stream_slot x) -> stream_slot { auto nslot = [](const stream_manager_map& x) -> stream_slot {
return x + 1; return x.rbegin()->first + 1;
}; };
if (!stream_managers_.empty()) if (!stream_managers_.empty())
result = std::max(nslot(stream_managers_.rbegin()->first.receiver), result); result = std::max(nslot(stream_managers_), result);
if (!pending_stream_managers_.empty()) if (!pending_stream_managers_.empty())
result = std::max(nslot(pending_stream_managers_.rbegin()->first), result); result = std::max(nslot(pending_stream_managers_), result);
return result; return result;
} }
bool scheduled_actor::add_stream_manager(stream_slots id, bool scheduled_actor::add_stream_manager(stream_slot id,
stream_manager_ptr ptr) { stream_manager_ptr ptr) {
CAF_LOG_TRACE(CAF_ARG(id)); CAF_LOG_TRACE(CAF_ARG(id));
if (stream_managers_.empty()) if (stream_managers_.empty())
...@@ -926,7 +927,7 @@ bool scheduled_actor::add_stream_manager(stream_slots id, ...@@ -926,7 +927,7 @@ bool scheduled_actor::add_stream_manager(stream_slots id,
return result; return result;
} }
void scheduled_actor::erase_stream_manager(stream_slots id) { void scheduled_actor::erase_stream_manager(stream_slot id) {
CAF_LOG_TRACE(CAF_ARG(id)); CAF_LOG_TRACE(CAF_ARG(id));
if (stream_managers_.erase(id) != 0 && stream_managers_.empty()) if (stream_managers_.erase(id) != 0 && stream_managers_.empty())
stream_ticks_.stop(); stream_ticks_.stop();
......
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