merge job rusage and pending reason support

This commit is contained in:
Rongsong Shen 2025-12-16 14:36:28 +08:00
parent 30824ae5ea
commit cb0c7fe9e7
82 changed files with 4280 additions and 2561 deletions

View file

@ -52,8 +52,10 @@ int xcecd_job_ckpted(uint64_t jobid, int status, const char* info);
int xcecd_job_remote_start(uint64_t jobid, const char* hostname,
const char* command);
const cc_callbacks_t* get_cc_callbacks(void);
int xcecd_task_fail(uint64_t jobid, int clientidx, int taskidx, int code);
int xcecd_jobpost_done(uint64_t jobid);
int xcecd_notify_job_fail(uint64_t jobid, int stage, int code);
int xcecd_task_fail(uint64_t jobid, int clientidx, int taskidx, int code);
int xcecd_jobpost_done(uint64_t jobid);
int xcecd_notify_job_fail(uint64_t jobid, int stage, int code);
int xcecd_jobrusage_update(xce_job_rusages_t* rusages);
bool xcecd_send_load_indices(xce_resval_t** vals, int count);
#endif // CONTROL_CONNECTOR_H

View file

@ -18,99 +18,23 @@
#define DEFAULT_AFTER_HOUR "19:00-7:00 5:19:00-1:7:00"
struct time_window {
char* window_name; // 窗口名称
char* window_name;
windows_t* week[8];
time_t window_edge;
};
struct node_status {
short node_id; // 节点ID
int max_cpu_cores; // CPU核心数
int max_memory; // 内存大小
int num_disk; // 磁盘数量
int max_swap; // 交换空间大小
int max_network; // 网络带宽
unsigned short port_number; // 端口号
char node_type[MAXXCENAMELEN]; // 节点类型
char node_arch[MAXXCENAMELEN]; // 节点架构
};
typedef struct host_node host_node_t;
struct host_node {
char* node_name; // 节点名称
short node_number; // 节点编号
short node_model_number; // 节点型号编号
short node_type_number; // 节点类型编号
in_addr_t* node_ip; // 节点IP地址
struct node_status node_status; // 节点状态
char info_valid; // 信息是否有效
unsigned char protocol_version; // 协议版本
short available_high; // 高可用性
short available_low; // 低可用性
short use; // 使用情况
char* windows; // 时间窗口
windows_t* week[8]; // 一周的时间窗口
time_t window_edge; // 时间窗口边界
time_t last_jack_time; // 最后一次插入时间
int node_inactivity_count; // 节点不活动计数
int* status; // 状态
float* busy_threshold; // 繁忙阈值
float* load_index; // 负载指数
float* user_load_index; // 用户负载指数
int info_mask; // 信息掩码
int load_mask; // 负载掩码
int* resource_bitmap; // 资源位图
int* dynamic_resource_bitmap; // 动态资源位图
int number_of_instances; // 实例数量
host_node_t* next; // 下一个节点
time_t expire_time; // 过期时间
uint8_t migrant;
};
struct cluster_node {
short cluster_number; // 集群编号
char* cluster_name; // 集群名称
int status; // 状态
in_addr_t candidate_address_list[MAXCANDHOSTS]; // 候选地址列表
char* primary_node_name; // 主节点名称
unsigned short primary_address; // 主地址
unsigned short primary_port; // 主端口
char primary_known; // 主节点可识别
int primary_inactivity_count; // 主节点不活动计数
host_node_t* primary_pointer; // 主节点指针
host_node_t* previous_primary_pointer; // 上一个主节点指针
unsigned short checksum; // 校验和
int number_of_nodes; // 节点数量
int number_of_clients; // 客户端数量
host_node_t* node_list; // 节点列表
host_node_t* client_list; // 客户端列表
int manager_id; // 管理器ID
char* manager_name; // 管理器名称
int channel_fd; // 通道文件描述符
int number_of_index; // 索引数量
int number_of_user_index; // 用户索引数量
int user_index_class; // 用户索引类
char** load_index_names; // 负载指数名称
int number_of_admins; // 管理员数量
int* admin_ids; // 管理员ID
char** admin_names; // 管理员名称
int number_of_resouces; // 资源数量
int* resource_bitmaps; // 资源位图
int* hosttype_bitmaps; // 主机类型位图
int* hostmodel_bitmaps; // 主机模型位图
int number_of_shared_resources; // 共享资源数量
char** shared_resources; // 共享资源
struct cluster_node* next;
};
// client_node 和 minSLimConfData 数据结构需要添加吗?
host_node_t* host_node_create(void);
int host_node_update(host_node_t* node_ptr, host_node_t* new_node);
void free_host_node(host_node_t* node_ptr);
int get_cpu_usage(int total_cores, int used_cores);
char* node_info_get_15s(void);
char* node_info_get_1m(void);
char* node_info_get_15m(void);
char* node_info_get_ut(void);
char* node_info_get_pg(void);
char* node_info_get_ls(void);
char* node_info_get_it(void);
char* node_info_get_swap(void);
char* node_info_get_mem(void);
char* node_info_get_tmp(void);
char* node_info_get_io(void);
#endif // NODE_INFORMATION_H_

View file

@ -1,29 +1,22 @@
#ifndef RESOURCE_SYNC_H
#define RESOURCE_SYNC_H
#include "hardware_monitor.h"
#include "node_information.h"
#include <stdbool.h>
// 资源同步模块结构体
typedef struct resource_sync resource_sync_t;
// 创建资源同步模块
resource_sync_t* resource_sync_create(void);
// 销毁资源同步模块
void resource_sync_destroy(resource_sync_t* sync);
// 启动资源同步模块
bool resource_sync_start(resource_sync_t* sync);
// 停止资源同步模块
void resource_sync_stop(resource_sync_t* sync);
// 更新资源同步模块
bool resource_sync_update(resource_sync_t* sync);
// 读取资源同步模块
void resource_sync_read(resource_sync_t* sync, host_node_t** node);
#ifndef RESOURCE_SYNC_H
#define RESOURCE_SYNC_H
#include "hardware_monitor.h"
#include "node_information.h"
#include <stdbool.h>
typedef struct resource_sync resource_sync_t;
resource_sync_t* resource_sync_create(void);
void resource_sync_destroy(resource_sync_t* sync);
bool resource_sync_start(resource_sync_t* sync);
void resource_sync_stop(resource_sync_t* sync);
bool resource_sync_update(resource_sync_t* sync);
void resource_sync_read(resource_sync_t* sync, void* data);
bool resource_sync_register_load_index(const char* name, int valtype,
char* (*getter_func)(void),
bool is_builtin);
bool resource_sync_unregister_load_index(const char* name);
#endif // RESOURCE_SYNC_H

View file

@ -11,12 +11,13 @@
extern "C" {
#endif
#define HOSTNAME_LENGTH 256
#define MESSAGE_LENGTH 256
#define COMMAND_LENGTH 4096
#define METHOD_LENGTH 64
#define PATH_LENGTH 1024
#define EXDATA_LENGTH 16 * 1024
#define HOSTNAME_LENGTH 256
#define MESSAGE_LENGTH 256
#define COMMAND_LENGTH 4096
#define METHOD_LENGTH 64
#define PATH_LENGTH 1024
#define EXDATA_LENGTH 16 * 1024
#define MAX_RUSAGE_PAYLOAD 4096
typedef enum {
MSG_GET_JOB,
@ -30,10 +31,22 @@ typedef enum {
MSG_JOB_CKPT_RESULT,
MSG_REMOTE_START_JOB,
MSG_REMOTE_START_JOB_ACK,
MSG_JOB_RUSAGES,
MSG_HEARTBEAT,
MSG_ERROR
} message_type_t;
/**
* @brief IPC rusage message structure
* As rusage data is variable-sized, we define a maximum payload size.
* This structure will be serialized and sent.
*/
typedef struct {
uint64_t job_id;
size_t payload_size;
char payload[MAX_RUSAGE_PAYLOAD];
} ipc_job_rusages_t;
typedef struct {
int type;
} heartbeat_message_t;
@ -103,6 +116,7 @@ typedef struct {
job_ckpt_result_t ckpt_result;
remote_start_job_t remote_start;
remote_start_job_ack_t remote_start_ack;
ipc_job_rusages_t rusages;
char error_message[MESSAGE_LENGTH];
} data;
int worker_pid;
@ -129,6 +143,7 @@ typedef void (*job_ckpt_callback_t)(uint64_t job_id, int kill,
const char* method);
typedef void (*job_ckpt_result_callback_t)(uint64_t job_id, int status,
const char* restore_cmd);
typedef void (*job_rusages_callback_t)(xce_job_rusages_t* rusages);
bool ipc_master_init(const char* socket_path);
void ipc_master_run(uv_run_mode mode);
@ -141,6 +156,7 @@ bool ipc_master_send_job_signal(uint64_t job_id, int signal, bool force);
bool ipc_master_send_job_ckpt(uint64_t job_id, int kill, const char* method);
void ipc_master_set_ckpt_callback(job_ckpt_callback_t callback);
void ipc_master_set_ckpt_result_callback(job_ckpt_result_callback_t callback);
void ipc_master_set_rusages_callback(job_rusages_callback_t callback);
bool ipc_worker_init(const char* socket_path);
void ipc_worker_connect(connection_callback_t callback);
@ -160,6 +176,7 @@ bool ipc_worker_send_ckpt_ack(uint64_t job_id, int kill, const char* method);
void ipc_worker_set_ckpt_callback(job_ckpt_callback_t callback);
bool ipc_worker_send_ckpt_result(uint64_t job_id, int status,
const char* restore_cmd);
bool ipc_worker_send_rusages(xce_job_rusages_t* rusages);
void on_write_complete(uv_write_t* req, int status);

View file

@ -92,7 +92,6 @@ int xcecd_init(const xcecd_config_t* config);
int xcecd_start(void);
void xcecd_stop(void);
void xcecd_cleanup(void);
int send_status_and_receive_tasks(void);
void xcecd_process_command(void* arg);
const char* handle_network_request(const char* request, void* user_data);
bool xcecd_config_is_valid(const xcecd_config_t* config);

View file

@ -1,6 +1,7 @@
#ifndef _XCECD_CGROUPS_H_
#define _XCECD_CGROUPS_H_
#include "xcecd_utils.h"
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
@ -93,4 +94,13 @@ int get_systemd_job_exit_code(const char* unit_name);
*/
char* get_systemd_job_status_text(const char* unit_name);
/**
* @brief Collect resource usage statistics for a systemd job.
* @param job_id The ID of the job.
* @return A pointer to an xce_job_rusages_t struct, or NULL on failure.
* The caller is responsible for freeing the returned struct
* using free_xce_job_rusages().
*/
xce_job_rusages_t* collect_job_rusages(uint64_t job_id);
#endif // _XCECD_CGROUPS_H_

View file

@ -1,8 +1,7 @@
#ifndef _XCECD_EXDATA_H
#define _XCECD_EXDATA_H
#include <cjson/cJSON.h>
#include "cjson/cJSON.h"
#include "xcecd_cgroups.h"
#include <stdbool.h>
#include <stdint.h>

View file

@ -4,7 +4,22 @@ xcecd_worker_srcs = [
'src/xcecd_exdata.c',
'src/uv_ipc.c',
'src/uv_proc.c',
'src/static_array.c'
'src/static_array.c',
'src/xcecd_cgroups.c',
'src/xcecd_worker_utils.c',
'src/controller_connector.c',
'src/xcecd_dbus_manager.c',
'src/xcecd_db.c',
'src/xcecd.c',
'src/connection_handler.c',
'src/resource_sync.c',
'src/node_information.c',
'src/thread_pool.c',
'src/hardware_monitor.c',
'src/system_monitor.c',
'src/process_manager.c',
'src/process_worker_manager.c',
'src/xcecd_sig.c',
]
incdir = include_directories('.', 'include', '../include', '../lib')
@ -13,10 +28,12 @@ libuv_dep = dependency('libuv')
elogind_dep = dependency('libelogind')
sqlite3_dep = dependency('sqlite3')
libcgroup_dep = dependency('libcgroup')
hwloc_dep = dependency('hwloc')
xcecd_worker_deps = [
libxce_dep,
libcgroup_dep,
hwloc_dep,
libuv_dep,
elogind_dep,
sqlite3_dep,
@ -69,9 +86,7 @@ xcecd_srcs = [
'src/job_information.c'
]
hwloc_dep = dependency('hwloc')
xcecd_deps = [xcecd_worker_deps,hwloc_dep]
xcecd_deps = [xcecd_worker_deps]
xcecd = executable('xcecd',
[xcecd_srcs,'src/main.c'],
c_args: ['-DDBUS_MANAGER'],
@ -85,47 +100,3 @@ xcecd_virtual = executable('xcecd_virtual',
include_directories: [incdir],
dependencies:[xcecd_deps],
install:true)
include_dir_test = include_directories('test')
xce_test_deps = xcecd_deps
xce_sched_test = executable('xce_sched_test',
[xcecd_srcs, 'test/xce_sched_test.c'],
include_directories: [incdir,include_dir_test],
dependencies: [xce_test_deps],
install: true)
unity_deps = dependency('unity')
xcecd_daemon_test = executable('xcecd_daemon_test',
[xcecd_srcs, 'test/xcecd_daemon_test.c'],
include_directories: [incdir, include_dir_test],
dependencies: [xce_test_deps, unity_deps],
install: true)
xcecd_worker_test = executable('xcecd_worker_test',
[xcecd_srcs, 'test/xcecd_worker_test.c'],
include_directories: [incdir, include_dir_test],
dependencies: [xce_test_deps, unity_deps],
install: true)
# cgroup_srcs = [
# 'src/xcecd_cgroups.c',
# 'test/cgroup_test.c'
# ]
# cgroup_test = executable('cgroup_test',
# cgroup_srcs,
# include_directories: [incdir, include_dir_test],
# dependencies: [xce_test_deps, unity_deps],
# install: true)
limit_test_srcs = [
'test/limit_test.c'
]
limit_test = executable('limit_test',
limit_test_srcs,
include_directories: [incdir, include_dir_test],
dependencies: [xce_test_deps, unity_deps, math_dep],
install: true)

48
src/worker/src/controller_connector.c Normal file → Executable file
View file

@ -1514,3 +1514,51 @@ int xcecd_jobpost_done(uint64_t jobid)
"Job post done and calling xcecli_notify_jobpostdone...");
return xcecli_notify_jobpostdone(xcecli_ctx, jobid);
}
int xcecd_jobrusage_update(xce_job_rusages_t* rusages)
{
if (!rusages) {
XCECD_LOG(XCE_LOG_ERROR,
"Invalid rusage data.");
return -1;
}
XCECD_LOG(XCE_LOG_DEBUG,
"Updating job rusage to controller for job %lu", rusages->jobid);
return xcecli_update_jusage(xcecli_ctx,
rusages->jobid,
rusages->npids,
rusages->pids,
rusages->npgids,
rusages->pgids,
rusages->count,
rusages->usages);
}
bool xcecd_send_load_indices(xce_resval_t** vals, int count) {
if (vals == NULL || count <= 0) {
XCECD_LOG(XCE_LOG_ERROR,
"Invalid parameters for sending load indices");
return false;
}
if (xcecli_ctx == NULL) {
XCECD_LOG(XCE_LOG_ERROR,
"xcecli_ctx is not initialized");
return false;
}
int result = xcecli_update_resvals(xcecli_ctx, count, &vals[0]);
if (result != 0) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to send load indices to controller, error code: %d",
result);
return false;
}
XCECD_LOG(XCE_LOG_DEBUG,
"Successfully sent %d load indices to controller", count);
return true;
}

