Skip to content

feat:Progressive compact #3229

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chejinge wants to merge 2 commits into
base: unstable
Choose a base branch
from
Open

feat:Progressive compact #3229

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b11fea6
feat:渐进式compact
brother-jin Mar 12, 2026
786b98a
feat:渐进式compact
brother-jin Mar 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions include/pika_server.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -520,6 +520,7 @@ class PikaServer : public pstd::noncopyable {
*/
void DoTimingTask();
void AutoCompactRange();
void AutoProgressiveCompact();
void AutoBinlogPurge();
void AutoServerlogPurge();
void AutoDeleteExpiredDump();
Expand Down Expand Up @@ -550,6 +551,7 @@ class PikaServer : public pstd::noncopyable {
*/
bool have_scheduled_crontask_ = false;
struct timeval last_check_compact_time_;
struct timeval last_progressive_compact_time_;

/*
* ResumeDB used
Expand Down
59 changes: 47 additions & 12 deletions src/pika_server.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ PikaServer::PikaServer()
: exit_(false),
slow_cmd_thread_pool_flag_(g_pika_conf->slow_cmd_pool()),
last_check_compact_time_({0, 0}),
last_progressive_compact_time_({0, 0}),
last_check_resume_time_({0, 0}),
repl_state_(PIKA_REPL_NO_CONNECT),
role_(PIKA_ROLE_SINGLE) {
Expand Down Expand Up @@ -1108,6 +1109,8 @@ int PikaServer::ClientPubSubChannelPatternSize(const std::shared_ptr<NetConn>& c
void PikaServer::DoTimingTask() {
// Maybe schedule compactrange
AutoCompactRange();
// Progressive compact
AutoProgressiveCompact();
// Purge serverlog
AutoServerlogPurge();
// Purge binlog
Expand Down Expand Up @@ -1140,6 +1143,33 @@ void PikaServer::StatDiskUsage() {
disk_statistic_.log_size_.store(pstd::Du(g_pika_conf->log_path()));
}

void PikaServer::AutoProgressiveCompact() {
struct timeval now;
gettimeofday(&now, nullptr);

// Execute progressive compact every 60 seconds
if (last_progressive_compact_time_.tv_sec == 0 ||
now.tv_sec - last_progressive_compact_time_.tv_sec >= 60) {
gettimeofday(&last_progressive_compact_time_, nullptr);

std::shared_lock db_rwl(dbs_rw_);
for (const auto& db_item : dbs_) {
db_item.second->DBLockShared();
auto storage = db_item.second->storage();
if (storage) {
storage::Status s = storage->LongestNotCompactionSstCompact(storage::DataType::kAll);
if (!s.ok()) {
LOG(WARNING) << "Progressive compact for DB: " << db_item.first
<< " failed: " << s.ToString();
} else {
LOG(INFO) << "Progressive compact for DB: " << db_item.first << " completed";
}
}
db_item.second->DBUnlockShared();
}
}
}

void PikaServer::AutoCompactRange() {
struct statfs disk_info;
int ret = statfs(g_pika_conf->db_path().c_str(), &disk_info);
Expand Down Expand Up @@ -1297,25 +1327,30 @@ void PikaServer::AutoServerlogPurge() {

// Process files for each log level
for (auto& [level, files] : log_files_by_level) {
// Sort by time in descending order
// Sort by time in descending order (newest first)
std::sort(files.begin(), files.end(),
[](const auto& a, const auto& b) { return a.second > b.second; });

bool has_recent_file = false;
// Keep the most recent file for each level, delete others that exceed retention_time
bool is_first = true;
for (const auto& [file, log_timestamp] : files) {
double diff_seconds = difftime(now_timestamp, log_timestamp);
int64_t interval_days = static_cast<int64_t>(diff_seconds / 86400);
if (interval_days <= retention_time) {
has_recent_file = true;
// Always keep the most recent file for each log level
if (is_first) {
is_first = false;
continue;
}
if (!has_recent_file) {
has_recent_file = true;
continue;

double diff_seconds = difftime(now_timestamp, log_timestamp);
int64_t interval_days = static_cast<int64_t>(diff_seconds / 86400);

// Delete files that exceed the retention time
if (interval_days > retention_time) {
std::string log_file = log_path + "/" + file;
LOG(INFO) << "Deleting out of date log file: " << log_file;
if(!pstd::DeleteFile(log_file)) {
Copy link

@coderabbitai coderabbitai bot Mar 12, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Minor formatting issue: missing space after if.

-        if(!pstd::DeleteFile(log_file)) {
+        if (!pstd::DeleteFile(log_file)) {
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
if(!pstd::DeleteFile(log_file)) {
if (!pstd::DeleteFile(log_file)) {
🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@src/pika_server.cc` at line 1350, There is a formatting issue: add a space
after the `if` keyword in the conditional calling pstd::DeleteFile so it reads
`if (!pstd::DeleteFile(log_file)) {`; update the occurrence around the use of
pstd::DeleteFile and log_file to follow this spacing convention (and optionally
scan nearby conditionals for the same pattern) to keep code style consistent.

LOG(ERROR) << "Failed to delete log file: " << log_file;
}
}
std::string log_file = log_path + "/" + file;
LOG(INFO) << "Deleting out of date log file: " << log_file;
if(!pstd::DeleteFile(log_file)) LOG(ERROR) << "Failed to delete log file: " << log_file;
}
}
}
Expand Down
Loading