Refactor code (#2987)

### What problem does this PR solve?

1. Refactor code
2. Make txn_context can be viewed in debug mode.

### Type of change

- [x] Refactoring

---------

Signed-off-by: Jin Hai <[email protected]>

Author: JinHai-CN

src/common/blocking_queue.cppm

@@ -40,7 +40,7 @@ public:
 
     void NotAllowEnqueue() { allow_enqueue_ = false; }
 
-    bool Enqueue(T &task) {
+    bool Enqueue(const T &task) {
         {
             if (!allow_enqueue_) {
                 return false;

src/executor/operator/physical_command_impl.cpp

@@ -299,7 +299,7 @@ bool PhysicalCommand::Execute(QueryContext *query_context, OperatorState *operat
                                 Status status = Status::InvalidCommand(fmt::format("Attempt to set compact segment interval: {}", interval));
                                 RecoverableError(status);
                             }
-                            query_context->storage()->periodic_trigger_thread()->compact_segment_trigger_->UpdateInternal(interval);
+                            query_context->storage()->periodic_trigger_thread()->compact_trigger_->UpdateInternal(interval);
                             config->SetCompactInterval(interval);
                             break;
                         }

src/storage/background_process.cppm

@@ -18,7 +18,6 @@ import :blocking_queue;
 
 namespace infinity {
 
-class CleanupPeriodicTrigger;
 class BGTask;
 
 export class BGTaskProcessor {

src/storage/background_process_impl.cpp

@@ -75,7 +75,7 @@ void BGTaskProcessor::Process() {
                     running = false;
                     break;
                 }
-                case BGTaskType::kNewCheckpoint: {
+                case BGTaskType::kCheckpoint: {
                     StorageMode storage_mode = InfinityContext::instance().storage()->GetStorageMode();
                     if (storage_mode == StorageMode::kUnInitialized) {
                         UnrecoverableError("Uninitialized storage mode");
@@ -102,7 +102,7 @@ void BGTaskProcessor::Process() {
                     }
                     break;
                 }
-                case BGTaskType::kNewCleanup: {
+                case BGTaskType::kCleanup: {
                     StorageMode storage_mode = InfinityContext::instance().storage()->GetStorageMode();
                     if (storage_mode == StorageMode::kUnInitialized) {
                         UnrecoverableError("Uninitialized storage mode");
@@ -122,8 +122,8 @@ void BGTaskProcessor::Process() {
                         CleanupTxnStore *cleanup_txn_store = static_cast<CleanupTxnStore *>(new_txn_shared->GetTxnStore());
                         if (cleanup_txn_store != nullptr) {
                             TxnTimeStamp clean_ts = cleanup_txn_store->timestamp_;
-                            std::shared_ptr<BGTaskInfo> bg_task_info = std::make_shared<BGTaskInfo>(BGTaskType::kNewCleanup);
-                            std::string task_text = fmt::format("NewCleanup task, cleanup timestamp: {}", clean_ts);
+                            std::shared_ptr<BGTaskInfo> bg_task_info = std::make_shared<BGTaskInfo>(BGTaskType::kCleanup);
+                            std::string task_text = fmt::format("Cleanup task, cleanup timestamp: {}", clean_ts);
                             bg_task_info->task_info_list_.emplace_back(task_text);
                             if (status.ok()) {
                                 bg_task_info->status_list_.emplace_back("OK");
@@ -133,7 +133,7 @@ void BGTaskProcessor::Process() {
                             }
                             new_txn_mgr->AddTaskInfo(bg_task_info);
                         }
-                        LOG_DEBUG("NewCleanup task in background done");
+                        LOG_DEBUG("Cleanup task in background done");
                     }
                     break;
                 }

src/storage/bg_task/bg_task.cppm

@@ -50,7 +50,7 @@ export struct BGTask {
     bool async_{false};
 
     bool complete_{false};
-    std::mutex mutex_{};
+    mutable std::mutex mutex_{};
     std::condition_variable cv_{};
 
     void Wait() {
@@ -70,6 +70,11 @@ export struct BGTask {
         cv_.notify_one();
     }
 
+    bool IsComplete() const {
+        std::unique_lock<std::mutex> locker(mutex_);
+        return complete_;
+    }
+
     Status result_status_{};
 
     virtual std::string ToString() const = 0;
@@ -88,7 +93,7 @@ export struct CheckpointTaskBase : public BGTask {
 };
 
 export struct NewCheckpointTask final : public CheckpointTaskBase {
-    NewCheckpointTask(i64 wal_size) : CheckpointTaskBase(BGTaskType::kNewCheckpoint, false), wal_size_(wal_size) {}
+    NewCheckpointTask(i64 wal_size) : CheckpointTaskBase(BGTaskType::kCheckpoint, false), wal_size_(wal_size) {}
 
     std::string ToString() const final { return "New catalog"; }
 
@@ -99,11 +104,11 @@ export struct NewCheckpointTask final : public CheckpointTaskBase {
     i64 wal_size_{};
 };
 
-export class NewCleanupTask final : public BGTask {
+export class CleanupTask final : public BGTask {
 public:
-    NewCleanupTask() : BGTask(BGTaskType::kNewCleanup, false) {}
+    CleanupTask() : BGTask(BGTaskType::kCleanup, false) {}
 
-    std::string ToString() const override { return "NewCleanupTask"; }
+    std::string ToString() const override { return "CleanupTask"; }
 
     Status Execute(TxnTimeStamp last_cleanup_ts, TxnTimeStamp &cur_cleanup_ts);
 
@@ -132,14 +137,11 @@ public:
 
 export class NotifyOptimizeTask final : public BGTask {
 public:
-    NotifyOptimizeTask(bool new_optimize = false) : BGTask(BGTaskType::kNotifyOptimize, true), new_optimize_(new_optimize) {}
+    NotifyOptimizeTask(bool new_optimize = false) : BGTask(BGTaskType::kNotifyOptimize, true) {}
 
     ~NotifyOptimizeTask() override = default;
 
     std::string ToString() const override { return "NotifyOptimizeTask"; }
-
-public:
-    bool new_optimize_ = false;
 };
 
 export class DumpMemIndexTask final : public BGTask {

src/storage/bg_task/bg_task_impl.cpp

@@ -57,7 +57,7 @@ Status NewCheckpointTask::ExecuteWithNewTxn() {
 
         auto *ckp_idx_store = static_cast<CheckpointTxnStore *>(new_txn_shared->GetTxnStore());
         if (ckp_idx_store != nullptr) {
-            std::shared_ptr<BGTaskInfo> bg_task_info = std::make_shared<BGTaskInfo>(BGTaskType::kNewCheckpoint);
+            std::shared_ptr<BGTaskInfo> bg_task_info = std::make_shared<BGTaskInfo>(BGTaskType::kCheckpoint);
             for (const std::shared_ptr<FlushDataEntry> &flush_data_entry : ckp_idx_store->entries_) {
                 std::string task_text = fmt::format("Txn: {}, commit: {}, checkpoint data: {}.{}.{}.{} {}",
                                                     new_txn_shared->TxnID(),
@@ -76,7 +76,7 @@ Status NewCheckpointTask::ExecuteWithNewTxn() {
     return status;
 }
 
-Status NewCleanupTask::Execute(TxnTimeStamp last_cleanup_ts, TxnTimeStamp &cur_cleanup_ts) {
+Status CleanupTask::Execute(TxnTimeStamp last_cleanup_ts, TxnTimeStamp &cur_cleanup_ts) {
     auto *new_txn_mgr = InfinityContext::instance().storage()->new_txn_manager();
     auto *txn = new_txn_mgr->BeginTxn(std::make_unique<std::string>("cleanup"), TransactionType::kCleanup);
     Status status = txn->Cleanup();
@@ -88,7 +88,7 @@ Status NewCleanupTask::Execute(TxnTimeStamp last_cleanup_ts, TxnTimeStamp &cur_c
 }
 
 NewCompactTask::NewCompactTask(NewTxn *new_txn, std::string db_name, std::string table_name)
-    : BGTask(BGTaskType::kNewCompact, false), new_txn_(new_txn), db_name_(db_name), table_name_(table_name) {}
+    : BGTask(BGTaskType::kCompact, false), new_txn_(new_txn), db_name_(db_name), table_name_(table_name) {}
 
 DumpMemIndexTask::DumpMemIndexTask(const std::string &db_name,
                                    const std::string &table_name,

src/storage/bg_task/bg_task_type.cppm

@@ -27,11 +27,11 @@ namespace infinity {
 
 export enum class BGTaskType {
     kStopProcessor,
-    kNewCheckpoint,
+    kCheckpoint,
     kNotifyCompact,
-    kNewCompact,
+    kCompact,
     kNotifyOptimize,
-    kNewCleanup,
+    kCleanup,
     kUpdateSegmentBloomFilterData, // Not used
     kDumpMemIndex,
     kAppendMemIndex,

src/storage/bg_task/bg_task_type_impl.cpp

@@ -34,24 +34,24 @@ std::string ToString(BGTaskType type) {
             return "StopProcessor";
             break;
         }
-        case BGTaskType::kNewCheckpoint: {
-            return "NewCheckpoint";
+        case BGTaskType::kCheckpoint: {
+            return "Checkpoint";
             break;
         }
         case BGTaskType::kNotifyCompact: {
             return "NotifyCompact";
             break;
         }
-        case BGTaskType::kNewCompact: {
-            return "NewCompact";
+        case BGTaskType::kCompact: {
+            return "Compact";
             break;
         }
         case BGTaskType::kNotifyOptimize: {
             return "NotifyOptimize";
             break;
         }
-        case BGTaskType::kNewCleanup: {
-            return "NewCleanup";
+        case BGTaskType::kCleanup: {
+            return "Cleanup";
             break;
         }
         case BGTaskType::kUpdateSegmentBloomFilterData: {

src/storage/bg_task/periodic_trigger.cppm

@@ -26,11 +26,13 @@ import global_resource_usage;
 
 namespace infinity {
 
-class NewCleanupTask;
+class CleanupTask;
+class NotifyOptimizeTask;
+class NotifyCompactTask;
 
 export class PeriodicTrigger {
 public:
-    explicit PeriodicTrigger(i64 interval) : interval_(interval), last_check_(std::chrono::system_clock::now()) {
+    explicit PeriodicTrigger(const i64 interval) : interval_(interval), last_check_(std::chrono::system_clock::now()) {
 #ifdef INFINITY_DEBUG
         GlobalResourceUsage::IncrObjectCount("PeriodicTrigger");
 #endif
@@ -44,7 +46,7 @@ public:
 
     bool Check();
 
-    void UpdateInternal(i64 new_interval) { interval_ = new_interval; }
+    void UpdateInternal(const i64 new_interval) { interval_ = new_interval; }
 
     virtual void Trigger() = 0;
 
@@ -56,51 +58,48 @@ protected:
     std::atomic_int64_t duration_{0};
 };
 
-export class NewCleanupPeriodicTrigger final : public PeriodicTrigger {
+export class CleanupPeriodicTrigger final : public PeriodicTrigger {
 public:
-    NewCleanupPeriodicTrigger(i64 interval) : PeriodicTrigger(interval) {}
+    explicit CleanupPeriodicTrigger(const i64 interval) : PeriodicTrigger(interval) {}
 
-    std::shared_ptr<NewCleanupTask> CreateNewCleanupTask();
+    std::shared_ptr<CleanupTask> CreateCleanupTask();
 
-    virtual void Trigger() override;
-
-private:
-    //
+    void Trigger() override;
 };
 
 export class CheckpointPeriodicTrigger final : public PeriodicTrigger {
 public:
-    explicit CheckpointPeriodicTrigger(i64 interval) : PeriodicTrigger(interval) {}
+    explicit CheckpointPeriodicTrigger(const i64 interval) : PeriodicTrigger(interval) {}
 
-    virtual void Trigger() override;
+    void Trigger() override;
 };
 
-export class CompactSegmentPeriodicTrigger final : public PeriodicTrigger {
+export class CompactPeriodicTrigger final : public PeriodicTrigger {
 public:
-    explicit CompactSegmentPeriodicTrigger(i64 interval, CompactionProcessor *compact_processor)
+    explicit CompactPeriodicTrigger(const i64 interval, CompactionProcessor *compact_processor)
         : PeriodicTrigger(interval), compact_processor_(compact_processor) {}
 
-    explicit CompactSegmentPeriodicTrigger(i64 interval) : PeriodicTrigger(interval), new_compaction_(true) {}
+    explicit CompactPeriodicTrigger(const i64 interval) : PeriodicTrigger(interval) {}
 
-    virtual void Trigger() override;
+    void Trigger() override;
 
 private:
-    CompactionProcessor *const compact_processor_{};
-    bool new_compaction_ = false;
+    CompactionProcessor *compact_processor_{};
+    std::shared_ptr<NotifyCompactTask> compact_task_{};
 };
 
 export class OptimizeIndexPeriodicTrigger final : public PeriodicTrigger {
 public:
-    explicit OptimizeIndexPeriodicTrigger(i64 interval, CompactionProcessor *compact_processor)
+    explicit OptimizeIndexPeriodicTrigger(const i64 interval, CompactionProcessor *compact_processor)
         : PeriodicTrigger(interval), compact_processor_(compact_processor) {}
 
-    explicit OptimizeIndexPeriodicTrigger(i64 interval) : PeriodicTrigger(interval), new_optimize_(true) {}
+    explicit OptimizeIndexPeriodicTrigger(const i64 interval) : PeriodicTrigger(interval) {}
 
-    virtual void Trigger() override;
+    void Trigger() override;
 
 private:
-    CompactionProcessor *const compact_processor_ = nullptr;
-    bool new_optimize_ = false;
+    CompactionProcessor *compact_processor_{};
+    std::shared_ptr<NotifyOptimizeTask> optimize_task_{};
 };
 
 } // namespace infinity

src/storage/bg_task/periodic_trigger_impl.cpp

@@ -45,7 +45,7 @@ bool PeriodicTrigger::Check() {
     return true;
 }
 
-std::shared_ptr<NewCleanupTask> NewCleanupPeriodicTrigger::CreateNewCleanupTask() {
+std::shared_ptr<CleanupTask> CleanupPeriodicTrigger::CreateCleanupTask() {
     auto *bg_processor = InfinityContext::instance().storage()->bg_processor();
     auto *new_txn_mgr = InfinityContext::instance().storage()->new_txn_manager();
 
@@ -55,11 +55,11 @@ std::shared_ptr<NewCleanupTask> NewCleanupPeriodicTrigger::CreateNewCleanupTask(
         return nullptr;
     }
 
-    return std::make_shared<NewCleanupTask>();
+    return std::make_shared<CleanupTask>();
 }
 
-void NewCleanupPeriodicTrigger::Trigger() {
-    auto cleanup_task = CreateNewCleanupTask();
+void CleanupPeriodicTrigger::Trigger() {
+    auto cleanup_task = CreateCleanupTask();
     if (!cleanup_task) {
         return;
     }
@@ -90,23 +90,25 @@ void CheckpointPeriodicTrigger::Trigger() {
     bg_processor->Submit(std::move(checkpoint_task));
 }
 
-void CompactSegmentPeriodicTrigger::Trigger() {
-    LOG_DEBUG(fmt::format("Trigger compact segment task, after {} seconds", duration_.load()));
-    auto compact_task = std::make_shared<NotifyCompactTask>();
-    auto *compact_processor = InfinityContext::instance().storage()->compaction_processor();
-    compact_processor->Submit(std::move(compact_task));
+void CompactPeriodicTrigger::Trigger() {
+    if (compact_task_ != nullptr and !compact_task_->IsComplete()) {
+        LOG_DEBUG(fmt::format("Skipping compact task, after {} seconds", duration_.load()));
+    }
+
+    LOG_DEBUG(fmt::format("Trigger compact task, after {} seconds", duration_.load()));
+    compact_task_ = std::make_shared<NotifyCompactTask>();
+    compact_processor_->Submit(compact_task_);
 }
 
 void OptimizeIndexPeriodicTrigger::Trigger() {
-    LOG_DEBUG(fmt::format("Trigger optimize index task, after {} seconds", duration_.load()));
-    if (!new_optimize_) {
-        auto optimize_task = std::make_shared<NotifyOptimizeTask>();
-        compact_processor_->Submit(std::move(optimize_task));
-    } else {
-        auto optimize_task = std::make_shared<NotifyOptimizeTask>(true);
-        auto *compact_processor = InfinityContext::instance().storage()->compaction_processor();
-        compact_processor->Submit(std::move(optimize_task));
+
+    if (optimize_task_ != nullptr and !optimize_task_->IsComplete()) {
+        LOG_DEBUG(fmt::format("Skipping optimize index task, after {} seconds", duration_.load()));
     }
+
+    LOG_DEBUG(fmt::format("Trigger optimize index task, after {} seconds", duration_.load()));
+    optimize_task_ = std::make_shared<NotifyOptimizeTask>();
+    compact_processor_->Submit(optimize_task_);
 }
 
 } // namespace infinity