effective_replication_map: erase from factory when destroyed

The effective_replication_map_factory keeps nakes pointers
to outstanding effective_replication_map:s.
These are kept valid using a shared effective_replication_map_ptr.

When the last shared ptr reference is dropped the effective_replication_map
object is destroyed, therefore the raw pointer to it in the factory
must be erased.

This now happens in ~effective_replication_map when the object
is marked as registered.

Registration happens when effective_replication_map_factory inserts
the newly created effective_replication_map to its _replication_maps
map, and the factory calles effective_replication_map::set_factory..

Note that effective_replication_map may be created temporarily
and not be inserted to the factory's map, therefore erase
is called only when required.

Signed-off-by: Benny Halevy <bhalevy@scylladb.com>

locator/abstract_replication_strategy.cc

@@ -345,6 +345,12 @@ future<> effective_replication_map::clear_gently() noexcept {
     co_await utils::clear_gently(_tmptr);
 }
 
+effective_replication_map::~effective_replication_map() {
+    if (is_registered()) {
+        _factory->erase_effective_replication_map(this);
+    }
+}
+
 effective_replication_map::factory_key effective_replication_map::make_factory_key(const abstract_replication_strategy::ptr_type& rs, const token_metadata_ptr& tmptr) {
     return factory_key(rs->get_type(), rs->get_config_options(), tmptr->get_ring_version());
 }
@@ -377,13 +383,30 @@ effective_replication_map_ptr effective_replication_map_factory::insert_effectiv
     auto [it, inserted] = _effective_replication_maps.insert({key, erm.get()});
     if (inserted) {
         rslogger.debug("insert_effective_replication_map: inserted {} [{}]", key, fmt::ptr(erm.get()));
+        erm->set_factory(*this, std::move(key));
         return erm;
     }
     auto res = it->second->shared_from_this();
     rslogger.debug("insert_effective_replication_map: found {} [{}]", key, fmt::ptr(res.get()));
     return res;
 }
 
+bool effective_replication_map_factory::erase_effective_replication_map(effective_replication_map* erm) {
+    const auto& key = erm->get_factory_key();
+    auto it = _effective_replication_maps.find(key);
+    if (it == _effective_replication_maps.end()) {
+        rslogger.warn("Could not unregister effective_replication_map {} [{}]: key not found", key, fmt::ptr(erm));
+        return false;
+    }
+    if (it->second != erm) {
+        rslogger.warn("Could not unregister effective_replication_map {} [{}]: different instance [{}] is currently registered", key, fmt::ptr(erm), fmt::ptr(it->second));
+        return false;
+    }
+    rslogger.debug("erase_effective_replication_map: erased {} [{}]", key, fmt::ptr(erm));
+    _effective_replication_maps.erase(it);
+    return true;
+}
+
 } // namespace locator
 
 std::ostream& operator<<(std::ostream& os, locator::replication_strategy_type t) {

locator/abstract_replication_strategy.hh

@@ -56,6 +56,7 @@ using replication_strategy_config_options = std::map<sstring, sstring>;
 using replication_map = std::unordered_map<token, inet_address_vector_replica_set>;
 
 class effective_replication_map;
+class effective_replication_map_factory;
 
 class abstract_replication_strategy {
     friend class effective_replication_map;
@@ -163,8 +164,11 @@ private:
     token_metadata_ptr _tmptr;
     replication_map _replication_map;
     size_t _replication_factor;
+    std::optional<factory_key> _factory_key = std::nullopt;
+    effective_replication_map_factory* _factory = nullptr;
 
     friend class abstract_replication_strategy;
+    friend class effective_replication_map_factory;
 public:
     explicit effective_replication_map(abstract_replication_strategy::ptr_type rs, token_metadata_ptr tmptr, replication_map replication_map, size_t replication_factor) noexcept
         : _rs(std::move(rs))
@@ -174,6 +178,7 @@ public:
     { }
     effective_replication_map() = delete;
     effective_replication_map(effective_replication_map&&) = default;
+    ~effective_replication_map();
 
     const token_metadata_ptr& get_token_metadata_ptr() const noexcept {
         return _tmptr;
@@ -227,6 +232,19 @@ private:
 
 public:
     static factory_key make_factory_key(const abstract_replication_strategy::ptr_type& rs, const token_metadata_ptr& tmptr);
+
+    const factory_key& get_factory_key() const noexcept {
+        return *_factory_key;
+    }
+
+    void set_factory(effective_replication_map_factory& factory, factory_key key) noexcept {
+        _factory = &factory;
+        _factory_key.emplace(std::move(key));
+    }
+
+    bool is_registered() const noexcept {
+        return _factory != nullptr;
+    }
 };
 
 using effective_replication_map_ptr = lw_shared_ptr<const effective_replication_map>;
@@ -315,7 +333,9 @@ private:
     effective_replication_map_ptr find_effective_replication_map(const effective_replication_map::factory_key& key) const;
     effective_replication_map_ptr insert_effective_replication_map(mutable_effective_replication_map_ptr erm, effective_replication_map::factory_key key);
 
-    // TODO: erase effective_replication_map when destroyed
+    bool erase_effective_replication_map(effective_replication_map* erm);
+
+    friend class effective_replication_map;
 };
 
 }