View file

@ -25,22 +25,13 @@ struct hardware_monitor {
time_t last_update_time;
};
static void write_to_acct_file(hardware_monitor_t* monitor, int cpu_cores)
bool hwloc_disabled = true;
bool hwloc_available = false;
static int get_cpu_cores_fallback(void)
{
XCECD_LOG(XCE_LOG_INFO, "Writing to acct file...");
FILE* file = fopen(monitor->acct_file, "a");
if (!file) {
return;
}
time_t now;
time(&now);
struct tm* timeinfo = localtime(&now);
char timestamp[64];
strftime(timestamp, sizeof(timestamp), "%Y-%m-%d %H:%M:%S", timeinfo);
fprintf(file, "%s\t%d\n", timestamp, cpu_cores);
fclose(file);
long cores = sysconf(_SC_NPROCESSORS_ONLN);
return (cores > 0) ? (int)cores : 1;
}
hardware_monitor_t* hardware_monitor_create(const char* acct_file)
@ -75,16 +66,20 @@ bool hardware_monitor_update(hardware_monitor_t* monitor)
}
hwloc_topology_t topology;
if (hwloc_topology_init(&topology) < 0
|| hwloc_topology_load(topology) < 0) {
return false;
int current_cores = -1;
if (!hwloc_disabled && hwloc_topology_init(&topology) >= 0
&& hwloc_topology_load(topology) >= 0) {
current_cores = hwloc_get_nbobjs_by_type(topology, HWLOC_OBJ_CORE);
hwloc_available = true;
} else {
XCECD_LOG(
XCE_LOG_DEBUG3,
"hwloc disabled or not available, using fallback for CPU cores");
current_cores = get_cpu_cores_fallback();
}
int current_cores = hwloc_get_nbobjs_by_type(topology, HWLOC_OBJ_CORE);
// 如果CPU核心数发生变化,记录到文件
if (current_cores != monitor->last_cpu_cores) {
write_to_acct_file(monitor, current_cores);
monitor->last_cpu_cores = current_cores;
}
@ -94,7 +89,10 @@ bool hardware_monitor_update(hardware_monitor_t* monitor)
monitor->free_memory = free_memory;
}
hwloc_topology_destroy(topology);
if (hwloc_available) {
hwloc_topology_destroy(topology);
}
monitor->last_update_time = time(NULL);
return true;
}
@ -136,23 +134,31 @@ bool hardware_monitor_get_memory_info(int64_t* total_memory,
return false;
}
hwloc_topology_t topology;
if (hwloc_topology_init(&topology) < 0
|| hwloc_topology_load(topology) < 0) {
return false;
}
if (!hwloc_disabled && hwloc_available) {
hwloc_topology_t topology;
if (hwloc_topology_init(&topology) < 0
|| hwloc_topology_load(topology) < 0) {
return false;
}
hwloc_obj_t root = hwloc_get_root_obj(topology);
*total_memory = root->total_memory;
hwloc_obj_t root = hwloc_get_root_obj(topology);
*total_memory = root->total_memory;
hwloc_topology_destroy(topology);
}
struct sysinfo info;
if (sysinfo(&info) != 0) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to get sysinfo: %s", strerror(errno));
return false;
}
*free_memory = (size_t)info.freeram * info.mem_unit;
hwloc_topology_destroy(topology);
if (!hwloc_available) {
*total_memory = (int64_t)info.totalram * info.mem_unit;
}
*free_memory = (int64_t)info.freeram * info.mem_unit;
return true;
}
#endif

View file

@ -1,9 +1,6 @@
#include "sqlite3.h"
#include "xcecd.h"
#include <errno.h>
#include <libgen.h>
#include <limits.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@ -15,16 +12,13 @@
#include <windows.h>
#else
#include <fcntl.h>
#include <libgen.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#endif
#if !defined(DBUS_MANAGER)
#include "process_worker_manager.h"
#endif
/* TEMP SOLUTION: use config later */
#define LOG_SIZE (1024 * 1024 * 100) // 100 MB
#define LOG_LEVEL 2 // DEBUG level
@ -116,6 +110,27 @@ static void handle_exited_children(process_manager_t* manager, StaticArray pids)
clear_static(&pids);
}
static void free_xce_job_rusages(xce_job_rusages_t* rusages)
{
if (!rusages) {
return;
}
FREEUP(rusages->pids);
FREEUP(rusages->pgids);
if (rusages->usages) {
for (int i = 0; i < rusages->count; i++) {
if (rusages->usages[i]) {
FREEUP(rusages->usages[i]->resname);
FREEUP(rusages->usages[i]);
}
}
FREEUP(rusages->usages);
}
FREEUP(rusages);
}
static void job_status_cleanup(xcecd_job_t* job)
{
job->job_id = 0;
@ -219,6 +234,20 @@ static void master_ckpt_result_callback(uint64_t job_id, int status,
xcecd_job_ckpted(job_id, status, restore_cmd);
}
static void master_rusages_callback(xce_job_rusages_t* rusages)
{
if (!rusages) {
return;
}
XCECD_LOG(XCE_LOG_DEBUG, "Received rusage for job %lu with %d metrics.",
rusages->jobid, rusages->count);
xcecd_jobrusage_update(rusages);
free_xce_job_rusages(rusages);
}
static int check_xce_server_dir(char* myfile)
{
char* xce_server_dir = getenv("XCE_SERVER_DIR");
@ -242,7 +271,6 @@ static int check_xce_server_dir(char* myfile)
}
XCECD_LOG(XCE_LOG_INFO, "XCE_SERVER_DIR set to `/opt/xce/bin'");
setenv("XCE_SERVER_DIR", "/opt/xce/bin", 0);
return 1;
}
@ -525,6 +553,7 @@ int main(int argc, char* argv[])
ipc_master_set_heartbeat_callback(master_heartbeat_callback);
ipc_master_set_disconnect_callback(master_disconnect_callback);
ipc_master_set_ckpt_callback(master_ckpt_result_callback);
ipc_master_set_rusages_callback(master_rusages_callback);
XCECD_LOG(XCE_LOG_INFO, "XCECD started successfully.");

File diff suppressed because it is too large Load diff

View file

