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

import worker code

Browse source
This commit is contained in:
Rongsong Shen 2025-11-24 15:53:31 +08:00
parent 61654a06e8
commit 8831040043
64 changed files with 20853 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

101
src/worker/README.md Normal file
View file

@ -0,0 +1,101 @@
# Cross-Platform Xcecd Process
This is a cross-platform process manager implementation in C that supports both Linux and Windows platforms. The xcecd provides the following features:
- Sending status to the remote controller module at port 6789 and receive job requests
- Maintains a configurable thread pool for request processing
- Periodically collects and updates system resource status
- Manages child processes spawned from received commands
- Caches job status information
- Uses Unix Domain Sockets for communication between the main process and worker processes
## Building
### Prerequisites
- C compiler with C11 support
- Windows: Visual Studio
- Linux: GCC and development tools
### Build Instructions
On Linux
```bash
make
```
On Windows
```bash
nmake
```
## Configuration
The xcecd can be configured through the following parameters in `main.c`:
- `thread_pool_size`: Number of threads in the thread pool (default: 4)
- `resource_update_interval`: Interval for updating system resource status in seconds (default: 1)
- `job_status_update_interval`: Interval for updating job status in seconds (default: 1)
- `port`: Port number to listen on (default: 7869)
- `acct_file_path`: Path to the accounting file (default: `/tmp/xcecd.acct`)
- `log_file_path`: Path to the log file (default: `/tmp/xcecd.log`)
- `domain_sock_path`: Path to the Unix Domain Socket file (default: `/tmp/xcecd.sock`)
- `remote_host`: Remote server address (default: `lchen02`)
- `remote_port`: Remote server port (default: 6789)
- `update_interval`: Interval for sending status updates to the remote server (default: 1 second)
## Usage
### Starting the Xcecd
On Windows:
```bash
xcecd.exe
```
On Linux:
```bash
./xcecd
```
## Features
1. **Cross-Platform Support**
- Works on both Windows and Linux
- Abstracts platform-specific implementations
2. **Thread Pool**
- Configurable number of worker threads
- Efficient request handling
3. **System Monitoring**
- CPU usage monitoring
- Memory usage monitoring
- Disk space monitoring
4. **Process Management**
- Spawns child processes for received commands
- Tracks process status
- Manages process lifecycle
5. **Network Communication**
- TCP socket-based communication
- Simple text-based protocol
6. **Unix Domain Socket Communication**
- Uses Unix Domain Sockets for communication between the main process and worker processes
## Architecture
The xcecd is composed of several key components:
- **Main Process**: Coordinates all components and manages the lifecycle
- **Thread Pool**: Handles concurrent request processing
- **System Monitor**: Collects and updates system resource information
- **Process Manager**: Manages child processes and their status
- **Network Handler**: Handles network communication and request receiving
- **Resource Sync**: Synchronizes and aggregates resource collection thread, system monitor, and job information
## Contributing
Feel free to submit issues and enhancement requests!

6
src/worker/conf/worker.yaml Normal file
View file

@ -0,0 +1,6 @@
---
workerconf:
management_list: ["hostA", "hostB"]
controller_port: 9990
log_level: "info"
log_dir: "/tmp"

29
src/worker/include/connection_handler.h Normal file
View file

@ -0,0 +1,29 @@
/* 此模块为预留功能，它提供了xcecd处理远程连接的可能。xcecd默认会打开一个监听端口并能接受外部请求。
* 此保留功能是为了提供LSF兼容性，默认不使能。
*/
#ifndef CONNECTION_HANDLER_H
#define CONNECTION_HANDLER_H
#include <stdint.h>
typedef struct network_handler network_handler_t;
// 网络请求回调函数类型
typedef const char* (*request_handler_t)(const char* request, void* user_data);
// 创建网络处理器
network_handler_t* network_handler_create(uint16_t port);
// 设置请求处理回调
void network_handler_set_callback(network_handler_t* handler, request_handler_t callback, void* user_data);
// 启动网络监听
int network_handler_start(network_handler_t* handler);
// 停止网络监听
void network_handler_stop(network_handler_t* handler);
// 销毁网络处理器
void network_handler_destroy(network_handler_t* handler);
#endif // CONNECTION_HANDLER_H

62
src/worker/include/controller_connector.h Normal file
View file

@ -0,0 +1,62 @@
#ifndef CONTROL_CONNECTOR_H
#define CONTROL_CONNECTOR_H
#include "thread_pool.h"
#include "xcecd_utils.h"
#include "client.h"
#include "xcectx.h"
#include "controller.capnp.h"
typedef struct {
xce_client_config_t* conf;
endpoint_t ep;
xce_ctx_t ctx;
} api_ctx_t;
typedef struct controller_connector_ctx_s {
void *thread_pool;
void *process_manager;
char *remote_host;
char *virtual_name;
void **worker_list;
int remote_port;
int join_interval;
bool running;
} controller_connector_ctx_t;
extern controller_connector_ctx_t *cc_ctx;
typedef struct {
int (*job_finished)(xcecd_job_t *job);
int (*job_stopped)(uint64_t jobid);
int (*job_resumed)(uint64_t jobid);
int (*job_start_error)(uint64_t jobid);
int (*job_status_error)(uint64_t jobid);
int (*job_ckpted)(uint64_t jobid, int status, const char *info);
int (*job_remote_start)(uint64_t jobid, const char *hostname, const char *command);
bool (*reply_startjob)(xcecd_job_t *job);
} cc_callbacks_t;
extern void xcecd_process_command(void* arg);
int start_controller_connector_thread(task_func_t function, void* arg);
int xcecd_join_cluster(hostattrs_t* attrs);
bool controller_connector_start(void);
bool xcecd_reply_startjob(xcecd_job_t *job);
int xcecd_job_finished(xcecd_job_t *job);
int xcecd_job_stopped(uint64_t jobid);
int xcecd_job_resumed(uint64_t jobid);
int checking_connection();
int xcecd_job_start_error(uint64_t jobid);
int xcecd_job_status_error(uint64_t jobid);
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);
#endif // CONTROL_CONNECTOR_H

28
src/worker/include/hardware_monitor.h Normal file
View file

@ -0,0 +1,28 @@
#ifndef HARDWARE_MONITOR_H
#define HARDWARE_MONITOR_H
#include <stdbool.h>
#include <sys/types.h>
#include <time.h>
typedef struct hardware_monitor hardware_monitor_t;
// 创建硬件监控器
hardware_monitor_t* hardware_monitor_create(const char* acct_file);
// 销毁硬件监控器
void hardware_monitor_destroy(hardware_monitor_t* monitor);
// 更新硬件信息
bool hardware_monitor_update(hardware_monitor_t* monitor);
// 获取CPU核心数
int hardware_monitor_get_cpu_cores(hardware_monitor_t* monitor);
double hardware_monitor_get_total_memory(hardware_monitor_t* monitor);
double hardware_monitor_get_free_memory(hardware_monitor_t* monitor);
bool hardware_monitor_get_memory_info(int64_t* total_memory, int64_t* free_memory);
#endif // HARDWARE_MONITOR_H

23
src/worker/include/inter_job_utils.h Normal file
View file

@ -0,0 +1,23 @@
#ifndef _INTER_JOB_UTILS_H
#define _INTER_JOB_UTILS_H
#define CMD_MAX_LENGTH 4096
#define URL_MAX_LENGTH 256
#define TTY_END_PORT 10000
#define XJOB_END_PORT 5999
typedef struct {
int tty_port;
int xjob_port;
} port_info_t;
int check_port_availability(int port_number);
int find_available_port(int start_port, int end_port);
int check_port(port_info_t port, int interactive_mode);
char* generate_cmd(char* job_cmd, port_info_t port, int interactive_mode);
char* generate_url(port_info_t port, int interactive_mode);
int export_display_env(int port);
#endif /* _INTER_JOB_UTILS_H */

211
src/worker/include/job_information.h Normal file
View file

@ -0,0 +1,211 @@
#ifndef JOB_INFORMATION_H
#define JOB_INFORMATION_H
#include "xcecd_utils.h"
#include <sys/types.h>
#include <netinet/in.h>
#define NBUILTINDEX 11
#define MAX_LSB_NAME_LEN 60
#define MAXFILENAMELEN 256
struct job_specs {
char *job_name;
char *job_owner;
char *job_group;
char *job_project;
char *job_class;
char *job_type;
char *job_queue;
char *job_start_time;
char *job_end_time;
char *job_window;
char *job_flags;
char *job_exec_host;
char *job_exec_port;
char *job_exec_path;
char *job_exec_args;
char *job_exec_env;
char *job_exec_dir;
char *job_exec_user;
char *job_exec_group;
char *job_exec_shell;
char *job_exec_type;
char *job_exec_flags;
char *job_exec_window;
char *job_exec_start_time;
char *job_exec_end_time;
char *job_exec_queue;
char *job_exec_project;
char *job_exec_class;
};
struct xce_rusage {
float ru_utime;
float ru_stime;
float ru_maxrss;
float ru_ixrss;
float ru_idrss;
float ru_isrss;
float ru_minflt;
float ru_majflt;
float ru_nswap;
float ru_inblock;
float ru_oublock;
float ru_msgsnd;
float ru_msgrcv;
float ru_nsignals;
float ru_nvcsw;
float ru_nivcsw;
};
struct job_rusage {
float cpu_time;
float max_rss;
float ixrss;
float idrss;
float isrss;
float minflt;
float majflt;
float nswap;
float inblock;
float oublock;
float msgsnd;
float msgrcv;
float nsignals;
float nvcsw;
float nivcsw;
};
struct res_val {
char *selectStr;
int nphase;
int order[NBUILTINDEX];
int genClass;
float *val;
int nindex;
int *indicies;
int *rusgBitMaps;
int duration;
float decay;
int numHosts;
int maxNumHosts;
int pTile;
int options;
int selectStrSize;
char **xorExprs;
};
// 以上结构体定义为了方便前期调试，后续可能会根据实际情况进行修改，或者从新定义在控制模块中.
// 它们包含 job_specs、xce_rusage、job_rusage、res_val 四个结构体，分别用于作业规格、资源使用情况、作业资源使用情况、资源值的定义
struct client_node {
struct client_node *prev;
struct client_node *next;
int chanfd;
struct sockaddr_in from;
int job_type;
long job_id; // 作业ID，暂定为长整型，后续可能会修改
struct job_card *jp;
};
struct job_card {
struct job_card *next;
struct job_card *prev;
gid_t exec_gid; // 执行用户组ID
char exec_user_name[MAX_LSB_NAME_LEN]; // 执行用户名
int not_reported;
time_t wind_edge; // 时间窗口边界
windows_t *week[8]; // 一周的时间窗口
char active;
char time_expire;
char missing;
char controller_restarted;
time_t wind_warn_time;
int runtime;
int w_status;
/* pre-exec report flag */
float cpu_time;
time_t last_chkpnt_time;
int mig_count;
struct job_specs job_specs;
struct xce_rusage xce_rusage;
int need_report_rusage;
int cleanup_pid;
int collected_child;
int exec_job_flag;
#define JOB_EXEC_QPRE_OK 0x1
#define JOB_EXEC_QPRE_KNOWN 0x2
#define JOB_EXEC_STARTED 0x4
char *stdin_file; // 标准输入文件
time_t last_controller_time; // 最后一次和控制器连接上的时间
struct job_rusage run_rusage; // 运行时资源使用情况
struct job_rusage alloc_rusage; // 控制器资源分配情况
struct job_rusage max_rusage; // 最大资源使用情况
int delievered_message_id;
struct client_node *client;
int reg_op_flag;
struct job_rusage work_rusage;
int job_sup_status;
#define REG_CHKPNT 0x1
#define REG_SIGNAL 0x2
#define REG_RUSAGE 0x4
#define REG_NICE 0x8
struct res_val *resume_cond_val;
struct res_val *stop_cond_val;
int act_flags;
int act_status;
int act_reasons;
int act_sub_reasons;
int exit_pid;
int job_done;
time_t last_check;
char *act_cmd;
char *exit_file;
char *cluster_name;
int server_socket;
int cross_platforms;
char *spooled_exec;
char post_job_started;
char user_job_succeed;
};
struct job_setup {
long job_id;
int job_status;
float cpu_time;
int w_status;
struct xce_rusage xce_rusage;
int reason;
int job_pid;
int job_pgid;
int exec_gid;
int exec_uid;
char exec_user_name[MAX_LSB_NAME_LEN];
char exec_home[MAXFILENAMELEN];
char exec_cwd[MAXFILENAMELEN];
int exec_job_flag;
#define LSB_PRE_ABORT 99
};
void init_job_setup(struct job_setup* setup, const char* data);
void init_job_card(struct job_card* card, const char* data);
int update_job_card(struct job_card* card, const char* data);
void cleanup_job_card(struct job_card* card);
struct client_node* init_client_node(long id);
#endif // JOB_INFORMATION_H

117
src/worker/include/node_information.h Normal file
View file

