Skip to content

[Bugfix][Store] Interrupt in-flight snapshot children during shutdown #3942

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
waizuichougou wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[Bugfix][Store] Interrupt in-flight snapshot children during shutdown #3942

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
8 changes: 5 additions & 3 deletions mooncake-store/include/master_snapshot_manager.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,9 +45,9 @@ struct MasterSnapshotManagerOptions {
* timeout handling, payload upload, catalog publish, retention cleanup, and
* snapshot metrics updates.
*
* This is a behavior-preserving refactor that moves snapshot orchestration
* logic out of MasterService without changing snapshot format, restore
* behavior, storage layout, flags, or locking semantics.
* Snapshot format, restore behavior, storage layout, flags, and locking
* semantics are unchanged; shutdown additionally interrupts an in-flight
* snapshot child so the manager can terminate promptly.
*/
class MasterSnapshotManager {
friend class test::MasterServiceSnapshotTestBase; // Allow test access to
Expand All @@ -72,6 +72,8 @@ class MasterSnapshotManager {
void WaitForSnapshotChild(pid_t pid, const std::string& snapshot_id,
int log_pipe_fd);
void HandleChildTimeout(pid_t pid, const std::string& snapshot_id);
void TerminateSnapshotChild(pid_t pid, const std::string& snapshot_id,
const char* reason);
void HandleChildExit(pid_t pid, int status, const std::string& snapshot_id);

tl::expected<void, SerializationError> PersistState(
Expand Down
85 changes: 60 additions & 25 deletions mooncake-store/src/master_snapshot_manager.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,12 @@ void MasterSnapshotManager::SnapshotThreadFunc() {
close(log_pipe[1]);
} else if (pid == 0) {
// Child process
// The master installs handlers that only update parent-process
// shutdown state. A forked snapshot child must remain terminable
// by the timeout and shutdown paths below.
signal(SIGINT, SIG_DFL);
signal(SIGTERM, SIG_DFL);

// Close read end, set write end for logging
close(log_pipe[0]);
g_snapshot_log_pipe_fd = log_pipe[1];
Expand Down Expand Up @@ -255,6 +261,12 @@ void MasterSnapshotManager::WaitForSnapshotChild(pid_t pid,
return;
} else if (result == 0) {
// Child process is still running
if (!snapshot_running_) {
flush_child_logs();
TerminateSnapshotChild(pid, snapshot_id, "manager shutdown");
return;
}

auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(
std::chrono::steady_clock::now() - start_time)
.count();
Expand All @@ -270,8 +282,11 @@ void MasterSnapshotManager::WaitForSnapshotChild(pid_t pid,
return;
}

// Brief sleep before checking again
std::this_thread::sleep_for(std::chrono::seconds(2));
// Briefly wait before checking again, but wake promptly on Stop().
std::unique_lock<std::mutex> lock(snapshot_thread_mutex_);
snapshot_thread_cv_.wait_for(lock, std::chrono::seconds(2), [this] {
return !snapshot_running_.load();
});
} else {
// Child process has exited
// Flush remaining logs from child
Expand All @@ -294,40 +309,60 @@ void MasterSnapshotManager::WaitForSnapshotChild(pid_t pid,

void MasterSnapshotManager::HandleChildTimeout(pid_t pid,
const std::string& snapshot_id) {
LOG(WARNING) << "[Snapshot] Child process timeout, snapshot_id="
<< snapshot_id << ", child_pid=" << pid
<< ", killing child process";
TerminateSnapshotChild(pid, snapshot_id, "snapshot timeout");
}

void MasterSnapshotManager::TerminateSnapshotChild(
pid_t pid, const std::string& snapshot_id, const char* reason) {
LOG(WARNING) << "[Snapshot] Terminating child process due to " << reason
<< ", snapshot_id=" << snapshot_id << ", child_pid=" << pid;

// Try to gracefully terminate the child process
if (kill(pid, SIGTERM) == 0) {
// Wait a few seconds to see if it exits gracefully
std::this_thread::sleep_for(std::chrono::seconds(5));
if (kill(pid, SIGTERM) == -1 && errno != ESRCH) {
LOG(ERROR) << "[Snapshot] Failed to send SIGTERM to child process, "
"snapshot_id="
<< snapshot_id << ", child_pid=" << pid
<< ", error=" << strerror(errno);
}

// Check if it has exited
int status;
if (waitpid(pid, &status, WNOHANG) == 0) {
// Child process still not exited, force kill
LOG(WARNING) << "[Snapshot] Child process still running, force "
"killing, snapshot_id="
<< snapshot_id << ", child_pid=" << pid;
kill(pid, SIGKILL);

// Wait for force termination to complete
waitpid(pid, &status, 0);
LOG(WARNING)
<< "[Snapshot] Child process force killed, snapshot_id="
<< snapshot_id << ", child_pid=" << pid;
} else {
constexpr auto kTerminationGracePeriod = std::chrono::seconds(5);
constexpr auto kReapPollInterval = std::chrono::milliseconds(50);
const auto deadline =
std::chrono::steady_clock::now() + kTerminationGracePeriod;
int status;
while (std::chrono::steady_clock::now() < deadline) {
const pid_t result = waitpid(pid, &status, WNOHANG);
if (result == pid) {
LOG(INFO) << "[Snapshot] Child process terminated gracefully after "
"SIGTERM, snapshot_id="
<< snapshot_id << ", child_pid=" << pid;
return;
}
} else {
LOG(ERROR) << "[Snapshot] Failed to send SIGTERM to child process, "
if (result == -1 && errno != EINTR) {
if (errno != ECHILD) {
LOG(ERROR) << "[Snapshot] Failed to wait for child process, "
"snapshot_id="
<< snapshot_id << ", child_pid=" << pid
<< ", error=" << strerror(errno);
}
return;
}
std::this_thread::sleep_for(kReapPollInterval);
}

LOG(WARNING) << "[Snapshot] Child process still running, force killing, "
"snapshot_id="
<< snapshot_id << ", child_pid=" << pid;
if (kill(pid, SIGKILL) == -1 && errno != ESRCH) {
LOG(ERROR) << "[Snapshot] Failed to send SIGKILL to child process, "
"snapshot_id="
<< snapshot_id << ", child_pid=" << pid
<< ", error=" << strerror(errno);
}
while (waitpid(pid, &status, 0) == -1 && errno == EINTR) {
}
LOG(WARNING) << "[Snapshot] Child process force killed, snapshot_id="
<< snapshot_id << ", child_pid=" << pid;
}

void MasterSnapshotManager::HandleChildExit(pid_t pid, int status,
Expand Down
49 changes: 49 additions & 0 deletions mooncake-store/tests/ha/snapshot/snapshot_child_process_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
#include <chrono>
#include <cstdlib>
#include <filesystem>
#include <future>
#include <memory>
#include <optional>
#include <regex>
Expand Down Expand Up @@ -176,6 +177,17 @@ class SnapshotChildProcessTest : public ::testing::Test {
}
}

void CallWaitForSnapshotChild(MasterSnapshotManager& manager, pid_t pid,
const std::string& snapshot_id,
int log_pipe_fd) {
manager.snapshot_running_ = true;
manager.WaitForSnapshotChild(pid, snapshot_id, log_pipe_fd);
}

std::unique_ptr<MasterSnapshotManager> CreateSnapshotManagerForTest() {
return CreateTempSnapshotManager();
}

void CallCleanupOldSnapshot(int keep_count,
const std::string& snapshot_id) {
if (service_->snapshot_manager_) {
Expand Down Expand Up @@ -413,6 +425,43 @@ TEST_F(SnapshotChildProcessTest, HandleChildTimeout_KillsSleepingChild) {
EXPECT_NE(result, 0) << "Child process should have been terminated";
}

TEST_F(SnapshotChildProcessTest, StopInterruptsRunningSnapshotChild) {
CreateDefaultService();
auto manager = CreateSnapshotManagerForTest();

int log_pipe[2];
ASSERT_EQ(pipe(log_pipe), 0);
pid_t pid = fork();
ASSERT_NE(pid, -1) << "fork failed";
if (pid == 0) {
close(log_pipe[0]);
sleep(300);
_exit(0);
}
close(log_pipe[1]);

auto waiter = std::async(std::launch::async, [&] {
CallWaitForSnapshotChild(*manager, pid, "test_snapshot_stop",
log_pipe[0]);
});
std::this_thread::sleep_for(std::chrono::milliseconds(100));
manager->Stop();

const auto wait_status = waiter.wait_for(std::chrono::seconds(2));
if (wait_status != std::future_status::ready) {
// Keep a regression from leaving this test blocked until the manager's
// configured 60-second snapshot timeout.
kill(pid, SIGTERM);
}
EXPECT_EQ(wait_status, std::future_status::ready);
waiter.get();
close(log_pipe[0]);

int status;
EXPECT_EQ(waitpid(pid, &status, WNOHANG), -1)
<< "Snapshot child should already be reaped";
}

// ========== CleanupOldSnapshot ==========

TEST_F(SnapshotChildProcessTest, CleanupOldSnapshot_KeepsRecentDeletesOld) {
Expand Down
Loading