@ -10,27 +10,326 @@
#include <stdlib.h>
#include <unistd.h>
#define SYNC_INTERVAL 1
#define SYNC_INTERVAL 1
#define LOAD_INDICES_INTERVAL_DEFAULT 15
#define MAX_LOAD_INDICES 64
#define BUILTIN_LOAD_INDICES_COUNT 11
struct resource_sync {
pthread_t thread;
_Atomic bool running;
_Atomic bool updating;
struct cluster_node* my_cluster;
struct host_node* my_node;
struct client_node* my_client;
pthread_t thread;
_Atomic bool running;
_Atomic bool updating;
int load_report_counter;
};
typedef struct {
char* name;
int valtype;
char* (*getter_func)(void);
bool is_builtin;
} load_index_info_t;
static load_index_info_t* g_load_indices = NULL;
static int g_load_indices_count = 0;
static int g_load_indices_capacity = 0;
static pthread_mutex_t g_load_indices_mutex = PTHREAD_MUTEX_INITIALIZER;
static void free_resvals(xce_resval_t* vals, int count)
{
if (vals == NULL)
return;
for (int i = 0; i < count; i++) {
if (vals[i].name) {
FREEUP(vals[i].name);
}
if (vals[i].host) {
FREEUP(vals[i].host);
}
if (vals[i].val) {
if (vals[i].val->which == Value_u_svalue && vals[i].val->u.svalue) {
FREEUP(vals[i].val->u.svalue);
}
FREEUP(vals[i].val);
}
}
FREEUP(vals);
}
static void init_builtin_load_indices(void)
{
static bool initialized = false;
if (initialized)
return;
g_load_indices_capacity = MAX_LOAD_INDICES;
g_load_indices = (load_index_info_t*)calloc(g_load_indices_capacity,
sizeof(load_index_info_t));
// register built-in load indices
resource_sync_register_load_index("r15s", Value_u_fvalue, node_info_get_15s,
true);
resource_sync_register_load_index("r1m", Value_u_fvalue, node_info_get_1m,
true);
resource_sync_register_load_index("r15m", Value_u_fvalue, node_info_get_15m,
true);
resource_sync_register_load_index("ut", Value_u_fvalue, node_info_get_ut,
true);
resource_sync_register_load_index("pg", Value_u_fvalue, node_info_get_pg,
true);
resource_sync_register_load_index("ls", Value_u_fvalue, node_info_get_ls,
true);
resource_sync_register_load_index("it", Value_u_fvalue, node_info_get_it,
true);
resource_sync_register_load_index("swp", Value_u_fvalue, node_info_get_swap,
true);
resource_sync_register_load_index("mem", Value_u_fvalue, node_info_get_mem,
true);
resource_sync_register_load_index("tmp", Value_u_fvalue, node_info_get_tmp,
true);
resource_sync_register_load_index("io", Value_u_fvalue, node_info_get_io,
true);
initialized = true;
}
/**
* @brief Registers a new load index.
* @param name The name of the load index.
* @param valtype The type of the load index value.
* @param getter_func Function pointer to retrieve the load index value.
* @param is_builtin Whether the load index is built-in or custom.
* @return true on success, false on failure.
*/
bool resource_sync_register_load_index(const char* name, int valtype,
char* (*getter_func)(void),
bool is_builtin)
{
if (name == NULL || getter_func == NULL) {
return false;
}
pthread_mutex_lock(&g_load_indices_mutex);
for (int i = 0; i < g_load_indices_count; i++) {
if (strcmp(g_load_indices[i].name, name) == 0) {
XCECD_LOG(XCE_LOG_WARN, "Load index '%s' already exists", name);
pthread_mutex_unlock(&g_load_indices_mutex);
return false;
}
}
if (g_load_indices_count >= g_load_indices_capacity) {
XCECD_LOG(XCE_LOG_ERROR, "Maximum load indices capacity reached (%d)",
g_load_indices_capacity);
pthread_mutex_unlock(&g_load_indices_mutex);
return false;
}
g_load_indices[g_load_indices_count].name = strdup(name);
g_load_indices[g_load_indices_count].valtype = valtype;
g_load_indices[g_load_indices_count].getter_func = getter_func;
g_load_indices[g_load_indices_count].is_builtin = is_builtin;
g_load_indices_count++;
pthread_mutex_unlock(&g_load_indices_mutex);
XCECD_LOG(XCE_LOG_DEBUG3,
"Registered load index '%s' (type: %d, builtin: %s)", name,
valtype, is_builtin ? "yes" : "no");
return true;
}
/**
* @brief Unregisters a load index.
* @param name The name of the load index to unregister.
* @return true on success, false on failure.
*/
bool resource_sync_unregister_load_index(const char* name)
{
if (name == NULL) {
return false;
}
pthread_mutex_lock(&g_load_indices_mutex);
for (int i = 0; i < g_load_indices_count; i++) {
if (strcmp(g_load_indices[i].name, name) == 0) {
if (g_load_indices[i].is_builtin) {
XCECD_LOG(XCE_LOG_WARN,
"Cannot unregister builtin load index '%s'", name);
pthread_mutex_unlock(&g_load_indices_mutex);
return false;
}
FREEUP(g_load_indices[i].name);
// Shift remaining indices
for (int j = i; j < g_load_indices_count - 1; j++) {
g_load_indices[j] = g_load_indices[j + 1];
}
g_load_indices_count--;
pthread_mutex_unlock(&g_load_indices_mutex);
XCECD_LOG(XCE_LOG_DEBUG3, "Unregistered load index '%s'", name);
return true;
}
}
pthread_mutex_unlock(&g_load_indices_mutex);
XCECD_LOG(XCE_LOG_WARN, "Load index '%s' not found", name);
return false;
}
static xce_resval_t* collect_load_indices(int* count)
{
if (g_load_indices == NULL) {
init_builtin_load_indices();
}
pthread_mutex_lock(&g_load_indices_mutex);
int rescount = g_load_indices_count;
if (rescount == 0) {
pthread_mutex_unlock(&g_load_indices_mutex);
*count = 0;
return NULL;
}
xce_resval_t* vals = (xce_resval_t*)calloc(rescount, sizeof(xce_resval_t));
if (vals == NULL) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for load indices");
pthread_mutex_unlock(&g_load_indices_mutex);
*count = 0;
return NULL;
}
char hostname[256];
if (gethostname(hostname, sizeof(hostname)) != 0) {
strcpy(hostname, "unknown");
}
for (int i = 0; i < rescount; i++) {
vals[i].name = strdup(g_load_indices[i].name);
vals[i].host = strdup(hostname);
vals[i].val = (xce_val_t*)calloc(1, sizeof(xce_val_t));
if (vals[i].name == NULL || vals[i].host == NULL
|| vals[i].val == NULL) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to allocate memory for load index %s",
g_load_indices[i].name);
free_resvals(vals, i + 1);
pthread_mutex_unlock(&g_load_indices_mutex);
*count = 0;
return NULL;
}
char* val_str = g_load_indices[i].getter_func();
switch (g_load_indices[i].valtype) {
case Value_u_fvalue:
vals[i].val->which = g_load_indices[i].valtype;
if (val_str == NULL) {
vals[i].val->u.fvalue = 0.0;
} else {
vals[i].val->u.fvalue = atof(val_str);
FREEUP(val_str);
}
break;
case Value_u_ivalue32:
vals[i].val->which = g_load_indices[i].valtype;
if (val_str == NULL) {
vals[i].val->u.ivalue32 = 0;
} else {
vals[i].val->u.ivalue32 = atoi(val_str);
FREEUP(val_str);
}
break;
case Value_u_svalue:
vals[i].val->which = g_load_indices[i].valtype;
if (val_str == NULL) {
vals[i].val->u.svalue = strdup("");
} else {
vals[i].val->u.svalue = val_str;
}
break;
default:
XCECD_LOG(XCE_LOG_WARN, "Unsupported value type for %s",
g_load_indices[i].name);
vals[i].val->which = Value_u_fvalue;
vals[i].val->u.fvalue = 0.0;
if (val_str)
FREEUP(val_str);
break;
}
}
pthread_mutex_unlock(&g_load_indices_mutex);
*count = rescount;
return vals;
}
/**
* @brief Cleans up all registered load indices.
*/
static void resource_sync_cleanup_load_indices(void)
{
pthread_mutex_lock(&g_load_indices_mutex);
if (g_load_indices != NULL) {
for (int i = 0; i < g_load_indices_count; i++) {
FREEUP(g_load_indices[i].name);
}
FREEUP(g_load_indices);
g_load_indices = NULL;
g_load_indices_count = 0;
g_load_indices_capacity = 0;
}
pthread_mutex_unlock(&g_load_indices_mutex);
}
static void* resource_sync_thread(void* arg)
{
resource_sync_t* sync = (resource_sync_t*)arg;
resource_sync_t* sync = (resource_sync_t*)arg;
xce_resval_t* vals = NULL;
xce_resval_t** valptrs = NULL;
int vals_count = 0;
while (atomic_load(&sync->running)) {
/* TODO:资源更新,添加逻辑 */
resource_sync_update(sync);
/* TODO:资源汇总,添加逻辑 */
sync->load_report_counter++;
if (sync->load_report_counter >= LOAD_INDICES_INTERVAL_DEFAULT) {
vals = collect_load_indices(&vals_count);
if (vals == NULL || vals_count <= 0) {
XCECD_LOG(XCE_LOG_ERROR, "No load indices to report");
sync->load_report_counter = 0;
sleep(SYNC_INTERVAL);
continue;
}
valptrs = (xce_resval_t**)calloc(vals_count, sizeof(xce_resval_t*));
if (valptrs == NULL) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to allocate memory for value pointers");
free_resvals(vals, vals_count);
sync->load_report_counter = 0;
sleep(SYNC_INTERVAL);
continue;
}
for (int i = 0; i < vals_count; i++) {
valptrs[i] = &vals[i];
}
xcecd_send_load_indices(valptrs, vals_count);
FREEUP(valptrs);
free_resvals(vals, vals_count);
sync->load_report_counter = 0;
}
sleep(SYNC_INTERVAL);
}
@ -48,10 +347,8 @@ resource_sync_t* resource_sync_create(void)
}
atomic_store(&sync->running, false);
atomic_store(&sync->updating, false);
sync->my_cluster = NULL;
sync->my_node = host_node_create();
sync->my_client = NULL;
sync->thread = 0;
sync->thread = 0;
sync->load_report_counter = 0;
XCECD_LOG(XCE_LOG_DEBUG, "Resource sync module created successfully");
return sync;
@ -64,9 +361,9 @@ void resource_sync_destroy(resource_sync_t* sync)
XCECD_LOG(XCE_LOG_ERROR, "Invalid resource sync module");
return;
}
/* 资源销毁 TODO:先释放host,client等结构 */
FREEUP(sync->my_node);
resource_sync_stop(sync);
resource_sync_cleanup_load_indices();
XCECD_LOG(XCE_LOG_DEBUG, "Resource sync module destroyed successfully");
FREEUP(sync);
}
@ -104,7 +401,6 @@ void resource_sync_stop(resource_sync_t* sync)
bool resource_sync_update(resource_sync_t* sync)
{
// XCECD_LOG(XCE_LOG_DEBUG, "Updating resource sync module...");
if (sync == NULL) {
XCECD_LOG(XCE_LOG_ERROR, "Invalid resource sync module");
return false;
@ -112,29 +408,23 @@ bool resource_sync_update(resource_sync_t* sync)
atomic_store(&sync->updating, true);
/* TODO:资源更新,添加逻辑 */
sync->my_node->node_status.max_cpu_cores
= hardware_monitor_get_cpu_cores(xcecd_ctx.hardware_monitor);
/* Currently there is no resource need to update */
atomic_store(&sync->updating, false);
// XCECD_LOG(XCE_LOG_DEBUG, "Resource sync module updated successfully");
return true;
}
void resource_sync_read(resource_sync_t* sync, host_node_t** node)
void resource_sync_read(resource_sync_t* sync, void* data)
{
if (!sync) {
return;
}
while (atomic_load(&sync->updating)) {
/* 等待更新完成 */
/* wait sync to complete */
usleep(1000);
}
/* 读取资源同步状态
* TODO: 添加读取逻辑
*/
*node = sync->my_node;
/* Currently there is no resource need to read */
}

View file

@ -32,6 +32,7 @@ static worker_reconnect_callback_t worker_reconnect_callback = NULL;
static job_signal_callback_t worker_signal_callback = NULL;
static job_ckpt_callback_t worker_ckpt_callback = NULL;
static job_ckpt_result_callback_t master_ckpt_result_callback = NULL;
static job_rusages_callback_t master_rusages_callback = NULL;
static uv_timer_t* exit_timer = NULL;
@ -506,6 +507,61 @@ static void on_master_read(uv_stream_t* client, ssize_t nread,
msg->data.ckpt_result.status,
msg->data.ckpt_result.restore_cmd);
}
} else if (msg->type == MSG_JOB_RUSAGES) {
XCECD_LOG(XCE_LOG_DEBUG, "Received rusage info for job %lu",
msg->data.rusages.job_id);
if (master_rusages_callback) {
// deserialize payload
xce_job_rusages_t* rusages
= (xce_job_rusages_t*)calloc(1, sizeof(xce_job_rusages_t));
if (rusages) {
char* p = msg->data.rusages.payload;
rusages->jobid = msg->data.rusages.job_id;
memcpy(&rusages->npids, p, sizeof(int));
p += sizeof(int);
if (rusages->npids > 0) {
rusages->pids
= (int*)calloc(rusages->npids, sizeof(int));
memcpy(rusages->pids, p, rusages->npids * sizeof(int));
p += rusages->npids * sizeof(int);
}
memcpy(&rusages->npgids, p, sizeof(int));
p += sizeof(int);
if (rusages->npgids > 0) {
rusages->pgids
= (int*)calloc(rusages->npgids, sizeof(int));
memcpy(rusages->pgids, p,
rusages->npgids * sizeof(int));
p += rusages->npgids * sizeof(int);
}
memcpy(&rusages->count, p, sizeof(int));
p += sizeof(int);
if (rusages->count > 0) {
rusages->usages = (xce_rusage_t**)calloc(
rusages->count, sizeof(xce_rusage_t*));
for (int i = 0; i < rusages->count; i++) {
rusages->usages[i] = (xce_rusage_t*)calloc(
1, sizeof(xce_rusage_t));
size_t name_len;
memcpy(&name_len, p, sizeof(size_t));
p += sizeof(size_t);
rusages->usages[i]->resname
= (char*)calloc(name_len + 1, sizeof(char));
memcpy(rusages->usages[i]->resname, p, name_len);
p += name_len;
memcpy(&rusages->usages[i]->value, p,
sizeof(float));
p += sizeof(float);
}
}
master_rusages_callback(rusages);
// note: master_rusages_callback needs to free rusages and
// its internal pointers
}
}
} else if (msg->type == MSG_REMOTE_START_JOB) {
XCECD_LOG(XCE_LOG_DEBUG,
"Received remote start request for host '%s'",
@ -1361,7 +1417,21 @@ void ipc_master_set_ckpt_callback(job_ckpt_callback_t callback)
}
/**
*
* @brief master set rusages callback
* @param callback
* @return void
*/
void ipc_master_set_rusages_callback(job_rusages_callback_t callback)
{
master_rusages_callback = callback;
}
/**
* @brief worker send checkpoint result
* @param job_id - job ID
* @param status - checkpoint status (0 success, non-zero failure)
* @param restore_cmd - command to restore the job
* @return bool - true if sent successfully
*/
bool ipc_worker_send_ckpt_result(uint64_t job_id, int status,
const char* restore_cmd)
@ -1396,3 +1466,79 @@ bool ipc_worker_send_ckpt_result(uint64_t job_id, int status,
return true;
}
/**
* @brief worker send rusage info to master
* @param rusages - rusage data
* @return bool - true if sent successfully
*/
bool ipc_worker_send_rusages(xce_job_rusages_t* rusages)
{
if (!worker_connected || !rusages)
return false;
ipc_message_t msg;
memset(&msg, 0, sizeof(msg));
msg.type = MSG_JOB_RUSAGES;
msg.data.rusages.job_id = rusages->jobid;
// serialize payload
char* p = msg.data.rusages.payload;
char* end = p + MAX_RUSAGE_PAYLOAD;
// serialize npids and pids
memcpy(p, &rusages->npids, sizeof(int));
p += sizeof(int);
if (p + rusages->npids * sizeof(int) > end)
return false; // check buffer overflow
if (rusages->npids > 0) {
memcpy(p, rusages->pids, rusages->npids * sizeof(int));
p += rusages->npids * sizeof(int);
}
// serialize npgids and pgids
memcpy(p, &rusages->npgids, sizeof(int));
p += sizeof(int);
if (p + rusages->npgids * sizeof(int) > end)
return false;
if (rusages->npgids > 0) {
memcpy(p, rusages->pgids, rusages->npgids * sizeof(int));
p += rusages->npgids * sizeof(int);
}
// serialize count and usages
memcpy(p, &rusages->count, sizeof(int));
p += sizeof(int);
for (int i = 0; i < rusages->count; i++) {
xce_rusage_t* usage = rusages->usages[i];
size_t name_len = strlen(usage->resname);
if (p + sizeof(size_t) + name_len + sizeof(float) > end)
return false;
memcpy(p, &name_len, sizeof(size_t));
p += sizeof(size_t);
memcpy(p, usage->resname, name_len);
p += name_len;
memcpy(p, &usage->value, sizeof(float));
p += sizeof(float);
}
msg.data.rusages.payload_size = p - msg.data.rusages.payload;
uv_write_t* req = (uv_write_t*)calloc(1, sizeof(uv_write_t));
if (req) {
uv_buf_t buf = uv_buf_init((char*)&msg, sizeof(msg));
int r = uv_write(req, (uv_stream_t*)&worker_pipe, &buf, 1,
on_write_complete);
if (r < 0) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to send rusage info: %s",
uv_strerror(r));
free(req);
return false;
}
}
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Sent rusage info for job %lu",
rusages->jobid);
return true;
}

View file

@ -1,8 +1,5 @@
#include "uv_proc.h"
#include "utils.h"
#include "xcecd_utils.h"
#include "xcelog.h"
#include <errno.h>
#include <signal.h>
#include <stdio.h>
@ -11,6 +8,8 @@
#include <sys/wait.h>
#include <unistd.h>
#include "utils.h"
static int make_job_output_path(char* buf, size_t bufsize, const char* dir,
uint64_t jobid, int task)
{
@ -25,6 +24,7 @@ static int make_job_output_path(char* buf, size_t bufsize, const char* dir,
}
fmt_jobid(idstring, sizeof(idstring) - 1, jobid);
return snprintf(buf, bufsize, "%s/job_%s_task%d.out", dir, idstring, task);
}
@ -261,9 +261,13 @@ int uv_proc_exec(uv_proc_manager_t* manager, const char* command,
}
char job_output[256];
#ifdef AUTOSCALER_TEST
make_job_output_path(job_output, sizeof(job_output) - 1, "/shared/log",
manager->job_id, i);
#else
make_job_output_path(job_output, sizeof(job_output) - 1, "/tmp",
manager->job_id, i);
#endif
uv_stdio_container_t stdio[3];
uv_fs_t file_req;

View file