@ -0,0 +1,117 @@
#ifndef NODE_INFORMATION_H_
#define NODE_INFORMATION_H_
#include <stdint.h>
#include <arpa/inet.h>
#include "xcecd_utils.h"
#include "connection_handler.h"
#include "process_manager.h"
#include "system_monitor.h"
#include "hardware_monitor.h"
#include "thread_pool.h"
#define MAXXCENAMELEN 40
#define MAXNODENAMELEN 256
#define MAXCANDHOSTS 10
#define DEFAULT_AFTER_HOUR "19:00-7:00 5:19:00-1:7:00"
struct time_window {
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);
#endif // NODE_INFORMATION_H_

128
src/worker/include/process_manager.h Normal file
View file

@ -0,0 +1,128 @@
#ifndef PROCESS_MANAGER_H
#define PROCESS_MANAGER_H
#include <stddef.h>
#include <stdbool.h>
#ifdef _WIN32
#include <windows.h>
typedef HANDLE process_handle_t;
#else
#include <pthread.h>
#include <unistd.h>
typedef pid_t process_handle_t;
#endif
#include "sqlite3.h"
typedef struct {
process_handle_t handle;
char* command; // command line
char* work_dir; // working directory
int status; // process status (0: stopped, 1: running)
int exit_code; // exit code
int output_fd; // output file descriptor
char* log_file; // log file path
#ifdef _WIN32
HANDLE pipe_read; // pipe read handle
HANDLE monitor_thread; // monitor thread handle
bool monitor_active; // monitor thread active flag
#endif
} process_info_t;
typedef struct process_manager {
process_info_t* processes;
size_t process_count;
int update_interval;
bool running;
bool initialized;
#ifdef _WIN32
HANDLE thread;
CRITICAL_SECTION lock;
#else
pthread_t thread;
pthread_mutex_t lock;
#endif
} process_manager_t;
/**
* @brief initialize process info structure
* @param proc process info structure
* @return void
*/
void process_info_init(process_info_t* proc);
/**
* @brief initialize process info structure
* @return process_manager_t* if success, NULL if failed
*/
process_manager_t* process_manager_create(void);
/**
* @brief start a new process
* @param manager process manager without resource limits
* @param command command with arguments
* @return 0 if success, -1 if failed
*/
int process_manager_start_process(process_manager_t* manager, const char* command);
/**
* @brief get the number of processes
* @param manager process manager
* @return the number of processes
*/
int get_process_count(process_manager_t* manager);
/**
* @brief get the current process pid
* @param manager process manager
* @return the current process pid
*/
int get_current_process_pid(process_manager_t* manager);
/**
* @brief update the process status in the list
* @param manager process manager
* @return true if success, false if failed
*/
bool process_manager_update(process_manager_t* manager);
/**
* @brief get process info by handle
* @param manager process manager
* @param handle process handle
* @return process_info_t* if success, NULL if failed
*/
process_info_t* process_manager_get_process_info(process_manager_t* manager, process_handle_t handle);
/**
* @brief terminate a process
* @param manager process manager
* @param handle process handle
* @return 0 if success, -1 if failed
*/
int process_manager_terminate_process(process_manager_t* manager, process_handle_t handle);
/**
* @brief destroy process manager
* @param manager process manager
* @return void
*/
void process_manager_destroy(process_manager_t* manager);
/**
* @brief remove a process from the list
* @param manager process manager
* @param handle process handle
* @return 0 if success, -1 if failed
*/
int process_manager_remove_process(process_manager_t* manager, process_handle_t handle);
/**
* @brief free process manager
* @param manager process manager
* @return void
*/
void process_manager_free(process_manager_t* manager);
#endif // PROCESS_MANAGER_H

35
src/worker/include/process_worker_manager.h Normal file
View file

@ -0,0 +1,35 @@
/**
* @file process_worker_manager.h
* @brief This file contains the declaration of the worker process manager functions.
* It uses to manage worker process with resource limits. And it relies on
* process_manager.h to manage the worker process. This module disabled by default.
*/
#ifndef _PROCESS_WORKER_MANAGER_H_
#define _PROCESS_WORKER_MANAGER_H_
#include "process_manager.h"
#include "xcecd_worker_utils.h"
#include "xcecd_utils.h"
#include "xcelog.h"
/**
* @brief start worker without resource limits (disabled by default)
* @param command command with arguments
* @return 0 if success, -1 if failed
*/
int process_manager_start_worker(process_manager_t* manager);
/**
* @brief update the process status in the list
* @param manager process manager
* @param worker worker process
* @param job XCE job information
* @param db database information
* @return 0 if success, -1 if failed
*/
bool worker_process_update(process_manager_t* manager,
xcecd_worker_t* worker,
xcecd_job_t* job,
sqlite3* db);
#endif // _PROCESS_WORKER_MANAGER_H_

29
src/worker/include/resource_sync.h Normal file
View file

@ -0,0 +1,29 @@
#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);
#endif // RESOURCE_SYNC_H

33
src/worker/include/static_array.h Normal file
View file

@ -0,0 +1,33 @@
#ifndef _STATIC_ARRAY_H_
#define _STATIC_ARRAY_H_
#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <string.h>
#include <pthread.h>
#define MAX_ARRAY_SIZE 128
typedef struct {
unsigned char data[MAX_ARRAY_SIZE * 64];
size_t element_size;
size_t count;
pthread_mutex_t lock;
} StaticArray;
void init_static_array(StaticArray* arr, size_t element_size);
bool push_static(StaticArray* arr, void* value);
bool pop_static(StaticArray* arr, void* out_value);
bool pop_static_from_bottom(StaticArray* arr, void* out_value);
void clear_static(StaticArray* arr);
bool get_element(StaticArray* arr, size_t index, void* out_value);
#endif // _STATIC_ARRAY_H_

29
src/worker/include/system_monitor.h Normal file
View file

@ -0,0 +1,29 @@
#ifndef SYSTEM_MONITOR_H
#define SYSTEM_MONITOR_H
#include <stdint.h>
#include <stdbool.h>
typedef struct {
double cpu_usage; // CPU使用率（百分比）
uint64_t memory_total; // 总内存（字节）
uint64_t memory_used; // 已使用内存（字节）
uint64_t disk_total; // 总磁盘空间（字节）
uint64_t disk_used; // 已使用磁盘空间（字节）
} system_status_t;
typedef struct system_monitor system_monitor_t;
// 创建系统监控器
system_monitor_t* system_monitor_create(void);
// 获取当前系统状态
const system_status_t* system_monitor_get_status(system_monitor_t* monitor);
// 更新系统状态
bool system_monitor_update(system_monitor_t* monitor);
// 销毁系统监控器
void system_monitor_destroy(system_monitor_t* monitor);
#endif // SYSTEM_MONITOR_H

100
src/worker/include/thread_pool.h Normal file
View file

@ -0,0 +1,100 @@
#ifndef THREAD_POOL_H
#define THREAD_POOL_H
#include "xcecd_utils.h"
#include "xcecd_cgroups.h"
#include <stddef.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdint.h>
#define IPC_BUFFER_SIZE 256
#define MAX_MSG_SIZE 4096
#define MAX_QUEUE_SIZE 1024
// 任务函数类型
typedef void (*task_func_t)(void* arg);
// 作业执行任务消息队列
typedef struct {
pid_t target_pid;
uint32_t ntasks;
uint32_t clientidx;
uint64_t xce_jobid;
task_func_t func;
job_startinfo_t* startinfo;
res_limits_t* limits;
#ifndef DBUS_MANAGER
void* manager;
#endif
} fork_task_t;
//作业控制消息队列
typedef struct {
uint64_t jobid;
pid_t target_pid;
int signal;
int force;
bool checkpoint;
int ckpt_kill;
char* ckpt_method;
task_func_t func;
} job_control_task_t;
// 消息队列结构
typedef struct {
void *messages;
task_func_t* func;
size_t msg_size;
size_t capacity;
size_t head;
size_t tail;
size_t count;
pthread_mutex_t lock;
pthread_cond_t not_empty;
pthread_cond_t not_full;
} message_queue_t;
typedef struct {
long system_tid; // 系统线程ID
size_t pool_tid; // 线程池中的线程ID (1-N)
} thread_id_map_t;
typedef struct {
size_t num_threads;
pthread_t* threads;
bool running;
thread_id_map_t* thread_map;
message_queue_t fork_queue;
message_queue_t control_queue;
} thread_pool_t;
enum {
JOB_CONTROL_STOP,
JOB_CONTROL_PAUSE,
JOB_CONTROL_RESUME,
JOB_CONTROL_RESTART,
JOB_CONTROL_KILL,
JOB_CONTROL_STATUS
};
thread_pool_t* thread_pool_create(size_t num_threads);
void thread_pool_destroy(thread_pool_t* pool);
message_queue_t* message_queue_create(size_t capacity, size_t msg_size);
void message_queue_destroy(message_queue_t* queue);
int queue_push(message_queue_t* queue, void* message);
int queue_pop(message_queue_t* queue, void* message);
int submit_fork_task(thread_pool_t* pool, fork_task_t* task);
int submit_job_control_task(thread_pool_t* pool, job_control_task_t* task);
void job_control_signal(void* arg);
#endif // THREAD_POOL_H

170
src/worker/include/uv_ipc.h Normal file
View file

@ -0,0 +1,170 @@
#ifndef UV_IPC_H
#define UV_IPC_H
#include "xcecd_utils.h"
#include "uv.h"
#include <stdint.h>
#include <stdbool.h>
#include <time.h>
#ifdef __cplusplus
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
typedef enum {
MSG_GET_JOB,
MSG_JOB_ASSIGNED,
MSG_JOB_STATUS,
MSG_JOB_COMPLETED,
MSG_JOB_SIGNAL,
MSG_JOB_SIGNAL_ACK,
MSG_JOB_CKPT,
MSG_JOB_CKPT_ACK,
MSG_JOB_CKPT_RESULT,
MSG_REMOTE_START_JOB,
MSG_REMOTE_START_JOB_ACK,
MSG_HEARTBEAT,
MSG_ERROR
} message_type_t;
typedef struct {
int type;
} heartbeat_message_t;
typedef struct {
uint64_t job_id;
int job_pid;
int status_code;
int worker_pid;
int worker_status;
int exit_code;
int stage;
bool post_done;
char status_message[MESSAGE_LENGTH];
char start_exdata[MAXLINELEN];
} job_status_t;
typedef struct {
uint64_t job_id;
int ntasks;
char cmd[COMMAND_LENGTH];
char q_pre_exec[COMMAND_LENGTH];
char q_post_exec[COMMAND_LENGTH];
char job_pre_exec[COMMAND_LENGTH];
char job_post_exec[COMMAND_LENGTH];
char exdata[EXDATA_LENGTH];
} job_assignment_t;
typedef struct {
uint64_t job_id;
int signal;
bool force;
} job_signal_t;
typedef struct {
uint64_t job_id;
int kill; // 0: do not kill process
char method[METHOD_LENGTH];
char img_path[PATH_LENGTH];
char parent_path[PATH_LENGTH];
} job_ckpt_t;
typedef struct {
uint64_t job_id;
int status; // 0: success, 1: fail
char restore_cmd[COMMAND_LENGTH];
} job_ckpt_result_t;
typedef struct {
uint64_t jobid;
char hostname[HOSTNAME_LENGTH];
char command[COMMAND_LENGTH];
} remote_start_job_t;
typedef struct {
int success; // 1: success, 0: failure
char error_message[MESSAGE_LENGTH];
} remote_start_job_ack_t;
typedef struct {
message_type_t type;
union {
job_status_t status;
job_assignment_t assignment;
job_signal_t signal;
job_ckpt_t ckpt;
job_ckpt_result_t ckpt_result;
remote_start_job_t remote_start;
remote_start_job_ack_t remote_start_ack;
char error_message[MESSAGE_LENGTH];
} data;
int worker_pid;
} ipc_message_t;
//#undef FREEUP
//#define FREEUP(ptr) do { if (ptr) { free(ptr); ptr = NULL; } } while(0)
extern int child_processes;
extern int active_jobs;
extern int exit_flag;
extern uv_loop_t* loop;
typedef void (*job_callback_t)(job_assignment_t* info);
typedef void (*connection_callback_t)(bool success);
typedef void (*worker_reconnect_callback_t)(bool connected);
typedef void (*job_status_callback_t)(job_status_t* job_status);
typedef void (*client_disconnect_callback_t)(uint64_t client_id);
typedef void (*heartbeat_callback_t)(uv_stream_t* client);
typedef void (*job_signal_callback_t)(uint64_t job_id, int signal, bool force);
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);
bool ipc_master_init(const char* socket_path);
void ipc_master_run(uv_run_mode mode);
void ipc_master_stop();
bool ipc_master_check_connection();
void ipc_master_set_status_callback(job_status_callback_t callback);
void ipc_master_set_disconnect_callback(client_disconnect_callback_t callback);
void ipc_master_set_heartbeat_callback(heartbeat_callback_t callback);
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);
bool ipc_worker_init(const char* socket_path);
void ipc_worker_connect(connection_callback_t callback);
void ipc_worker_reconnect(worker_reconnect_callback_t callback);
void ipc_worker_run(uv_run_mode mode);
void ipc_worker_stop();
void ipc_worker_set_job_callback(job_callback_t callback);
void ipc_worker_set_reconnection_callback(worker_reconnect_callback_t callback);
void ipc_worker_request_job();
void ipc_worker_send_status(job_runtime_info_t* job_info,
const char* status_message);
void ipc_worker_job_completed(uint64_t job_id);
bool ipc_worker_check_connection();
void ipc_worker_send_signal_ack(uint64_t job_id, int signal, bool force);
void ipc_worker_set_signal_callback(job_signal_callback_t callback);
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);
void on_write_complete(uv_write_t* req,
int status);
uv_loop_t* ipc_get_event_loop();
#ifdef __cplusplus
}
#endif
#endif /* UV_IPC_H */

