diff --git a/CMakeLists.txt b/CMakeLists.txt index 06e8d310..64215d04 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -302,15 +302,11 @@ if(WITH_DATA_STORE STREQUAL "ELOQDSS_ROCKSDB" OR WITH_DATA_STORE STREQUAL "ELOQD elseif(WITH_DATA_STORE STREQUAL "ELOQDSS_ROCKSDB_CLOUD_S3") SET(DATA_SUBSTRATE_SOURCES ${DATA_SUBSTRATE_SOURCES} store_handler/eloq_data_store_service/rocksdb_cloud_data_store.cpp - store_handler/eloq_data_store_service/purger_event_listener.cpp - store_handler/eloq_data_store_service/purger_sliding_window.cpp store_handler/eloq_data_store_service/s3_file_downloader.cpp ) elseif(WITH_DATA_STORE STREQUAL "ELOQDSS_ROCKSDB_CLOUD_GCS") SET(DATA_SUBSTRATE_SOURCES ${DATA_SUBSTRATE_SOURCES} store_handler/eloq_data_store_service/rocksdb_cloud_data_store.cpp - store_handler/eloq_data_store_service/purger_event_listener.cpp - store_handler/eloq_data_store_service/purger_sliding_window.cpp ) endif() elseif(WITH_DATA_STORE STREQUAL "ELOQDSS_ELOQSTORE") diff --git a/store_handler/eloq_data_store_service/CMakeLists.txt b/store_handler/eloq_data_store_service/CMakeLists.txt index 7e9c8f3e..9d5eeccf 100644 --- a/store_handler/eloq_data_store_service/CMakeLists.txt +++ b/store_handler/eloq_data_store_service/CMakeLists.txt @@ -397,8 +397,6 @@ if ((WITH_DATA_STORE STREQUAL "ELOQDSS_ROCKSDB_CLOUD_S3") OR rocksdb_data_store_common.cpp rocksdb_cloud_data_store.cpp rocksdb_config.cpp - purger_event_listener.cpp - purger_sliding_window.cpp ) elseif (WITH_DATA_STORE STREQUAL "ELOQDSS_ROCKSDB") SET(RESELOQ_SOURCES ${RESELOQ_SOURCES} diff --git a/store_handler/eloq_data_store_service/purger_event_listener.cpp b/store_handler/eloq_data_store_service/purger_event_listener.cpp deleted file mode 100644 index 40148f8e..00000000 --- a/store_handler/eloq_data_store_service/purger_event_listener.cpp +++ /dev/null @@ -1,231 +0,0 @@ -/** - * Copyright (C) 2025 EloqData Inc. - * - * This program is free software: you can redistribute it and/or modify - * it under either of the following two licenses: - * 1. GNU Affero General Public License, version 3, as published by the Free - * Software Foundation. - * 2. GNU General Public License as published by the Free Software - * Foundation; version 2 of the License. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Affero General Public License or GNU General Public License for more - * details. - * - * You should have received a copy of the GNU Affero General Public License - * and GNU General Public License V2 along with this program. If not, see - * . - * - */ - -#include "purger_event_listener.h" - -#include - -#include - -namespace EloqDS -{ - -PurgerEventListener::PurgerEventListener( - const std::string &epoch, - const std::string &bucket_name, - const std::string &s3_object_path, - std::shared_ptr storage_provider, - std::chrono::milliseconds entry_duration, - std::chrono::milliseconds s3_update_interval) - : bucket_name_(bucket_name), s3_object_path_(s3_object_path) -{ - sliding_window_ = std::make_unique(entry_duration, - s3_update_interval, - epoch, - bucket_name_, - s3_object_path_, - storage_provider); - - LOG(INFO) << "PurgerEventListener created for epoch " << epoch - << ", bucket: " << bucket_name_ - << ", object_path: " << s3_object_path_ - << ", window_duration: " << entry_duration.count() << "ms" - << ", s3_update_interval: " << s3_update_interval.count() << "ms"; -} - -PurgerEventListener::~PurgerEventListener() -{ - Stop(); -} - -void PurgerEventListener::SetEpoch(const std::string &epoch) -{ - if (sliding_window_) - { - LOG(INFO) << "PurgerEventListener epoch updated from " - << (sliding_window_->GetEpoch().empty() - ? "empty" - : sliding_window_->GetEpoch()) - << " to " << epoch; - sliding_window_->SetEpoch(epoch); - } -} - -void PurgerEventListener::BlockPurger() -{ - if (sliding_window_) - { - sliding_window_->BlockPurger(); - } -} - -void PurgerEventListener::OnFlushBegin( - rocksdb::DB *db, const rocksdb::FlushJobInfo &flush_job_info) -{ - // Log flush begin event (similar to existing RocksDBEventListener) - if (flush_job_info.triggered_writes_slowdown || - flush_job_info.triggered_writes_stop) - { - LOG(INFO) << "[PurgerEventListener] Flush begin, file: " - << flush_job_info.file_path - << ", job_id: " << flush_job_info.job_id - << ", thread: " << flush_job_info.thread_id - << ", file_number: " << flush_job_info.file_number - << ", triggered_writes_slowdown: " - << flush_job_info.triggered_writes_slowdown - << ", triggered_writes_stop: " - << flush_job_info.triggered_writes_stop - << ", smallest_seqno: " << flush_job_info.smallest_seqno - << ", largest_seqno: " << flush_job_info.largest_seqno - << ", flush_reason: " - << GetFlushReason(flush_job_info.flush_reason); - } - - // Update sliding window with current max file number - UpdateSlidingWindow(db, flush_job_info.thread_id, flush_job_info.job_id); -} - -void PurgerEventListener::OnFlushCompleted( - rocksdb::DB *db, const rocksdb::FlushJobInfo &flush_job_info) -{ - // Log flush completion event - if (flush_job_info.triggered_writes_slowdown || - flush_job_info.triggered_writes_stop) - { - LOG(INFO) << "[PurgerEventListener] Flush completed, file: " - << flush_job_info.file_path - << ", job_id: " << flush_job_info.job_id - << ", thread: " << flush_job_info.thread_id - << ", file_number: " << flush_job_info.file_number - << ", triggered_writes_slowdown: " - << flush_job_info.triggered_writes_slowdown - << ", triggered_writes_stop: " - << flush_job_info.triggered_writes_stop - << ", smallest_seqno: " << flush_job_info.smallest_seqno - << ", largest_seqno: " << flush_job_info.largest_seqno - << ", flush_reason: " - << GetFlushReason(flush_job_info.flush_reason); - } - - // Remove the entry from sliding window - if (sliding_window_) - { - sliding_window_->RemoveFileNumber(flush_job_info.thread_id, - flush_job_info.job_id); - } -} - -void PurgerEventListener::OnCompactionBegin( - rocksdb::DB *db, const rocksdb::CompactionJobInfo &ci) -{ - DLOG(INFO) << "[PurgerEventListener] Compaction begin, job_id: " - << ci.job_id << ", thread: " << ci.thread_id - << ", output_level: " << ci.output_level - << ", input_files_size: " << ci.input_files.size() - << ", compaction_reason: " - << static_cast(ci.compaction_reason); - - // Update sliding window with current max file number - UpdateSlidingWindow(db, ci.thread_id, ci.job_id); -} - -void PurgerEventListener::OnCompactionCompleted( - rocksdb::DB *db, const rocksdb::CompactionJobInfo &ci) -{ - DLOG(INFO) << "[PurgerEventListener] Compaction completed, job_id: " - << ci.job_id << ", thread: " << ci.thread_id - << ", output_level: " << ci.output_level - << ", input_files_size: " << ci.input_files.size() - << ", output_files_size: " << ci.output_files.size() - << ", compaction_reason: " - << static_cast(ci.compaction_reason); - - // Remove the entry from sliding window - if (sliding_window_) - { - sliding_window_->RemoveFileNumber(ci.thread_id, ci.job_id); - } -} - -void PurgerEventListener::Stop() -{ - if (sliding_window_) - { - sliding_window_->Stop(); - sliding_window_.reset(); - } -} - -std::string PurgerEventListener::GetFlushReason( - rocksdb::FlushReason flush_reason) -{ - switch (flush_reason) - { - case rocksdb::FlushReason::kOthers: - return "Others"; - case rocksdb::FlushReason::kGetLiveFiles: - return "GetLiveFiles"; - case rocksdb::FlushReason::kShutDown: - return "ShutDown"; - case rocksdb::FlushReason::kExternalFileIngestion: - return "ExternalFileIngestion"; - case rocksdb::FlushReason::kManualCompaction: - return "ManualCompaction"; - case rocksdb::FlushReason::kWriteBufferManager: - return "WriteBufferManager"; - case rocksdb::FlushReason::kWriteBufferFull: - return "WriteBufferFull"; - case rocksdb::FlushReason::kTest: - return "Test"; - case rocksdb::FlushReason::kDeleteFiles: - return "DeleteFiles"; - case rocksdb::FlushReason::kAutoCompaction: - return "AutoCompaction"; - case rocksdb::FlushReason::kManualFlush: - return "ManualFlush"; - case rocksdb::FlushReason::kErrorRecovery: - return "ErrorRecovery"; - default: - return "Unknown"; - } -} - -void PurgerEventListener::UpdateSlidingWindow(rocksdb::DB *db, - uint64_t thread_id, - uint64_t job_id) -{ - if (!db) - { - LOG(ERROR) << "[PurgerEventListener] DB pointer is null"; - return; - } - - // Get current max file number from RocksDB - uint64_t max_file_number = db->GetNextFileNumber() - 1; - - if (sliding_window_) - { - sliding_window_->AddFileNumber(max_file_number, thread_id, job_id); - } -} - -} // namespace EloqDS diff --git a/store_handler/eloq_data_store_service/purger_event_listener.h b/store_handler/eloq_data_store_service/purger_event_listener.h deleted file mode 100644 index 0ca5c31c..00000000 --- a/store_handler/eloq_data_store_service/purger_event_listener.h +++ /dev/null @@ -1,146 +0,0 @@ -/** - * Copyright (C) 2025 EloqData Inc. - * - * This program is free software: you can redistribute it and/or modify - * it under either of the following two licenses: - * 1. GNU Affero General Public License, version 3, as published by the Free - * Software Foundation. - * 2. GNU General Public License as published by the Free Software - * Foundation; version 2 of the License. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Affero General Public License or GNU General Public License for more - * details. - * - * You should have received a copy of the GNU Affero General Public License - * and GNU General Public License V2 along with this program. If not, see - * . - * - */ - -#pragma once - -#include -#include -#include - -#include -#include -#include - -#include "purger_sliding_window.h" - -namespace EloqDS -{ - -/** - * @brief Enhanced EventListener for tracking file numbers to improve purger - * safety - * - * This listener subscribes to FlushBegin and CompactionBegin events to capture - * the maximum file number at the time of these operations. The file numbers are - * fed into a sliding window which periodically updates S3 with the smallest - * file number threshold to prevent premature deletion by the purger. - */ -class PurgerEventListener : public rocksdb::EventListener -{ -public: - /** - * @brief Constructor for PurgerEventListener - * @param epoch The epoch string for this DB instance - * @param bucket_name S3 bucket name - * @param s3_object_path S3 object path - * @param storage_provider Cloud storage provider for S3 operations - * @param entry_duration Duration to keep entries in sliding window even it - * is deleted (default: 15 seconds, should be less than purger interval, - * indicating the minimum update frequency, it prevents too frequent - * updates) - * @param s3_update_interval Interval for updating S3 file (default: 30 - * seconds, indicating the maximum update frequency) - */ - PurgerEventListener( - const std::string &epoch, - const std::string &bucket_name, - const std::string &s3_object_path, - std::shared_ptr storage_provider, - std::chrono::milliseconds entry_duration = std::chrono::seconds(15), - std::chrono::milliseconds s3_update_interval = - std::chrono::seconds(30)); - - /** - * @brief Destructor - stops the sliding window - */ - ~PurgerEventListener(); - - /** - * @brief Update the epoch string - * @param epoch The new epoch string - */ - void SetEpoch(const std::string &epoch); - - void BlockPurger(); - - /** - * @brief Called when a flush operation begins - * @param db Pointer to the database instance - * @param flush_job_info Information about the flush operation - */ - void OnFlushBegin(rocksdb::DB *db, - const rocksdb::FlushJobInfo &flush_job_info) override; - - /** - * @brief Called when a flush operation completes - * @param db Pointer to the database instance - * @param flush_job_info Information about the flush operation - */ - void OnFlushCompleted(rocksdb::DB *db, - const rocksdb::FlushJobInfo &flush_job_info) override; - - /** - * @brief Called when a compaction operation begins - * @param db Pointer to the database instance - * @param ci Information about the compaction operation - */ - void OnCompactionBegin(rocksdb::DB *db, - const rocksdb::CompactionJobInfo &ci) override; - - /** - * @brief Called when a compaction operation completes - * @param db Pointer to the database instance - * @param ci Information about the compaction operation - */ - void OnCompactionCompleted(rocksdb::DB *db, - const rocksdb::CompactionJobInfo &ci) override; - - /** - * @brief Stop the event listener and cleanup resources - */ - void Stop(); - - /** - * @brief Get flush reason string for logging - * @param flush_reason The flush reason enum value - * @return String representation of the flush reason - */ - std::string GetFlushReason(rocksdb::FlushReason flush_reason); - -private: - std::string bucket_name_; - std::string s3_object_path_; - - std::unique_ptr sliding_window_; - - /** - * @brief Update sliding window with current max file number from DB - * @param db Pointer to the database instance - * @param thread_id The thread ID of the operation (default: 0) - * @param job_id The job ID of the operation (default: 0) - */ - void UpdateSlidingWindow(rocksdb::DB *db, - uint64_t thread_id = 0, - uint64_t job_id = 0); -}; - -} // namespace EloqDS diff --git a/store_handler/eloq_data_store_service/purger_sliding_window.cpp b/store_handler/eloq_data_store_service/purger_sliding_window.cpp deleted file mode 100644 index edd50fd0..00000000 --- a/store_handler/eloq_data_store_service/purger_sliding_window.cpp +++ /dev/null @@ -1,368 +0,0 @@ -/** - * Copyright (C) 2025 EloqData Inc. - * - * This program is free software: you can redistribute it and/or modify - * it under either of the following two licenses: - * 1. GNU Affero General Public License, version 3, as published by the Free - * Software Foundation. - * 2. GNU General Public License as published by the Free Software - * Foundation; version 2 of the License. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Affero General Public License or GNU General Public License for more - * details. - * - * You should have received a copy of the GNU Affero General Public License - * and GNU General Public License V2 along with this program. If not, see - * . - * - */ - -#include "purger_sliding_window.h" - -#include -#include -#include -#include -#include -#include - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -namespace EloqDS -{ -using std::make_unique; - -// S3FileNumberUpdater implementation - -S3FileNumberUpdater::S3FileNumberUpdater( - const std::string &bucket_name, - const std::string &s3_object_path, - std::shared_ptr storage_provider) - : bucket_name_(bucket_name), - s3_object_path_(s3_object_path), - storage_provider_(storage_provider) -{ -} - -void S3FileNumberUpdater::UpdateSmallestFileNumber(uint64_t file_number, - const std::string &epoch) -{ - std::string content = std::to_string(file_number); - std::string object_key = GetS3ObjectKey(epoch); - - // Write to temp local file at first - char tmp_template[] = - "/tmp/smallest_file_number_upload_XXXXXX"; // Xs will be replaced - int fd = mkstemp(tmp_template); - if (fd == -1) - { - LOG(ERROR) << "Failed to open temp file for writing: " << tmp_template; - return; - } - - std::string temp_file_path = tmp_template; - - // write content to the temp file - if (write(fd, content.c_str(), content.size()) == -1) - { - LOG(ERROR) << "Failed to write to temp file: " << temp_file_path; - close(fd); - // Remove the temp file - if (std::remove(temp_file_path.c_str()) != 0) - { - LOG(WARNING) << "Failed to remove temp file: " << temp_file_path; - } - return; - } - close(fd); // We will open it later for reading - - if (!storage_provider_) - { - LOG(ERROR) << "Cloud storage provider is not initialized"; - // Remove the temp file - if (std::remove(temp_file_path.c_str()) != 0) - { - LOG(WARNING) << "Failed to remove temp file: " << temp_file_path; - } - return; - } - - rocksdb::IOStatus s = storage_provider_->PutCloudObject( - temp_file_path, bucket_name_, object_key); - - if (!s.ok()) - { - LOG(ERROR) << "Failed to update smallest file number to S3: " - << s.ToString() << " ,bucket_name: " << bucket_name_ - << ", object_key: " << object_key - << ", file_number: " << file_number; - } - else - { - DLOG(INFO) << "Updated smallest file number in S3: " - << " bucket_name: " << bucket_name_ - << ", file_number: " << file_number - << ", object_key: " << object_key; - } - - // Remove the temp file - if (std::remove(temp_file_path.c_str()) != 0) - { - LOG(WARNING) << "Failed to remove temp file: " << temp_file_path; - } -} - -void S3FileNumberUpdater::BlockPurger(const std::string &epoch) -{ - DLOG(INFO) << "Wrote 0 as file number to S3 to block purger, epoch: " - << epoch; - UpdateSmallestFileNumber(std::numeric_limits::min(), epoch); - // Don't update last_published_smallest_ in sliding window, - // so that future smaller file number can still be updated -} - -std::string S3FileNumberUpdater::GetS3ObjectKey(const std::string &epoch) const -{ - std::ostringstream oss; - oss << s3_object_path_; - if (!s3_object_path_.empty() && s3_object_path_.back() != '/') - { - oss << "/"; - } - oss << "smallest_new_file_number-" << epoch; - return oss.str(); -} - -// SlidingWindow implementation - -SlidingWindow::SlidingWindow( - std::chrono::milliseconds entry_duration, - std::chrono::milliseconds s3_update_interval, - const std::string &epoch, - const std::string &bucket_name, - const std::string &s3_object_path, - std::shared_ptr storage_provider) - : entry_duration_(entry_duration), - s3_update_interval_(s3_update_interval), - epoch_(epoch), - last_published_smallest_(std::numeric_limits::max()), - should_stop_(false) -{ - s3_updater_ = std::make_unique( - bucket_name, s3_object_path, storage_provider); - - // Start the timer thread - timer_thread_ = make_unique(&SlidingWindow::TimerWorker, this); - - DLOG(INFO) << "SlidingWindow started for epoch " << epoch_ - << ", window_duration: " << entry_duration_.count() << "ms" - << ", s3_update_interval: " << s3_update_interval_.count() - << "ms"; -} - -SlidingWindow::~SlidingWindow() -{ - Stop(); -} - -void SlidingWindow::SetEpoch(const std::string &epoch) -{ - std::lock_guard lock(window_mutex_); - epoch_ = epoch; -} - -std::string SlidingWindow::GetEpoch() -{ - std::lock_guard lock(window_mutex_); - return epoch_; -} - -void SlidingWindow::AddFileNumber(uint64_t file_number, - uint64_t thread_id, - uint64_t job_id) -{ - std::lock_guard lock(window_mutex_); - if (file_number < last_published_smallest_) - { - DLOG(INFO) << "Immediate S3 update with smaller file number: " - << file_number << ", thread_id: " << thread_id - << ", job_id: " << job_id << ", epoch: " << epoch_; - s3_updater_->UpdateSmallestFileNumber(file_number, epoch_); - last_published_smallest_ = file_number; - } - std::string key = GenerateKey(thread_id, job_id); - window_entries_.emplace(key, WindowEntry(file_number)); - DLOG(INFO) << "Added file number to sliding window: " << file_number - << ", thread_id: " << thread_id << ", job_id: " << job_id - << ", epoch: " << epoch_ - << ", window size: " << window_entries_.size(); -} - -void SlidingWindow::RemoveFileNumber(uint64_t thread_id, uint64_t job_id) -{ - std::lock_guard lock(window_mutex_); - - std::string key = GenerateKey(thread_id, job_id); - auto it = window_entries_.find(key); - - if (it != window_entries_.end()) - { - uint64_t removed_file_number = it->second.file_number_; - auto now = std::chrono::steady_clock::now(); - // Mark the entry as deleted, but do not remove it immediately - // to avoid frequent S3 updates, and give some time for Manifests - // file get updated. - // - // The entry will be removed when it expires in GetSmallestFileNumber() - it->second.deleted_ = true; - it->second.timestamp_ = now; - - DLOG(INFO) << "Removed file number from sliding window: " - << removed_file_number << ", thread_id: " << thread_id - << ", job_id: " << job_id << ", epoch: " << epoch_ - << ", window size: " << window_entries_.size(); - } - else - { - DLOG(WARNING) - << "Attempted to remove non-existent entry from sliding window: " - << "thread_id: " << thread_id << ", job_id: " << job_id - << ", epoch: " << epoch_; - } -} - -uint64_t SlidingWindow::GetSmallestFileNumber() -{ - if (window_entries_.empty()) - { - return std::numeric_limits::max(); - } - - uint64_t smallest = std::numeric_limits::max(); - auto now = std::chrono::steady_clock::now(); - for (auto it = window_entries_.begin(); it != window_entries_.end();) - { - // To avoid frequent S3 updates, do not remove entries that are - // marked deleted until they expire - if (it->second.deleted_) - { - if (now - it->second.timestamp_ >= entry_duration_) - { - // Entry is expired, remove it - it = window_entries_.erase(it); - continue; - } - } - - if (it->second.file_number_ < smallest) - { - smallest = it->second.file_number_; - } - - it++; - } - - DLOG(INFO) << "Current smallest file number: " << smallest - << ", epoch: " << epoch_ - << ", window size: " << window_entries_.size(); - - return smallest; -} - -void SlidingWindow::BlockPurger() -{ - std::lock_guard lock(window_mutex_); - if (epoch_.empty()) - { - LOG(WARNING) - << "Cannot block purger, epoch is not set in sliding window"; - return; - } - s3_updater_->BlockPurger(epoch_); -} - -void SlidingWindow::Stop() -{ - // Signal the timer thread to stop - { - std::lock_guard lock(window_mutex_); - if (should_stop_) - { - return; // Already stopped - } - should_stop_ = true; - } - cv_.notify_all(); - - // Wait for the timer thread to finish - if (timer_thread_ && timer_thread_->joinable()) - { - timer_thread_->join(); - timer_thread_.reset(); - } - - DLOG(INFO) << "SlidingWindow stopped for epoch " << epoch_; -} - -void SlidingWindow::TimerWorker() -{ - std::unique_lock lock(window_mutex_); - - while (!should_stop_) - { - DLOG(INFO) << "SlidingWindow timer tick for epoch " << epoch_; - // Wait for the specified interval or stop signal - cv_.wait_for( - lock, s3_update_interval_, [this] { return should_stop_; }); - - if (should_stop_) - { - break; - } - - // do not attempt S3 update if epoch is empty (indicates epoch is still - // being set) - if (epoch_.empty()) - { - continue; - } - DLOG(INFO) << "SlidingWindow timer processing for epoch " << epoch_; - - uint64_t smallest = GetSmallestFileNumber(); - FlushToS3(smallest); - } - - DLOG(INFO) << "SlidingWindow timer thread exiting for epoch " << epoch_; -} - -void SlidingWindow::FlushToS3(uint64_t smallest) -{ - s3_updater_->UpdateSmallestFileNumber(smallest, epoch_); - last_published_smallest_ = smallest; - DLOG(INFO) << "Updated S3 with smallest file number: " << smallest - << ", epoch: " << epoch_; -} - -std::string SlidingWindow::GenerateKey(uint64_t thread_id, - uint64_t job_id) const -{ - std::ostringstream oss; - oss << thread_id << "-" << job_id; - return oss.str(); -} - -} // namespace EloqDS diff --git a/store_handler/eloq_data_store_service/purger_sliding_window.h b/store_handler/eloq_data_store_service/purger_sliding_window.h deleted file mode 100644 index 96fd0a91..00000000 --- a/store_handler/eloq_data_store_service/purger_sliding_window.h +++ /dev/null @@ -1,195 +0,0 @@ -/** - * Copyright (C) 2025 EloqData Inc. - * - * This program is free software: you can redistribute it and/or modify - * it under either of the following two licenses: - * 1. GNU Affero General Public License, version 3, as published by the Free - * Software Foundation. - * 2. GNU General Public License as published by the Free Software - * Foundation; version 2 of the License. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Affero General Public License or GNU General Public License for more - * details. - * - * You should have received a copy of the GNU Affero General Public License - * and GNU General Public License V2 along with this program. If not, see - * . - * - */ - -#pragma once - -#include -#include - -#include -#include -#include -#include -#include -#include -#include -#include - -namespace EloqDS -{ - -/** - * @brief S3 file updater for writing smallest file number to S3 - */ -class S3FileNumberUpdater -{ -public: - S3FileNumberUpdater( - const std::string &bucket_name, - const std::string &s3_object_path, - std::shared_ptr storage_provider); - - ~S3FileNumberUpdater() = default; - - /** - * @brief Update the smallest file number in S3 - * @param file_number The smallest file number to write - */ - void UpdateSmallestFileNumber(uint64_t file_number, - const std::string &epoch); - - /** - * @brief Block purger temporarily - */ - void BlockPurger(const std::string &epoch); - -private: - std::string bucket_name_; - std::string s3_object_path_; - std::shared_ptr storage_provider_; - - std::string GetS3ObjectKey(const std::string &epoch) const; -}; - -/** - * @brief Time-based sliding window for tracking file numbers with automatic S3 - * updates - */ -class SlidingWindow -{ -public: - /** - * @brief Constructor for sliding window - * @param window_duration Duration to keep entries in the window - * @param s3_update_interval Interval for updating S3 file - * @param epoch The epoch string for this DB instance - * @param bucket_name S3 bucket name - * @param s3_object_path S3 object path - * @param storage_provider Cloud storage provider for S3 operations - */ - SlidingWindow( - std::chrono::milliseconds entry_duration, - std::chrono::milliseconds s3_update_interval, - const std::string &epoch, - const std::string &bucket_name, - const std::string &s3_object_path, - std::shared_ptr storage_provider); - - /** - * @brief Destructor - stops the timer thread - */ - ~SlidingWindow(); - - SlidingWindow(const SlidingWindow &) = delete; - SlidingWindow &operator=(const SlidingWindow &) = delete; - SlidingWindow(SlidingWindow &&) = delete; - SlidingWindow &operator=(SlidingWindow &&) = delete; - - void SetEpoch(const std::string &epoch); - - std::string GetEpoch(); - - /** - * @brief Add a file number to the sliding window - * @param file_number The file number to add - * @param thread_id The thread ID of the operation - * @param job_id The job ID of the operation - */ - void AddFileNumber(uint64_t file_number, - uint64_t thread_id, - uint64_t job_id); - - /** - * @brief Remove a file number entry from the sliding window - * @param thread_id The thread ID of the operation - * @param job_id The job ID of the operation - */ - void RemoveFileNumber(uint64_t thread_id, uint64_t job_id); - - /** - * @brief Get the smallest file number in the current window - * @return The smallest file number, or UINT64_MAX if window is empty - */ - uint64_t GetSmallestFileNumber(); - - /** - * @brief Block purger temporarily - */ - void BlockPurger(); - - /** - * @brief Stop the sliding window and cleanup - */ - void Stop(); - -private: - struct WindowEntry - { - uint64_t file_number_; - std::chrono::steady_clock::time_point timestamp_; - bool deleted_; - - WindowEntry(uint64_t num) - : file_number_(num), - timestamp_(std::chrono::steady_clock::now()), - deleted_(false) - { - } - }; - - // Map of (thread_id + job_id) -> WindowEntry - std::unordered_map window_entries_; - std::chrono::milliseconds entry_duration_; - std::chrono::milliseconds s3_update_interval_; - std::string epoch_; - - // Last published smallest file number to avoid conflicting updates - uint64_t last_published_smallest_{std::numeric_limits::max()}; - - std::unique_ptr s3_updater_; - - // Threading - std::unique_ptr timer_thread_; - std::mutex window_mutex_; - std::condition_variable cv_; - bool should_stop_; - - /** - * @brief Timer thread worker function - */ - void TimerWorker(); - - /** - * @brief Flush current minimum file number to S3 - */ - void FlushToS3(uint64_t smallest); - - /** - * @brief Generate a key string for the window_entries_ map - * @param thread_id The thread ID of the operation - * @param job_id The job ID of the operation - * @return A string key combining thread_id and job_id - */ - std::string GenerateKey(uint64_t thread_id, uint64_t job_id) const; -}; - -} // namespace EloqDS diff --git a/store_handler/eloq_data_store_service/rocksdb_cloud_data_store.cpp b/store_handler/eloq_data_store_service/rocksdb_cloud_data_store.cpp index f0a298f1..4973ce89 100644 --- a/store_handler/eloq_data_store_service/rocksdb_cloud_data_store.cpp +++ b/store_handler/eloq_data_store_service/rocksdb_cloud_data_store.cpp @@ -49,7 +49,6 @@ #include "data_store_service.h" #include "ds_request.pb.h" #include "internal_request.h" -#include "purger_event_listener.h" #include "rocksdb/cloud/cloud_file_system_impl.h" #include "rocksdb/cloud/cloud_storage_provider.h" @@ -352,6 +351,7 @@ bool RocksDBCloudDataStore::StartDB(int64_t term) // Disable cloud file deletion totally, since we defer file deletion to // purger cfs_options_.disable_cloud_file_deletion = true; + cfs_options_.publish_file_number_guard = true; // keep invisible files in cloud storage since they can be referenced // by other nodes with old valid cloud manifest files during leader transfer @@ -650,23 +650,6 @@ bool RocksDBCloudDataStore::OpenCloudDB( max_bytes_for_level_multiplier_; } - // Add event listener for purger - rocksdb::CloudFileSystemImpl *cfs_impl = - dynamic_cast(cloud_fs_.get()); - if (cfs_impl == nullptr) - { - LOG(ERROR) << "Fail to get CloudFileSystemImpl from cloud_fs_"; - return false; - } - std::string bucket_name = - cloud_config_.bucket_prefix_ + cloud_config_.bucket_name_; - auto db_event_listener = std::make_shared( - "", /*We still don't know the epoch now*/ - bucket_name, - cloud_config_.object_path_, - cfs_impl->GetStorageProvider()); - options.listeners.emplace_back(db_event_listener); - // The max_open_files default value is -1, it cause DB open all files on // DB::Open() This behavior causes 2 effects, // 1. DB::Open() will be slow @@ -757,7 +740,7 @@ bool RocksDBCloudDataStore::OpenCloudDB( return false; } - // set epoch for purger event listener + // The file number guard published by DBCloud requires a valid epoch std::string current_epoch; status = db_->GetCurrentEpoch(¤t_epoch); if (!status.ok()) @@ -776,8 +759,6 @@ bool RocksDBCloudDataStore::OpenCloudDB( db_->ContinueBackgroundWork(); return false; } - db_event_listener->SetEpoch(current_epoch); - db_event_listener->BlockPurger(); // Resume background work db_->ContinueBackgroundWork();