@ -12,8 +12,9 @@
#include <sys/socket.h>
#include <sys/un.h>
#if !defined(DBUS_MANAGER)
#include "process_worker_manager.h"
#include "utils.h"
#endif
xcecd_context_t xcecd_ctx = { 0 };
hostattrs_t host_attrs = { 0 };
@ -32,7 +33,7 @@ static int virtual_pid = 1000;
void xcecd_process_command(void* arg)
{
pid_t pid = 0;
char unit_name[256];
char unit_name[256];
fork_task_t* ctx = (fork_task_t*)arg;
if (!ctx || !ctx->startinfo || !ctx->startinfo->cmdline) {
@ -129,6 +130,7 @@ void xcecd_process_command(void* arg)
make_job_sd_unit_name(unit_name, sizeof(unit_name) - 1, "xcecd-worker",
ctx->xce_jobid);
/* get worker pid */
pid = dbus_get_worker_pid(unit_name);
#else
@ -245,136 +247,13 @@ const char* handle_network_request(const char* request, void* user_data)
return strdup("Task submitted successfully\n");
}
// TODO: next phase task - send workload and host info to controller
int send_status_and_receive_tasks(void)
{
// disabled currently
return 0;
int sockfd;
struct sockaddr_in server_addr;
char buffer[BUFFER_SIZE];
static int error_count = 0;
int error_log_interval = 10;
// 创建套接字
sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd < 0) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to create socket");
return -1;
}
// 设置服务器地址
memset(&server_addr, 0, sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(xcecd_ctx.remote_port);
// 打印远程主机名和端口以进行调试
XCECD_LOG(XCE_LOG_DEBUG, "Connecting to remote host: %s, port: %d",
xcecd_ctx.remote_host, xcecd_ctx.remote_port);
// 使用getaddrinfo解析主机名
struct addrinfo hints, *res;
memset(&hints, 0, sizeof(struct addrinfo));
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
int err = getaddrinfo(xcecd_ctx.remote_host, NULL, &hints, &res);
if (err != 0) {
XCECD_LOG(XCE_LOG_ERROR, "getaddrinfo: %s", gai_strerror(err));
close(sockfd);
return -1;
}
struct sockaddr_in* addr_in = (struct sockaddr_in*)res->ai_addr;
server_addr.sin_addr = addr_in->sin_addr;
freeaddrinfo(res);
// 打印解析后的IP地址
char ip_str[INET_ADDRSTRLEN];
inet_ntop(AF_INET, &(server_addr.sin_addr), ip_str, INET_ADDRSTRLEN);
XCECD_LOG(XCE_LOG_DEBUG, "Resolved IP address: %s", ip_str);
// 连接到服务器
if (connect(sockfd, (struct sockaddr*)&server_addr, sizeof(server_addr))
< 0) {
error_count++;
if ((error_count == 1) || (error_count % error_log_interval == 0)) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed connecting to remote host: %s (attempt %d)",
strerror(errno), error_count);
}
close(sockfd);
return -1;
}
// 准备状态信息
int running_processes = get_process_count(xcecd_ctx.process_manager);
host_node_t* node = NULL;
resource_sync_read(xcecd_ctx.resource_sync, &node);
int slots
= get_cpu_usage(node->node_status.max_cpu_cores, running_processes);
snprintf(buffer, BUFFER_SIZE, "{\"slots\":%d}", slots);
// 发送状态信息
if (send(sockfd, buffer, strlen(buffer), 0) < 0) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to send status");
close(sockfd);
return -1;
}
/* 接收任务请求 */
ssize_t bytes_received;
while ((bytes_received = recv(sockfd, buffer, BUFFER_SIZE - 1, 0)) > 0) {
buffer[bytes_received] = '\0';
XCECD_LOG(XCE_LOG_DEBUG, "Received task: %s", buffer);
/* 创建任务上下文并添加到fork线程池队列 */
fork_task_t* ctx = calloc(1, sizeof(fork_task_t));
if (ctx) {
ctx->startinfo = calloc(1, sizeof(job_startinfo_t));
if (!ctx->startinfo) {
FREEUP(ctx);
XCECD_LOG(XCE_LOG_ERROR,
"Failed to allocate memory for start_info");
continue;
}
ctx->startinfo->cmdline = strdup(buffer);
if (!ctx->startinfo->cmdline) {
FREEUP(ctx->startinfo);
FREEUP(ctx);
XCECD_LOG(XCE_LOG_ERROR,
"Failed to allocate memory for command");
continue;
}
#ifndef DBUS_MANAGER
ctx->manager = (void*)xcecd_ctx.process_manager;
#endif
ctx->func = xcecd_process_command;
if (submit_fork_task(xcecd_ctx.thread_pool, ctx) != 0) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to add task to thread pool");
FREEUP(ctx->startinfo->cmdline);
FREEUP(ctx->startinfo);
FREEUP(ctx);
} else {
XCECD_LOG(XCE_LOG_DEBUG, "Task added to thread pool queue");
}
}
}
close(sockfd);
return 0;
}
static void connector_to_controller(void* arg)
{
while (xcecd_ctx.running && cc_ctx->running) {
checking_connection();
if (xcecd_ctx.running && cc_ctx->running) {
send_status_and_receive_tasks();
// continue running
} else {
XCECD_LOG(XCE_LOG_ERROR,
"XCECD daemon is not running, exiting thread");
@ -612,7 +491,10 @@ int xcecd_start(void)
if (sqlite3_open(xcecd_ctx.db_path, &xcecd_ctx.db) != SQLITE_OK) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to open database: %s",
sqlite3_errmsg(xcecd_ctx.db));
FREEUP(xcecd_ctx.db);
if (xcecd_ctx.db != NULL) {
sqlite3_close(xcecd_ctx.db);
}
return false;
}

View file

@ -1,5 +1,4 @@
#include "xcecd_cgroups.h"
#include "xcecd_utils.h"
#include "xcelog.h"
#include <sched.h>
@ -27,9 +26,11 @@ char* start_systemd_job(const char* command, const res_limits_t* limits,
char service_name[256];
// 生成唯一的unit名
make_job_sd_unit_name(service_name, sizeof(service_name) - 1,
unit_name_prefix ? unit_name_prefix : "xcecd-worker",
const char* prefix = unit_name_prefix ? unit_name_prefix : "xcecd-worker";
make_job_sd_unit_name(service_name, sizeof(service_name) - 1, prefix,
job_id);
// 连接到system bus
int r = sd_bus_open_system(&bus);
if (r < 0) {
@ -241,6 +242,56 @@ char* start_systemd_job(const char* command, const res_limits_t* limits,
if (r < 0)
goto finish;
// 设置 TasksAccounting = yes
r = sd_bus_message_open_container(m, 'r', "sv");
if (r < 0)
goto finish;
r = sd_bus_message_append(m, "s", "TasksAccounting");
if (r < 0)
goto finish;
r = sd_bus_message_open_container(m, 'v', "b");
if (r < 0)
goto finish;
r = sd_bus_message_append(m, "b", 1);
if (r < 0)
goto finish;
r = sd_bus_message_close_container(m); // 关闭变体
if (r < 0)
goto finish;
r = sd_bus_message_close_container(m); // 关闭属性
if (r < 0)
goto finish;
// 设置 MemoryAccounting = yes
r = sd_bus_message_open_container(m, 'r', "sv");
if (r < 0)
goto finish;
r = sd_bus_message_append(m, "s", "MemoryAccounting");
if (r < 0)
goto finish;
r = sd_bus_message_open_container(m, 'v', "b");
if (r < 0)
goto finish;
r = sd_bus_message_append(m, "b", 1);
if (r < 0)
goto finish;
r = sd_bus_message_close_container(m); // 关闭变体
if (r < 0)
goto finish;
r = sd_bus_message_close_container(m); // 关闭属性
if (r < 0)
goto finish;
// 如果提供了资源限制,设置相应的属性
if (limits) {
@ -1225,3 +1276,454 @@ char* get_systemd_job_status_text(const char* unit_name)
return status_text;
}
/**
* @brief Comparison function for qsort.
*/
static int compare_ints(const void* a, const void* b)
{
int int_a = *((int*)a);
int int_b = *((int*)b);
if (int_a < int_b)
return -1;
if (int_a > int_b)
return 1;
return 0;
}
/**
* @brief Get a specific property from a systemd unit.
* This is a generic helper function.
* @param bus The sd_bus connection.
* @param unit_path The object path of the unit.
* @param interface The D-Bus interface.
* @param property The name of the property to get.
* @param type The expected D-Bus type character.
* @param ret A pointer to store the result.
* @return 0 on success, negative error code on failure.
*/
static int get_unit_property(sd_bus* bus, const char* unit_path,
const char* interface, const char* property,
char type, void* ret)
{
sd_bus_error error = SD_BUS_ERROR_NULL;
int r = sd_bus_get_property_trivial(bus, "org.freedesktop.systemd1",
unit_path, interface, property, &error,
type, ret);
if (r < 0) {
XCECD_LOG(XCE_LOG_DEBUG, "Failed to get property '%s': %s", property,
error.message);
sd_bus_error_free(&error);
}
return r;
}
/**
* @brief From cgroup.procs read all PIDs
* @param control_group for example "/system.slice/xcecd-worker-123.service"
*/
static int read_pids_from_cgroup(const char* control_group, int** pids_out,
int* npids_out)
{
if (!control_group || !pids_out || !npids_out)
return -EINVAL;
char path[MAXPATHLEN];
snprintf(path, sizeof(path), "/sys/fs/cgroup%s/cgroup.procs",
control_group);
FILE* fp = fopen(path, "r");
if (!fp) {
XCECD_LOG(XCE_LOG_DEBUG, "Open %s failed: %s", path, strerror(errno));
return -errno;
}
int capacity = 32;
int* arr = (int*)calloc(capacity, sizeof(int));
if (!arr) {
fclose(fp);
return -ENOMEM;
}
int pid, count = 0;
while (fscanf(fp, "%d", &pid) == 1) {
if (count >= capacity) {
capacity *= 2;
int* tmp = (int*)realloc(arr, capacity * sizeof(int));
if (!tmp) {
FREEUP(arr);
fclose(fp);
return -ENOMEM;
}
arr = tmp;
}
arr[count++] = pid;
}
fclose(fp);
if (count == 0) {
FREEUP(arr);
return 1; // no pid
}
*pids_out = arr;
*npids_out = count;
return 0;
}
/**
* @brief Collect resource usage statistics for a systemd job.
* @param job_id The ID of the job.
* @return A pointer to an xce_job_rusages_t struct, or NULL on failure.
* The caller is responsible for freeing the returned struct
* using free_xce_job_rusages().
*/
xce_job_rusages_t* collect_job_rusages(uint64_t job_id)
{
sd_bus* bus = NULL;
int r;
char unit_name[256];
char unit_path[512];
xce_job_rusages_t* rusages = NULL;
rusages = (xce_job_rusages_t*)calloc(1, sizeof(xce_job_rusages_t));
if (!rusages) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to allocate memory for rusages struct.");
return NULL;
}
rusages->jobid = job_id;
r = sd_bus_open_system(&bus);
if (r < 0) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to connect to system bus: %s",
strerror(-r));
FREEUP(rusages);
return NULL;
}
snprintf(unit_name, sizeof(unit_name), "xcecd-worker-%lu.service", job_id);
char* escaped_unit_name = replace_chars(unit_name);
if (!escaped_unit_name) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to escape unit name.");
sd_bus_unref(bus);
FREEUP(rusages);
return NULL;
}
snprintf(unit_path, sizeof(unit_path), "/org/freedesktop/systemd1/unit/%s",
escaped_unit_name);
FREEUP(escaped_unit_name);
const char* res_names[]
= { "cpu_time_sec", "memory_rss", "memory_swap", "tasks_current" };
const int num_res = sizeof(res_names) / sizeof(res_names[0]);
rusages->count = num_res;
rusages->usages = (xce_rusage_t**)calloc(num_res, sizeof(xce_rusage_t*));
if (!rusages->usages) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to allocate memory for usages array.");
goto cleanup;
}
for (int i = 0; i < num_res; i++) {
rusages->usages[i] = (xce_rusage_t*)calloc(1, sizeof(xce_rusage_t));
if (!rusages->usages[i]) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to allocate memory for rusage item.");
goto cleanup;
}
rusages->usages[i]->resname = strdup(res_names[i]);
}
uint64_t u64_val = 0;
// CPU Time (CPUUsageNSec)
if (get_unit_property(bus, unit_path, "org.freedesktop.systemd1.Service",
"CPUUsageNSec", 't', &u64_val)
>= 0) {
rusages->usages[0]->value = (float)u64_val / 1e9; // 从纳秒转换为秒
}
// Memory RSS (MemoryCurrent)
if (get_unit_property(bus, unit_path, "org.freedesktop.systemd1.Service",
"MemoryCurrent", 't', &u64_val)
>= 0) {
rusages->usages[1]->value
= (float)u64_val / (1024 * 1024); // 从字节转换为MB
}
// Memory Swap (MemorySwapCurrent)
if (get_unit_property(bus, unit_path, "org.freedesktop.systemd1.Service",
"MemorySwapCurrent", 't', &u64_val)
>= 0) {
rusages->usages[2]->value
= (float)u64_val / (1024 * 1024); // 从字节转换为MB
}
// Tasks/Threads (TasksCurrent)
if (get_unit_property(bus, unit_path, "org.freedesktop.systemd1.Service",
"TasksCurrent", 't', &u64_val)
>= 0) {
rusages->usages[3]->value = (float)u64_val;
}
// get all PIDs from cgroup.procs
char* control_group = NULL;
sd_bus_error cg_error = SD_BUS_ERROR_NULL;
sd_bus_message* reply = NULL;
r = sd_bus_call_method(bus, "org.freedesktop.systemd1", unit_path,
"org.freedesktop.DBus.Properties", "Get", &cg_error,
&reply, "ss", "org.freedesktop.systemd1.Service",
"ControlGroup");
if (r < 0) {
XCECD_LOG(XCE_LOG_DEBUG, "Failed to call Get ControlGroup: %s",
cg_error.message ? cg_error.message : "unknown");
sd_bus_error_free(&cg_error);
sd_bus_message_unref(reply);
reply = NULL;
} else {
const char* cgroup_str = NULL;
r = sd_bus_message_enter_container(reply, 'v', "s");
if (r >= 0) {
r = sd_bus_message_read(reply, "s", &cgroup_str);
if (r >= 0 && cgroup_str) {
control_group = strdup(cgroup_str);
XCECD_LOG(XCE_LOG_DEBUG, "Got ControlGroup: %s", control_group);
sd_bus_message_exit_container(reply);
} else {
XCECD_LOG(XCE_LOG_DEBUG,
"Failed to read ControlGroup from reply: %s",
strerror(-r));
}
} else {
XCECD_LOG(XCE_LOG_DEBUG, "Failed to enter variant container: %s",
strerror(-r));
}
sd_bus_message_unref(reply);
reply = NULL;
}
if (control_group) {
r = read_pids_from_cgroup(control_group, &rusages->pids,
&rusages->npids);
if (r == 0 && rusages->npids > 0) {
// filter out xcecd_worker processes
int* filtered_pids = (int*)calloc(rusages->npids, sizeof(int));
int filtered_count = 0;
if (filtered_pids) {
for (int i = 0; i < rusages->npids; i++) {
pid_t pid = rusages->pids[i];
// check if xcecd_worker exists in cmdline
char proc_path[256];
snprintf(proc_path, sizeof(proc_path), "/proc/%d/cmdline",
pid);
FILE* fp = fopen(proc_path, "r");
if (fp) {
char cmdline[1024] = { 0 };
size_t len = fread(cmdline, 1, sizeof(cmdline) - 1, fp);
fclose(fp);
// replace \0 with space for searching
for (size_t j = 0; j < len; j++) {
if (cmdline[j] == '\0')
cmdline[j] = ' ';
}
// if xcecd_worker is not found, keep this PID
if (!strstr(cmdline, "xcecd_worker")) {
filtered_pids[filtered_count++] = pid;
XCECD_LOG(XCE_LOG_DEBUG3, "Including PID %d: %s",
pid, cmdline);
} else {
XCECD_LOG(XCE_LOG_DEBUG,
"Excluding xcecd_worker PID %d", pid);
}
} else {
// If cmdline cannot be read, include this PID for
// safety
filtered_pids[filtered_count++] = pid;
XCECD_LOG(
XCE_LOG_DEBUG3,
"Cannot read cmdline for PID %d, including it",
pid);
}
}
// Replace original PIDs array
FREEUP(rusages->pids);
rusages->pids = filtered_pids;
int original_count = rusages->npids;
rusages->npids = filtered_count;
XCECD_LOG(XCE_LOG_DEBUG,
"Filtered PIDs: %d -> %d (excluded %d xcecd_worker "
"processes)",
original_count, filtered_count,
original_count - filtered_count);
}
// collect PGID (based on filtered PID)
if (rusages->npids > 0) {
int* all_pgids = (int*)calloc(rusages->npids, sizeof(int));
if (all_pgids) {
int valid = 0;
XCECD_LOG(XCE_LOG_DEBUG3,
"Collecting PGIDs for %d processes",
rusages->npids);
for (int i = 0; i < rusages->npids; i++) {
pid_t pid = rusages->pids[i];
// Read PGID from /proc/<pid>/stat
char stat_path[256];
snprintf(stat_path, sizeof(stat_path), "/proc/%d/stat",
pid);
FILE* fp = fopen(stat_path, "r");
if (fp) {
int stat_pid, stat_ppid, stat_pgrp;
char comm[256];
char state;
if (fscanf(fp, "%d %s %c %d %d", &stat_pid, comm,
&state, &stat_ppid, &stat_pgrp)
>= 5) {
all_pgids[valid++] = stat_pgrp;
XCECD_LOG(XCE_LOG_DEBUG3, "PID %d -> PGID %d",
pid, stat_pgrp);
}
fclose(fp);
} else {
pid_t pgid = getpgid(pid);
if (pgid != -1) {
all_pgids[valid++] = (int)pgid;
XCECD_LOG(XCE_LOG_DEBUG3,
"PID %d -> PGID %d (getpgid)", pid,
(int)pgid);
} else {
XCECD_LOG(XCE_LOG_DEBUG,
"Failed to get PGID for PID %d: %s",
pid, strerror(errno));
}
}
}
if (valid > 0) {
qsort(all_pgids, valid, sizeof(int), compare_ints);
int unique = 1;
for (int j = 1; j < valid; j++) {
if (all_pgids[j] != all_pgids[j - 1]) {
unique++;
}
}
rusages->pgids = (int*)calloc(unique, sizeof(int));
if (rusages->pgids) {
rusages->npgids = unique;
rusages->pgids[0] = all_pgids[0];
int k = 1;
for (int j = 1; j < valid; j++) {
if (all_pgids[j] != all_pgids[j - 1]) {
rusages->pgids[k++] = all_pgids[j];
}
}
XCECD_LOG(XCE_LOG_DEBUG,
"Found %d unique PGIDs from %d PIDs",
unique, valid);
} else {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to allocate memory for PGIDs");
}
} else {
XCECD_LOG(XCE_LOG_DEBUG,
"No valid PGIDs found after filtering");
}
FREEUP(all_pgids);
} else {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to allocate memory for PGID collection");
}
}
} else if (r < 0) {
XCECD_LOG(XCE_LOG_DEBUG, "Read cgroup.procs failed (r=%d) for %s",
r, control_group);
}
FREEUP(control_group);
} else {
XCECD_LOG(XCE_LOG_DEBUG,
"ControlGroup is null, skipping PID collection");
}
sd_bus_error_free(&cg_error);
/* consider to disable this in product for performance reasons */
#if 1
char pids_str[1024] = "";
char pgids_str[1024] = "";
if (rusages->npids > 0 && rusages->pids) {
char temp[32];
for (int i = 0; i < rusages->npids; i++) {
if (i > 0)
strcat(pids_str, ",");
snprintf(temp, sizeof(temp), "%d", rusages->pids[i]);
strcat(pids_str, temp);
if (strlen(pids_str) > 900) {
strcat(pids_str, "...");
break;
}
}
} else {
strcpy(pids_str, "none");
}
if (rusages->npgids > 0 && rusages->pgids) {
char temp[32];
for (int i = 0; i < rusages->npgids; i++) {
if (i > 0)
strcat(pgids_str, ",");
snprintf(temp, sizeof(temp), "%d", rusages->pgids[i]);
strcat(pgids_str, temp);
if (strlen(pgids_str) > 900) {
strcat(pgids_str, "...");
break;
}
}
} else {
strcpy(pgids_str, "none");
}
XCECD_LOG(XCE_LOG_DEBUG3,
"Job %lu rusage: CPU: %.2fs, Mem: %.2fMB, Swap: %.2fMB, Tasks: "
"%d, PIDs: %d [%s], PGIDs: %d [%s]",
job_id, rusages->usages[0]->value, rusages->usages[1]->value,
rusages->usages[2]->value, (int)rusages->usages[3]->value,
rusages->npids, pids_str, rusages->npgids, pgids_str);
#endif
sd_bus_unref(bus);
return rusages;
cleanup:
if (rusages) {
if (rusages->usages) {
for (int i = 0; i < num_res; i++) {
if (rusages->usages[i]) {
FREEUP(rusages->usages[i]->resname);
FREEUP(rusages->usages[i]);
}
}
FREEUP(rusages->usages);
}
FREEUP(rusages->pids);
FREEUP(rusages);
}
sd_bus_unref(bus);
return NULL;
}