182
src/worker/include/uv_proc.h Executable file
View file

@ -0,0 +1,182 @@
#ifndef UV_PROC_H
#define UV_PROC_H
#include "uv.h"
#include <stdbool.h>
#include <stdint.h>
/**
* process status enumeration
*/
typedef enum {
UV_PROC_IDLE,
UV_PROC_RUNNING,
UV_PROC_COMPLETED,
UV_PROC_ERROR,
UV_PROC_TERMINATED,
UV_PROC_UNKNOWN
} uv_proc_status_t;
/**
* process exit callback function type
* @param pid process ID
* @param exit_status exit status code
* @param term_signal termination signal (if terminated by signal)
* @param user_data user data passed to the callback
*/
typedef void (*uv_proc_exit_cb)(int pid, int exit_status,
int term_signal, void* user_data);
/**
* process manager structure
*/
typedef struct uv_proc_manager_s {
uint64_t job_id; // job ID
uv_loop_t* loop; // event loop shared with ipc
uv_process_t* active_proc; // current active process
uv_proc_status_t status; // current process status
int last_exit_code; // last exit code
int last_term_signal; // last termination signal
void* user_data; // user data for callback
uv_proc_exit_cb exit_cb; // exit callback function
uv_process_t** proc_list;
int proc_count;
int proc_capacity;
int remaining_procs;
} uv_proc_manager_t;
/**
* initialize the process manager
*
* @param manager process manager pointer
* @param loop event loop pointer
* @return 0 on success, non-zero error code on failure
*/
int uv_proc_init(uv_proc_manager_t* manager,
uv_loop_t* loop);
/**
* pre-execute a command (blocking execution, not managed by event loop)
*
* @param manager process manager pointer
* @param command pre-execute command
* @param args command arguments
* @return 0 on success, non-zero error code on failure
*/
int uv_proc_pre_exec(uv_proc_manager_t* manager,
const char* command, char* const args[]);
/**
* asynchronously execute a command (managed by event loop)
*
* @param manager process manager pointer
* @param command command to execute
* @param args command arguments
* @param exit_cb exit callback function
* @param user_data user data passed to the callback
* @return 0 on success, non-zero error code on failure
*/
int uv_proc_exec(uv_proc_manager_t* manager, const char* command, char* const args[],
uv_proc_exit_cb exit_cb, void* user_data, int ntasks);
/**
* post-execute a command (blocking execution, not managed by event loop)
*
* @param manager process manager pointer
* @param command post-execute command
* @param args command arguments
* @return 0 on success, non-zero error code on failure
*/
int uv_proc_post_exec(uv_proc_manager_t* manager,
const char* command, char* const args[]);
/**
* send a signal to the current process
*
* @param manager process manager pointer
* @param signum signal number
* @return 0 on success, non-zero error code on failure
* @note this function does not wait for the process to exit
*/
int uv_proc_kill(uv_proc_manager_t* manager, int signum);
/**
* get the current process status
*
* @param manager process manager pointer
* @return current process status
*/
uv_proc_status_t uv_proc_get_status(uv_proc_manager_t* manager);
/**
* get the current process ID
*
* @param manager process manager pointer
* @return process ID
*/
int uv_proc_get_pid(uv_proc_manager_t* manager);
/**
* get the last exit code
*
* @param manager process manager pointer
* @return exit code
*/
int uv_proc_get_exit_code(uv_proc_manager_t* manager);
/**
* get the last termination signal
*
* @param manager process manager pointer
* @return termination signal
*/
int uv_proc_get_term_signal(uv_proc_manager_t* manager);
/**
* cleanup the process manager
*
* @param manager process manager pointer
*/
void uv_proc_cleanup(uv_proc_manager_t* manager);
/**
* wait for the process to exit
*
* @param manager process manager pointer
* @param timeout_ms timeout in milliseconds (0 for infinite wait)
* @return 0 on success, non-zero error code on failure
*/
int uv_proc_wait(uv_proc_manager_t* manager, int timeout_ms);
/**
* pre-execute a command in background mode
*
* @param manager process manager pointer
* @param command command to execute
* @param args command arguments
* @param background whether to run in background mode (1 for yes, 0 for no)
* @return 0 on success, non-zero error code on failure
*/
int uv_proc_pre_exec_background(uv_proc_manager_t* manager,
const char* command,
char* const args[],
int background);
/**
* execute a command in single-task mode
*
* @param manager process manager pointer
* @param command command to execute
* @param args command arguments
* @param exit_cb exit callback function
* @param user_data user data passed to the callback
* @return 0 on success, non-zero error code on failure
*/
int uv_proc_exec_single(uv_proc_manager_t* manager,
const char* command,
char* const args[],
uv_proc_exit_cb exit_cb,
void* user_data);
#endif /* UV_PROC_H */

104
src/worker/include/xcecd.h Normal file
View file

@ -0,0 +1,104 @@
#ifndef XCECD_H
#define XCECD_H
#include <stdint.h>
#include <stdlib.h>
#include <stdbool.h>
#include "xcecd_conf.h"
#include "thread_pool.h"
#include "controller_connector.h"
#include "process_manager.h"
#include "system_monitor.h"
#include "hardware_monitor.h"
#include "resource_sync.h"
#include "node_information.h"
#include "job_information.h"
#include "xcecd_worker_utils.h"
#include "xcecd_dbus_manager.h"
#include "static_array.h"
#include "uv_ipc.h"
#include "xcecd_db.h"
#include "xcelog.h"
#include "xcectx.h"
#ifdef _WIN32
#include <windows.h>
#include <winsock2.h>
#include <pdh.h>
typedef HANDLE process_handle_t;
#else
#include <unistd.h>
#include <sys/types.h>
#include <netdb.h>
#include <netinet/in.h>
#include <arpa/inet.h>
typedef pid_t process_handle_t;
#endif
#define REMOTE_PORT 6789
#define DEFAULT_PORT 0 // disable by default
#define BUFFER_SIZE 4096
#define RESTART_FLAG_FILE "/tmp/.xcecd.restarting"
typedef struct {
int thread_pool_size;
int resource_update_interval;
int job_status_update_interval;
uint16_t port;
const char* acct_file;
char* log_file;
const char* db_path;
const char* work_dir;
const char* domain_sock_file;
const char* remote_host;
const char* virutal_name;
int remote_port;
int join_interval;
int update_interval;
int domain_sock;
int debug_level;
} xcecd_config_t;
typedef struct {
thread_pool_t* thread_pool;
network_handler_t* network;
process_manager_t* process_manager;
system_monitor_t* system_monitor;
hardware_monitor_t* hardware_monitor;
resource_sync_t* resource_sync;
xcecd_worker_t* worker_list;
xcecd_db_t *db;
xce_ctx_t* xce_ctx;
bool running;
bool normal_restart;
char* remote_host;
int local_port;
int remote_port;
int update_interval;
int join_interval;
int domain_sock;
int debug_level;
char* work_dir;
char* log_dir;
char* db_path;
char* virtual_name;
} xcecd_context_t;
extern xcecd_context_t xcecd_ctx;
extern hostattrs_t host_attrs;
int xcecd_config_init(xcecd_config_t* config);
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);
void xcecd_prepare_restart(void);
bool xcecd_check_restart_type(void);
#endif // XCECD_H

99
src/worker/include/xcecd_cgroups.h Normal file
View file

@ -0,0 +1,99 @@
#ifndef _XCECD_CGROUPS_H_
#define _XCECD_CGROUPS_H_
#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <string.h>
#include <unistd.h>
#include <systemd/sd-bus.h>
typedef struct {
// CPU limit
uint64_t cpu_quota; // CPU quota in microseconds (e.g. 200000 = 20%, cpu_limit)
uint64_t cpu_period; // CPU period in microseconds (e.g. 1000000 = 1 second)
uint32_t cpu_weight; // CPU weight (e.g. 100 = default)
uint64_t cpu_limit; // CPU limit (run time limit in seconds)
uint64_t core_limit; // Core limit (in number of cores, e.g. 2 = 2 cores)
char *cpu_affinity; // CPU affinity list (e.g. "0,1,2")
// Memory limit
uint64_t memory_max; // Maximum memory usage (rss_limit)
uint64_t memory_high; // Memory high usage level (in bytes)
uint64_t memory_swap_max; // Maximum swap space usage (swap_limit)
// IO limit
uint64_t io_weight; // IO weight (e.g. 100 = default)
uint64_t io_read_bps_max; // Maximum read speed (bytes/second)
uint64_t io_write_bps_max; // Maximum write speed (bytes/second)
char *io_device_path; // IO device path (e.g. "/dev/sda")
// Tasks limit
uint64_t tasks_max; // Max number of threads (thread_limit)
uint64_t run_limit; // Max run time (in seconds)
uint64_t core_size_limit; // Max core file size (in bytes)
uint64_t data_limit; // Max data size (in bytes)
uint64_t fsize_limit; // Max file size (in bytes)
uint64_t stack_limit; // Max stack size (in bytes)
uint64_t process_limit; // Max number of processes
// OOM score
int32_t oom_score_adjust; // OOM (-1000 to 1000)
char *user_name; // User name
} res_limits_t;
/**
* start worker with resource limits
* @param command command with arguments
* @param limits resource limits (NULL if no limits)
* @param unit_name_prefix unit name prefix, NULL for default prefix
* @param job_id job id
* @param extra_args extra command line arguments (NULL if none)
* @return unit name if success, NULL if failed
*/
char* start_systemd_job(const char* command,
const res_limits_t* limits,
const char* unit_name_prefix,
uint64_t job_id,
char* extra_args);
/**
* Get PID of a systemd job
* @param unit_name unit name
* @return PID if success, -1 if failed
*/
pid_t get_systemd_job_pid(const char* unit_name);
/**
* Stop a systemd job
* @param unit_name unit name
* @return 0 if success, -1 if failed
*/
int stop_systemd_job(const char* unit_name);
/**
* Check if a systemd job is active
* @param unit_name unit name
* @return true if active, false if not active
*/
bool is_systemd_job_active(const char* unit_name);
/**
* Get systemd job exit code
* @param unit_name unit name
* @return exit code if success, -1 if failed
*/
int get_systemd_job_exit_code(const char* unit_name);
/**
* @brief Get systemd job status text
* @param unit_name unit name
* @return status text if success, NULL if failed.
* Caller must free the returned string.
*/
char* get_systemd_job_status_text(const char* unit_name);
#endif // _XCECD_CGROUPS_H_

47
src/worker/include/xcecd_conf.h Normal file
View file

@ -0,0 +1,47 @@
#ifndef XCECD_CONF_H
#define XCECD_CONF_H
#include <cyaml/cyaml.h>
#include <stdint.h>
#include <stdbool.h>
#define DEFAULT_CONFIG_PATH "/opt/xce/conf/worker.yaml"
#define MAX_HOSTNAME_LEN 256
#define MAX_PATH_LEN 512
#define MAX_LOG_LEVEL_LEN 16
typedef enum {
CONFIG_SUCCESS = 0,
CONFIG_ERROR_FILE_NOT_FOUND,
CONFIG_ERROR_PARSE_FAILED,
CONFIG_ERROR_INVALID_DATA,
CONFIG_ERROR_MEMORY,
CONFIG_ERROR_UNKNOWN
} config_error_t;
typedef struct {
char *hostname;
} management_host_t;
typedef struct {
management_host_t *management_list;
uint32_t management_list_count;
uint16_t controller_port;
char *log_level;
char *log_dir;
} worker_config_t;
typedef struct {
config_error_t error_code;
char error_message[512];
worker_config_t *config;
} config_result_t;
config_result_t* config_load(const char *config_path);
config_result_t* config_load_default(void);
void config_free(config_result_t *result);
const char* config_error_string(config_error_t error);
void config_print(const worker_config_t *config);
int level_to_int(const char *level);
#endif // XCECD_CONF_H

97
src/worker/include/xcecd_db.h Normal file
View file

