Hinted handoff: use after free during shutdown

[vladzcloudius]

Installation details
HEAD: 77686ab

Description
ASAN caught a use-after-free during regular shut-down process during unit testing the fix for #4685.
(I had to patch space_watchdog's code in order to simulate the race, see the patch below).
It turns out that the resource manager shutdown is not performed in the correct way. It shuts down managers (which in turn shut down all underlying end_point_managers and deletes all end_point_managers instances) and only after that it shuts down a space_watchdog which may have been blocked waiting for an <some ep_manager object>::file_update_mutex().

The proper shut down procedure should be:

1. Prevent hints from being generated.
2. Stop a resource_manager which should be stopping a space_watchdog.
3. Stop all managers.

Patch used to trigger the race

diff --git a/db/hints/resource_manager.hh b/db/hints/resource_manager.hh
index 453d51b51..864db57d0 100644
--- a/db/hints/resource_manager.hh
+++ b/db/hints/resource_manager.hh
@@ -51,7 +51,7 @@ class manager;
 class space_watchdog {
 private:
     using ep_key_type = gms::inet_address;
-    static const std::chrono::seconds _watchdog_period;
+    static const std::chrono::microseconds _watchdog_period;
 
     enum class state {
         stopping,               // stop() was called
diff --git a/db/hints/resource_manager.cc b/db/hints/resource_manager.cc
index 2ec4d67f0..dd990eaa2 100644
--- a/db/hints/resource_manager.cc
+++ b/db/hints/resource_manager.cc
@@ -60,7 +60,7 @@ future<semaphore_units<semaphore_default_exception_factory>> resource_manager::g
     return get_units(_send_limiter, hint_memory_budget);
 }
 
-const std::chrono::seconds space_watchdog::_watchdog_period = std::chrono::seconds(1);
+const std::chrono::microseconds space_watchdog::_watchdog_period = std::chrono::microseconds(1);
 
 space_watchdog::space_watchdog(shard_managers_set& managers, per_device_limits_map& per_device_limits_map)
     : _shard_managers(managers)
@@ -144,8 +144,10 @@ void space_watchdog::on_timer() {
                     // continue to enumeration - there is no one to change them.
                     auto it = shard_manager.find_ep_manager(de.name);
                     if (it != shard_manager.ep_managers_end()) {
-                        return with_lock(it->second.file_update_mutex(), [this, &shard_manager, dir = std::move(dir), ep_name = std::move(de.name)]() mutable {
-                            return scan_one_ep_dir(dir / ep_name, shard_manager, ep_key_type(ep_name));
+                        return with_lock(it->second.file_update_mutex(), [this, &shard_manager, dir = std::move(dir), ep_name = std::move(de.name)] () mutable {
+                            return sleep_abortable(std::chrono::seconds(5)).then([this, &shard_manager, dir = std::move(dir), ep_name = std::move(ep_name)] () mutable {
+                                return scan_one_ep_dir(dir / ep_name, shard_manager, ep_key_type(ep_name));
+                            });
                         });
                     } else {
                         return scan_one_ep_dir(dir / de.name, shard_manager, ep_key_type(de.name));

[bhalevy]

Refs #5089