View file

@ -31,6 +31,7 @@ bool xcecd_db_init(xcecd_db_t** db, const char* db_path)
if (sqlite3_open(db_path, db) != SQLITE_OK) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to open database: %s",
sqlite3_errmsg(*db));
if ((*db) != NULL) {
sqlite3_close(*db);
}

View file

@ -3,7 +3,6 @@
* @brief This file contains the implementation of the D-Bus manager functions.
*/
#include "xcecd_dbus_manager.h"
#include "utils.h"
#include "xcecd_utils.h"
#include "xcelog.h"
@ -65,6 +64,7 @@ bool dbus_start_worker(const res_limits_t* limits, uint64_t job_id,
char myname[256];
make_job_sd_unit_name(myname, sizeof(myname) - 1, "xcecd-worker", job_id);
/* checking if service name is exist before start service and it's a
* negative case */
dbus_stop_worker(myname);
@ -387,6 +387,7 @@ void dbus_stop_all_workers(xcecd_worker_t** worker_list, sqlite3* db)
make_job_sd_unit_name(unit_name, sizeof(unit_name) - 1, "xcecd-worker",
worker->job_info.jobid);
job.job_id = worker->job_info.jobid;
job.job_pid = worker->worker_pid;
job.job_status = JOB_EXITED;

View file

@ -9,9 +9,6 @@
#include "xcelog.h"
#include <errno.h>
#include <stdint.h>
#include <sys/resource.h>
#include "utils.h"
/**
* @brief Parses the JSON string, populates an environment variable linked list,
@ -282,15 +279,9 @@ bool parse_limits(const char* exdata, res_limits_t* limits)
bool success = false;
cJSON* limits_obj = cJSON_GetObjectItemCaseSensitive(root, "limits");
if (limits_obj == NULL) {
if (!limits_obj || !cJSON_IsObject(limits_obj)) {
// limits not exist
success = true;
memset(limits, 0x0, sizeof(*limits));
goto cleanup;
}
if (!cJSON_IsObject(limits_obj)) {
// invalidate limits
goto cleanup;
}
@ -414,7 +405,7 @@ void apply_env_vars(env_var_t** env_vars, uint64_t jobid)
// Set some default XCE environment variables
char jobid_str[32];
snprintf(jobid_str, sizeof(jobid_str), "%lu", array_jobid(jobid));
snprintf(jobid_str, sizeof(jobid_str), "%lu", jobid);
if (setenv("XCE_JOBID", jobid_str, 1) != 0) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Error setting environment variable: %s",
@ -426,7 +417,7 @@ void apply_env_vars(env_var_t** env_vars, uint64_t jobid)
}
char jobindex_str[8];
snprintf(jobindex_str, sizeof(jobindex_str), "%lu", array_idx(jobid));
snprintf(jobindex_str, sizeof(jobindex_str), "%d", 0);
if (setenv("XCE_JOBINDEX", jobindex_str, 1) != 0) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Error setting environment variable: %s",
@ -501,4 +492,4 @@ char* generate_exdata(const start_exdata_t* start_info)
}
return exdata_str;
}
}

View file

@ -3,6 +3,7 @@
#include "static_array.h"
#include "uv_ipc.h"
#include "uv_proc.h"
#include "xcecd_cgroups.h"
#include "xcecd_exdata.h"
#include "xcecd_utils.h"
#include "xcelog.h"
@ -45,6 +46,8 @@ static int return_code = 0;
static job_runtime_info_t job_runtime_info = { 0 };
char* log_file = NULL;
start_exdata_t* start_exdata = NULL;
static uv_timer_t rusage_timer;
static int job_rusage_collect_interval = 10000; // 10 seconds
// process manager will share the same event loop with ipc
uv_loop_t* uv_loop = NULL;
@ -56,6 +59,57 @@ static void xjob_stop_vnc();
static bool check_process_status(pid_t target_pid, int expected_signal,
int timeout_seconds);
static void free_xce_job_rusages(xce_job_rusages_t* rusages)
{
if (!rusages)
return;
FREEUP(rusages->pids);
FREEUP(rusages->pgids);
if (rusages->usages) {
for (int i = 0; i < rusages->count; i++) {
if (rusages->usages[i]) {
FREEUP(rusages->usages[i]->resname);
FREEUP(rusages->usages[i]);
}
}
FREEUP(rusages->usages);
}
FREEUP(rusages);
}
static void on_rusage_timer(uv_timer_t* timer)
{
if (jobid > 0) {
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Collecting rusage for job %lu", jobid);
xce_job_rusages_t* rusages = collect_job_rusages(jobid);
if (rusages) {
ipc_worker_send_rusages(rusages);
free_xce_job_rusages(rusages);
} else {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to collect job rusages.");
}
}
}
static void start_rusage_collection()
{
uv_loop_t* loop = ipc_get_event_loop();
if (loop) {
uv_timer_init(loop, &rusage_timer);
uv_timer_start(&rusage_timer, on_rusage_timer,
job_rusage_collect_interval,
job_rusage_collect_interval);
}
}
static void stop_rusage_collection()
{
uv_timer_stop(&rusage_timer);
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Job rusage collection stoped.");
}
static void signal_children_recursive(int signo, pid_t target_pid)
{
DIR* dir;
@ -412,6 +466,8 @@ static void process_exit_callback(int pid, int exit_status, int term_signal,
{
check_and_reconnect();
stop_rusage_collection();
if (term_signal == 0) {
XCECD_WORKER_LOG(XCE_LOG_INFO, "Child %d exited with status %d\n", pid,
@ -540,14 +596,14 @@ static void xjob_start_vnc()
if (access(XJOB_PRE_EXEC, F_OK) != 0) {
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "VNC pre-exec command not found: %s",
XJOB_PRE_EXEC);
char* server_dir = getenv("XCE_SERVICE_DIR");
char* server_dir = getenv("XCE_SERVER_DIR");
if (server_dir) {
snprintf(vnc_start_cmd, sizeof(vnc_start_cmd), "%s/startvnc %d",
server_dir, port_info.xjob_port);
} else {
XCECD_WORKER_LOG(
XCE_LOG_ERROR,
"XCE_SERVICE_DIR not set, cannot find VNC pre-exec command");
"XCE_SERVER_DIR not set, cannot find VNC pre-exec command");
exit(1);
}
} else {
@ -608,14 +664,14 @@ static void xjob_stop_vnc()
if (access(XJOB_POST_EXEC, F_OK) != 0) {
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "VNC post-exec command not found: %s",
XJOB_POST_EXEC);
char* server_dir = getenv("XCE_SERVICE_DIR");
char* server_dir = getenv("XCE_SERVER_DIR");
if (server_dir) {
snprintf(vnc_stop_cmd, sizeof(vnc_stop_cmd), "%s/stopvnc %d",
server_dir, port_info.xjob_port);
} else {
XCECD_WORKER_LOG(
XCE_LOG_ERROR,
"XCE_SERVICE_DIR not set, cannot find VNC post-exec command");
"XCE_SERVER_DIR not set, cannot find VNC post-exec command");
exit(1);
}
} else {
@ -778,6 +834,9 @@ static void worker_job_callback(job_assignment_t* info)
"Process started successfully to run task: %s",
info->cmd);
start_rusage_collection();
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Starting job rusage collection...");
if (mydata.transfer_info->interactive_mode > 0) {
url = generate_url(port_info,
mydata.transfer_info->interactive_mode);
@ -1168,16 +1227,18 @@ static void do_criu_ckpt(uint64_t job_id, int kill)
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Checkpoint directory created: %s",
ckpt_dir);
char criu_cmd[MAXLINELEN*2];
char info[MAXLINELEN*2];
char criu_cmd[MAXLINELEN * 2];
char info[MAXLINELEN * 2];
if (kill) {
snprintf(criu_cmd, sizeof(criu_cmd),
"cd %s; %s dump -t %d -vvv -o dump.log", ckpt_dir, path,
target_pid);
"cd %s; %s dump -t %d -vvv --tcp-established --manage-cgroups "
"-o dump.log",
ckpt_dir, path, target_pid);
} else {
snprintf(criu_cmd, sizeof(criu_cmd),
"cd %s; %s dump -t %d -R -vvv -o dump.log", ckpt_dir, path,
target_pid);
"cd %s; %s dump -t %d -R -vvv --tcp-established "
"--manage-cgroups -o dump.log",
ckpt_dir, path, target_pid);
}
int ret = system(criu_cmd);
@ -1228,7 +1289,7 @@ static void do_criu_ckpt(uint64_t job_id, int kill)
snprintf(info, sizeof(info),
"{\"ckpt_pid\": %d, \"restart_command\": \"cd %s; %s restore -vvv "
"-o restore.log\"}",
"--tcp-established --manage-cgroups -o restore.log\"}",
target_pid, ckpt_dir, path);
ipc_worker_send_ckpt_result(job_id, 0, info);

View file

@ -1,5 +1,4 @@
#include "xcecd_worker_utils.h"
#include "utils.h"
#include "uv_ipc.h"
#include "xcecd_dbus_manager.h"
#include "xcecd_utils.h"
@ -525,5 +524,6 @@ int make_job_sd_unit_name(char* buf, size_t bufsize, const char* prefix,
}
fmt_jobid_ex(idstring, sizeof(idstring) - 1, jobid, '_', '_');
return snprintf(buf, bufsize, "%s-%s.service", prefix, idstring);
}

View file

@ -1,795 +0,0 @@
/**
* @file test_xcecd_exdata.c
* @brief Test program for resource limits in xcecd worker
*
* This program tests resource limits by gradually increasing resource usage
* and reporting progress. It can be used to test limits parsing functionality
* in xcecd_exdata.c.
*/
#include <errno.h>
#include <getopt.h>
#include <math.h>
#include <pthread.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/resource.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>
// Default values
#define DEFAULT_STEP_SIZE_KB 1024 // Default memory step size (1MB)
#define DEFAULT_STEP_INTERVAL_MS 100 // 100ms between steps
#define DEFAULT_FILE_PATH "/tmp/xcecd_test_file"
#define MAX_THREADS 10000
// Global variables for cleanup
volatile sig_atomic_t terminate = 0;
void** memory_blocks = NULL;
int num_memory_blocks = 0;
pthread_t* thread_list = NULL;
int num_threads = 0;
FILE* test_file = NULL;
char* file_path = NULL;
pid_t* process_list = NULL;
int num_processes = 0;
// Thread function prototype
void* thread_function(void* arg);
void process_function(void);
// Signal handler
void signal_handler(int sig)
{
terminate = 1;
printf("\nReceived signal %d, cleaning up and exiting...\n", sig);
}
// Function to get current timestamp in milliseconds
long long current_time_ms()
{
struct timeval tv;
gettimeofday(&tv, NULL);
return (long long)(tv.tv_sec) * 1000 + (tv.tv_usec / 1000);
}
// Helper function to print a specific resource limit
static void print_limit(const char* name, int resource)
{
struct rlimit rlim;
if (getrlimit(resource, &rlim) == 0) {
printf(" - %-25s: ", name);
if (rlim.rlim_cur == RLIM_INFINITY) {
printf("soft (unlimited), ");
} else {
printf("soft (%-10lld), ", (long long)rlim.rlim_cur);
}
if (rlim.rlim_max == RLIM_INFINITY) {
printf("hard (unlimited)\n");
} else {
printf("hard (%lld)\n", (long long)rlim.rlim_max);
}
} else {
printf(" - %-25s: Failed to get limit (%s)\n", name, strerror(errno));
}
}
// Helper function to print all relevant limits
static void print_all_limits()
{
printf("System Resource Limits:\n");
print_limit("CPU time (seconds)", RLIMIT_CPU);
print_limit("Max file size (bytes)", RLIMIT_FSIZE);
print_limit("Max data segment (bytes)", RLIMIT_DATA);
print_limit("Max stack size (bytes)", RLIMIT_STACK);
print_limit("Max core file size (bytes)", RLIMIT_CORE);
print_limit("Max resident set (bytes)", RLIMIT_RSS);
print_limit("Max processes", RLIMIT_NPROC);
print_limit("Max open files", RLIMIT_NOFILE);
print_limit("Max locked memory (bytes)", RLIMIT_MEMLOCK);
print_limit("Max address space (bytes)", RLIMIT_AS);
printf("\n");
}
// Function to print current resource usage
void print_resource_usage()
{
// Print limits only on the first call
static int first_call = 1;
if (first_call) {
print_all_limits();
first_call = 0;
}
struct rusage usage;
if (getrusage(RUSAGE_SELF, &usage) == 0) {
printf("Resource usage - User CPU: %.2fs, System CPU: %.2fs, RSS: %ld "
"KB\n",
usage.ru_utime.tv_sec + usage.ru_utime.tv_usec / 1000000.0,
usage.ru_stime.tv_sec + usage.ru_stime.tv_usec / 1000000.0,
usage.ru_maxrss);
}
}
// Memory test function
void test_memory(size_t target_mb, size_t step_kb, int interval_ms)
{
printf("Starting memory test: Target=%zu MB, Step=%zu KB, Interval=%d ms\n",
target_mb, step_kb, interval_ms);
size_t target_kb = target_mb * 1024;
size_t allocated_kb = 0;
long long start_time = current_time_ms();
long long next_step_time = start_time;
// Pre-allocate array for memory block pointers
num_memory_blocks = (target_kb / step_kb) + 1;
memory_blocks = calloc(num_memory_blocks, sizeof(void*));
if (!memory_blocks) {
printf("Failed to allocate memory for tracking blocks\n");
exit(EXIT_FAILURE);
}
int block_index = 0;
while (allocated_kb < target_kb && !terminate) {
long long current_time = current_time_ms();
// Check if it's time for the next allocation
if (current_time >= next_step_time) {
// Allocate memory block
void* block = malloc(step_kb * 1024);
if (block) {
// Write to the memory to ensure it's actually allocated
memset(block, 1, step_kb * 1024);
// Store pointer to free later
if (block_index < num_memory_blocks) {
memory_blocks[block_index++] = block;
}
allocated_kb += step_kb;
// Print progress
printf("[%6.2f s] Memory allocated: %zu KB (%zu MB)\n",
(current_time - start_time) / 1000.0, allocated_kb,
allocated_kb / 1024);
// Update next step time
next_step_time = current_time + interval_ms;
// Print resource usage every 10MB
if (allocated_kb % (10 * 1024) == 0) {
print_resource_usage();
}
} else {
printf("Memory allocation failed at %zu KB\n", allocated_kb);
exit(EXIT_FAILURE);
}
}
// Small sleep to prevent CPU hogging
usleep(10000); // 10ms
}
printf("Memory test completed: Allocated %zu KB (%zu MB)\n", allocated_kb,
allocated_kb / 1024);
print_resource_usage();
}
// CPU test function
void test_cpu(int duration_seconds)
{
printf("Starting CPU test: Duration=%d seconds\n", duration_seconds);
long long start_time = current_time_ms();
long long end_time = start_time + (duration_seconds * 1000);
long long next_report_time = start_time + 1000; // Report every second
// Variables for busy work
double result = 0.0;
unsigned long iterations = 0;
while (current_time_ms() < end_time && !terminate) {
long long current_time = current_time_ms();
// Do some CPU-intensive calculation
for (int i = 0; i < 1000000 && !terminate; i++) {
result += sin(result + i) * cos(result + i);
iterations++;
}
// Check if it's time to report progress
if (current_time >= next_report_time) {
double elapsed_seconds = (current_time - start_time) / 1000.0;
double remaining_seconds = duration_seconds - elapsed_seconds;
printf("[%6.2f s] CPU test progress: %.1f%% complete (%.1f seconds "
"remaining)\n",
elapsed_seconds,
(elapsed_seconds / duration_seconds) * 100.0,
remaining_seconds);
print_resource_usage();
next_report_time = current_time + 1000; // Next report in 1 second
}
}
printf("CPU test completed: Performed %lu iterations\n", iterations);
print_resource_usage();
}
// Thread creation function
void* thread_function(void* arg)
{
int thread_id = *(int*)arg;
free(arg); // Free the memory allocated for the argument
printf("Thread %d started\n", thread_id);
// Keep the thread alive until termination signal
while (!terminate) {
sleep(1);
}
printf("Thread %d exiting\n", thread_id);
return NULL;
}
// Thread test function
void test_threads(int target_threads, int interval_ms)
{
printf("Starting thread test: Target=%d threads, Interval=%d ms\n",
target_threads, interval_ms);
int created_threads = 0;
long long start_time = current_time_ms();
long long next_step_time = start_time;
// Allocate array for thread IDs
thread_list = calloc(target_threads, sizeof(pthread_t));
if (!thread_list) {
printf("Failed to allocate memory for thread tracking\n");
exit(EXIT_FAILURE);
}
while (created_threads < target_threads && !terminate) {
long long current_time = current_time_ms();
// Check if it's time to create a new thread
if (current_time >= next_step_time) {
// Allocate memory for thread ID argument
int* thread_id = malloc(sizeof(int));
if (!thread_id) {
printf("Failed to allocate memory for thread ID\n");
break;
}
*thread_id = created_threads + 1;
// Create a new thread
if (pthread_create(&thread_list[created_threads], NULL,
thread_function, thread_id)
== 0) {
created_threads++;
// Print progress
printf("[%6.2f s] Threads created: %d\n",
(current_time - start_time) / 1000.0, created_threads);
// Update next step time
next_step_time = current_time + interval_ms;
// Print resource usage every 10 threads
if (created_threads % 10 == 0) {
print_resource_usage();
}
} else {
printf("Thread creation failed at %d threads: %s\n",
created_threads, strerror(errno));
free(thread_id); // Free the ID memory if thread creation failed
exit(EXIT_FAILURE);
}
}
// Small sleep to prevent CPU hogging
usleep(10000); // 10ms
}
printf("Thread test completed: Created %d threads\n", created_threads);
print_resource_usage();
num_threads = created_threads;
}
void process_function()
{
printf("Child process %d started\n", getpid());
volatile sig_atomic_t local_terminate = 0;
void child_signal_handler(int sig)
{
local_terminate = 1;
printf("Child process %d received signal %d, exiting...\n", getpid(),
sig);
}
struct sigaction sa;
sa.sa_handler = child_signal_handler;
sigemptyset(&sa.sa_mask);
sa.sa_flags = 0;
sigaction(SIGTERM, &sa, NULL);
sigaction(SIGINT, &sa, NULL);
while (!local_terminate) {
sleep(1);
}
printf("Child process %d exiting\n", getpid());
exit(EXIT_SUCCESS);
}
void test_processes(int target_processes, int interval_ms)
{
printf("Starting process test: Target=%d processes, Interval=%d ms\n",
target_processes, interval_ms);
int created_processes = 0;
long long start_time = current_time_ms();
long long next_step_time = start_time;
process_list = calloc(target_processes, sizeof(pid_t));
if (!process_list) {
printf("Failed to allocate memory for process tracking\n");
exit(EXIT_FAILURE);
}
while (created_processes < target_processes && !terminate) {
long long current_time = current_time_ms();
if (current_time >= next_step_time) {
pid_t pid = fork();
if (pid == -1) {
printf("Process creation failed at %d processes: %s\n",
created_processes, strerror(errno));
exit(EXIT_FAILURE);
} else if (pid == 0) {
process_function();
exit(EXIT_SUCCESS);
} else {
process_list[created_processes] = pid;
created_processes++;
printf("[%6.2f s] Processes created: %d\n",
(current_time - start_time) / 1000.0, created_processes);
next_step_time = current_time + interval_ms;
if (created_processes % 10 == 0) {
print_resource_usage();
}
}
}
usleep(10000); // 10ms
}
printf("Process test completed: Created %d processes\n", created_processes);
print_resource_usage();
num_processes = created_processes;
}
// File size test function
void test_file_size(size_t target_mb, size_t step_kb, int interval_ms)
{
printf(
"Starting file size test: Target=%zu MB, Step=%zu KB, Interval=%d ms\n",
target_mb, step_kb, interval_ms);
size_t target_kb = target_mb * 1024;
size_t written_kb = 0;
long long start_time = current_time_ms();
long long next_step_time = start_time;
// Create buffer for writing
char* buffer = malloc(step_kb * 1024);
if (!buffer) {
printf("Failed to allocate buffer for file writing\n");
exit(EXIT_FAILURE);
}
// Fill buffer with random data
memset(buffer, 'A', step_kb * 1024);
// Open file for writing
test_file = fopen(file_path, "wb");
if (!test_file) {
printf("Failed to open file %s: %s\n", file_path, strerror(errno));
free(buffer);
exit(EXIT_FAILURE);
}
while (written_kb < target_kb && !terminate) {
long long current_time = current_time_ms();
// Check if it's time for the next write
if (current_time >= next_step_time) {
// Write to file
size_t bytes_written = fwrite(buffer, 1, step_kb * 1024, test_file);
if (bytes_written > 0) {
written_kb += bytes_written / 1024;
// Flush to ensure data is actually written
fflush(test_file);
// Print progress
printf("[%6.2f s] File size: %zu KB (%zu MB)\n",
(current_time - start_time) / 1000.0, written_kb,
written_kb / 1024);
// Update next step time
next_step_time = current_time + interval_ms;
// Print resource usage every 10MB
if (written_kb % (10 * 1024) == 0) {
print_resource_usage();
}
} else {
printf("File write failed at %zu KB: %s\n", written_kb,
strerror(errno));
exit(EXIT_FAILURE);
}
}
// Small sleep to prevent CPU hogging
usleep(10000); // 10ms
}
free(buffer);
printf("File size test completed: Wrote %zu KB (%zu MB)\n", written_kb,
written_kb / 1024);
print_resource_usage();
}
// Runtime test function (simple sleep)
void test_runtime(int duration_seconds)
{
printf("Starting runtime test: Duration=%d seconds\n", duration_seconds);
long long start_time = current_time_ms();
long long end_time = start_time + (duration_seconds * 1000);
long long next_report_time = start_time + 1000; // Report every second
while (current_time_ms() < end_time && !terminate) {
long long current_time = current_time_ms();
// Check if it's time to report progress
if (current_time >= next_report_time) {
double elapsed_seconds = (current_time - start_time) / 1000.0;
double remaining_seconds = duration_seconds - elapsed_seconds;
printf("[%6.2f s] Runtime test progress: %.1f%% complete (%.1f "
"seconds remaining)\n",
elapsed_seconds,
(elapsed_seconds / duration_seconds) * 100.0,
remaining_seconds);
print_resource_usage();
next_report_time = current_time + 1000; // Next report in 1 second
}
// Sleep for a short time
usleep(100000); // 100ms
}
printf("Runtime test completed: Slept for %.2f seconds\n",
(current_time_ms() - start_time) / 1000.0);
print_resource_usage();
}
// Cleanup function
void cleanup()
{
printf("Cleaning up resources...\n");
// Free memory blocks
if (memory_blocks) {
for (int i = 0; i < num_memory_blocks && memory_blocks[i]; i++) {
free(memory_blocks[i]);
}
free(memory_blocks);
memory_blocks = NULL;
}
// Wait for threads to terminate
if (thread_list) {
for (int i = 0; i < num_threads; i++) {
pthread_join(thread_list[i], NULL);
}
free(thread_list);
thread_list = NULL;
}
// Terminate and wait for child processes
if (process_list) {
printf("Terminating %d child processes...\n", num_processes);
for (int i = 0; i < num_processes; i++) {
if (process_list[i] > 0) {
kill(process_list[i], SIGTERM);
}
}
printf("Waiting for child processes to exit...\n");
int timeout_seconds = 5;
time_t start_wait = time(NULL);
int remaining = num_processes;
while (remaining > 0 && time(NULL) - start_wait < timeout_seconds) {
for (int i = 0; i < num_processes; i++) {
if (process_list[i] > 0) {
int status;
pid_t result = waitpid(process_list[i], &status, WNOHANG);
if (result > 0) {
if (WIFEXITED(status)) {
printf("Process %d exited with status %d\n",
(int)process_list[i], WEXITSTATUS(status));
} else if (WIFSIGNALED(status)) {
printf("Process %d killed by signal %d\n",
(int)process_list[i], WTERMSIG(status));
}
process_list[i] = -1;
remaining--;
}
}
}
usleep(100000);
}
if (remaining > 0) {
printf("Force killing %d remaining processes...\n", remaining);
for (int i = 0; i < num_processes; i++) {
if (process_list[i] > 0) {
kill(process_list[i], SIGKILL);
waitpid(process_list[i], NULL, 0);
printf("Process %d force killed\n", (int)process_list[i]);
}
}
}
free(process_list);
process_list = NULL;
}
// Close and remove test file
if (test_file) {
fclose(test_file);
test_file = NULL;
if (file_path) {
unlink(file_path);
}
}
// Free file path
if (file_path && strcasecmp(file_path, DEFAULT_FILE_PATH) != 0) {
free(file_path);
file_path = NULL;
}
printf("Cleanup completed\n");
}
// Print usage
void print_usage(const char* program_name)
{
printf("Usage: %s <sleep_time> [options]\n", program_name);
printf(" <sleep_time> Required: Duration to run in seconds\n");
printf("Options:\n");
printf(" -m, --memory MB Target memory usage in MB\n");
printf(" -c, --cpu Enable CPU stress test\n");
printf(" -t, --threads N Number of threads to create\n");
printf(" -p, --processes N Number of processes to create\n");
printf(" -f, --filesize MB Target file size in MB\n");
printf(" -P, --filepath PATH Path for file size test (default: %s)\n",
DEFAULT_FILE_PATH);
printf(" -s, --step KB Step size in KB (default: %d)\n",
DEFAULT_STEP_SIZE_KB);
printf(" -i, --interval MS Interval between steps in ms (default: %d)\n",
DEFAULT_STEP_INTERVAL_MS);
printf(" -h, --help Show this help message\n");
printf("\nExample:\n");
printf(" %s 60 -m 500 -t 50 -p 10 -s 1024 -i 100\n", program_name);
printf(" (Run for 60 seconds, allocate up to 500MB memory, create 50 "
"threads,\n");
printf(" create 10 processes, with step size of 1MB every 100ms)\n");
}
int main(int argc, char* argv[])
{
// Default parameters
int sleep_time = 0;
size_t memory_mb = 0;
int cpu_test = 0;
int thread_count = 0;
int process_count = 0;
size_t file_size_mb = 0;
size_t step_size_kb = DEFAULT_STEP_SIZE_KB;
int step_interval_ms = DEFAULT_STEP_INTERVAL_MS;
file_path = (char*)DEFAULT_FILE_PATH;
// Define long options
static struct option long_options[] = {
{ "memory", required_argument, 0, 'm' },
{ "cpu", no_argument, 0, 'c' },
{ "threads", required_argument, 0, 't' },
{ "processes", required_argument, 0, 'p' },
{ "filesize", required_argument, 0, 'f' },
{ "filepath", required_argument, 0, 'P' },
{ "step", required_argument, 0, 's' },
{ "interval", required_argument, 0, 'i' },
{ "help", no_argument, 0, 'h' },
{ 0, 0, 0, 0 }
};
// Check if sleep_time is provided
if (argc < 2) {
print_usage(argv[0]);
return EXIT_FAILURE;
}
// Parse the sleep_time argument
sleep_time = atoi(argv[1]);
if (sleep_time <= 0) {
printf("Error: Sleep time must be a positive integer\n");
print_usage(argv[0]);
return EXIT_FAILURE;
}
// Shift the arguments to make getopt work
for (int i = 1; i < argc - 1; i++) {
argv[i] = argv[i + 1];
}
argc--;
// Parse options
int option_index = 0;
int opt;
while ((opt = getopt_long(argc, argv, "m:ct:p:f:P:s:i:h", long_options,
&option_index))
!= -1) {
switch (opt) {
case 'm':
memory_mb = atoi(optarg);
break;
case 'c':
cpu_test = 1;
break;
case 't':
thread_count = atoi(optarg);
break;
case 'p':
process_count = atoi(optarg);
break;
case 'f':
file_size_mb = atoi(optarg);
break;
case 'P':
file_path = strdup(optarg);
break;
case 's':
step_size_kb = atoi(optarg);
break;
case 'i':
step_interval_ms = atoi(optarg);
break;
case 'h':
print_usage(argv[0]);
return EXIT_SUCCESS;
case '?':
// getopt_long already printed an error message
print_usage(argv[0]);
return EXIT_FAILURE;
default:
break;
}
}
// Validate parameters
if (step_size_kb <= 0) {
printf("Warning: Invalid step size, using default %d KB\n",
DEFAULT_STEP_SIZE_KB);
step_size_kb = DEFAULT_STEP_SIZE_KB;
}
if (step_interval_ms <= 0) {
printf("Warning: Invalid interval, using default %d ms\n",
DEFAULT_STEP_INTERVAL_MS);
step_interval_ms = DEFAULT_STEP_INTERVAL_MS;
}
// Set up signal handlers
signal(SIGINT, signal_handler);
signal(SIGTERM, signal_handler);
// Set up cleanup handler
atexit(cleanup);
// Print test configuration
printf("Resource Limit Test Configuration:\n");
printf("- Sleep time: %d seconds\n", sleep_time);
printf("- Memory test: %s", memory_mb > 0 ? "Enabled" : "Disabled");
if (memory_mb > 0)
printf(" (Target: %zu MB)", memory_mb);
printf("\n");
printf("- CPU test: %s\n", cpu_test ? "Enabled" : "Disabled");
printf("- Thread test: %s", thread_count > 0 ? "Enabled" : "Disabled");
if (thread_count > 0)
printf(" (Target: %d threads)", thread_count);
printf("\n");
printf("- Process test: %s", process_count > 0 ? "Enabled" : "Disabled");
if (process_count > 0)
printf(" (Target: %d processes)", process_count);
printf("\n");
printf("- File size test: %s", file_size_mb > 0 ? "Enabled" : "Disabled");
if (file_size_mb > 0)
printf(" (Target: %zu MB, Path: %s)", file_size_mb, file_path);
printf("\n");
printf("- Step size: %zu KB\n", step_size_kb);
printf("- Step interval: %d ms\n", step_interval_ms);
printf("\n");
// Record start time
long long start_time = current_time_ms();
// Run tests based on configuration
if (memory_mb > 0) {
test_memory(memory_mb, step_size_kb, step_interval_ms);
}
if (thread_count > 0) {
test_threads(thread_count, step_interval_ms);
}
if (process_count > 0) {
test_processes(process_count, step_interval_ms);
}
if (file_size_mb > 0) {
test_file_size(file_size_mb, step_size_kb, step_interval_ms);
}
if (cpu_test) {
test_cpu(sleep_time);
} else {
// If no CPU test, just sleep for the specified duration
test_runtime(sleep_time);
}
terminate = 1;
if (thread_count > 0) {
printf("Waiting for threads to exit...\n");
usleep(500000);
}
// Print overall runtime
printf("\nTest completed in %.2f seconds\n",
(current_time_ms() - start_time) / 1000.0);
return EXIT_SUCCESS;
}

View file

@ -1,390 +0,0 @@
#include "uv_ipc.h"
#include <errno.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>
#define SOCKET_PATH "/tmp/ipc_test.sock"
#define CHILD_PROCESS_COUNT 5
// 进程类型
typedef enum {
PROCESS_MASTER,
PROCESS_WORKER
} process_type_t;
static process_type_t process_type;
// 全局变量,需要从uv_ipc.c中访问
int child_processes = 0;
int active_jobs = 0;
int exit_flag = 0;
void run_worker();
// 处理闹钟超时
void alarm_handler(int sig)
{
if (sig == SIGALRM) {
printf("\n[主进程] 警告:程序运行超时,强制退出\n");
// 强制杀死所有子进程
kill(0, SIGKILL);
// 打印当前状态
printf("[主进程] 强制退出时状态:child_processes=%d, active_jobs=%d, "
"exit_flag=%d\n",
child_processes, active_jobs, exit_flag);
// 直接退出,不进行任何清理
_exit(1);
}
}
// 主进程: 处理子进程任务状态回调
void master_job_status_callback(uint64_t job_id, int status_code,
const char* status_message)
{
if (!status_message) {
printf("[主进程] 警告: 收到空消息,job_id=%lu, status_code=%d\n",
job_id, status_code);
return;
}
printf("[主进程] 收到任务状态: Job ID %lu, 状态码 %d, 消息: %s\n", job_id,
status_code, status_message);
// 任务完成的情况有两种:
// 1. MSG_JOB_COMPLETED 消息(status_code=0, message="Completed")
// 2. 最后一个状态消息(status_code=1, message="Completed")
if ((status_code == 0 && strncmp(status_message, "Completed", 9) == 0)
|| (status_code == 1 && strncmp(status_message, "Completed", 9) == 0)) {
// 使用静态变量跟踪已完成的任务,避免重复计数
static uint64_t completed_jobs[1024] = { 0 };
static int completed_count = 0;
// 检查是否已记录过这个任务
bool already_completed = false;
for (int i = 0; i < completed_count; i++) {
if (completed_jobs[i] == job_id) {
already_completed = true;
break;
}
}
if (!already_completed && completed_count < 1024) {
completed_jobs[completed_count++] = job_id;
active_jobs--;
printf("[主进程] 任务 %lu 已完成, 剩余活跃任务: %d\n", job_id,
active_jobs);
}
}
}
// 主进程: 处理子进程断开连接回调
void master_disconnect_callback(uint64_t client_id)
{
printf("[主进程] 子进程 %lu 已断开连接\n", client_id);
child_processes--;
printf("[主进程] 剩余子进程: %d, 剩余活跃任务: %d\n", child_processes,
active_jobs);
// 如果所有子进程都已断开连接,可以考虑设置退出标志
if (child_processes <= 0 && active_jobs <= 0) {
printf("[主进程] 所有子进程已断开连接且没有活跃任务,设置退出标志\n");
exit_flag = 1;
}
}
// 子进程: 连接回调
void worker_connection_callback(bool success)
{
if (success) {
printf("[子进程 %d] 已连接到主进程\n", getpid());
// 连接成功后请求任务
ipc_worker_request_job();
} else {
printf("[子进程 %d] 连接到主进程失败\n", getpid());
_exit(1);
}
}
// 子进程: 任务回调
void worker_job_callback(uint64_t job_id, const char* cmd)
{
printf("[子进程 %d] 收到任务: Job ID %lu, 命令: %s\n", getpid(), job_id,
cmd);
// 模拟任务进行
printf("[子进程 %d] 开始执行任务 %lu\n", getpid(), job_id);
// 发送状态更新
ipc_worker_send_status(job_id, 1, "Running");
// 模拟随机执行时间(较短以便于测试)
// int execution_time = rand() % 3 + 1;
int execution_time = 3;
printf("[子进程 %d] 任务 %lu 将运行 %d 秒\n", getpid(), job_id,
execution_time);
// 模拟任务进度报告
for (int i = 1; i <= execution_time; i++) {
usleep(10 * 1000); // 10ms
char status[256];
int progress = (i * 100) / execution_time;
// 清除status缓冲区,避免垃圾数据
memset(status, 0, sizeof(status));
if (progress < 100) {
snprintf(status, sizeof(status), "Progress %d%%", progress);
ipc_worker_send_status(job_id, 1, status);
}
}
// 发送任务完成消息 - 使用 MSG_JOB_COMPLETED 类型
printf("[子进程 %d] 任务 %lu 已完成\n", getpid(), job_id);
// 先发送一个最终状态,确保主进程知道我们完成了
ipc_worker_send_status(job_id, 1, "Completed");
// 等待一小段时间确保上面的消息已发送
usleep(50000); // 50ms
// 发送正式的完成消息
ipc_worker_job_completed(job_id);
// 等待另一小段时间确保完成消息被发送
usleep(100000); // 100ms
// 关闭连接,模拟子进程完成后退出
printf("[子进程 %d] 准备断开连接并退出\n", getpid());
// 停止libuv相关资源(不会尝试关闭或删除事件循环)
ipc_worker_stop();
printf("[子进程 %d] 即将退出\n", getpid());
fflush(stdout); // 确保所有输出都被写出
// 最后一次等待,确保所有消息都已发送
usleep(200000); // 200ms
_exit(0); // 立即退出,不执行任何清理
}
void signal_handler(int sig)
{
if (sig == SIGINT || sig == SIGTERM) {
printf("\n收到信号 %d,准备退出\n", sig);
fflush(stdout);
if (process_type == PROCESS_MASTER) {
exit_flag = 1;
// 向所有子进程发送中断信号
printf("[主进程] 向子进程发送终止信号\n");
kill(0, SIGTERM); // 使用SIGTERM而不是SIGINT
// 如果没有子进程,立即停止事件循环
if (child_processes <= 0) {
printf("[主进程] 没有子进程,直接退出\n");
if (loop)
uv_stop(loop);
return;
}
} else {
// 子进程直接退出,不做任何清理
printf("[子进程 %d] 收到信号,直接退出\n", getpid());
fflush(stdout);
_exit(0);
}
}
}
void run_master()
{
process_type = PROCESS_MASTER;
printf("[主进程] 启动中...\n");
// 设置随机种子
srand((unsigned int)time(NULL));
// 初始化主进程
if (!ipc_master_init(SOCKET_PATH)) {
fprintf(stderr, "[主进程] 初始化失败\n");
exit(1);
}
// 设置回调
ipc_master_set_status_callback(master_job_status_callback);
ipc_master_set_disconnect_callback(master_disconnect_callback);
printf("[主进程] 初始化完成,等待子进程连接...\n");
// 启动子进程
for (int i = 0; i < CHILD_PROCESS_COUNT; i++) {
pid_t pid = fork();
if (pid < 0) {
fprintf(stderr, "[主进程] 创建子进程失败\n");
exit(1);
} else if (pid == 0) {
// 子进程 - 确保不继承父进程的状态
child_processes = 0; // 重置子进程中的计数器
active_jobs = 0;
exit_flag = 0;
run_worker();
_exit(0); // 使用_exit确保不执行父进程的atexit处理程序
} else {
// 主进程 - 记录子进程
printf("[主进程] 子进程 %d 已创建\n", pid);
child_processes++;
active_jobs++;
}
}
// 运行主进程事件循环
ipc_master_run();
printf("[主进程] 事件循环已退出,等待子进程结束...\n");
// 确保所有子进程退出
int status;
pid_t wpid;
// 先等待一小段时间,让子进程有机会自然退出
usleep(100000); // 等待100ms
// 等待所有子进程退出
int remaining = 0;
while ((wpid = waitpid(-1, &status, WNOHANG)) > 0) {
printf("[主进程] 子进程 %d 已退出,状态码: %d\n", wpid,
WIFEXITED(status) ? WEXITSTATUS(status) : -1);
}
// 检查是否还有子进程
pid_t* child_pids = malloc(sizeof(pid_t) * 100);
if (child_pids) {
int remaining = 0;
FILE* fp = popen("ps -o pid -p $$ | tail -n +2", "r"); // 使用ps更可靠
if (fp) {
char line[32];
while (fgets(line, sizeof(line), fp)) {
pid_t child_pid = atoi(line);
if (child_pid > 0) {
printf("[主进程] 发现仍在运行的子进程: %d,发送终止信号\n",
child_pid);
kill(child_pid, SIGTERM);
remaining++;
}
}
pclose(fp);
}
free(child_pids);
if (remaining > 0) {
printf("[主进程] 等待 %d 个子进程终止...\n", remaining);
sleep(1); // 给子进程一些时间来处理SIGTERM
// 使用ps再次检查并用SIGKILL终止顽固的进程
fp = popen("ps -o pid -p $$ | tail -n +2", "r");
if (fp) {
char line[32];
while (fgets(line, sizeof(line), fp)) {
pid_t child_pid = atoi(line);
if (child_pid > 0) {
printf(
"[主进程] 子进程 %d 未响应 SIGTERM, 发送 SIGKILL\n",
child_pid);
kill(child_pid, SIGKILL);
}
}
pclose(fp);
}
}
}
printf("[主进程] 所有子进程已退出,测试完成\n");
}
void run_worker()
{
process_type = PROCESS_WORKER;
// 设置随机种子,使不同子进程行为不同
srand(getpid() + (unsigned int)time(NULL));
printf("[子进程 %d] 启动中...\n", getpid());
// 初始化子进程
if (!ipc_worker_init(SOCKET_PATH)) {
fprintf(stderr, "[子进程 %d] 初始化失败\n", getpid());
_exit(1);
}
// 设置回调
ipc_worker_set_job_callback(worker_job_callback);
// 连接到主进程
ipc_worker_connect(worker_connection_callback);
// 运行子进程事件循环
ipc_worker_run();
printf("[子进程 %d] 事件循环已退出\n", getpid());
// 确保子进程立即退出,不执行任何清理操作
fflush(stdout);
_exit(0);
}
int main()
{
// 初始化全局变量
child_processes = 0;
active_jobs = 0;
exit_flag = 0;
// 忽略SIGPIPE信号,它可能在写入已关闭的pipe时发生
signal(SIGPIPE, SIG_IGN);
// 设置信号处理
struct sigaction sa;
memset(&sa, 0, sizeof(sa));
sa.sa_handler = signal_handler;
sa.sa_flags = SA_RESTART; // 使被中断的系统调用自动重启
sigaction(SIGINT, &sa, NULL);
sigaction(SIGTERM, &sa, NULL);
// 设置闹钟信号处理
struct sigaction sa_alrm;
memset(&sa_alrm, 0, sizeof(sa_alrm));
sa_alrm.sa_handler = alarm_handler;
sigaction(SIGALRM, &sa_alrm, NULL);
// 设置超时机制,防止程序永久挂起
alarm(30); // 如果30秒后程序仍未退出,将发送SIGALRM
// 处理子进程退出,避免僵尸进程
struct sigaction sa_chld;
memset(&sa_chld, 0, sizeof(sa_chld));
sa_chld.sa_handler = SIG_DFL;
sa_chld.sa_flags = SA_NOCLDWAIT; // 不创建僵尸进程
sigaction(SIGCHLD, &sa_chld, NULL);
// 作为主进程运行
run_master();
// 确保清理任何残留资源
printf("[主进程] 程序退出\n");
return 0;
}

View file

@ -1,142 +0,0 @@
#include <arpa/inet.h>
#include <netinet/in.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>
#define BUFFER_SIZE 1024
static volatile int running = 1;
static int server_fd = -1;
static void signal_handler(int signo)
{
if (signo == SIGINT || signo == SIGTERM) {
printf("\nReceived signal %d, shutting down...\n", signo);
running = 0;
if (server_fd != -1) {
close(server_fd);
}
}
}
// 获取当前时间戳字符串
static void get_timestamp(char* timestamp, size_t size)
{
time_t now;
struct tm* timeinfo;
time(&now);
timeinfo = localtime(&now);
strftime(timestamp, size, "%Y-%m-%d %H:%M:%S", timeinfo);
}
// 从JSON字符串中解析CPU核心数
static int parse_cpu_cores(const char* json)
{
int cores = 0;
if (sscanf(json, "{\"slots\":%d}", &cores) == 1) {
return cores;
}
return -1;
}
int main()
{
int client_fd;
struct sockaddr_in address;
int opt = 1;
socklen_t addrlen = sizeof(address);
char buffer[BUFFER_SIZE] = { 0 };
char timestamp[64];
const char* tasks[] = { "sleep 7", "sleep 5", "sleep 6", "date" };
// 设置随机数种子
srand(time(NULL));
// 设置信号处理
signal(SIGINT, signal_handler);
signal(SIGTERM, signal_handler);
// 创建服务器套接字
server_fd = socket(AF_INET, SOCK_STREAM, 0);
if (server_fd == 0) {
perror("socket failed");
exit(EXIT_FAILURE);
}
// 设置套接字选项
if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt,
sizeof(opt))) {
perror("setsockopt failed");
close(server_fd);
exit(EXIT_FAILURE);
}
// 绑定套接字
address.sin_family = AF_INET;
address.sin_addr.s_addr = INADDR_ANY;
address.sin_port = htons(6789);
if (bind(server_fd, (struct sockaddr*)&address, sizeof(address)) < 0) {
perror("bind failed");
close(server_fd);
exit(EXIT_FAILURE);
}
// 监听连接
if (listen(server_fd, 3) < 0) {
perror("listen failed");
close(server_fd);
exit(EXIT_FAILURE);
}
printf("Server listening on port 6789\n");
while (running) {
client_fd = accept(server_fd, (struct sockaddr*)&address, &addrlen);
if (client_fd < 0) {
if (running) {
perror("accept failed");
}
continue;
}
// 处理客户端请求
int bytes_read = read(client_fd, buffer, BUFFER_SIZE);
if (bytes_read > 0) {
buffer[bytes_read] = '\0';
get_timestamp(timestamp, sizeof(timestamp));
printf("[%s] Received: %s\n", timestamp, buffer);
// 解析CPU核心数
int cpu_cores = parse_cpu_cores(buffer);
if (cpu_cores > 0) {
printf("Parsed CPU cores: %d\n", cpu_cores);
// 随机选择一个任务
const char* task = tasks[rand() % 4];
printf("Sending task: %s\n", task);
// 发送任务到远端
if (send(client_fd, task, strlen(task), 0) < 0) {
perror("send failed");
}
} else if (cpu_cores == 0) {
printf("The server is running at full capacity\n");
} else {
printf("Failed to parse CPU cores\n");
}
}
close(client_fd);
}
close(server_fd);
printf("Server shut down\n");
return 0;
}