@ -0,0 +1,97 @@
#ifndef _XCECD_DB_H
#define _XCECD_DB_H
#include "sqlite3.h"
#include "xcecd_worker_utils.h"
#include "xcecd_utils.h"
#include <stdint.h>
#include <stdbool.h>
typedef struct sqlite3 xcecd_db_t;
/**
* @brief Initialize the database
*
* @param db Pointer to the database structure
* @param db_path Path to the database file
* @return true if success, false if failed
*/
bool xcecd_db_init(xcecd_db_t** db,
const char* db_path);
/**
* @brief close the database
* @param db Pointer to the database structure
*/
void xcecd_db_close(xcecd_db_t* db);
/**
* @brief Insert a worker into the database
*
* @param db Pointer to the database structure
* @param worker_pid PID of the worker
* @param worker_status Status of the worker
* @param job_id Job ID associated with the worker
* @param job_status Status of the job associated with the worker
* @return true if success, false if failed
*/
bool xcecd_db_insert_worker(xcecd_db_t* db,
int worker_pid,
int worker_status,
uint64_t job_id,
int job_status);
/**
* @brief Update a worker in the database
*
* @param db Pointer to the database structure
* @param worker_pid PID of the worker
* @param worker_status Status of the worker
* @param job_status Status of the job associated with the worker
* @return true if success, false if failed
*/
bool xcecd_db_update_worker(sqlite3* db,
int worker_pid,
int worker_status,
int job_status);
/**
* @brief Delete a worker from the database
*
* @param db Pointer to the database structure
* @param worker_pid PID of the worker
* @return true if success, false if failed
*/
bool xcecd_db_delete_worker(xcecd_db_t* db,
int worker_pid);
/**
* @brief Get a worker from the database
*
* @param db Pointer to the database structure
* @param worker_pid PID of the worker
* @return worker status if success, -1 if failed
*/
int xcecd_db_get_worker(xcecd_db_t* db,
int worker_pid);
/**
* @brief Get running workers from the database
*
* @param db Pointer to the database structure
* @return Pointer to the worker list (db->worker_list) if success, NULL if failed
*/
bool xcecd_db_get_running_workers(xcecd_db_t* db,
xcecd_worker_t** worker_list);
/**
* @brief Check if the database is available
* @param db Pointer to the database structure
* @param db_path Path to the database file
* @return void and directly exit if the database is not available
*/
void xcecd_db_is_available(xcecd_db_t* db,
const char* db_path);
#endif

82
src/worker/include/xcecd_dbus_manager.h Normal file
View file

@ -0,0 +1,82 @@
/**
* @file xcecd_dbus_manager.h
* @brief This file contains the declaration of the D-Bus manager functions.
* It uses to manage worker and its child process with resource limits.
*/
#ifndef _XCECD_DBUS_MANAGER_H_
#define _XCECD_DBUS_MANAGER_H_
#include "sqlite3.h"
#include "xcecd_cgroups.h"
#include "xcecd_worker_utils.h"
#include "xcecd_utils.h"
#include <stdbool.h>
/**
* @brief Start worker with resource limits
*
* @param limits resource limits (NULL if no limits)
* @param job_id job id
* @param extra_args extra command line arguments (NULL if none)
* @return true if success, false if failed
*/
bool dbus_start_worker(const res_limits_t* limits,
uint64_t job_id, char* extra_args);
/**
* @brief Stop a worker service
*
* @param unit_name worker service name (xcecd-worker-<job_id>.service)
* @return true if success, false if failed
*/
bool dbus_stop_worker(const char* unit_name);
/**
* @brief Get PID of a worker service
*
* @param unit_name worker service name (xcecd-worker-<job_id>.service)
* @return PID if success, -1 if failed
*/
pid_t dbus_get_worker_pid(const char* unit_name);
/**
* @brief Check if a worker service is active
*
* @param unit_name worker service name (xcecd-worker-<job_id>.service)
* @return true if active, false if not active
*/
bool dbus_is_worker_active(const char* unit_name);
/**
* @brief Update worker status
*
* @param worker_list worker list
* @param job job information
* @param db database information
* @return XCECD_OK if success, others in execd_error if failed
*/
int dbus_workers_update(xcecd_worker_t** worker_list,
xcecd_job_t* job,
sqlite3* db);
/**
* @brief Clean worker service
*
* @param worker worker information
* @param job job information
* @return true if success, false if failed
*/
bool dbus_worker_clean(xcecd_worker_t* worker,
xcecd_job_t* job);
/**
* @brief Go through worker list and stop all workers
* @param worker_list worker list
* @param db database information
* @return void
*/
void dbus_stop_all_workers(xcecd_worker_t** worker_list,
sqlite3* db);
#endif // _XCECD_DBUS_MANAGER_H_

52
src/worker/include/xcecd_exdata.h Normal file
View file

@ -0,0 +1,52 @@
#ifndef _XCECD_EXDATA_H
#define _XCECD_EXDATA_H
#include <cjson/cJSON.h>
#include "xcecd_cgroups.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include <stdbool.h>
typedef struct env_var {
char* key;
char* value;
struct env_var* next;
} env_var_t;
typedef struct {
char* local_file;
char* remote_file;
char* op;
} transfer_file_t;
typedef struct {
transfer_file_t* ops;
int interactive_mode;
int count;
} transfer_info_t;
typedef struct {
char* url;
} start_exdata_t;
bool parse_exdata(const char* exdata,
env_var_t** env_vars,
transfer_info_t* transfer_infos,
uint64_t jobid);
bool parse_limits(const char* exdata,
res_limits_t *limits);
void apply_env_vars(env_var_t** env_vars,
uint64_t jobid);
void free_transfer_info(transfer_info_t* transfer_info);
char* generate_exdata(const start_exdata_t* start_info);
#endif /* _XCECD_EXDATA_H */

127
src/worker/include/xcecd_utils.h Normal file
View file

@ -0,0 +1,127 @@
#ifndef XCECD_UTILS_H
#define XCECD_UTILS_H
#include <stdbool.h>
#include "xce/types.h"
#include "utils.h"
/* This header file has some utilities and structure for compatibility with
* other scheduler systems. */
#define XCE_DEFAULT_SOCKS 15
#define MAXPATHLEN 1024
#define MAXHOSTNAMELEN 256
#define MAXLINELEN 4096
#define MAXDATASIZE (16 * 1024)
#define MAXXCENAMELEN 40
#define MAXSRES 32
#define MAXRESDESLEN 256
#define NBUILTINDEX 11
#define MAXTYPES 128
#define MAXMODELS 128
#define MAXTYPES_31 25
#define MAXMODELS_31 30
#define MAXFILENAMELEN 256
#define MAX_CONCURRENT_JOBS 200
struct windows {
struct windows * nextwind;
float opentime;
float closetime;
};
typedef struct windows windows_t;
enum value_type {LS_BOOLEAN, LS_NUMERIC, LS_STRING, LS_EXTERNAL};
#define BOOLEAN LS_BOOLEAN
#define NUMERIC LS_NUMERIC
#define STRING LS_STRING
#define EXTERNAL LS_EXTERNAL
enum order_type {INCR, DECR, NA};
struct res_item {
char name[MAXXCENAMELEN];
char des[MAXRESDESLEN];
enum value_type value_type;
enum order_type order_type;
int flags;
int interval;
};
struct is_info {
int num_res;
struct res_item *res_table;
int num_types;
char host_types[MAXTYPES][MAXXCENAMELEN];
int num_models;
char host_models[MAXMODELS][MAXXCENAMELEN];
char host_archs[MAXMODELS][MAXXCENAMELEN];
int model_refs[MAXMODELS];
float cpu_factor[MAXMODELS];
int num_indx;
int num_usr_indx;
};
#define GET_INTNUM(i) ((i)/INTEGER_BITS + 1)
enum job_status {
JOB_PENDING,
JOB_STARTING,
JOB_RUNNING,
JOB_SUSPENDED,
JOB_DONE,
JOB_REMOVED,
JOB_EXITED,
JOB_FAILED,
JOB_UNKNOWN
};
enum {
WORKER_RUNNING = 0,
WORKER_DONE,
WORKER_EXITED,
WORKER_UNKNOWN
};
enum execd_error {
XCECD_OK = 0,
XCECD_ERROR,
XCECD_TIMEOUT,
XCECD_NO_MEMORY,
XCECD_INVALID_PARAM,
XCECD_NOT_FOUND,
XCECD_DB_ERROR,
XCECD_NOT_SUPPORTED
};
typedef struct {
uint64_t job_id;
int job_pid;
int job_status;
int exit_code;
int stage;
int interactive_port;
char exdata[MAXDATASIZE];
} xcecd_job_t;
typedef struct {
uint64_t job_id;
pid_t job_pid;
pid_t worker_pid;
int job_status;
int worker_status;
int exit_code;
int stage;
bool post_done;
int interactive_port;
char q_pre_exec[MAXLINELEN];
char q_post_exec[MAXLINELEN];
char job_pre_exec[MAXLINELEN];
char job_post_exec[MAXLINELEN];
char exdata[MAXDATASIZE];
} job_runtime_info_t;
void block_signal(int sig);
void unblock_signal(int sig);
#endif

101
src/worker/include/xcecd_worker_utils.h Normal file
View file

@ -0,0 +1,101 @@
#ifndef XCED_WORKER_UTILS_H
#define XCED_WORKER_UTILS_H
#include "sqlite3.h"
#include "job_information.h"
#include "xcecd_utils.h"
#include "xcelog.h"
#include <stdbool.h>
#include <stdint.h>
#include <time.h>
struct xcecd_worker {
int worker_pid;
int worker_status;
int job_status;
int job_exit_code;
int ntasks;
char *exechome;
char *execcwd;
char *execuser;
char *job_cmd;
char *job_exdata;
char *launch_type;
char *q_pre_exec;
char *q_post_exec;
char *job_pre_exec;
char *job_post_exec;
struct xcecd_worker *prev;
struct xcecd_worker *next;
xce_job_runinfo_t job_info;
#ifdef VIRTUAL_CLUSTER
time_t start_time;
int sleep_time;
int killed_signal;
#endif
};
struct xcecd_db_t;
typedef struct xcecd_worker xcecd_worker_t;
/* TODO: New Worker states */
typedef enum {
SERVICE_UNKNOWN = 0,
SERVICE_ACTIVE,
SERVICE_INACTIVE,
SERVICE_ACTIVATING,
SERVICE_DEACTIVATING,
SERVICE_FAILED,
SERVICE_RELOADING
} service_state_t;
void add_worker(xcecd_worker_t** worker_list,
int worker_pid,
int worker_status,
uint64_t job_id,
int ntasks,
job_startinfo_t* startinfo);
void remove_worker(xcecd_worker_t** worker_list,
int worker_pid);
int update_worker_status(xcecd_worker_t* worker_list,
int worker_pid,
int worker_status,
xcecd_job_t* job,
sqlite3* db);
/* Cleanup the worker in the list which finished work */
void cleanup_worker_list(xcecd_worker_t** worker_list,
sqlite3* db);
xcecd_worker_t* find_worker_by_pid(void* worker_list,
int pid);
xcecd_worker_t* find_worker_by_jobid(xcecd_worker_t* worker_list,
uint64_t jobid);
extern bool xcecd_reply_startjob(xcecd_job_t *job);
extern int xcecd_job_stopped(uint64_t job_id);
extern int xcecd_job_resumed(uint64_t job_id);
extern int xcecd_job_finished(xcecd_job_t *job);
extern bool xcecd_db_update_worker(sqlite3* db,
int worker_pid,
int worker_status,
int job_status);
extern bool xcecd_db_delete_worker(sqlite3* db,
int worker_pid);
bool resend_job_status(xcecd_worker_t* worker_list);
#ifdef VIRTUAL_CLUSTER
bool check_worker_sleep_time_elapsed(xcecd_worker_t* worker);
#endif
#endif // XCED_WORKER_UTILS_H

20
src/worker/include/xrun.h Normal file
View file

@ -0,0 +1,20 @@
#ifndef XRUN_H
#define XRUN_H
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <stdbool.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <errno.h>
int xrun_ipc_send_remote_start(const char *hostname, const char *command);
int run_remote_command(const char *hostname, const char *command);
void setup_signal_handlers(void);
void handle_signal(int sig);
#endif // XRUN_H

86
src/worker/meson.build Normal file
View file

@ -0,0 +1,86 @@
xcecd_worker_srcs = [
'src/xcecd_worker.c',
'src/inter_job_utils.c',
'src/xcecd_exdata.c',
'src/uv_ipc.c',
'src/uv_proc.c',
'src/static_array.c'
]
incdir = include_directories('.', 'include', '../include', '../lib')
libuv_dep = dependency('libuv')
elogind_dep = dependency('libelogind')
sqlite3_dep = dependency('sqlite3')
xcecd_worker_deps = [
libxce_dep,
libuv_dep,
elogind_dep,
sqlite3_dep,
libcjson_dep,
yaml_deps
]
xcecd_worker = executable('xcecd_worker',
xcecd_worker_srcs,
c_args:['-DIPC_WORKER'],
include_directories: [incdir],
dependencies: [xcecd_worker_deps],
install: true)
xrun_srcs = [
'src/xrun.c'
]
xrun_deps = [libxce_dep]
xrun = executable('xrun',
xrun_srcs,
c_args: ['-DXRUN'],
include_directories: [incdir],
dependencies: [xrun_deps],
install: true)
xcecd_srcs = [
'src/static_array.c',
'src/xcecd_conf.c',
'src/inter_job_utils.c',
'src/xcecd_sig.c',
'src/uv_ipc.c',
'src/xcecd_db.c',
'src/xcecd_exdata.c',
'src/xcecd_cgroups.c',
'src/xcecd_dbus_manager.c',
'src/main.c',
'src/xcecd.c',
'src/thread_pool.c',
'src/system_monitor.c',
'src/process_manager.c',
'src/process_worker_manager.c',
'src/connection_handler.c',
'src/controller_connector.c',
'src/resource_sync.c',
'src/xcecd_worker_utils.c',
'src/hardware_monitor.c',
'src/node_information.c',
'src/job_information.c'
]
hwloc_dep = dependency('hwloc')
xcecd_deps = [xcecd_worker_deps,hwloc_dep]
xcecd = executable('xcecd',
xcecd_srcs,
c_args: ['-DDBUS_MANAGER'],
include_directories: [incdir],
dependencies: [xcecd_deps],
install: true)
xcecd_virtual = executable('xcecd_virtual',
xcecd_srcs,
c_args: ['-DVIRTUAL_CLUSTER', '-DDBUS_MANAGER'],
include_directories: [incdir],
dependencies:[xcecd_deps],
install:true)

