lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(build): no fork in multi-threaded XCE processes, enforced by check-no-fork

Browse source 
Design constraint (tech.md "Process Creation"): the controller, scheduler,
xcecd and anything that starts libxce threads never call fork, vfork,
forkpty, daemon, system, popen or uv_spawn; children start with posix_spawn()
or a systemd unit. A process proven single-threaded may fork, marked
/* xce-fork-ok: <why> */. Tests, harness, autoscaler (not finalized; bug-1969)
and the Go frontend are out of scope.

- scripts/check-no-fork.py scans src/{controller,scheduler,lib,worker}
  (comment/string/#if 0 aware); runs as meson test "no-fork rule" in the
  lib/controller/scheduler/worker suites and in the pre-commit hook.
- worker: drop xcecd's unused fork-based worker start paths
  (process_manager_start_process, process_manager_start_worker; xcecd
  starts xcecd_worker over D-Bus) and a dead #if 0 daemonize fork.
- xcecd_worker's fork/uv_spawn/system calls are marked: its call graph
  from main(), callbacks included, reaches no pthread_create or libuv
  thread-pool call.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-10 13:35:52 +00:00
parent c4fee1077c
commit 6dd70f1091
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
15 changed files with 182 additions and 444 deletions
Download patch file Download diff file Expand all files Collapse all files

10
src/meson.build
View file

@ -72,6 +72,16 @@ if component != 'xcecli'
subdir('frontend/apiserver')
subdir('xceadmin')
# No-fork rule (.kiro/steering/tech.md, "Process creation"): fails if a
# covered component's C code calls fork()/system()/popen()/uv_spawn()
# outside a call marked xce-fork-ok. In each covered suite, so
# `xce-build.sh <component>` runs it too.
test('no-fork rule',
find_program('python3'),
args: [meson.project_source_root() / '..' / 'scripts' / 'check-no-fork.py',
meson.project_source_root() / '..'],
suite: ['lib', 'controller', 'scheduler', 'worker'])
configure_file(
input: 'tools/appimage/entry.sh',
output: 'xce',

2
src/tools/xce-build.sh
View file

@ -333,7 +333,7 @@ case "$COMPONENT" in
meson compile -C "$BUILD_DIR" "${MESON_COMPILE_JOB_OPTS[@]}" xcelib_test
if [ "$RUN_TESTS" = 1 ]; then
echo "=== Running lib tests ==="
meson test -C "$BUILD_DIR" "${MESON_TEST_JOB_OPTS[@]}" "libxce test" --print-errorlogs
meson test -C "$BUILD_DIR" "${MESON_TEST_JOB_OPTS[@]}" "libxce test" "no-fork rule" --print-errorlogs
fi
;;
controller)

3
src/worker/CLAUDE.md
View file

@ -11,7 +11,8 @@ C daemon running on each compute node. Manages job execution, hardware monitorin
## Architecture
- `src/main.c` — Main daemon entry point (`int main`); links against `xcecd_srcs` from `src/xcecd.c` and friends
- `src/hardware_monitor.c` — CPU/memory resource monitoring (no GPU)
- `src/process_worker_manager.c` + `src/uv_proc.c` — Job process lifecycle (fork `xcecd_worker`, spawn/exec the job command); `src/process_manager.c` is unused dead code for job spawn (its `process_manager_start_process` is never called) but is still wired up for generic create/update/destroy/reap in `xcecd.c`
- `src/uv_proc.c` — job process lifecycle inside `xcecd_worker` (pre/post-exec, spawn/exec the job command). `xcecd` starts `xcecd_worker` as a systemd unit over D-Bus (`xcecd_dbus_manager.c`); `process_manager.c`/`process_worker_manager.c` only track and reap worker pids (their fork-based start paths were removed)
- **No fork in `xcecd`**: the daemon is multi-threaded, so it never forks (no `fork`/`system`/`popen`/`uv_spawn`). `xcecd_worker` is single-threaded and may fork; each such call carries `/* xce-fork-ok: ... */`. Keep `xcecd_worker` thread-free, or move its child starts to `posix_spawn()` first. See `.kiro/steering/tech.md` "Process Creation"
Links against `libxce` from `src/lib/`.

9
src/worker/include/process_manager.h
View file

@ -57,15 +57,6 @@ void process_info_init(process_info_t* proc);
*/
process_manager_t* process_manager_create(void);
/**
* @brief start a new process
* @param manager process manager without resource limits
* @param command command with arguments
* @return 0 if success, -1 if failed
*/
int process_manager_start_process(process_manager_t* manager,
const char* command);
/**
* @brief get the number of processes
* @param manager process manager

7
src/worker/include/process_worker_manager.h
View file

@ -13,13 +13,6 @@
#include "xcecd_worker_utils.h"
#include "xcelog.h"
/**
* @brief start worker without resource limits (disabled by default)
* @param command command with arguments
* @return 0 if success, -1 if failed
*/
int process_manager_start_worker(process_manager_t* manager);
/**
* @brief update the process status in the list
* @param manager process manager

15
src/worker/src/main.c
View file

@ -588,21 +588,6 @@ int main(int argc, char* argv[])
init_static_array(&pids, sizeof(int));
#ifndef _WIN32
#if 0
pid_t pid = fork();
if (pid < 0) {
fprintf(stderr, "First fork failed\n");
return 1;
}
if (pid > 0) {
exit(0);
}
if (setsid() < 0) {
fprintf(stderr, "setsid failed\n");
return 1;
}
#endif
char current_dir[1024];
getcwd(current_dir, sizeof(current_dir));

199
src/worker/src/process_manager.c
View file

@ -89,205 +89,6 @@ process_manager_t* process_manager_create(void)
return manager;
}
int process_manager_start_process(process_manager_t* manager,
const char* command)
{
if (!manager || !command) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Invalid process manager or command");
return -1;
}
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Starting process: %s", command);
#ifdef _WIN32
EnterCriticalSection(&manager->lock);
#else
pthread_mutex_lock(&manager->lock);
#endif
// 创建新的进程信息结构
process_info_t* new_processes
= realloc(manager->processes,
(manager->process_count + 1) * sizeof(process_info_t));
if (!new_processes) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Failed to allocate memory for process info");
#ifdef _WIN32
LeaveCriticalSection(&manager->lock);
#else
pthread_mutex_unlock(&manager->lock);
#endif
return -1;
}
manager->processes = new_processes;
process_info_t* proc = &manager->processes[manager->process_count];
process_info_init(proc);
proc->command = strdup(command);
char* jobid_str = getenv("XCE_JOBID");
uint64_t jobid = 0;
if (jobid_str != NULL) {
jobid = atoll(jobid_str);
}
// 创建日志文件
char log_path[256];
time_t now;
time(&now);
struct tm* timeinfo = localtime(&now);
if (jobid != 0) {
char idstring[128];
fmt_jobid(idstring, sizeof(idstring) - 1, jobid);
snprintf(log_path, sizeof(log_path), "/tmp/job_%lu_%s.out",
(unsigned long)now, idstring);
XCECD_WORKER_LOG(XCE_LOG_INFO, "Job output file path is: %s", log_path);
} else {
snprintf(log_path, sizeof(log_path),
"/tmp/job_%d_%02d%02d_%02d%02d%02d.out", (int)now,
timeinfo->tm_mon + 1, timeinfo->tm_mday, timeinfo->tm_hour,
timeinfo->tm_min, timeinfo->tm_sec);
}
proc->log_file = strdup(log_path);
// 打开日志文件
proc->output_fd = open(proc->log_file, O_WRONLY | O_CREAT | O_APPEND, 0644);
if (proc->output_fd == -1) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to open log file: %s",
strerror(errno));
FREEUP(proc->log_file);
FREEUP(proc->command);
#ifdef _WIN32
LeaveCriticalSection(&manager->lock);
#else
pthread_mutex_unlock(&manager->lock);
#endif
return -1;
}
#ifdef _WIN32
// Windows实现
SECURITY_ATTRIBUTES sa
= { .nLength = sizeof(SECURITY_ATTRIBUTES), .bInheritHandle = TRUE };
// 创建管道
HANDLE pipe_read, pipe_write;
if (!CreatePipe(&pipe_read, &pipe_write, &sa, 0)) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to create pipe: %lu",
GetLastError());
close(proc->output_fd);
FREEUP(proc->log_file);
FREEUP(proc->command);
LeaveCriticalSection(&manager->lock);
return -1;
}
// 确保读取端不被继承
SetHandleInformation(pipe_read, HANDLE_FLAG_INHERIT, 0);
STARTUPINFOA si = { 0 };
si.cb = sizeof(STARTUPINFO);
si.dwFlags = STARTF_USESTDHANDLES;
si.hStdInput = GetStdHandle(STD_INPUT_HANDLE);
si.hStdOutput = pipe_write;
si.hStdError = pipe_write;
PROCESS_INFORMATION pi;
if (!CreateProcessA(NULL, proc->command, NULL, NULL, TRUE, CREATE_NO_WINDOW,
NULL, NULL, &si, &pi)) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to create process: %lu",
GetLastError());
CloseHandle(pipe_read);
CloseHandle(pipe_write);
close(proc->output_fd);
FREEUP(proc->log_file);
FREEUP(proc->command);
LeaveCriticalSection(&manager->lock);
return -1;
}
CloseHandle(pipe_write); // 关闭写端
proc->pipe_read = pipe_read;
proc->handle = pi.hProcess;
proc->status = 1; // 运行中
proc->monitor_active = true;
// 创建监控线程
proc->monitor_thread
= CreateThread(NULL, 0, output_monitor_thread, proc, 0, NULL);
if (!proc->monitor_thread) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to create monitor thread: %lu",
GetLastError());
CloseHandle(pipe_read);
CloseHandle(pi.hProcess);
close(proc->output_fd);
FREEUP(proc->log_file);
FREEUP(proc->command);
LeaveCriticalSection(&manager->lock);
return -1;
}
CloseHandle(pi.hThread);
#else
// Linux实现
pid_t pid = fork();
if (pid == -1) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Fork failed: %s", strerror(errno));
close(proc->output_fd);
FREEUP(proc->log_file);
FREEUP(proc->command);
pthread_mutex_unlock(&manager->lock);
return -1;
}
if (pid == 0) {
// 子进程
dup2(proc->output_fd, STDOUT_FILENO);
dup2(proc->output_fd, STDERR_FILENO);
close(proc->output_fd);
char* args[] = { "/bin/sh", "-c", proc->command, NULL };
execv("/bin/sh", args);
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to execute command: %s",
strerror(errno));
_exit(1);
} else {
// 父进程
XCECD_WORKER_LOG(
XCE_LOG_DEBUG,
"Process started successfully, output redirected to %s",
proc->log_file);
proc->handle = pid;
proc->status = 1; // 运行中
// 非阻塞等待子进程退出，避免僵尸进程
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "The pid of the child is: %d\n", pid);
}
#endif
manager->process_count++;
manager->initialized = true;
#ifdef _WIN32
LeaveCriticalSection(&manager->lock);
#else
pthread_mutex_unlock(&manager->lock);
#endif
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Process started successfully");
FREEUP(proc->log_file);
return 0;
}
bool process_manager_update(process_manager_t* manager)
{
// XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Updating process manager...");

182
src/worker/src/process_worker_manager.c
View file

@ -27,188 +27,6 @@
#include <unistd.h>
#endif
/**
* @brief start worker without resource limits (disabled by default)
* @param command command with arguments
* @return 0 if success, -1 if failed
*/
int process_manager_start_worker(process_manager_t* manager)
{
XCECD_LOG(XCE_LOG_DEBUG, "Starting worker process");
if (!manager) {
XCECD_LOG(XCE_LOG_ERROR,
"Invalid process manager and quit starting process.");
return -1;
}
#ifdef _WIN32
EnterCriticalSection(&manager->lock);
#else
pthread_mutex_lock(&manager->lock);
#endif
process_info_t* new_processes
= realloc(manager->processes,
(manager->process_count + 1) * sizeof(process_info_t));
if (!new_processes) {
#ifdef _WIN32
LeaveCriticalSection(&manager->lock);
#else
pthread_mutex_unlock(&manager->lock);
#endif
return -1;
}
manager->processes = new_processes;
process_info_t* proc = &manager->processes[manager->process_count];
process_info_init(proc);
proc->command = strdup("xcecd_worker");
proc->log_file = strdup("/tmp/xcecd_worker.log");
/* bug-1869: fixed name in /tmp, opened as root */
proc->output_fd = xcecd_open_owned_file(proc->log_file, O_APPEND, 0644);
if (proc->output_fd < 0)
proc->output_fd = -1;
if (proc->output_fd == -1) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to open log file: %s",
strerror(errno));
FREEUP(proc->log_file);
FREEUP(proc->command);
#ifdef _WIN32
LeaveCriticalSection(&manager->lock);
#else
pthread_mutex_unlock(&manager->lock);
#endif
return -1;
}
#ifdef _WIN32
SECURITY_ATTRIBUTES sa
= { .nLength = sizeof(SECURITY_ATTRIBUTES), .bInheritHandle = TRUE };
HANDLE pipe_read, pipe_write;
if (!CreatePipe(&pipe_read, &pipe_write, &sa, 0)) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to create pipe: %lu", GetLastError());
close(proc->output_fd);
FREEUP(proc->log_file);
FREEUP(proc->command);
LeaveCriticalSection(&manager->lock);
return -1;
}
SetHandleInformation(pipe_read, HANDLE_FLAG_INHERIT, 0);
STARTUPINFOA si = { 0 };
si.cb = sizeof(STARTUPINFO);
si.dwFlags = STARTF_USESTDHANDLES;
si.hStdInput = GetStdHandle(STD_INPUT_HANDLE);
si.hStdOutput = pipe_write;
si.hStdError = pipe_write;
PROCESS_INFORMATION pi;
if (!CreateProcessA(NULL, proc->command, NULL, NULL, TRUE, CREATE_NO_WINDOW,
NULL, NULL, &si, &pi)) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to create process: %lu",
GetLastError());
CloseHandle(pipe_read);
CloseHandle(pipe_write);
close(proc->output_fd);
FREEUP(proc->log_file);
FREEUP(proc->command);
LeaveCriticalSection(&manager->lock);
return -1;
}
CloseHandle(pipe_write);
proc->pipe_read = pipe_read;
proc->handle = pi.hProcess;
proc->status = 1;
proc->monitor_active = true;
proc->monitor_thread
= CreateThread(NULL, 0, output_monitor_thread, proc, 0, NULL);
if (!proc->monitor_thread) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to create monitor thread: %lu",
GetLastError());
CloseHandle(pipe_read);
CloseHandle(pi.hProcess);
close(proc->output_fd);
FREEUP(proc->log_file);
FREEUP(proc->command);
LeaveCriticalSection(&manager->lock);
return -1;
}
CloseHandle(pi.hThread);
#else
pid_t pid = fork();
if (pid == -1) {
XCECD_LOG(XCE_LOG_ERROR, "Fork failed: %s", strerror(errno));
close(proc->output_fd);
FREEUP(proc->log_file);
FREEUP(proc->command);
pthread_mutex_unlock(&manager->lock);
return -1;
}
if (pid == 0) {
dup2(proc->output_fd, STDOUT_FILENO);
dup2(proc->output_fd, STDERR_FILENO);
close(proc->output_fd);
char* xce_server_dir = getenv("XCE_SERVER_DIR");
char worker_path[1024];
if (xce_server_dir != NULL) {
snprintf(worker_path, sizeof(worker_path), "%s/xcecd_worker",
xce_server_dir);
if (access(worker_path, X_OK) != 0) {
XCECD_LOG(XCE_LOG_WARN,
"xcecd_worker not found or not executable in "
"XCE_SERVER_DIR: %s",
xce_server_dir);
}
} else {
XCECD_LOG(
XCE_LOG_WARN,
"XCE_SERVER_DIR not set, using default path for xcecd_worker");
}
execl(worker_path, proc->command, NULL);
XCECD_LOG(XCE_LOG_ERROR, "Failed to execute command: %s",
strerror(errno));
_exit(1);
} else {
/* parent */
XCECD_LOG(XCE_LOG_DEBUG,
"Process started successfully, output redirected to %s",
proc->log_file);
proc->handle = pid;
proc->status = 1;
}
close(proc->output_fd);
#endif
manager->process_count++;
manager->initialized = true;
#ifdef _WIN32
LeaveCriticalSection(&manager->lock);
#else
pthread_mutex_unlock(&manager->lock);
#endif
XCECD_LOG(XCE_LOG_DEBUG, "Worker process started successfully");
return 0;
}
/**
* @brief update the process status in the list
* @param manager process manager

4
src/worker/src/uv_proc.c
View file

@ -96,6 +96,7 @@ static pid_t start_background_service(const char* command)
pid_t pid;
int status;
/* xce-fork-ok: xcecd_worker is single-threaded (tech.md) */
pid = fork();
if (pid < 0) {
@ -109,6 +110,7 @@ static pid_t start_background_service(const char* command)
}
// 2. double fork to ensure the process is not a session leader
/* xce-fork-ok: xcecd_worker is single-threaded (tech.md) */
pid_t child_pid = fork();
if (child_pid < 0) {
exit(EXIT_FAILURE);
@ -153,6 +155,7 @@ static int execute_blocking_command(const char* command, char* const args[])
pid_t pid;
int status;
/* xce-fork-ok: xcecd_worker is single-threaded (tech.md) */
pid = fork();
if (pid < 0) {
@ -285,6 +288,7 @@ int uv_proc_exec(uv_proc_manager_t* manager, const char* command,
options.stdio = stdio;
options.flags = UV_PROCESS_DETACHED;
/* xce-fork-ok: xcecd_worker is single-threaded (tech.md) */
int result = uv_spawn(manager->loop, process, &options);
close(fd);

37
src/worker/src/xcecd.c
View file

@ -33,8 +33,10 @@ static int virtual_pid = 1000;
*/
void xcecd_process_command(void* arg)
{
#ifdef DBUS_MANAGER
pid_t pid = 0;
char unit_name[256];
#endif
char jobidStr[64];
fork_task_t* ctx = (fork_task_t*)arg;
@ -67,34 +69,13 @@ void xcecd_process_command(void* arg)
}
#ifndef DBUS_MANAGER
if (process_manager_start_worker(ctx->manager) != 0) {
XCECD_LOG(XCE_LOG_FATAL, "Failed to start worker process");
xcecd_job_start_error(ctx->xce_jobid);
return;
}
/* adding worker to worker list，and update worker pid */
pid = get_current_process_pid(ctx->manager);
if (pid != -1) {
add_worker(&xcecd_ctx.worker_list, pid, WORKER_RUNNING, ctx->xce_jobid,
ctx->ntasks, ctx->startinfo);
if (!xcecd_db_insert_worker(xcecd_ctx.db, pid, WORKER_RUNNING,
ctx->xce_jobid, JOB_RUNNING)) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to insert worker into database, XCECD exit...");
dbus_stop_all_workers(&xcecd_ctx.worker_list, NULL);
xcecd_db_is_available(xcecd_ctx.db, xcecd_ctx.db_path);
}
} else {
XCECD_LOG(XCE_LOG_ERROR, "Failed to get worker pid");
xcecd_job_start_error(ctx->xce_jobid);
return;
}
/* send command to worker */
/* Starting xcecd_worker by fork() from this multi-threaded daemon was
* removed (no-fork rule, .kiro/steering/tech.md): workers start only as
* systemd units over D-Bus, which every shipped xcecd build uses. */
XCECD_LOG(XCE_LOG_FATAL, "xcecd built without DBUS_MANAGER cannot start "
"workers");
xcecd_job_start_error(ctx->xce_jobid);
return;
#else
#ifndef VIRTUAL_CLUSTER

1
src/worker/src/xcecd_worker.c
View file

@ -1343,6 +1343,7 @@ static void do_criu_ckpt(uint64_t job_id, int kill)
ckpt_dir, path, target_pid);
}
/* xce-fork-ok: xcecd_worker is single-threaded (tech.md) */
int ret = system(criu_cmd);
if (ret != 0) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to run criu command: %s",