View file

@ -1,165 +0,0 @@
#include "../include/connection_handler.h"
#include "../include/process_manager.h"
#include "../include/system_monitor.h"
#include "../include/thread_pool.h"
#include "../include/xcecd.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity.h>
// Mock function declarations
static int mock_network_start_called = 0;
static int mock_thread_pool_create_called = 0;
static bool mock_network_start_return = true;
void setUp(void)
{
mock_network_start_called = 0;
mock_thread_pool_create_called = 0;
mock_network_start_return = true;
}
void tearDown(void) { xcecd_cleanup(); }
// Test successful xcecd initialization
void test_xcecd_init_success(void)
{
xcecd_config_t config = { .thread_pool_size = 4,
.resource_update_interval = 5,
.job_status_update_interval = 1,
.port = 8080,
.acct_file = "/tmp/xcecd.acct",
.log_file = "/tmp/xcecd.log",
.domain_sock_file = "/tmp/xcecd.sock",
.remote_host = "localhost",
.remote_port = 6789,
.update_interval = 2 };
int result = xcecd_init(&config);
TEST_ASSERT_EQUAL_INT(0, result);
}
// Test xcecd initialization with null config
void test_xcecd_init_null_config(void)
{
int result = xcecd_init(NULL);
TEST_ASSERT_EQUAL_INT(-1, result);
}
// Test xcecd initialization with invalid parameters
void test_xcecd_init_invalid_params(void)
{
xcecd_config_t config = { .thread_pool_size = 0, // Invalid size
.resource_update_interval = 0,
.job_status_update_interval = 0,
.port = 0,
.acct_file = NULL,
.log_file = NULL,
.domain_sock_file = NULL,
.remote_host = NULL,
.remote_port = 0,
.update_interval = 0 };
int result = xcecd_init(&config);
TEST_ASSERT_EQUAL_INT(-1, result);
}
// Test xcecd start without initialization
void test_xcecd_start_without_init(void)
{
int result = xcecd_start();
TEST_ASSERT_EQUAL_INT(-1, result);
}
// Test xcecd stop functionality
void test_xcecd_stop(void)
{
xcecd_config_t config = { .thread_pool_size = 4,
.resource_update_interval = 5,
.job_status_update_interval = 1,
.port = 8080,
.acct_file = "/tmp/xcecd.acct",
.log_file = "/tmp/xcecd.log",
.domain_sock_file = "/tmp/xcecd.sock",
.remote_host = "localhost",
.remote_port = 6789,
.update_interval = 2 };
xcecd_init(&config);
xcecd_stop();
// Verify xcecd is stopped by trying to start it
int result = xcecd_start();
TEST_ASSERT_EQUAL_INT(-1, result);
}
// Test cleanup functionality
void test_xcecd_cleanup(void)
{
xcecd_config_t config = { .thread_pool_size = 4,
.resource_update_interval = 5,
.job_status_update_interval = 1,
.port = 8080,
.acct_file = "/tmp/xcecd.acct",
.log_file = "/tmp/xcecd.log",
.domain_sock_file = "/tmp/xcecd.sock",
.remote_host = "localhost",
.remote_port = 6789,
.update_interval = 2 };
xcecd_init(&config);
xcecd_cleanup();
// Verify cleanup was successful by trying to start
int result = xcecd_start();
TEST_ASSERT_EQUAL_INT(-1, result);
}
// Test multiple init/cleanup cycles
void test_xcecd_multiple_init_cleanup(void)
{
xcecd_config_t config = { .thread_pool_size = 4,
.resource_update_interval = 5,
.job_status_update_interval = 1,
.port = 8080,
.acct_file = "/tmp/xcecd.acct",
.log_file = "/tmp/xcecd.log",
.domain_sock_file = "/tmp/xcecd.sock",
.remote_host = "localhost",
.remote_port = 6789,
.update_interval = 2 };
for (int i = 0; i < 5; i++) {
int init_result = xcecd_init(&config);
TEST_ASSERT_EQUAL_INT(0, init_result);
xcecd_cleanup();
}
}
// Test initialization with minimum values
void test_xcecd_init_minimum_values(void)
{
xcecd_config_t config = { .thread_pool_size = 1,
.resource_update_interval = 1,
.job_status_update_interval = 1,
.port = 1024,
.acct_file = "/tmp/xcecd.acct",
.log_file = "/tmp/xcecd.log",
.domain_sock_file = "/tmp/xcecd.sock",
.remote_host = "localhost",
.remote_port = 6789,
.update_interval = 1 };
int result = xcecd_init(&config);
TEST_ASSERT_EQUAL_INT(0, result);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_xcecd_init_success);
RUN_TEST(test_xcecd_init_null_config);
RUN_TEST(test_xcecd_init_invalid_params);
RUN_TEST(test_xcecd_start_without_init);
RUN_TEST(test_xcecd_stop);
RUN_TEST(test_xcecd_cleanup);
RUN_TEST(test_xcecd_multiple_init_cleanup);
RUN_TEST(test_xcecd_init_minimum_values);
return UNITY_END();
}