73
src/worker/scripts/startvnc Executable file
View file

@ -0,0 +1,73 @@
#!/bin/bash
# Script: startvnc
# Purpose: Start TigerVNC server with specified password and port
# Usage: startvnc <port>
# Exit codes: 0 = success, 1 = invalid arguments, 2 = vnc setup failed, 3 = vnc start failed
# Check if exactly 1 argument is provided
if [ $# -ne 1 ]; then
echo "Usage: $0 <port>" >&2
echo "Example: $0 5901" >&2
exit 1
fi
PORT="$1"
# Validate port number
if ! [[ "$PORT" =~ ^[0-9]+$ ]] || [ "$PORT" -lt 1024 ] || [ "$PORT" -gt 65535 ]; then
echo "Error: Port must be a number between 1024 and 65535" >&2
exit 1
fi
# Calculate display number (VNC display = port - 5900)
DISPLAY_NUM=$((PORT - 5900))
if [ $DISPLAY_NUM -lt 1 ]; then
echo "Error: Port must be 5901 or higher for VNC display" >&2
exit 1
fi
# Create VNC directory if it doesn't exist
VNC_DIR="$HOME/.vnc"
mkdir -p "$VNC_DIR"
# Create xstartup file if it doesn't exist
XSTARTUP_FILE="$VNC_DIR/xstartup"
if [ ! -f "$XSTARTUP_FILE" ]; then
cat > "$XSTARTUP_FILE" << 'EOF'
#!/bin/sh
unset SESSION_MANAGER
unset DBUS_SESSION_BUS_ADDRESS
[ -f ~/.Xresources ] && xrdb ~/.Xresources
xsetroot -solid grey
vncconfig -iconic &
# Start Xfce4 desktop
if type startxfce4 >/dev/null 2>&1; then
startxfce4
elif type xfce4-session >/dev/null 2>&1; then
xfce4-session
fi
EOF
chmod +x "$XSTARTUP_FILE"
fi
# Kill any existing VNC server on this display
vncserver -kill ":$DISPLAY_NUM" 2>/dev/null || true
# Start VNC server (remove -passwd parameter as password file is already set)
vncserver ":$DISPLAY_NUM" -PlainUsers=$USER -pam_service=login -SecurityTypes=TLSPlain
if [ $? -ne 0 ]; then
echo "Error: Failed to start VNC server on port $PORT (display :$DISPLAY_NUM)" >&2
exit 3
fi
# Verify the server is running
sleep 2
if ! netstat -ln | grep -q ":$PORT "; then
echo "Warning: VNC server may not be listening on port $PORT" >&2
fi
echo "VNC server started successfully on port $PORT (display :$DISPLAY_NUM)"
echo "Password file: $VNC_DIR/passwd"

85
src/worker/scripts/stopvnc Executable file
View file

@ -0,0 +1,85 @@
#!/bin/bash
# Script: stopvnc
# Purpose: Stop TigerVNC server running on specified port
# Usage: stopvnc <port>
# Exit codes: 0 = success, 1 = invalid arguments, 2 = vnc server not found, 3 = stop failed
# Check if exactly 1 argument is provided
if [ $# -ne 1 ]; then
echo "Usage: $0 <port>" >&2
echo "Example: $0 5900" >&2
exit 1
fi
PORT="$1"
# Validate port number
if ! [[ "$PORT" =~ ^[0-9]+$ ]] || [ "$PORT" -lt 1024 ] || [ "$PORT" -gt 65535 ]; then
echo "Error: Port must be a number between 1024 and 65535" >&2
exit 1
fi
# Calculate display number (VNC display = port - 5900)
DISPLAY_NUM=$((PORT - 5900))
if [ $DISPLAY_NUM -lt 0 ]; then
echo "Error: Port must be 5900 or higher for VNC display" >&2
exit 1
fi
# Check if VNC server is running on this display
VNC_LOCK_FILE="$HOME/.vnc/$(hostname):${DISPLAY_NUM}.pid"
if [ ! -f "$VNC_LOCK_FILE" ]; then
# Also check if process is actually running on the port
if ! netstat -tuln 2>/dev/null | grep -q ":$PORT "; then
echo "No VNC server found running on port $PORT (display :$DISPLAY_NUM)" >&2
exit 2
fi
fi
# Try to kill VNC server gracefully
echo "Stopping VNC server on port $PORT (display :$DISPLAY_NUM)..."
vncserver -kill ":$DISPLAY_NUM"
KILL_RESULT=$?
# Check if the kill command succeeded
if [ $KILL_RESULT -eq 0 ]; then
echo "VNC server on port $PORT stopped successfully"
exit 0
else
# If graceful shutdown failed, try to find and kill the process by port
echo "Graceful shutdown failed, attempting forceful termination..."
# Find process using the port
PID=$(netstat -tulnp 2>/dev/null | grep ":$PORT " | awk '{print $7}' | cut -d'/' -f1)
if [ -n "$PID" ] && [ "$PID" != "-" ]; then
kill -TERM "$PID" 2>/dev/null
sleep 2
# Check if process is still running
if kill -0 "$PID" 2>/dev/null; then
# Force kill if still running
kill -KILL "$PID" 2>/dev/null
sleep 1
fi
# Verify the process is gone
if ! kill -0 "$PID" 2>/dev/null; then
echo "VNC server on port $PORT forcefully terminated"
# Clean up lock file if it exists
if [ -f "$VNC_LOCK_FILE" ]; then
rm -f "$VNC_LOCK_FILE"
fi
exit 0
else
echo "Error: Failed to stop VNC server on port $PORT" >&2
exit 3
fi
else
echo "Error: Could not find process using port $PORT" >&2
exit 2
fi
fi

239
src/worker/src/connection_handler.c Normal file
View file

@ -0,0 +1,239 @@
#include "connection_handler.h"
#include "xcelog.h"
#include <stdlib.h>
#include <string.h>
#include <stdbool.h>
#ifdef _WIN32
#include <winsock2.h>
#include <ws2tcpip.h>
#pragma comment(lib, "ws2_32.lib")
#else
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <fcntl.h>
#include <pthread.h>
#endif
#define BUFFER_SIZE 1024
#define MAX_PENDING 10
struct network_handler {
uint16_t port;
bool running;
request_handler_t callback;
void* user_data;
#ifdef _WIN32
SOCKET server_socket;
HANDLE thread;
#else
int server_socket;
pthread_t thread;
#endif
};
#ifdef _WIN32
static DWORD WINAPI server_thread(LPVOID arg)
#else
static void* server_thread(void* arg)
#endif
{
network_handler_t* handler = (network_handler_t*)arg;
char buffer[BUFFER_SIZE];
while (handler->running) {
#ifdef _WIN32
SOCKET client_socket = accept(handler->server_socket, NULL, NULL);
if (client_socket == INVALID_SOCKET) {
continue;
}
#else
int client_socket = accept(handler->server_socket, NULL, NULL);
if (client_socket < 0) {
continue;
}
#endif
// 读取请求
memset(buffer, 0, BUFFER_SIZE);
#ifdef _WIN32
int bytes_received = recv(client_socket, buffer, BUFFER_SIZE - 1, 0);
if (bytes_received != SOCKET_ERROR && handler->callback) {
#else
int bytes_received = recv(client_socket, buffer, BUFFER_SIZE - 1, 0);
if (bytes_received > 0 && handler->callback) {
#endif
// 调用回调处理请求，并获取响应
const char* result = handler->callback(buffer, handler->user_data);
// 发送响应给客户端
if (result) {
#ifdef _WIN32
send(client_socket, result, (int)strlen(result), 0);
#else
send(client_socket, result, strlen(result), 0);
#endif
free((void*)result); // 释放回调函数分配的内存
} else {
const char* error_msg = "Command execution failed\n";
#ifdef _WIN32
send(client_socket, error_msg, (int)strlen(error_msg), 0);
#else
send(client_socket, error_msg, strlen(error_msg), 0);
#endif
}
}
#ifdef _WIN32
closesocket(client_socket);
#else
close(client_socket);
#endif
}
return 0;
}
network_handler_t* network_handler_create(uint16_t port) {
network_handler_t* handler = calloc(1, sizeof(network_handler_t));
if (!handler) {
return NULL;
}
handler->port = port;
#ifdef _WIN32
// 初始化Winsock
WSADATA wsa_data;
if (WSAStartup(MAKEWORD(2, 2), &wsa_data) != 0) {
free(handler);
return NULL;
}
#endif
return handler;
}
void network_handler_set_callback(network_handler_t* handler, request_handler_t callback, void* user_data) {
if (handler) {
handler->callback = callback;
handler->user_data = user_data;
}
}
int network_handler_start(network_handler_t* handler) {
if (!handler) {
return -1;
}
// 创建服务器socket
#ifdef _WIN32
handler->server_socket = socket(AF_INET, SOCK_STREAM, 0);
if (handler->server_socket == INVALID_SOCKET) {
return -1;
}
#else
handler->server_socket = socket(AF_INET, SOCK_STREAM, 0);
if (handler->server_socket < 0) {
return -1;
}
#endif
// 设置socket选项
int opt = 1;
#ifdef _WIN32
if (setsockopt(handler->server_socket, SOL_SOCKET, SO_REUSEADDR, (char*)&opt, sizeof(opt)) == SOCKET_ERROR) {
closesocket(handler->server_socket);
return -1;
}
#else
if (setsockopt(handler->server_socket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
close(handler->server_socket);
return -1;
}
#endif
// 绑定地址
struct sockaddr_in server_addr;
memset(&server_addr, 0, sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_addr.s_addr = INADDR_ANY;
server_addr.sin_port = htons(handler->port);
#ifdef _WIN32
if (bind(handler->server_socket, (struct sockaddr*)&server_addr, sizeof(server_addr)) == SOCKET_ERROR) {
closesocket(handler->server_socket);
return -1;
}
#else
if (bind(handler->server_socket, (struct sockaddr*)&server_addr, sizeof(server_addr)) < 0) {
close(handler->server_socket);
return -1;
}
#endif
// 开始监听
#ifdef _WIN32
if (listen(handler->server_socket, MAX_PENDING) == SOCKET_ERROR) {
closesocket(handler->server_socket);
return -1;
}
#else
if (listen(handler->server_socket, MAX_PENDING) < 0) {
close(handler->server_socket);
return -1;
}
#endif
handler->running = true;
// 创建服务器线程
#ifdef _WIN32
handler->thread = CreateThread(NULL, 0, server_thread, handler, 0, NULL);
if (!handler->thread) {
closesocket(handler->server_socket);
return -1;
}
#else
if (pthread_create(&handler->thread, NULL, server_thread, handler) != 0) {
close(handler->server_socket);
return -1;
}
#endif
return 0;
}
void network_handler_stop(network_handler_t* handler) {
if (!handler) {
return;
}
handler->running = false;
#ifdef _WIN32
closesocket(handler->server_socket);
WaitForSingleObject(handler->thread, INFINITE);
CloseHandle(handler->thread);
#else
close(handler->server_socket);
pthread_join(handler->thread, NULL);
#endif
}
void network_handler_destroy(network_handler_t* handler) {
if (!handler) {
return;
}
if (handler->running) {
network_handler_stop(handler);
}
#ifdef _WIN32
WSACleanup();
#endif
free(handler);
}

1561
src/worker/src/controller_connector.c Executable file
View file

File diff suppressed because it is too large Load diff

162
src/worker/src/hardware_monitor.c Normal file
View file

@ -0,0 +1,162 @@
#include "hardware_monitor.h"
#include "xcelog.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <hwloc.h>
#include <time.h>
#include <errno.h>
#ifdef _WIN32
#include <windows.h>
#else
#include <unistd.h>
#include <pthread.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/sysinfo.h>
#endif
struct hardware_monitor {
char* acct_file;
int last_cpu_cores;
double total_memory;
double free_memory;
time_t last_update_time;
};
static void write_to_acct_file(hardware_monitor_t* monitor, int cpu_cores) {
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);
}
hardware_monitor_t* hardware_monitor_create(const char* acct_file) {
hardware_monitor_t* monitor = calloc(1, sizeof(hardware_monitor_t));
if (!monitor) {
return NULL;
}
monitor->acct_file = strdup(acct_file);
monitor->last_cpu_cores = -1;
monitor->last_update_time = 0;
monitor->total_memory = 0;
monitor->free_memory = 0;
XCECD_LOG(XCE_LOG_INFO, "Hardware monitor created.");
return monitor;
}
void hardware_monitor_destroy(hardware_monitor_t* monitor) {
if (monitor) {
free(monitor->acct_file);
free(monitor);
}
}
bool hardware_monitor_update(hardware_monitor_t* monitor) {
if (!monitor) {
return false;
}
hwloc_topology_t topology;
if (hwloc_topology_init(&topology) < 0 ||
hwloc_topology_load(topology) < 0) {
return false;
}
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;
}
int64_t total_memory, free_memory;
if (hardware_monitor_get_memory_info(&total_memory, &free_memory)) {
monitor->total_memory = total_memory;
monitor->free_memory = free_memory;
}
hwloc_topology_destroy(topology);
monitor->last_update_time = time(NULL);
return true;
}
int hardware_monitor_get_cpu_cores(hardware_monitor_t* monitor) {
if (!monitor) {
return -1;
}
return monitor->last_cpu_cores;
}
#ifdef _WIN32
bool hardware_monitor_get_memory_info(int64_t* total_memory, int64_t* free_memory) {
if (!total_memory || !free_memory) {
return false;
}
MEMORYSTATUSEX status;
status.dwLength = sizeof(status);
if (!GlobalMemoryStatusEx(&status)) {
XCECD_LOG(XCE_LOG_ERROR, "Failed to get memory status: %d", GetLastError());
return false;
}
*total_memory = status.ullTotalPhys;
*free_memory = status.ullAvailPhys;
return true;
}
#else
bool hardware_monitor_get_memory_info(int64_t* total_memory, int64_t* free_memory) {
if (!total_memory || !free_memory) {
return false;
}
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;
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);
return true;
}
#endif
double hardware_monitor_get_total_memory(hardware_monitor_t* monitor) {
if (!monitor) {
return 0;
}
return monitor->total_memory;
}
double hardware_monitor_get_free_memory(hardware_monitor_t* monitor) {
if (!monitor) {
return 0;
}
return monitor->free_memory;
}

327
src/worker/src/inter_job_utils.c Normal file
View file

@ -0,0 +1,327 @@
#include "inter_job_utils.h"
#include "xcecd_utils.h"
#include "xcelog.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <ifaddrs.h>
#include <unistd.h>
/**
* @brief Check if a port is available for binding.
* @param port_number The port number to check.
* @return 0 if the port is available, 1 if it is in use,
* -1 if an error occurred.
*/
int check_port_availability(int port_number) {
int sockfd;
struct sockaddr_in server_addr;
if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Error creating socket");
return -1;
}
memset(&server_addr, 0, sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(port_number);
server_addr.sin_addr.s_addr = INADDR_ANY;
if (bind(sockfd, (struct sockaddr *)&server_addr, sizeof(server_addr)) == 0) {
// successfully bound to the port
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Port %d is available", port_number);
close(sockfd);
return 0;
} else {
if (errno == EADDRINUSE) {
// Error: Address already in use
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
"Port %d is already in use", port_number);
close(sockfd);
return 1;
} else {
// Other error occurred
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
"Error binding socket: %s", strerror(errno));
close(sockfd);
return -1;
}
}
}
/**
* @brief Find an available port within a specified range.
* @param start_port The starting port number of the range.
* @param end_port The ending port number of the range.
* @return An available port number if found, -1 if no ports are available,
* or -2 if an error occurred (e.g., invalid range).
*/
int find_available_port(int start_port, int end_port) {
if (start_port < 1 || end_port > 65535 || start_port > end_port) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Invalid port range: %d - %d\n",
start_port, end_port);
return -2;
}
for (int port = start_port; port <= end_port; ++port) {
int status = check_port_availability(port);
if (status == 0) {
return port;
} else if (status == 1) {
continue; // Port is in use, try next one
} else if (status == -2) {
// system error occurred
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"System error occurred while checking port %d. Aborting scan.\n", port);
return -2;
}
}
return -1;
}
/**
* @brief Generate a command string based on the job command and port.
* @param job_cmd The command to execute.
* @param port info to use for the job.
* @param interactive_mode The mode of interaction (TTY, DISPLAY, XJOB, etc.).
* @return A dynamically allocated command string, or NULL on failure.
*/
char* generate_cmd(char* job_cmd, port_info_t port, int interactive_mode) {
if (!job_cmd) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Job command is NULL");
return NULL;
}
if (interactive_mode == NORMAL) {
return strdup(job_cmd);
}
char* cmd = (char*)calloc(1, CMD_MAX_LENGTH);
if (!cmd) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Memory allocation failed for command string");
return NULL;
}
char launcher[MAXFILENAMELEN] = {0};
char exe_path[MAXFILENAMELEN - 5] = {0};
ssize_t len = readlink("/proc/self/exe", exe_path, sizeof(exe_path) - 1);
if (len > 0) {
exe_path[len] = '\0';
char* last_slash = strrchr(exe_path, '/');
if (last_slash) {
*last_slash = '\0';
snprintf(launcher, sizeof(launcher), "%s/ttyd", exe_path);
} else {
snprintf(launcher, sizeof(launcher), "/opt/xce/bin/ttyd");
}
} else {
snprintf(launcher, sizeof(launcher), "/opt/xce/bin/ttyd");
}
if (access(launcher, X_OK) != 0) {
if (interactive_mode == TTY ||
interactive_mode == DISPLAY) {
// For interactive modes, if ttyd is not found, return error
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"ttyd not found or not executable at %s, cannot run interactive job",
launcher);
free(cmd);
return NULL;
}
}
if (interactive_mode == TTY) {
snprintf(cmd, CMD_MAX_LENGTH,
"%s -WQ -p %d %s",
launcher, port.tty_port, job_cmd);
} else if (interactive_mode == DISPLAY) {
snprintf(cmd, CMD_MAX_LENGTH,
"%s -p %d -Q %s",
launcher, port.tty_port, job_cmd);
} else if (interactive_mode == XJOB) {
int dport = port.xjob_port - 5900; // Adjust port for VNC
if (dport < 1 || dport > 99) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Invalid XJOB port %d, must be in range 5901-5999",
port.xjob_port);
free(cmd);
return NULL;
}
snprintf(cmd, CMD_MAX_LENGTH, "%s", job_cmd);
} else {
// Default case for normal command execution
free(cmd);
return strdup(job_cmd);
}
return cmd;
}
/**
* @brief Get the local IP address of the machine.
* @return A dynamically allocated string containing the IP address,
* or NULL on failure. Caller should free the returned string.
*/
static char* get_local_ip() {
struct ifaddrs *ifaddrs_ptr = NULL;
struct ifaddrs *ifa = NULL;
char* ip_str = NULL;
if (getifaddrs(&ifaddrs_ptr) == -1) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to get network interfaces");
return NULL;
}
for (ifa = ifaddrs_ptr; ifa != NULL; ifa = ifa->ifa_next) {
if (ifa->ifa_addr == NULL) continue;
if (ifa->ifa_addr->sa_family == AF_INET &&
strcmp(ifa->ifa_name, "lo") != 0 &&
strcmp(ifa->ifa_name, "lo0") != 0) {
struct sockaddr_in* addr = (struct sockaddr_in*)ifa->ifa_addr;
char* temp_ip = inet_ntoa(addr->sin_addr);
if (strncmp(temp_ip, "169.254.", 8) != 0) {
ip_str = strdup(temp_ip);
break;
}
}
}
freeifaddrs(ifaddrs_ptr);
if (!ip_str) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Could not find valid IP address.");
}
return ip_str;
}
/**
* @brief Generate a URL based on the port and interaction mode.
* The url will be in the format:
* "http://<hostip>:<port>"
* @param port The port information to use.
* @param interactive_mode The mode of interaction (TTY, DISPLAY, XJOB, etc.).
* @return A dynamically allocated URL string, or NULL on failure.
*/
char* generate_url(port_info_t port, int interactive_mode) {
if (interactive_mode < 0 || interactive_mode > 3) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Invalid interactive mode specified");
return NULL; // Invalid mode
}
if (interactive_mode == NORMAL) {
return NULL; // No URL for normal command execution
}
char* url = (char*)calloc(1, URL_MAX_LENGTH);
if (!url) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Memory allocation failed for URL string");
return NULL;
}
char* local_ip = get_local_ip();
if (!local_ip) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Failed to get local IP address");
free(url);
return NULL;
}
if (interactive_mode == TTY || interactive_mode == DISPLAY) {
snprintf(url, CMD_MAX_LENGTH, "http://%s:%d",
local_ip, port.tty_port);
} else if (interactive_mode == XJOB) {
snprintf(url, CMD_MAX_LENGTH, "%s::%d",
local_ip, port.xjob_port);
} else {
free(url);
free(local_ip);
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Invalid interactive mode specified");
return NULL; // No URL for normal command execution
}
free(local_ip);
return url;
}
/**
* @brief Check if a port is available for the given port information.
* This function will call find_available_port to check
* if the port is available for the specified interaction mode.
* @param port The port information to check.
* @param interactive_mode The mode of interaction (TTY, DISPLAY, XJOB, etc.).
* @return port number if available, -1 if not available,
* or -2 if an error occurred.
*/
int check_port(port_info_t port, int interactive_mode) {
if (interactive_mode < 0 || interactive_mode > 3) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Invalid interactive mode specified");
return -2; // Invalid mode
}
if (interactive_mode == NORMAL) {
// For normal mode, we don't need to check a specific port
return 0; // No port needed for normal execution
}
if (port.tty_port == TTY_END_PORT) {
port.tty_port = 9000; // Default TTY port
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
"TTY port maxed, using default: %d", port.tty_port);
}
if (port.xjob_port == XJOB_END_PORT) {
port.xjob_port = 5901; // Default XJOB port
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
"XJOB port maxed, using default: %d", port.xjob_port);
}
if (interactive_mode == TTY) {
return find_available_port(port.tty_port, TTY_END_PORT);
} else if (interactive_mode == DISPLAY) {
return find_available_port(port.tty_port, TTY_END_PORT);
} else if (interactive_mode == XJOB) {
return find_available_port(port.xjob_port, XJOB_END_PORT);
} else {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Invalid interactive mode specified");
return -2; // Invalid mode
}
}
/**
* @brief This function accepts a xjob port number and exports it
* as DISPLAY environment variable.
* @param port The port number to export as DISPLAY.
* @return 0 on success, -1 on failure.
*/
int export_display_env(int port) {
char display_env[32];
snprintf(display_env, sizeof(display_env),
":%d", port - 5900);
if (setenv("DISPLAY", display_env, 1) != 0) {
XCECD_WORKER_LOG(XCE_LOG_ERROR,
"Failed to set DISPLAY environment variable");
return -1;
}
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
"Exported DISPLAY environment variable: %s", display_env);
return 0;
}

