sstables: make delete_atomically() throw a distinct exception when cancelled

Throwing a runtime_error makes it impossible to catch the cancellation
exception, so replace it with a distinct exception class.

sstables/sstables.cc

@@ -45,6 +45,7 @@
 #include <boost/filesystem/operations.hpp>
 #include <boost/algorithm/string.hpp>
 #include <boost/range/adaptor/map.hpp>
+#include <boost/range/adaptor/transformed.hpp>
 #include <boost/range/algorithm_ext/insert.hpp>
 #include <boost/range/algorithm_ext/push_back.hpp>
 #include <boost/range/algorithm/set_algorithm.hpp>
@@ -2048,7 +2049,9 @@ do_delete_atomically(std::vector<sstable_to_delete> atomic_deletion_set, unsigne
 
     if (g_atomic_deletions_cancelled) {
         deletion_logger.debug("atomic deletions disabled, erroring out");
-        throw std::runtime_error(sprint("atomic deletions disabled; not deleting %s", atomic_deletion_set));
+        using boost::adaptors::transformed;
+        throw atomic_deletion_cancelled(atomic_deletion_set
+                                        | transformed(std::mem_fn(&sstable_to_delete::name)));
     }
 
     // Insert atomic_deletion_set into the list of sets pending deletion.  If the new set
@@ -2143,11 +2146,20 @@ cancel_atomic_deletions() {
     g_atomic_deletions_cancelled = true;
     for (auto&& pd : g_atomic_deletion_sets) {
         for (auto&& c : pd->completions) {
-            c->set_exception(std::runtime_error(sprint("Atomic sstable deletions cancelled; not deleting %s", pd->names)));
+            c->set_exception(atomic_deletion_cancelled(pd->names));
         }
     }
     g_atomic_deletion_sets.clear();
     g_shards_agreeing_to_delete_sstable.clear();
 }
 
+atomic_deletion_cancelled::atomic_deletion_cancelled(std::vector<sstring> names)
+        : _msg(sprint("atomic deletions cancelled; not deleting %s", names)) {
+}
+
+const char*
+atomic_deletion_cancelled::what() const noexcept {
+    return _msg.c_str();
+}
+
 }

sstables/sstables.hh

@@ -605,12 +605,26 @@ struct sstable_to_delete {
 // shared among shard, so actual on-disk deletion of an sstable is deferred
 // until all shards agree it can be deleted.
 //
+// When shutting down, we will not be able to complete some deletions.
+// In that case, an atomic_deletion_cancelled exception is returned instead.
+//
 // This function only solves the second problem for now.
 future<> delete_atomically(std::vector<shared_sstable> ssts);
 future<> delete_atomically(std::vector<sstable_to_delete> ssts);
 
+class atomic_deletion_cancelled : public std::exception {
+    std::string _msg;
+public:
+    explicit atomic_deletion_cancelled(std::vector<sstring> names);
+    template <typename StringRange>
+    explicit atomic_deletion_cancelled(StringRange range)
+            : atomic_deletion_cancelled(std::vector<sstring>{range.begin(), range.end()}) {
+    }
+    const char* what() const noexcept override;
+};
+
 // Cancel any deletions scheduled by delete_atomically() and make their
-// futures complete
+// futures complete (with an atomic_deletion_cancelled exception).
 void cancel_atomic_deletions();
 
 }