View file

@ -1,322 +0,0 @@
#include "../include/process_manager.h"
#include "../include/xcecd.h"
#include "../include/xcecd_utils.h"
#include "unity.h"
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/un.h>
#include <unistd.h>
#define SOCKET_PATH "/tmp/xcecd.sock"
#define BUFFER_SIZE 4096
static bool reported_flag = false;
// Mock function declarations
static int mock_socket_fd = -1;
static int mock_connect_called = 0;
static int mock_send_called = 0;
static int mock_recv_called = 0;
static int mock_close_called = 0;
static int mock_select_called = 0;
static int mock_waitpid_called = 0;
static int mock_process_manager_start_process_called = 0;
static int mock_connect_return = 0;
static int mock_send_return = 0;
static int mock_recv_return = 0;
static int mock_select_return = 0;
static int mock_waitpid_return = 0;
static int mock_process_manager_start_process_return = 0;
extern int main(void);
void setUp(void)
{
mock_socket_fd = -1;
mock_connect_called = 0;
mock_send_called = 0;
mock_recv_called = 0;
mock_close_called = 0;
mock_select_called = 0;
mock_waitpid_called = 0;
mock_process_manager_start_process_called = 0;
mock_connect_return = 0;
mock_send_return = 0;
mock_recv_return = 0;
mock_select_return = 0;
mock_waitpid_return = 0;
mock_process_manager_start_process_return = 0;
}
void tearDown(void)
{
// 清理资源
}
// Mock functions
int socket(int domain, int type, int protocol) { return mock_socket_fd; }
int connect(int sockfd, const struct sockaddr* addr, socklen_t addrlen)
{
mock_connect_called++;
return mock_connect_return;
}
ssize_t send(int sockfd, const void* buf, size_t len, int flags)
{
mock_send_called++;
return mock_send_return;
}
ssize_t recv(int sockfd, void* buf, size_t len, int flags)
{
mock_recv_called++;
if (mock_recv_return > 0) {
strncpy(buf, "sleep 5", len);
}
return mock_recv_return;
}
int close(int fd)
{
mock_close_called++;
return 0;
}
int select(int nfds, fd_set* readfds, fd_set* writefds, fd_set* exceptfds,
struct timeval* timeout)
{
mock_select_called++;
return mock_select_return;
}
pid_t waitpid(pid_t pid, int* status, int options)
{
mock_waitpid_called++;
*status = 0;
return mock_waitpid_return;
}
__attribute__((weak)) int
process_manager_start_process(process_manager_t* manager, const char* command)
{
mock_process_manager_start_process_called++;
return mock_process_manager_start_process_return;
}
static char* get_task_command(void)
{
int sockfd;
struct sockaddr_un addr;
if ((sockfd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
perror("socket error");
return NULL;
}
memset(&addr, 0, sizeof(addr));
addr.sun_family = AF_UNIX;
strncpy(addr.sun_path, SOCKET_PATH, sizeof(addr.sun_path) - 1);
if (connect(sockfd, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
perror("connect error");
close(sockfd);
return NULL;
}
// int flags = fcntl(sockfd, F_GETFL, 0);
// fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
fd_set read_fds;
FD_ZERO(&read_fds);
FD_SET(sockfd, &read_fds);
struct timeval timeout = { 1, 0 };
int select_ret = select(sockfd + 1, &read_fds, NULL, NULL, &timeout);
if (select_ret > 0 && FD_ISSET(sockfd, &read_fds)) {
char command[BUFFER_SIZE] = { 0 };
ssize_t bytes_read = read(sockfd, command, sizeof(command) - 1);
if (bytes_read > 0) {
command[bytes_read] = '\0';
close(sockfd);
return strndup(command, sizeof(command));
} else if (bytes_read == -1 && errno != EAGAIN
&& errno != EWOULDBLOCK) {
perror("read error");
}
} else if (select_ret == -1) {
perror("select error");
} else {
printf("Socket not ready for reading\n");
}
close(sockfd);
return NULL;
}
static void report_status_to_master(int pid, int status)
{
int sockfd;
struct sockaddr_un addr;
if (reported_flag) {
return;
}
if ((sockfd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
perror("socket error");
return;
}
memset(&addr, 0, sizeof(addr));
addr.sun_family = AF_UNIX;
strncpy(addr.sun_path, SOCKET_PATH, sizeof(addr.sun_path) - 1);
if (connect(sockfd, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
perror("connect error");
close(sockfd);
return;
}
// int flags = fcntl(sockfd, F_GETFL, 0);
// fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
fd_set write_fds;
FD_ZERO(&write_fds);
FD_SET(sockfd, &write_fds);
struct timeval timeout = { 1, 0 };
int select_ret = select(sockfd + 1, NULL, &write_fds, NULL, &timeout);
if (select_ret > 0 && FD_ISSET(sockfd, &write_fds)) {
char buffer[BUFFER_SIZE] = { 0 };
snprintf(buffer, sizeof(buffer),
"WORKER_PID: %d JOB_PID: %d STATUS: %d", getpid(), pid,
status);
ssize_t bytes_written
= send(sockfd, buffer, strlen(buffer), MSG_NOSIGNAL);
if (bytes_written < 0) {
perror("send error");
} else {
printf("Reported status to master\n");
fflush(stdout);
reported_flag = true;
}
} else if (select_ret == -1) {
perror("select error");
} else {
printf("Socket not ready for writing\n");
}
close(sockfd);
}
static void handle_task_output(bool peek)
{
if (!peek) {
return;
}
printf("Succeed!");
fflush(stdout);
}
// Test cases
void test_xcecd_worker_init(void)
{
// 测试初始化
process_manager_t* manager = process_manager_create();
TEST_ASSERT_NOT_NULL(manager);
process_manager_destroy(manager);
}
void test_xcecd_worker_get_task_command(void)
{
// 测试获取任务命令
mock_socket_fd = 1;
mock_connect_return = 0;
mock_recv_return = 8;
mock_select_return = 1;
char* command = get_task_command();
TEST_ASSERT_NOT_NULL(command);
TEST_ASSERT_EQUAL_STRING("sleep 5", command);
free(command);
TEST_ASSERT_EQUAL_INT(1, mock_connect_called);
TEST_ASSERT_EQUAL_INT(1, mock_recv_called);
TEST_ASSERT_EQUAL_INT(1, mock_select_called);
TEST_ASSERT_EQUAL_INT(1, mock_close_called);
}
void test_xcecd_worker_handle_task_output(void)
{
// 测试处理任务输出
process_manager_t* manager = process_manager_create();
process_info_t proc;
proc.output_fd = 1;
manager->processes = &proc;
manager->process_count = 1;
handle_task_output(true);
process_manager_destroy(manager);
}
void test_xcecd_worker_report_status_to_master(void)
{
// 测试报告状态到主进程
mock_socket_fd = 1;
mock_connect_return = 0;
mock_send_return = 20;
mock_select_return = 1;
report_status_to_master(1234, WORKER_DONE);
TEST_ASSERT_EQUAL_INT(1, mock_connect_called);
TEST_ASSERT_EQUAL_INT(1, mock_send_called);
TEST_ASSERT_EQUAL_INT(1, mock_select_called);
TEST_ASSERT_EQUAL_INT(1, mock_close_called);
}
void test_xcecd_worker_main(void)
{
// 测试worker主函数
mock_socket_fd = 1;
mock_connect_return = 0;
mock_recv_return = 8;
mock_select_return = 1;
mock_process_manager_start_process_return = 0;
mock_waitpid_return = 1234;
int result = main();
TEST_ASSERT_EQUAL_INT(0, result);
TEST_ASSERT_EQUAL_INT(1, mock_connect_called);
TEST_ASSERT_EQUAL_INT(1, mock_recv_called);
TEST_ASSERT_EQUAL_INT(1, mock_select_called);
TEST_ASSERT_EQUAL_INT(1, mock_close_called);
TEST_ASSERT_EQUAL_INT(1, mock_process_manager_start_process_called);
TEST_ASSERT_EQUAL_INT(1, mock_waitpid_called);
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_xcecd_worker_init);
RUN_TEST(test_xcecd_worker_get_task_command);
RUN_TEST(test_xcecd_worker_handle_task_output);
RUN_TEST(test_xcecd_worker_report_status_to_master);
RUN_TEST(test_xcecd_worker_main);
return UNITY_END();
}