105
src/worker/src/job_information.c Normal file
View file

@ -0,0 +1,105 @@
#include "job_information.h"
#include "xcecd_utils.h"
#include "xcelog.h"
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
// 初始化job_setup结构体
void init_job_setup(struct job_setup* setup, const char* data) {
if (!setup || !data) {
return;
}
// 假设data是一个JSON字符串，解析并初始化job_setup结构体
// 这里使用简单的字符串解析，实际应用中建议使用JSON解析库
sscanf(data, "{\"job_id\":%ld,\"job_status\":%d,\"cpu_time\":%f,\"w_status\":%d}",
&setup->job_id, &setup->job_status, &setup->cpu_time, &setup->w_status);
// 初始化其他字段
setup->reason = 0;
setup->job_pid = 0;
setup->job_pgid = 0;
setup->exec_gid = 0;
setup->exec_uid = 0;
strcpy(setup->exec_user_name, "user01");
strcpy(setup->exec_home, "/home/user01");
strcpy(setup->exec_cwd, "/home/user01");
setup->exec_job_flag = 0;
}
// 初始化job_card结构体
void init_job_card(struct job_card* card, const char* data) {
if (!card || !data) {
return;
}
// 假设data是一个JSON字符串，解析并初始化job_card结构体
// 这里使用简单的字符串解析，实际应用中建议使用JSON解析库
sscanf(data, "{\"exec_user_name\":\"%[^\"]\",\"exec_gid\":%d,\"cpu_time\":%f}",
card->exec_user_name, &card->exec_gid, &card->cpu_time);
// 初始化其他字段
card->exec_job_flag = 0;
card->not_reported = 0;
card->wind_edge = 0;
card->active = 0;
card->time_expire = 0;
card->missing = 0;
card->controller_restarted = 0;
card->wind_warn_time = 0;
card->runtime = 0;
card->w_status = 0;
card->need_report_rusage = 0;
card->cleanup_pid = 0;
card->collected_child = 0;
card->stdin_file = NULL;
card->last_controller_time = 0;
card->delievered_message_id = 0;
card->client = NULL;
card->reg_op_flag = 0;
card->job_sup_status = 0;
card->act_flags = 0;
card->act_status = 0;
card->act_reasons = 0;
card->act_sub_reasons = 0;
card->exit_pid = 0;
card->job_done = 0;
card->last_check = 0;
card->act_cmd = NULL;
card->exit_file = NULL;
card->cluster_name = NULL;
card->server_socket = 0;
card->cross_platforms = 0;
card->spooled_exec = NULL;
card->post_job_started = 0;
card->user_job_succeed = 0;
}
void cleanup_job_card(struct job_card* card) {
if (!card) {
return;
}
FREEUP(card->stdin_file);
FREEUP(card->act_cmd);
FREEUP(card->exit_file);
FREEUP(card->cluster_name);
FREEUP(card->spooled_exec);
}
struct client_node* init_client_node(long id) {
struct client_node* node = (struct client_node*)calloc(1, sizeof(struct client_node));
if (!node) {
return NULL;
}
node->job_id = id;
node->job_type = 0;
node->jp = NULL;
node->prev = NULL;
node->next = NULL;
return node;
}

