From e9d94894f1d76c961d0fbb7367098fcd45ca4e9a Mon Sep 17 00:00:00 2001
From: Aleksandra Martyniuk <aleksandra.martyniuk@scylladb.com>
Date: Thu, 10 Aug 2023 10:40:46 +0200
Subject: [PATCH] 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 | 27 ++++++++++++++++++++++++---
 compaction/compaction_manager.hh |  1 +
 2 files changed, 25 insertions(+), 3 deletions(-)

diff --git a/compaction/compaction_manager.cc b/compaction/compaction_manager.cc
index d89d28d4778f..c9d310fbc946 100644
--- a/compaction/compaction_manager.cc
+++ b/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 {
diff --git a/compaction/compaction_manager.hh b/compaction/compaction_manager.hh
index 6e04f0515d1d..caed6797ea14 100644
--- a/compaction/compaction_manager.hh
+++ b/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