compaction: release resources of compaction executors

Before compaction task executors started inheriting from
compaction_task_impl, they were destructed immediately after
compaction finished. Destructors of executors and their
fields performed actions that affected global structures and
statistics and had impact on compaction process.

Currently, task executors are kept in memory much longer, as their
are tracked by task manager. Thus, destructors are not called just
after the compaction, which results in compaction stats not being
updated, which causes e.g. infinite cleanup loop.

Add release_resources() method which is called at the end
of compaction process and does what destructors used to.

Fixes: #14966.
Fixes: #15030.

Closes #15005

compaction/compaction_manager.cc

@@ -90,6 +90,11 @@ class compacting_sstable_registration {
         }
     }
 
+    void release_all() noexcept {
+        _cm.deregister_compacting_sstables(_compacting);
+        _compacting = {};
+    }
+
     class update_me : public compaction_task_executor::on_replacement {
         compacting_sstable_registration& _registration;
         public:
@@ -462,8 +467,9 @@ class sstables_task_executor : public compaction_task_executor, public sstables_
         set_sstables(std::move(sstables));
     }
 
-    virtual ~sstables_task_executor();
+    virtual ~sstables_task_executor() = default;
 
+    virtual void release_resources() noexcept override;
 
     virtual tasks::is_internal is_internal() const noexcept override {
         return tasks::is_internal::yes;
@@ -543,6 +549,7 @@ future<compaction_manager::compaction_stats_opt> compaction_manager::perform_com
     auto unregister_task = defer([this, task_executor] {
         _tasks.remove(task_executor);
         task_executor->switch_state(compaction_task_executor::state::none);
+        task_executor->release_resources();
     });
 
     if (parent_info) {
@@ -710,8 +717,10 @@ compaction::compaction_state::~compaction_state() {
     compaction_done.broken();
 }
 
-sstables_task_executor::~sstables_task_executor() {
+void sstables_task_executor::release_resources() noexcept {
     _cm._stats.pending_tasks -= _sstables.size() - (_state == state::pending);
+    _sstables = {};
+    compaction_task_executor::release_resources();
 }
 
 future<compaction_manager::compaction_stats_opt> compaction_task_executor::run_compaction() noexcept {
@@ -1427,6 +1436,12 @@ class rewrite_sstables_compaction_task_executor : public sstables_task_executor
         , _can_purge(can_purge)
     {}
 
+    virtual void release_resources() noexcept override {
+        _compacting.release_all();
+        _owned_ranges_ptr = nullptr;
+        sstables_task_executor::release_resources();
+    }
+
 protected:
     virtual future<compaction_manager::compaction_stats_opt> do_run() override {
         sstables::compaction_stats stats{};
@@ -1615,8 +1630,14 @@ class cleanup_sstables_compaction_task_executor : public compaction_task_executo
         _cm._stats.pending_tasks += _pending_cleanup_jobs.size();
     }
 
-    virtual ~cleanup_sstables_compaction_task_executor() {
+    virtual ~cleanup_sstables_compaction_task_executor() = default;
+
+    virtual void release_resources() noexcept override {
         _cm._stats.pending_tasks -= _pending_cleanup_jobs.size();
+        _pending_cleanup_jobs = {};
+        _compacting.release_all();
+        _owned_ranges_ptr = nullptr;
+        compaction_task_executor::release_resources();
     }
 
     virtual tasks::is_internal is_internal() const noexcept override {

compaction/compaction_manager.hh

@@ -480,6 +480,7 @@ public:
     compaction_task_executor(compaction_task_executor&&) = delete;
     compaction_task_executor(const compaction_task_executor&) = delete;
 
+    virtual void release_resources() noexcept {}
     virtual ~compaction_task_executor() = default;
 
     // called when a compaction replaces the exhausted sstables with the new set