542
src/worker/src/main.c Normal file
View file

@ -0,0 +1,542 @@
#include "xcecd.h"
#include "sqlite3.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <sys/select.h>
#include <errno.h>
#ifdef _WIN32
#include <windows.h>
#else
#include <unistd.h>
#include <fcntl.h>
#include <signal.h>
#include <sys/wait.h>
#include <sys/types.h>
#endif
/* TEMP SOLUTION: use config later */
#define LOG_SIZE (1024 * 1024 * 100) // 100 MB
#define LOG_LEVEL 2 // DEBUG level
extern xcecd_context_t xcecd_ctx;
volatile sig_atomic_t child_exited = 0;
/* job status using for communication with controller, should be a list? */
static xcecd_job_t job = {
.job_id = 0,
.job_pid = 0,
.job_status = JOB_STARTING,
.exit_code = 0,
.stage = JOB_EXEC_STAGE_MAIN,
.interactive_port = 0,
.exdata = {0}
};
static StaticArray pids;
static void signal_handler(int signo) {
if (signo == SIGINT || signo == SIGTERM) {
XCECD_LOG(XCE_LOG_WARN,
"Received signal %d, shutting down...",
signo);
exit_flag = 1;
/* mark to normal restart */
xcecd_prepare_restart();
#if 0
/* kill all jobs and update DB */
dbus_stop_all_workers(&xcecd_ctx.worker_list,
(sqlite3*)xcecd_ctx.db);
#endif
xcecd_ctx.running = false;
xcecd_stop();
}
}
static void alarm_handler(int sig) {
if (sig == SIGALRM) {
exit_flag = 1;
XCECD_LOG(XCE_LOG_ERROR,
"Program running timeout, force exit");
kill(0, SIGKILL);
_exit(1);
}
}
static void sigchld_handler(int signo) {
(void)signo;
int status;
pid_t pid;
int i = 0;
child_exited = 1;
while ((pid = waitpid(-1, &status, WNOHANG)) > 0) {
if (WIFEXITED(status) || WIFSIGNALED(status)) {
if (i++ < MAX_ARRAY_SIZE) {
/* just recording the exited pids and handle them in main loop */
if (!push_static(&pids, &pid)) {
XCECD_LOG(XCE_LOG_ERROR,
"Fatal error: failed to store exit pid %d",
pid);
}
} else {
/* reach array max limit and handle rest pid later */
XCECD_LOG(XCE_LOG_DEBUG,
"Too many workers exited, pid %d. Please reduce your slots on this host.",
pid);
break;
}
}
}
}
static void handle_exited_children(process_manager_t* manager,
StaticArray pids)
{
int pid;
for (int i = 0; i < pids.count ; i++) {
if (!get_element(&pids, i, &pid)) {
XCECD_LOG(XCE_LOG_ERROR,
"handle_exited_children: failed to get pid");
}
process_manager_remove_process(manager, pid);
}
clear_static(&pids);
}
static void job_status_cleanup(xcecd_job_t* job) {
job->job_id = 0;
job->job_pid = 0;
job->job_status = JOB_UNKNOWN;
job->exit_code = 0;
job->stage = JOB_EXEC_STAGE_MAIN;
job->interactive_port = 0;
memset(job->exdata, 0, sizeof(job->exdata));
}
static void handle_message_from_queue(void) {
thread_pool_t* pool = xcecd_ctx.thread_pool;
fork_task_t task;
if (queue_pop(&pool->fork_queue, &task) == 0) {
/* handle fork task */
task.func(&task);
}
}
static void master_disconnect_callback(uint64_t job_id) {
XCECD_LOG(XCE_LOG_DEBUG,
"Worker with client ID %lu disconnected",
job_id);
}
static void master_job_status_callback(job_status_t* job_status)
{
int ret = XCECD_OK;
XCECD_LOG(XCE_LOG_INFO,
"Job %lu status: %d, exdata: %s, %s",
job_status->job_id,
job_status->status_code,
job_status->start_exdata[0] ? job_status->start_exdata : "no exdata",
job_status->status_message);
job.job_status = job_status->status_code;
job.job_pid = job_status->job_pid;
job.exit_code = job_status->exit_code;
strncpy(job.exdata,
job_status->start_exdata,
sizeof(job.exdata) - 1);
job.exdata[sizeof(job.exdata) - 1] = '\0';
if ((job_status->stage == JOB_EXEC_STAGE_POST_EXEC) ||
(job_status->stage == JOB_EXEC_STAGE_Q_POST_EXEC)) {
/* Post-exec status update */
if (job_status->post_done) {
xcecd_jobpost_done(job_status->job_id);
} else {
xcecd_notify_job_fail(job_status->job_id,
job_status->stage,
job_status->exit_code);
}
job_status_cleanup(&job);
return;
}
/* update worker status and send job status to controller */
ret = update_worker_status(xcecd_ctx.worker_list,
job_status->worker_pid,
job_status->worker_status,
&job,
(sqlite3*)xcecd_ctx.db);
if (ret == XCECD_DB_ERROR) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to update worker status in database");
dbus_stop_all_workers(&xcecd_ctx.worker_list,
NULL);
xcecd_db_is_available(xcecd_ctx.db,
xcecd_ctx.db_path);
}
job_status_cleanup(&job);
}
static void master_heartbeat_callback(uv_stream_t* client) {
XCECD_LOG(XCE_LOG_DEBUG,
"Received heartbeat from worker, sending response.");
uv_write_t* req = (uv_write_t*)malloc(sizeof(uv_write_t));
if (req) {
ipc_message_t msg;
memset(&msg, 0, sizeof(msg));
msg.type = MSG_HEARTBEAT;
uv_buf_t send_buf = uv_buf_init((char*)&msg, sizeof(ipc_message_t));
uv_write(req, client, &send_buf, 1, on_write_complete);
}
}
static void master_ckpt_result_callback(uint64_t job_id,
int status,
const char* restore_cmd)
{
XCECD_LOG(XCE_LOG_DEBUG,
"Received checkpoint result for job %lu: status %d, restore_cmd: %s",
job_id, status, restore_cmd ? restore_cmd : "NULL");
if (job_id <= 0) {
XCECD_LOG(XCE_LOG_ERROR, "Job ID is not correct.");
xcecd_job_ckpted(job_id, 1, NULL);
return;
}
xcecd_job_ckpted(job_id, status, restore_cmd);
}
int main(int argc, char* argv[]) {
int opt;
config_result_t *config_result = NULL;
int ret = XCECD_OK;
int config_flag = 0;
char* custom_config_path = NULL;
__attribute__((unused)) char worker_log[1024] = {0};
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);
struct sigaction sa_chld;
memset(&sa_chld, 0, sizeof(sa_chld));
sa_chld.sa_handler = sigchld_handler;
sa_chld.sa_flags = SA_NOCLDSTOP | SA_NOCLDWAIT;
sigaction(SIGCHLD, &sa_chld, NULL);
xcecd_config_t config = {
.thread_pool_size = 1,
.job_status_update_interval = 1,
.port = 0,
.acct_file = "/tmp/xcecd.acct",
.log_file = "/tmp/xcecd.log",
.db_path = "/var/run/xcecd.db",
.work_dir = "/tmp",
.domain_sock_file = "/var/run/.xcecd.sock",
.virutal_name = "localhost01",
.remote_host = "localhost",
.remote_port = 6789,
.join_interval = 5,
.update_interval = 5,
.resource_update_interval = 1,
.debug_level = XCE_LOG_INFO
};
char logfile[1024] = {0};
char *logdir = NULL;
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "-c") == 0 && i + 1 < argc) {
custom_config_path = argv[i + 1];
config_flag = 1;
break;
}
}
if (config_flag == 0) {
custom_config_path = strdup(DEFAULT_CONFIG_PATH);
}
if (access(custom_config_path, F_OK) == 0) {
/* load config from custom file */
config_result = config_load(custom_config_path);
if (!config_result) {
fprintf(stderr, "Failed to load config from %s\n", custom_config_path);
return -1;
}
if (config_result->error_code == CONFIG_SUCCESS) {
if (config_result->config->controller_port > 0) {
config.remote_port = config_result->config->controller_port;
}
if (config_result->config->log_dir) {
logdir = strdup(config_result->config->log_dir);
}
if (config_result->config->log_level) {
config.debug_level = level_to_int(config_result->config->log_level);
}
if (config_result->config->management_list_count > 0) {
config.remote_host = config_result->config->management_list[0].hostname;
}
}
} else {
fprintf(stderr, "Configuration file %s not found.\n", custom_config_path);
return -1;
}
while ((opt = getopt(argc, argv, "c:p:m:n:l:h")) != -1) {
switch (opt) {
case 'c':
break; // Custom config file already handled above
case 'p':
if (config_flag) {
fprintf(stderr, "Custom config file already specified, ignoring -p option.\n");
break;
}
config.remote_port = atoi(optarg);
break;
case 'm':
if (config_flag) {
fprintf(stderr, "Custom config file already specified, ignoring -m option.\n");
break;
}
config.remote_host = optarg;
break;
case 'n':
config.virutal_name = optarg;
break;
case 'l':
if (config_flag) {
fprintf(stderr, "Custom config file already specified, ignoring -l option.\n");
break;
}
if (optarg) {
if (logdir) {
FREEUP(logdir);
}
logdir = strdup(optarg);
}
break;
case 'h':
default:
#ifdef VIRTUAL_CLUSTER
printf("Usage: %s -n virtual_name [-p remote_port] [-m remote_host] [-l log_dir] [-h help]\n", argv[0]);
#else
printf("Usage: %s [-p remote_port] [-m remote_host] [-l log_dir ] [-h help]\n", argv[0]);
#endif
return -1;
}
}
struct stat st = {0};
if (stat(logdir, &st) == -1) {
if (mkdir(logdir, 0755) != 0) {
fprintf(stderr, "Failed to create log directory %s: %s\n",
logdir, strerror(errno));
FREEUP(logdir);
return -1;
}
}
chmod(logdir, 0777);
#ifdef VIRTUAL_CLUSTER
snprintf(logfile, sizeof(logfile), "%s/xcecd-%s.log",
logdir ? logdir : "/tmp", config.virutal_name);
#else
snprintf(logfile, sizeof(logfile), "%s/xcecd.log",
logdir ? logdir : "/tmp");
/* we need create worker log here and set it to 0666 */
snprintf(worker_log, sizeof(worker_log), "%s/xcecd_worker.log",
logdir ? logdir : "/tmp");
#endif
if (logdir) {
FREEUP(logdir);
}
config.log_file = logfile;
/* array to store exit worker pids */
init_static_array(&pids, sizeof(int));
#ifndef _WIN32
#if 0
pid_t pid = fork();
if (pid < 0) {
fprintf(stderr, "First fork failed\n");
return 1;
}
if (pid > 0) {
exit(0);
}
if (setsid() < 0) {
fprintf(stderr, "setsid failed\n");
return 1;
}
#endif
char current_dir[1024];
getcwd(current_dir, sizeof(current_dir));
config.work_dir = current_dir;
chdir("/");
setbuf(stdout, NULL);
//int fd = open(config.log_file, O_WRONLY | O_CREAT | O_APPEND, 0644);
int fd = open("/home/user01/xcecd.log", O_WRONLY | O_CREAT | O_APPEND, 0644);
if (fd != -1) {
dup2(fd, STDOUT_FILENO);
dup2(fd, STDERR_FILENO);
close(fd);
}
#endif
if (xce_init_xcecd_log(config.log_file, LOG_SIZE, config.debug_level) != 0) {
/* FIXME: write error message to file. */
fprintf(stderr, "Failed to initialize log module.");
return 1;
}
#ifndef VIRTUAL_CLUSTER
/* create worker log file */
int worker_log_fd = open(worker_log, O_WRONLY | O_CREAT | O_APPEND, 0644);
if (worker_log_fd != -1) {
XCECD_LOG(XCE_LOG_INFO,
"Worker log file created: %s", worker_log);
}
/* close it and wait real worker to use */
close(worker_log_fd);
chmod(worker_log, 0666);
#endif
if (xcecd_config_init(&config) != 0) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to initialize daemon configuration.");
return 1;
}
if (xcecd_init(&config) != 0) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to initialize XCECD daemon.");
return 1;
}
if (xcecd_start() != 0) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to start XCECD daemon.");
xcecd_cleanup();
return 1;
}
#ifndef VIRTUAL_CLUSTER
if (!ipc_master_init(config.domain_sock_file)) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to initialize XCECD IPC.");
xcecd_cleanup();
return 1;
}
chmod(config.domain_sock_file, 0666);
#endif
ipc_master_set_status_callback(master_job_status_callback);
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);
XCECD_LOG(XCE_LOG_INFO, "XCECD started successfully.");
/* waiting for signals and housekeeping */
while (xcecd_ctx.running) {
/* cleanup must following with worker update */
cleanup_worker_list(&xcecd_ctx.worker_list,
(sqlite3*)xcecd_ctx.db);
handle_message_from_queue();
usleep(10 * 1000);
/* job should be running now */
ipc_master_run(UV_RUN_NOWAIT);
#ifndef DBUS_MANAGER
worker_process_update(xcecd_ctx.process_manager,
xcecd_ctx.worker_list,
&job,
(sqlite3*)xcecd_ctx.db);
#else
ret = dbus_workers_update(&xcecd_ctx.worker_list,
&job,
(sqlite3*)xcecd_ctx.db);
if (ret == XCECD_DB_ERROR) {
XCECD_LOG(XCE_LOG_ERROR,
"Failed to update worker status in database");
dbus_stop_all_workers(&xcecd_ctx.worker_list,
NULL);
xcecd_db_is_available(xcecd_ctx.db,
xcecd_ctx.db_path);
}
#endif
ipc_master_run(UV_RUN_NOWAIT);
/* handling child process exit in main process to prevent deadlock */
if (child_exited) {
handle_exited_children(xcecd_ctx.process_manager,
pids);
child_exited = 0;
}
}
xcecd_cleanup();
unlink(config.domain_sock_file);
XCECD_LOG(XCE_LOG_INFO, "XCECD is down.");
alarm(5);
xce_log_fini();
exit(0);
}

