Refactor periodic trigger

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/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;
@@ -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/periodic_trigger.cppm

@@ -27,10 +27,12 @@ import global_resource_usage;
 namespace infinity {
 
 class NewCleanupTask;
+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;
 
@@ -58,47 +60,49 @@ protected:
 
 export class NewCleanupPeriodicTrigger final : public PeriodicTrigger {
 public:
-    NewCleanupPeriodicTrigger(i64 interval) : PeriodicTrigger(interval) {}
+    explicit NewCleanupPeriodicTrigger(const i64 interval) : PeriodicTrigger(interval) {}
 
     std::shared_ptr<NewCleanupTask> CreateNewCleanupTask();
 
-    virtual void Trigger() override;
+    void Trigger() override;
 
 private:
     //
 };
 
 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) {}
+    explicit CompactPeriodicTrigger(const i64 interval) : PeriodicTrigger(interval) {}
 
-    virtual void Trigger() override;
+    void Trigger() override;
 
 private:
     CompactionProcessor *const 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) {}
+    explicit OptimizeIndexPeriodicTrigger(const i64 interval) : PeriodicTrigger(interval) {}
 
-    virtual void Trigger() override;
+    void Trigger() override;
 
 private:
     CompactionProcessor *const compact_processor_ = nullptr;
+    std::shared_ptr<NotifyOptimizeTask> optimize_task_{};
 };
 
 } // namespace infinity

src/storage/bg_task/periodic_trigger_impl.cpp

@@ -90,16 +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>();
-    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() {
+
+    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()));
-    auto optimize_task = std::make_shared<NotifyOptimizeTask>(true);
-    compact_processor_->Submit(std::move(optimize_task));
+    optimize_task_ = std::make_shared<NotifyOptimizeTask>();
+    compact_processor_->Submit(optimize_task_);
 }
 
 } // namespace infinity

src/storage/compaction_process.cppm

@@ -56,7 +56,7 @@ public:
 
     void Stop();
 
-    void Submit(std::shared_ptr<BGTask> bg_task);
+    void Submit(const std::shared_ptr<BGTask> &bg_task);
 
     void NewDoCompact();
 

src/storage/compaction_process_impl.cpp

@@ -78,8 +78,8 @@ void CompactionProcessor::Stop() {
     LOG_INFO("Compaction processor is stopped.");
 }
 
-void CompactionProcessor::Submit(std::shared_ptr<BGTask> bg_task) {
-    task_queue_.Enqueue(std::move(bg_task));
+void CompactionProcessor::Submit(const std::shared_ptr<BGTask>& bg_task) {
+    task_queue_.Enqueue(bg_task);
     ++task_count_;
 }
 

src/storage/periodic_trigger_thread.cppm

@@ -33,7 +33,7 @@ public:
 public:
     std::shared_ptr<NewCleanupPeriodicTrigger> new_cleanup_trigger_;
     std::shared_ptr<CheckpointPeriodicTrigger> checkpoint_trigger_;
-    std::shared_ptr<CompactSegmentPeriodicTrigger> compact_segment_trigger_;
+    std::shared_ptr<CompactPeriodicTrigger> compact_segment_trigger_;
     std::shared_ptr<OptimizeIndexPeriodicTrigger> optimize_index_trigger_;
 
 private:

src/storage/storage_impl.cpp

@@ -370,12 +370,12 @@ Status Storage::AdminToWriter() {
     periodic_trigger_thread_->new_cleanup_trigger_ = std::make_shared<NewCleanupPeriodicTrigger>(cleanup_interval);
 
     i64 optimize_interval = config_ptr_->OptimizeIndexInterval() > 0 ? config_ptr_->OptimizeIndexInterval() : 0;
-    periodic_trigger_thread_->optimize_index_trigger_ = std::make_shared<OptimizeIndexPeriodicTrigger>(optimize_interval);
+    periodic_trigger_thread_->optimize_index_trigger_ = std::make_shared<OptimizeIndexPeriodicTrigger>(optimize_interval, compact_processor_.get());
 
     i64 checkpoint_interval_sec = config_ptr_->CheckpointInterval() > 0 ? config_ptr_->CheckpointInterval() : 0;
     periodic_trigger_thread_->checkpoint_trigger_ = std::make_shared<CheckpointPeriodicTrigger>(checkpoint_interval_sec);
 
-    periodic_trigger_thread_->compact_segment_trigger_ = std::make_shared<CompactSegmentPeriodicTrigger>(compact_interval);
+    periodic_trigger_thread_->compact_segment_trigger_ = std::make_shared<CompactPeriodicTrigger>(compact_interval, compact_processor_.get());
 
     periodic_trigger_thread_->Start();
 
@@ -506,7 +506,7 @@ Status Storage::ReaderToWriter() {
     //                i64 cleanup_interval = config_ptr_->CleanupInterval() > 0 ? config_ptr_->CleanupInterval() : 0;
     i64 checkpoint_interval_sec = config_ptr_->CheckpointInterval() > 0 ? config_ptr_->CheckpointInterval() : 0;
     periodic_trigger_thread_->checkpoint_trigger_ = std::make_shared<CheckpointPeriodicTrigger>(checkpoint_interval_sec);
-    periodic_trigger_thread_->compact_segment_trigger_ = std::make_shared<CompactSegmentPeriodicTrigger>(compact_interval, compact_processor_.get());
+    periodic_trigger_thread_->compact_segment_trigger_ = std::make_shared<CompactPeriodicTrigger>(compact_interval, compact_processor_.get());
     periodic_trigger_thread_->optimize_index_trigger_ = std::make_shared<OptimizeIndexPeriodicTrigger>(optimize_interval, compact_processor_.get());
     periodic_trigger_thread_->Start();