Commit 9f40d339 authored by Dominik Charousset's avatar Dominik Charousset

Use only the local slot ID as scatterer path key

parent b52b3770
...@@ -36,11 +36,11 @@ public: ...@@ -36,11 +36,11 @@ public:
using cache_type = std::vector<T>; using cache_type = std::vector<T>;
/// Maps slot IDs to caches. /// Maps slot IDs to caches.
using cache_map_type = detail::unordered_flat_map<stream_slots, cache_type>; using cache_map_type = detail::unordered_flat_map<stream_slot, cache_type>;
typename super::path_ptr add_path(stream_slots slots, typename super::path_ptr add_path(stream_slots slots,
strong_actor_ptr target) override { strong_actor_ptr target) override {
auto res = caches_.emplace(slots, cache_type{}); auto res = caches_.emplace(slots.sender, cache_type{});
return res.second ? super::add_path(slots, target) : nullptr; return res.second ? super::add_path(slots, target) : nullptr;
} }
...@@ -59,7 +59,8 @@ public: ...@@ -59,7 +59,8 @@ public:
} }
void force_emit_batches() override { void force_emit_batches() override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE(CAF_ARG2("buffered", this->buffered())
<< CAF_ARG2("paths", this->paths_.size()));
// Handler for the last (underful) batch that sends it anyway. // Handler for the last (underful) batch that sends it anyway.
auto f = [&](outbound_path& p, cache_type& c) { auto f = [&](outbound_path& p, cache_type& c) {
p.emit_batch(this->self_, c.size(), make_message(std::move(c))); p.emit_batch(this->self_, c.size(), make_message(std::move(c)));
...@@ -70,7 +71,8 @@ public: ...@@ -70,7 +71,8 @@ public:
protected: protected:
void about_to_erase(typename super::map_type::iterator i, bool silent, void about_to_erase(typename super::map_type::iterator i, bool silent,
error* reason) override { error* reason) override {
caches_.erase(i->second->slots); CAF_LOG_DEBUG("remove cache:" << CAF_ARG2("slot", i->second->slots.sender));
caches_.erase(i->second->slots.sender);
super::about_to_erase(i, silent, reason); super::about_to_erase(i, silent, reason);
} }
...@@ -98,6 +100,8 @@ private: ...@@ -98,6 +100,8 @@ private:
if (this->paths_.empty()) if (this->paths_.empty())
return; return;
auto emit_impl = [&](outbound_path& p, cache_type& c) { auto emit_impl = [&](outbound_path& p, cache_type& c) {
CAF_LOG_DEBUG("emit up to" << c.size()
<< "items on slot" << p.slots.sender);
if (c.empty()) if (c.empty())
return; return;
auto dbs = p.desired_batch_size; auto dbs = p.desired_batch_size;
......
...@@ -44,7 +44,7 @@ public: ...@@ -44,7 +44,7 @@ public:
using path_unique_ptr = std::unique_ptr<path_type>; using path_unique_ptr = std::unique_ptr<path_type>;
using map_type = detail::unordered_flat_map<stream_slots, path_unique_ptr>; using map_type = detail::unordered_flat_map<stream_slot, path_unique_ptr>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
...@@ -64,10 +64,10 @@ public: ...@@ -64,10 +64,10 @@ public:
virtual path_ptr add_path(stream_slots slots, strong_actor_ptr target); virtual path_ptr add_path(stream_slots slots, strong_actor_ptr target);
/// Removes a path from the scatterer and returns it. /// Removes a path from the scatterer and returns it.
path_unique_ptr take_path(stream_slots slots) noexcept; path_unique_ptr take_path(stream_slot slots) noexcept;
/// Returns the path associated to `slots` or `nullptr`. /// Returns the path associated to `slots` or `nullptr`.
path_ptr path(stream_slots slots) noexcept; path_ptr path(stream_slot slots) noexcept;
/// Returns the stored state for `x` if `x` is a known path and associated to /// Returns the stored state for `x` if `x` is a known path and associated to
/// `sid`, otherwise `nullptr`. /// `sid`, otherwise `nullptr`.
......
...@@ -70,7 +70,7 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) { ...@@ -70,7 +70,7 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_open& x) {
void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) { void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(x));
auto path = out().path(slots.invert()); auto path = out().path(slots.receiver);
if (path != nullptr) { if (path != nullptr) {
path->open_credit += x.new_capacity; path->open_credit += x.new_capacity;
path->desired_batch_size = x.desired_batch_size; path->desired_batch_size = x.desired_batch_size;
...@@ -80,11 +80,11 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) { ...@@ -80,11 +80,11 @@ void stream_manager::handle(stream_slots slots, upstream_msg::ack_batch& x) {
} }
void stream_manager::handle(stream_slots slots, upstream_msg::drop&) { void stream_manager::handle(stream_slots slots, upstream_msg::drop&) {
out().take_path(slots.invert()); out().take_path(slots.receiver);
} }
void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) { void stream_manager::handle(stream_slots slots, upstream_msg::forced_drop& x) {
if (out().path(slots.invert()) != nullptr) if (out().path(slots.receiver) != nullptr)
abort(std::move(x.reason)); abort(std::move(x.reason));
} }
......
...@@ -47,7 +47,7 @@ stream_scatterer::~stream_scatterer() { ...@@ -47,7 +47,7 @@ stream_scatterer::~stream_scatterer() {
pointer stream_scatterer::add_path(stream_slots slots, pointer stream_scatterer::add_path(stream_slots slots,
strong_actor_ptr target) { strong_actor_ptr target) {
CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(target)); CAF_LOG_TRACE(CAF_ARG(slots) << CAF_ARG(target));
auto res = paths_.emplace(slots, nullptr); auto res = paths_.emplace(slots.sender, nullptr);
if (res.second) { if (res.second) {
auto ptr = new outbound_path(slots, std::move(target)); auto ptr = new outbound_path(slots, std::move(target));
res.first->second.reset(ptr); res.first->second.reset(ptr);
...@@ -56,9 +56,9 @@ pointer stream_scatterer::add_path(stream_slots slots, ...@@ -56,9 +56,9 @@ pointer stream_scatterer::add_path(stream_slots slots,
return nullptr; return nullptr;
} }
unique_pointer stream_scatterer::take_path(stream_slots slots) noexcept { unique_pointer stream_scatterer::take_path(stream_slot slot) noexcept {
unique_pointer result; unique_pointer result;
auto i = paths_.find(slots); auto i = paths_.find(slot);
if (i != paths_.end()) { if (i != paths_.end()) {
result.swap(i->second); result.swap(i->second);
paths_.erase(i); paths_.erase(i);
...@@ -66,8 +66,8 @@ unique_pointer stream_scatterer::take_path(stream_slots slots) noexcept { ...@@ -66,8 +66,8 @@ unique_pointer stream_scatterer::take_path(stream_slots slots) noexcept {
return result; return result;
} }
pointer stream_scatterer::path(stream_slots slots) noexcept { pointer stream_scatterer::path(stream_slot slot) noexcept {
auto i = paths_.find(slots); auto i = paths_.find(slot);
return i != paths_.end() ? i->second.get() : nullptr; return i != paths_.end() ? i->second.get() : nullptr;
} }
......
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