88
src/worker/src/node_information.c Normal file
View file

@ -0,0 +1,88 @@
#include "node_information.h"
#include "xcelog.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdbool.h>
#include <errno.h>
#include <pthread.h>
// 获取当前CPU使用情况
int get_cpu_usage(int total_cores, int used_cores) {
if (used_cores == 0) {
return total_cores;
}
if (used_cores >= total_cores) {
return 0;
}
if (used_cores > 0 && used_cores < total_cores) {
return total_cores - used_cores;
}
// 其它状况
return 0;
}
host_node_t* host_node_create(void) {
host_node_t* node_ptr = (host_node_t*)calloc(1, sizeof(host_node_t));
if (node_ptr == NULL) {
return NULL;
}
node_ptr->node_name = NULL;
node_ptr->node_number = 0;
node_ptr->node_model_number = 0;
node_ptr->node_type_number = 0;
node_ptr->node_ip = NULL;
node_ptr->node_status.node_id = 0;
node_ptr->node_status.max_cpu_cores = 0;
node_ptr->node_status.max_memory = 0;
node_ptr->node_status.num_disk = 0;
node_ptr->node_status.max_swap = 0;
node_ptr->node_status.max_network = 0;
node_ptr->node_status.port_number = 0;
/* TODO: initialize load_index and user_load_index */
node_ptr->use = -1;
node_ptr->expire_time = -1;
return node_ptr;
}
int host_node_update(host_node_t* node_ptr, host_node_t* new_node) {
if (node_ptr == NULL || new_node == NULL) {
return -1;
}
/* TODO: add implementation here */
return 0;
}
void free_host_node(host_node_t* node_ptr) {
if (node_ptr == NULL) {
return;
}
FREEUP(node_ptr->node_name);
FREEUP(node_ptr->node_ip);
FREEUP(node_ptr->windows);
for (int i = 0; i < 8; i++) {
FREEUP(node_ptr->week[i]);
}
FREEUP(node_ptr->load_index);
FREEUP(node_ptr->user_load_index);
FREEUP(node_ptr->status);
FREEUP(node_ptr->busy_threshold);
FREEUP(node_ptr->resource_bitmap);
FREEUP(node_ptr->dynamic_resource_bitmap);
FREEUP(node_ptr);
}

586
src/worker/src/process_manager.c Normal file
View file

@ -0,0 +1,586 @@
#include "process_manager.h"
#include "xcecd_utils.h"
#include "xcelog.h"
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <stdbool.h>
#include <time.h>
#include <errno.h>
#ifdef _WIN32
#include <windows.h>
#include <io.h>
#include <fcntl.h>
#else
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <fcntl.h>
#endif
#define MAX_PROCESSES 1024
#define BUFFER_SIZE 4096
void process_info_init(process_info_t* proc) {
proc->handle = 0;
proc->command = NULL;
proc->status = 0;
proc->exit_code = 0;
proc->output_fd = -1;
proc->log_file = NULL;
#ifdef _WIN32
proc->pipe_read = NULL;
proc->monitor_thread = NULL;
proc->monitor_active = false;
#endif
XCECD_LOG(XCE_LOG_DEBUG, "Process info initialized");
}
#ifdef _WIN32
// 监控线程函数，用于读取进程输出并写入日志文件
static DWORD WINAPI output_monitor_thread(LPVOID arg) {
process_info_t* proc = (process_info_t*)arg;
char buffer[BUFFER_SIZE];
DWORD bytes_read;
while (proc->monitor_active) {
if (ReadFile(proc->pipe_read, buffer, BUFFER_SIZE - 1, &bytes_read, NULL) && bytes_read > 0) {
buffer[bytes_read] = '\0';
_write(proc->output_fd, buffer, bytes_read);
} else {
if (GetLastError() != ERROR_NO_DATA) {
break;
}
Sleep(100); // 避免过度消耗 CPU
}
}
return 0;
}
#endif
process_manager_t* process_manager_create(void) {
process_manager_t* manager = calloc(1, sizeof(process_manager_t));
if (!manager) {
return NULL;
}
manager->processes = NULL;
manager->process_count = 0;
manager->running = true;
manager->initialized = false;
#ifdef _WIN32
InitializeCriticalSection(&manager->lock);
#else
pthread_mutex_init(&manager->lock, NULL);
#endif
XCECD_LOG(XCE_LOG_DEBUG, "Process manager created");
return manager;
}
int process_manager_start_process(process_manager_t* manager, const char* command) {
if (!manager || !command) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Invalid process manager or command");
return -1;
}
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Starting process: %s", command);
#ifdef _WIN32
EnterCriticalSection(&manager->lock);
#else
pthread_mutex_lock(&manager->lock);
#endif
// 创建新的进程信息结构
process_info_t* new_processes = realloc(manager->processes,
(manager->process_count + 1) * sizeof(process_info_t));
if (!new_processes) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to allocate memory for process info");
#ifdef _WIN32
LeaveCriticalSection(&manager->lock);
#else
pthread_mutex_unlock(&manager->lock);
#endif
return -1;
}
manager->processes = new_processes;
process_info_t* proc = &manager->processes[manager->process_count];
process_info_init(proc);
proc->command = strdup(command);
char *jobid_str = getenv("XCE_JOBID");
uint64_t jobid = 0;
if (jobid_str != NULL) {
jobid = atoll(jobid_str);
}
// 创建日志文件
char log_path[256];
time_t now;
time(&now);
struct tm* timeinfo = localtime(&now);
if (jobid != 0) {
snprintf(log_path, sizeof(log_path), "/tmp/job_%lu_%lu.out", (unsigned long)now, jobid);
XCECD_WORKER_LOG(XCE_LOG_INFO, "Job output file path is: %s", log_path);
} else {
snprintf(log_path, sizeof(log_path), "/tmp/job_%d_%02d%02d_%02d%02d%02d.out",
(int)now,
timeinfo->tm_mon + 1,
timeinfo->tm_mday,
timeinfo->tm_hour,
timeinfo->tm_min,
timeinfo->tm_sec);
}
proc->log_file = strdup(log_path);
// 打开日志文件
proc->output_fd = open(proc->log_file, O_WRONLY | O_CREAT | O_APPEND, 0644);
if (proc->output_fd == -1) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to open log file: %s", strerror(errno));
FREEUP(proc->log_file);
FREEUP(proc->command);
#ifdef _WIN32
LeaveCriticalSection(&manager->lock);
#else
pthread_mutex_unlock(&manager->lock);
#endif
return -1;
}
#ifdef _WIN32
// Windows实现
SECURITY_ATTRIBUTES sa = {
.nLength = sizeof(SECURITY_ATTRIBUTES),
.bInheritHandle = TRUE
};
// 创建管道
HANDLE pipe_read, pipe_write;
if (!CreatePipe(&pipe_read, &pipe_write, &sa, 0)) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to create pipe: %lu", GetLastError());
close(proc->output_fd);
FREEUP(proc->log_file);
FREEUP(proc->command);
LeaveCriticalSection(&manager->lock);
return -1;
}
// 确保读取端不被继承
SetHandleInformation(pipe_read, HANDLE_FLAG_INHERIT, 0);
STARTUPINFOA si = {0};
si.cb = sizeof(STARTUPINFO);
si.dwFlags = STARTF_USESTDHANDLES;
si.hStdInput = GetStdHandle(STD_INPUT_HANDLE);
si.hStdOutput = pipe_write;
si.hStdError = pipe_write;
PROCESS_INFORMATION pi;
if (!CreateProcessA(NULL,
proc->command,
NULL,
NULL,
TRUE,
CREATE_NO_WINDOW,
NULL,
NULL,
&si,
&pi)) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to create process: %lu", GetLastError());
CloseHandle(pipe_read);
CloseHandle(pipe_write);
close(proc->output_fd);
FREEUP(proc->log_file);
FREEUP(proc->command);
LeaveCriticalSection(&manager->lock);
return -1;
}
CloseHandle(pipe_write); // 关闭写端
proc->pipe_read = pipe_read;
proc->handle = pi.hProcess;
proc->status = 1; // 运行中
proc->monitor_active = true;
// 创建监控线程
proc->monitor_thread = CreateThread(NULL, 0, output_monitor_thread, proc, 0, NULL);
if (!proc->monitor_thread) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to create monitor thread: %lu", GetLastError());
CloseHandle(pipe_read);
CloseHandle(pi.hProcess);
close(proc->output_fd);
FREEUP(proc->log_file);
FREEUP(proc->command);
LeaveCriticalSection(&manager->lock);
return -1;
}
CloseHandle(pi.hThread);
#else
// Linux实现
pid_t pid = fork();
if (pid == -1) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Fork failed: %s", strerror(errno));
close(proc->output_fd);
FREEUP(proc->log_file);
FREEUP(proc->command);
pthread_mutex_unlock(&manager->lock);
return -1;
}
if (pid == 0) {
// 子进程
dup2(proc->output_fd, STDOUT_FILENO);
dup2(proc->output_fd, STDERR_FILENO);
close(proc->output_fd);
char* args[] = {"/bin/sh", "-c", proc->command, NULL};
execv("/bin/sh", args);
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Failed to execute command: %s", strerror(errno));
_exit(1);
} else {
// 父进程
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Process started successfully, output redirected to %s", proc->log_file);
proc->handle = pid;
proc->status = 1; // 运行中
// 非阻塞等待子进程退出，避免僵尸进程
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "The pid of the child is: %d\n", pid);
}
#endif
manager->process_count++;
manager->initialized = true;
#ifdef _WIN32
LeaveCriticalSection(&manager->lock);
#else
pthread_mutex_unlock(&manager->lock);
#endif
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Process started successfully");
FREEUP(proc->log_file);
return 0;
}
bool process_manager_update(process_manager_t* manager) {
//XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Updating process manager...");
if (!manager) {
XCECD_WORKER_LOG(XCE_LOG_ERROR, "Invalid process manager");
return false;
}
int status;
pid_t pid;
int i;
// 使用临时数组存储需要删除的进程索引
int* remove_indices = calloc(1, manager->process_count * sizeof(int));
int remove_count = 0;
#ifdef _WIN32
DWORD exit_code;
for (i = 0; i < manager->process_count; i++) {
if (GetExitCodeProcess(manager->processes[i].handle, &exit_code)) {
if (exit_code != STILL_ACTIVE) {
XCECD_WORKER_LOG(XCE_LOG_INFO, "Child %d exited with status %lu",
manager->processes[i].handle, exit_code);
remove_indices[remove_count++] = i;
}
} else {
perror("GetExitCodeProcess");
}
}
#else
for (i = 0; i < manager->process_count; i++) {
pid = waitpid(manager->processes[i].handle, &status, WNOHANG);
if (pid > 0) {
if (WIFEXITED(status)) {
XCECD_WORKER_LOG(XCE_LOG_INFO, "Child %d exited with status %d",
pid, WEXITSTATUS(status));
} else if (WIFSIGNALED(status)) {
XCECD_WORKER_LOG(XCE_LOG_INFO, "Child %d killed by signal %d",
pid, WTERMSIG(status));
}
remove_indices[remove_count++] = i;
manager->processes[i].status = 0;
}
}
/* handle this after receiving sigchld and avoid high cpu usage */
#if 0
while ((pid = waitpid(-1, &status, WNOHANG)) > 0) {
for (i = 0; i < manager->process_count; i++) {
if (manager->processes[i].handle == pid) {
if (WIFEXITED(status)) {
XCECD_WORKER_LOG(XCE_LOG_INFO, "Child %d exited with status %d",
pid, WEXITSTATUS(status));
} else if (WIFSIGNALED(status)) {
XCECD_WORKER_LOG(XCE_LOG_INFO, "Child %d killed by signal %d",
pid, WTERMSIG(status));
}
remove_indices[remove_count++] = i;
manager->processes[i].status = 0;
break;