Add X11 forwarding for interactive jobs (xcecli -I -X, ssh -X style)
Render -X jobs' windows on the user's own X server with zero client install, mirroring the ssh -X experience: the job command starts with DISPLAY pointed at a per-job TCP listener on the worker, and the user's xcecli relays that listener's byte stream to their local X server. - xcecli: new -X/--x11 flag (interactive=2 + exdata "x11":1). After submit it fetches the per-job MIT-MAGIC-COOKIE-1, registers it in the user's Xauthority, then bridges one xattach websocket to the local X server address parsed from $DISPLAY (unix socket or TCP). The cookie entry is removed when the session ends. Pre-flight checks report missing DISPLAY/xauth before submit. - apiserver: GET /xce/v1/jobs/:id/xattach relays websocket binary frames to the worker's forwarding port (raw TCP, no ttyd). Close codes match the ttyd attach (4000 unreachable, 4001 ended). The job-finished event ends the relay. Teardown closes both hops as soon as either pump exits, so a client disconnect propagates instead of deadlocking in a blocking read. - worker: XJOB jobs with exdata x11=1 skip VNC, listen on 6100-6199, generate a per-job xauth file (/tmp/job_<id>_xauth, worker-side since the worker has no session), export DISPLAY/XAUTHORITY to the command, and bridge the app connection (loopback) with the relay connection (non-loopback; roles classified by source address, not arrival order). A 60 s no-client watchdog reaps jobs nothing ever attached to; the v1 single-connection boundary refuses a second X connection loudly. t1-verified per the plan matrix: window rendering (xclock/xeyes on Xvfb), normal end -> DONE + xauth removed, no-client 60 s reap (137), kill -> immediate teardown with no residue, -I/-I --browse regression, daemon health, missing-DISPLAY error, second-connection refusal. (--xjob VNC on t1 fails for lack of a vncserver binary; pre-existing.) Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
lei.chen
parent
4f188310ba
commit
85a22271c1
19 changed files with 1744 additions and 50 deletions
|
|
@ -5,6 +5,11 @@
|
|||
#define URL_MAX_LENGTH 256
|
||||
#define TTY_END_PORT 10000
|
||||
#define XJOB_END_PORT 5999
|
||||
/* X11 forwarding (-X jobs, exdata "x11":1): the worker listens on this
|
||||
* segment and exports DISPLAY=:<port-6000> to the job; the app's X
|
||||
* connections are relayed to the user's X server through the apiserver. */
|
||||
#define X11_START_PORT 6100
|
||||
#define X11_END_PORT 6199
|
||||
|
||||
typedef struct {
|
||||
int tty_port;
|
||||
|
|
@ -17,5 +22,6 @@ 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);
|
||||
int export_x11_display_env(int port);
|
||||
|
||||
#endif /* _INTER_JOB_UTILS_H */
|
||||
|
|
|
|||
|
|
@ -28,11 +28,19 @@ typedef struct {
|
|||
* command (xcecli attach path). The worker may reap the ttyd when no
|
||||
* client ever connects; web-viewed jobs (0) keep it for a later browser. */
|
||||
int attach_mode;
|
||||
/* x11=1 (exdata key "x11"): XJOB job running as X11 forwarding -- the
|
||||
* worker exports DISPLAY to its forwarding port and reports a per-job
|
||||
* xauth cookie instead of starting VNC (xcecli -I -X). */
|
||||
int x11_mode;
|
||||
int count;
|
||||
} transfer_info_t;
|
||||
|
||||
typedef struct {
|
||||
char* url;
|
||||
/* Per-job MIT-MAGIC-COOKIE-1 (hex) for X11-forwarding jobs, generated
|
||||
* with xauth and reported to the apiserver so the user's client can
|
||||
* register it with their Xauthority. NULL otherwise. */
|
||||
char* x11cookie;
|
||||
} start_exdata_t;
|
||||
|
||||
bool parse_exdata(const char* exdata, env_var_t** env_vars,
|
||||
|
|
|
|||
|
|
@ -326,3 +326,35 @@ int export_display_env(int port)
|
|||
"Exported DISPLAY environment variable: %s", display_env);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* @brief Export DISPLAY for an X11-forwarding job: the app's X connections
|
||||
* go to the worker's forwarding port (rendering happens on the user's
|
||||
* own X server via the apiserver relay, ssh -X style).
|
||||
* @param port The forwarding port number (X11_START_PORT..X11_END_PORT).
|
||||
* @return 0 on success, -1 on failure.
|
||||
*/
|
||||
int export_x11_display_env(int port)
|
||||
{
|
||||
if (port < 6000 || port - 6000 > 255) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||
"Invalid X11 forwarding port %d for DISPLAY", port);
|
||||
return -1;
|
||||
}
|
||||
|
||||
char display_env[32];
|
||||
snprintf(display_env, sizeof(display_env), ":%d", port - 6000);
|
||||
|
||||
if (setenv("DISPLAY", display_env, 1) != 0) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||
"Failed to set DISPLAY environment variable for "
|
||||
"X11 forwarding");
|
||||
return -1;
|
||||
}
|
||||
|
||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG3,
|
||||
"Exported DISPLAY environment variable for X11 "
|
||||
"forwarding: %s",
|
||||
display_env);
|
||||
return 0;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -252,6 +252,22 @@ bool parse_exdata(const char* exdata, env_var_t** env_vars,
|
|||
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Parsed attach mode: %d",
|
||||
transfer_infos->attach_mode);
|
||||
}
|
||||
// --- Handle 'x11' ---
|
||||
else if (strcmp(key, "x11") == 0) {
|
||||
if (cJSON_IsString(item) && item->valuestring != NULL) {
|
||||
transfer_infos->x11_mode = atoi(item->valuestring);
|
||||
} else if (cJSON_IsNumber(item)) {
|
||||
transfer_infos->x11_mode = item->valueint;
|
||||
} else {
|
||||
XCECD_WORKER_LOG(
|
||||
XCE_LOG_ERROR,
|
||||
"Error: 'x11' is neither a string nor a number.");
|
||||
success = false;
|
||||
continue;
|
||||
}
|
||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Parsed x11 mode: %d",
|
||||
transfer_infos->x11_mode);
|
||||
}
|
||||
// --- Skip 'limits' ---
|
||||
else if (strcmp(key, "limits") == 0) {
|
||||
// limits will be handled in parse_limit
|
||||
|
|
@ -525,6 +541,9 @@ char* generate_exdata(const start_exdata_t* start_info)
|
|||
}
|
||||
|
||||
cJSON_AddStringToObject(root, "url", start_info->url);
|
||||
if (start_info->x11cookie) {
|
||||
cJSON_AddStringToObject(root, "x11cookie", start_info->x11cookie);
|
||||
}
|
||||
|
||||
char* exdata_str = cJSON_PrintUnformatted(root);
|
||||
cJSON_Delete(root);
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
#include "job_signal.h"
|
||||
#include "job_output_path.h"
|
||||
#include "inter_job_utils.h"
|
||||
#include "process_manager.h"
|
||||
#include "static_array.h"
|
||||
|
|
@ -8,12 +9,18 @@
|
|||
#include "xcecd_exdata.h"
|
||||
#include "xcecd_utils.h"
|
||||
#include "xcelog.h"
|
||||
#include <arpa/inet.h>
|
||||
#include <dirent.h>
|
||||
#include <errno.h>
|
||||
#include <fcntl.h>
|
||||
#include <netinet/in.h>
|
||||
#include <netinet/tcp.h>
|
||||
#include <signal.h>
|
||||
#include <stdbool.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/socket.h>
|
||||
#include <sys/types.h>
|
||||
#include <sys/wait.h>
|
||||
#include <unistd.h>
|
||||
|
|
@ -26,6 +33,12 @@
|
|||
#define LOG_SIZE (1024 * 1024 * 100) // 100 MB
|
||||
#define LOG_LEVEL 3
|
||||
|
||||
/* X11 forwarding (-X jobs): the worker listens for the app's X connections
|
||||
* and exports DISPLAY to the listening port; rendering happens on the
|
||||
* user's own X server via the apiserver xattach relay (ssh -X model). One
|
||||
* relayed websocket carries one X connection. */
|
||||
#define X11_NO_CLIENT_MS 60000 /* reap when no X connection ever arrived */
|
||||
|
||||
typedef struct exdata {
|
||||
env_var_t* env_vars;
|
||||
transfer_info_t* transfer_info;
|
||||
|
|
@ -69,6 +82,37 @@ static bool tty_attach_saw_child = false;
|
|||
static int64_t tty_attach_child_gone_ms = 0;
|
||||
static const char* tty_reap_reason = NULL;
|
||||
|
||||
/* One end of a bridged X11 session: either the job app's X connection or
|
||||
* the apiserver xattach relay connection. The two are told apart by source
|
||||
* address, not arrival order: the app always runs on this host and its
|
||||
* DISPLAY is always ":N", so app connections come from loopback, while the
|
||||
* relay dials the reported x11://<local_ip>:<port> URL from the apiserver
|
||||
* host (which is never loopback, even when apiserver and worker share a
|
||||
* host). Byte pumps are identical for both roles. */
|
||||
#define X11_STASH_SIZE 32768
|
||||
typedef struct {
|
||||
int fd; /* connection fd, -1 = open slot */
|
||||
bool relay; /* true = apiserver relay leg */
|
||||
int len; /* bytes stashed awaiting the peer */
|
||||
char stash[X11_STASH_SIZE];
|
||||
} x11_leg_t;
|
||||
|
||||
/* X11 forwarding session state (one per-job worker process, so a single
|
||||
* static session is enough -- same shape as the VNC branch). */
|
||||
typedef struct {
|
||||
int listen_fd; /* forwarding-port listener, -1 = off */
|
||||
int x11_port; /* TCP port DISPLAY points at */
|
||||
volatile sig_atomic_t accepts; /* total accepted X connections */
|
||||
bool saw_client; /* app+relay bridge ever established */
|
||||
uv_timer_t no_client_timer;
|
||||
bool timer_ready;
|
||||
char display[32]; /* ":N" exported to the job */
|
||||
char cookie_hex[65]; /* per-job MIT-MAGIC-COOKIE-1 */
|
||||
char auth_file[256]; /* per-job Xauthority file, "" = none */
|
||||
x11_leg_t legs[2]; /* the bridged pair */
|
||||
} x11_forward_t;
|
||||
static x11_forward_t x11_fwd = { .listen_fd = -1 };
|
||||
|
||||
// process manager will share the same event loop with ipc
|
||||
uv_loop_t* uv_loop = NULL;
|
||||
|
||||
|
|
@ -77,6 +121,12 @@ static void worker_reconnect_callback(bool connected);
|
|||
static int xjob_start_vnc();
|
||||
static void xjob_stop_vnc();
|
||||
static void tty_stop_attach_session();
|
||||
static int x11_start_forwarding();
|
||||
static void x11_stop_forwarding();
|
||||
static void x11_accept_and_pump();
|
||||
static void x11_no_client_callback(uv_timer_t* timer);
|
||||
static void x11_no_client_timer_start();
|
||||
static void x11_no_client_timer_stop();
|
||||
static bool check_process_status(pid_t target_pid, int expected_signal,
|
||||
int timeout_seconds);
|
||||
static pid_t ttyd_real_pid(pid_t monitored_pid);
|
||||
|
|
@ -537,7 +587,12 @@ static void process_exit_callback(int pid, int exit_status, int term_signal,
|
|||
|
||||
/* Job post-exec start here (blocked) */
|
||||
if (XJOB == mydata.transfer_info->interactive_mode) {
|
||||
xjob_stop_vnc();
|
||||
if (1 == mydata.transfer_info->x11_mode) {
|
||||
x11_no_client_timer_stop();
|
||||
x11_stop_forwarding();
|
||||
} else {
|
||||
xjob_stop_vnc();
|
||||
}
|
||||
}
|
||||
|
||||
/* attach-mode TTY job: the session ends with the command (web-viewed
|
||||
|
|
@ -630,6 +685,450 @@ static void check_and_reconnect()
|
|||
reconnect_retry_limit = 5;
|
||||
}
|
||||
|
||||
/* ---- X11 forwarding (XJOB + exdata x11=1, the xcecli -I -X path) ----
|
||||
*
|
||||
* The job command starts immediately (no daemon waits for a client) with
|
||||
* DISPLAY pointing at the forwarding port. The app's XOpenDisplay connects
|
||||
* to the listener; the apiserver's /xattach relay picks each accepted
|
||||
* connection up as one websocket and pumps the bytes to the user's X
|
||||
* server, where the windows render. Auth follows ssh -X: a per-job
|
||||
* MIT-MAGIC-COOKIE-1 is generated with xauth (the only X-aware helper
|
||||
* guaranteed around an X11 user setup) and reported through the start
|
||||
* exdata so the user's client can register it with their Xauthority.
|
||||
*
|
||||
* The no-client timer covers the one residue shape an -X job has: a job
|
||||
* whose X connection never arrives (the CLI died before relaying) would
|
||||
* otherwise sit in RUN forever -- the app blocks inside XOpenDisplay and
|
||||
* never exits on its own. Once any connection was accepted the session is
|
||||
* the app's and X's to end (a dropped relay surfaces as an X IO error and
|
||||
* the app exits), so the timer is cancelled on the first accept. */
|
||||
|
||||
/* 32 hex chars = 128 bits of cookie, exactly xauth's MIT-MAGIC-COOKIE-1
|
||||
* format. Reads /dev/urandom; retries a few times before giving up. */
|
||||
static bool x11_generate_cookie(char* out, size_t outlen)
|
||||
{
|
||||
unsigned char raw[16];
|
||||
static const char hex[] = "0123456789abcdef";
|
||||
|
||||
if (outlen < sizeof(raw) * 2 + 1) {
|
||||
return false;
|
||||
}
|
||||
|
||||
int fd = open("/dev/urandom", O_RDONLY);
|
||||
if (fd < 0) {
|
||||
return false;
|
||||
}
|
||||
size_t got = 0;
|
||||
bool ok = true;
|
||||
while (got < sizeof(raw)) {
|
||||
ssize_t n = read(fd, (char*)raw + got, sizeof(raw) - got);
|
||||
if (n <= 0) {
|
||||
if (n < 0 && errno == EINTR) {
|
||||
continue;
|
||||
}
|
||||
ok = false;
|
||||
break;
|
||||
}
|
||||
got += (size_t)n;
|
||||
}
|
||||
close(fd);
|
||||
if (!ok) {
|
||||
return false;
|
||||
}
|
||||
|
||||
for (size_t i = 0; i < sizeof(raw); i++) {
|
||||
out[i * 2] = hex[raw[i] >> 4];
|
||||
out[i * 2 + 1] = hex[raw[i] & 0xF];
|
||||
}
|
||||
out[sizeof(raw) * 2] = '\0';
|
||||
return true;
|
||||
}
|
||||
|
||||
/* Create a fresh Xauthority file for this job and register the generated
|
||||
* cookie in it under our display number. The per-job worker runs on a headless
|
||||
* node -- there is no session Xauthority to inherit (getenv("XAUTHORITY") is
|
||||
* normally empty here), so the file is ours from scratch, named after the job
|
||||
* output dir convention in /tmp. The job processes inherit the worker's
|
||||
* environment, so exporting XAUTHORITY alongside DISPLAY (the same channel
|
||||
* export_x11_display_env uses) makes the app's XOpenDisplay present exactly
|
||||
* this cookie. */
|
||||
static int x11_install_cookie(uint64_t jobid)
|
||||
{
|
||||
char cmd[MAXLINELEN];
|
||||
char jobidStr[64];
|
||||
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
|
||||
|
||||
snprintf(x11_fwd.auth_file, sizeof(x11_fwd.auth_file),
|
||||
XCECD_JOB_OUTPUT_DIR "/job_%s_xauth", jobidStr);
|
||||
unlink(x11_fwd.auth_file); /* a re-run job must not inherit an old entry */
|
||||
|
||||
snprintf(cmd, sizeof(cmd),
|
||||
"xauth -f '%s' add ':%d' . %s 2>/dev/null && chmod 600 '%s'",
|
||||
x11_fwd.auth_file, x11_fwd.x11_port - 6000, x11_fwd.cookie_hex,
|
||||
x11_fwd.auth_file);
|
||||
/* xce-fork-ok: xcecd_worker is single-threaded (tech.md) */
|
||||
int rc = system(cmd);
|
||||
if (rc != 0 || access(x11_fwd.auth_file, R_OK) != 0) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||
"xauth add failed for X11 forwarding display :%d "
|
||||
"(rc=%d)",
|
||||
x11_fwd.x11_port - 6000, rc);
|
||||
unlink(x11_fwd.auth_file);
|
||||
x11_fwd.auth_file[0] = '\0';
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (setenv("XAUTHORITY", x11_fwd.auth_file, 1) != 0) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||
"Failed to set XAUTHORITY for X11 forwarding: %s",
|
||||
strerror(errno));
|
||||
unlink(x11_fwd.auth_file);
|
||||
x11_fwd.auth_file[0] = '\0';
|
||||
return -1;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void x11_remove_cookie()
|
||||
{
|
||||
if (x11_fwd.auth_file[0] != '\0') {
|
||||
unlink(x11_fwd.auth_file);
|
||||
x11_fwd.auth_file[0] = '\0';
|
||||
}
|
||||
x11_fwd.cookie_hex[0] = '\0';
|
||||
}
|
||||
|
||||
static int x11_start_forwarding()
|
||||
{
|
||||
char jobidStr[64];
|
||||
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
|
||||
|
||||
x11_fwd.x11_port = port_info.xjob_port; /* X11 rides the XJOB slot */
|
||||
x11_fwd.listen_fd = -1;
|
||||
x11_fwd.accepts = 0;
|
||||
x11_fwd.saw_client = false;
|
||||
x11_fwd.cookie_hex[0] = '\0';
|
||||
x11_fwd.auth_file[0] = '\0';
|
||||
x11_fwd.legs[0].fd = -1;
|
||||
x11_fwd.legs[1].fd = -1;
|
||||
x11_fwd.legs[0].relay = false;
|
||||
x11_fwd.legs[1].relay = false;
|
||||
x11_fwd.legs[0].len = 0;
|
||||
x11_fwd.legs[1].len = 0;
|
||||
|
||||
if (!x11_generate_cookie(x11_fwd.cookie_hex,
|
||||
sizeof(x11_fwd.cookie_hex))) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||
"Failed to generate the X11 auth cookie for job %s",
|
||||
jobidStr);
|
||||
return -1;
|
||||
}
|
||||
|
||||
/* The listen socket stays bound for the job's lifetime; the accept
|
||||
* handshake for the app's X connection must already succeed before the
|
||||
* apiserver relay dials (X11 needs no protocol on accept). */
|
||||
int fd = socket(AF_INET, SOCK_STREAM, 0);
|
||||
if (fd < 0) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR, "X11 forwarding socket failed: %s",
|
||||
strerror(errno));
|
||||
return -1;
|
||||
}
|
||||
int one = 1;
|
||||
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
|
||||
/* accept() must never block: the main loop shares this fd with the uv
|
||||
* loop and a 1 ms cadence, so a blocking accept would freeze the whole
|
||||
* worker (timers included) the moment the backlog runs dry */
|
||||
int fl = fcntl(fd, F_GETFL, 0);
|
||||
if (fl < 0 || fcntl(fd, F_SETFL, fl | O_NONBLOCK) != 0) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||
"X11 forwarding listener fcntl failed: %s",
|
||||
strerror(errno));
|
||||
close(fd);
|
||||
return -1;
|
||||
}
|
||||
|
||||
struct sockaddr_in addr;
|
||||
memset(&addr, 0, sizeof(addr));
|
||||
addr.sin_family = AF_INET;
|
||||
addr.sin_port = htons(x11_fwd.x11_port);
|
||||
addr.sin_addr.s_addr = INADDR_ANY;
|
||||
if (bind(fd, (struct sockaddr*)&addr, sizeof(addr)) != 0 || listen(fd, 8) != 0) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||
"X11 forwarding listen on port %d failed: %s",
|
||||
x11_fwd.x11_port, strerror(errno));
|
||||
close(fd);
|
||||
return -1;
|
||||
}
|
||||
x11_fwd.listen_fd = fd;
|
||||
|
||||
if (x11_install_cookie(jobid) != 0) {
|
||||
x11_stop_forwarding();
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (export_x11_display_env(x11_fwd.x11_port) != 0) {
|
||||
x11_stop_forwarding();
|
||||
return -1;
|
||||
}
|
||||
|
||||
XCECD_WORKER_LOG(XCE_LOG_INFO,
|
||||
"X11 forwarding ready for job %s: port %d, display :%d, "
|
||||
"authority %s",
|
||||
jobidStr, x11_fwd.x11_port, x11_fwd.x11_port - 6000,
|
||||
x11_fwd.auth_file);
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void x11_stop_forwarding()
|
||||
{
|
||||
if (x11_fwd.listen_fd >= 0) {
|
||||
close(x11_fwd.listen_fd);
|
||||
x11_fwd.listen_fd = -1;
|
||||
}
|
||||
for (int i = 0; i < 2; i++) {
|
||||
if (x11_fwd.legs[i].fd >= 0) {
|
||||
close(x11_fwd.legs[i].fd);
|
||||
x11_fwd.legs[i].fd = -1;
|
||||
}
|
||||
x11_fwd.legs[i].len = 0;
|
||||
}
|
||||
x11_remove_cookie();
|
||||
}
|
||||
|
||||
/* One pump step for a bridged connection: drain its stash toward the peer
|
||||
* and pull new bytes into it. Non-blocking; short counts and EAGAIN are
|
||||
* normal -- the stash absorbs bursts the peer cannot take right away, and
|
||||
* a full stash applies backpressure (we simply stop recv'ing until the
|
||||
* peer drains it) instead of dropping or killing the session.
|
||||
* Returns false once the peer is gone (EOF or error), which ends the
|
||||
* session: X11 has no notion of a half-closed connection, so when either
|
||||
* leg dies the whole bridge dies with it (the job then exits on IO error,
|
||||
* or the relay teardown reports the end to the CLI). */
|
||||
static bool x11_pump_leg(x11_leg_t* leg, x11_leg_t* peer)
|
||||
{
|
||||
if (leg->fd < 0) {
|
||||
return true;
|
||||
}
|
||||
|
||||
/* flush stashed bytes toward the peer first */
|
||||
if (leg->len > 0) {
|
||||
ssize_t n = send(peer->fd, leg->stash, leg->len,
|
||||
MSG_DONTWAIT | MSG_NOSIGNAL);
|
||||
if (n < 0) {
|
||||
if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG,
|
||||
"X11 bridge leg %d send failed: %s", leg->fd,
|
||||
strerror(errno));
|
||||
return false;
|
||||
}
|
||||
} else if (n > 0) {
|
||||
if (n < leg->len) {
|
||||
memmove(leg->stash, leg->stash + n, leg->len - n);
|
||||
}
|
||||
leg->len -= n;
|
||||
}
|
||||
}
|
||||
|
||||
char buf[8192];
|
||||
for (;;) {
|
||||
if (leg->len >= X11_STASH_SIZE) {
|
||||
return true; /* backpressure: wait for the peer to drain */
|
||||
}
|
||||
ssize_t n = recv(leg->fd, buf, sizeof(buf), MSG_DONTWAIT);
|
||||
if (n < 0) {
|
||||
if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
|
||||
return true;
|
||||
}
|
||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "X11 bridge leg %d recv failed: %s",
|
||||
leg->fd, strerror(errno));
|
||||
return false;
|
||||
}
|
||||
if (n == 0) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "X11 bridge leg %d closed by peer",
|
||||
leg->fd);
|
||||
return false;
|
||||
}
|
||||
int off = 0;
|
||||
while (off < n) {
|
||||
int room = X11_STASH_SIZE - leg->len;
|
||||
if (room <= 0) {
|
||||
break; /* stash full -- drain on a later tick */
|
||||
}
|
||||
int take = (n - off) < room ? (n - off) : room;
|
||||
memcpy(leg->stash + leg->len, buf + off, take);
|
||||
leg->len += take;
|
||||
off += take;
|
||||
|
||||
ssize_t w = send(peer->fd, leg->stash, leg->len,
|
||||
MSG_DONTWAIT | MSG_NOSIGNAL);
|
||||
if (w < 0) {
|
||||
if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
|
||||
continue;
|
||||
}
|
||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG,
|
||||
"X11 bridge leg %d send failed: %s", leg->fd,
|
||||
strerror(errno));
|
||||
return false;
|
||||
}
|
||||
if (w < leg->len) {
|
||||
memmove(leg->stash, leg->stash + w, leg->len - w);
|
||||
}
|
||||
leg->len -= w;
|
||||
}
|
||||
if (off < n) {
|
||||
return true; /* stash filled mid-buffer -- drain on a later tick */
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* Classify an accepted connection by source address: the app runs on this
|
||||
* host and its DISPLAY is ":N" (loopback), the apiserver relay dials the
|
||||
* worker's non-loopback IP from the apiserver host. */
|
||||
static bool x11_conn_is_relay(int fd)
|
||||
{
|
||||
struct sockaddr_in sa;
|
||||
socklen_t sl = sizeof(sa);
|
||||
if (getpeername(fd, (struct sockaddr*)&sa, &sl) != 0) {
|
||||
return false;
|
||||
}
|
||||
uint32_t ip = ntohl(sa.sin_addr.s_addr);
|
||||
return ip != INADDR_LOOPBACK;
|
||||
}
|
||||
|
||||
/* Non-blocking accept/pump step for the X11 listener, called from the main
|
||||
* loop. v1 carries exactly one X connection: one app leg (loopback) plus
|
||||
* one apiserver relay leg, bridged once both are in. Anything beyond that
|
||||
* is refused loudly -- xim/clipboard-style secondary connections are out of
|
||||
* scope and silent window-event loss would be worse than a clear failure
|
||||
* (plan 2.4). Classification is by source address so connection arrival
|
||||
* order cannot mispair two racing app connections with each other. */
|
||||
static void x11_accept_and_pump()
|
||||
{
|
||||
if (x11_fwd.listen_fd < 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
bool bridged = (x11_fwd.legs[0].fd >= 0 && x11_fwd.legs[1].fd >= 0);
|
||||
|
||||
for (;;) {
|
||||
int fd = accept(x11_fwd.listen_fd, NULL, NULL);
|
||||
if (fd < 0) {
|
||||
if (errno != EAGAIN && errno != EWOULDBLOCK && errno != EINTR
|
||||
&& errno != ECONNABORTED) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||
"X11 forwarding accept failed: %s",
|
||||
strerror(errno));
|
||||
}
|
||||
break;
|
||||
}
|
||||
int fl = fcntl(fd, F_GETFL, 0);
|
||||
if (fl >= 0) {
|
||||
fcntl(fd, F_SETFL, fl | O_NONBLOCK);
|
||||
}
|
||||
/* TCP_NODELAY: X11 turns interactive (cursor moves, key events) --
|
||||
* tiny packets must not wait for Nagle on either leg */
|
||||
int one = 1;
|
||||
setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
|
||||
|
||||
if (bridged) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_NOTICE,
|
||||
"X11 job app opened a second X connection; "
|
||||
"v1 supports one -- closing it");
|
||||
close(fd);
|
||||
break; /* one refusal per tick keeps the loop responsive */
|
||||
}
|
||||
|
||||
bool relay = x11_conn_is_relay(fd);
|
||||
/* slot 0 = app leg, slot 1 = relay leg; a duplicate of a role we
|
||||
* already hold (e.g. a second app racing in) is refused */
|
||||
int slot = relay ? 1 : 0;
|
||||
if (x11_fwd.legs[slot].fd >= 0) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_NOTICE,
|
||||
"X11 job app opened a second X connection; "
|
||||
"v1 supports one -- closing it");
|
||||
close(fd);
|
||||
break;
|
||||
}
|
||||
x11_fwd.legs[slot].fd = fd;
|
||||
x11_fwd.legs[slot].relay = relay;
|
||||
x11_fwd.legs[slot].len = 0;
|
||||
x11_fwd.accepts++;
|
||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG, "X11 forwarding %s connection accepted",
|
||||
relay ? "relay" : "app");
|
||||
|
||||
if (x11_fwd.legs[0].fd >= 0 && x11_fwd.legs[1].fd >= 0) {
|
||||
bridged = true;
|
||||
x11_fwd.saw_client = true;
|
||||
x11_no_client_timer_stop();
|
||||
XCECD_WORKER_LOG(XCE_LOG_INFO,
|
||||
"X11 bridge established on port %d",
|
||||
x11_fwd.x11_port);
|
||||
}
|
||||
}
|
||||
|
||||
if (bridged) {
|
||||
if (!x11_pump_leg(&x11_fwd.legs[0], &x11_fwd.legs[1])) {
|
||||
x11_stop_forwarding();
|
||||
return;
|
||||
}
|
||||
if (!x11_pump_leg(&x11_fwd.legs[1], &x11_fwd.legs[0])) {
|
||||
x11_stop_forwarding();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static void x11_no_client_callback(uv_timer_t* timer)
|
||||
{
|
||||
(void)timer;
|
||||
|
||||
if (worker_exit_flag
|
||||
|| XJOB != mydata.transfer_info->interactive_mode
|
||||
|| 1 != mydata.transfer_info->x11_mode) {
|
||||
return;
|
||||
}
|
||||
if (x11_fwd.saw_client) {
|
||||
/* an X connection arrived: the session belongs to the app now */
|
||||
x11_no_client_timer_stop();
|
||||
return;
|
||||
}
|
||||
|
||||
pid_t pid = uv_proc_get_pid(&uv_proc_manager);
|
||||
if (pid <= 0) {
|
||||
x11_no_client_timer_stop();
|
||||
return; /* process_exit_callback owns the job */
|
||||
}
|
||||
|
||||
char jobidStr[64];
|
||||
fmt_jobid(jobidStr, sizeof(jobidStr), jobid);
|
||||
XCECD_WORKER_LOG(XCE_LOG_NOTICE,
|
||||
"X11 job %s: no X client connected within %d s, reaping "
|
||||
"the job",
|
||||
jobidStr, (int)(X11_NO_CLIENT_MS / 1000));
|
||||
|
||||
tty_reap_reason = "No X11 client connected within 60 s";
|
||||
x11_no_client_timer_stop();
|
||||
/* the listener is not the app's parent: killing the job tree directly
|
||||
* lands us in the regular process_exit_callback path */
|
||||
signal_children_recursive(SIGKILL, pid);
|
||||
kill(pid, SIGKILL);
|
||||
}
|
||||
|
||||
static void x11_no_client_timer_start()
|
||||
{
|
||||
if (!x11_fwd.timer_ready) {
|
||||
return;
|
||||
}
|
||||
uv_timer_start(&x11_fwd.no_client_timer, x11_no_client_callback,
|
||||
X11_NO_CLIENT_MS, X11_NO_CLIENT_MS);
|
||||
}
|
||||
|
||||
static void x11_no_client_timer_stop()
|
||||
{
|
||||
if (x11_fwd.timer_ready) {
|
||||
uv_timer_stop(&x11_fwd.no_client_timer);
|
||||
}
|
||||
}
|
||||
|
||||
static int xjob_start_vnc()
|
||||
{
|
||||
char jobidStr[64];
|
||||
|
|
@ -1090,8 +1589,19 @@ static void worker_job_callback(job_assignment_t* info)
|
|||
"Parsed exdata for job %s: interactive=%d", jobidStr,
|
||||
mydata.transfer_info->interactive_mode);
|
||||
|
||||
int job_port
|
||||
= check_port(port_info, mydata.transfer_info->interactive_mode);
|
||||
/* X11-forwarding jobs are interactive=XJOB with exdata x11=1: they take
|
||||
* their port from the forwarding segment (6100-6199) instead of the VNC
|
||||
* one, and generate_cmd passes the raw command through (the job needs no
|
||||
* ttyd wrapper). */
|
||||
int x11_backend = (mydata.transfer_info->interactive_mode == XJOB)
|
||||
&& (1 == mydata.transfer_info->x11_mode);
|
||||
|
||||
int job_port;
|
||||
if (x11_backend) {
|
||||
job_port = find_available_port(X11_START_PORT, X11_END_PORT);
|
||||
} else {
|
||||
job_port = check_port(port_info, mydata.transfer_info->interactive_mode);
|
||||
}
|
||||
if (job_port < 0) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||
"Failed to check port availability for job %s",
|
||||
|
|
@ -1102,6 +1612,8 @@ static void worker_job_callback(job_assignment_t* info)
|
|||
|| mydata.transfer_info->interactive_mode == DISPLAY) {
|
||||
port_info.tty_port = job_port;
|
||||
} else if (mydata.transfer_info->interactive_mode == XJOB) {
|
||||
/* X11 forwarding and VNC share the xjob_port slot; the backend
|
||||
* branch decides what listens on it */
|
||||
port_info.xjob_port = job_port;
|
||||
}
|
||||
XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Assigned port %d for job %s",
|
||||
|
|
@ -1114,8 +1626,13 @@ static void worker_job_callback(job_assignment_t* info)
|
|||
exit(1);
|
||||
}
|
||||
|
||||
job_cmd = generate_cmd((char*)info->cmd, port_info,
|
||||
mydata.transfer_info->interactive_mode);
|
||||
if (x11_backend) {
|
||||
/* no ttyd wrapper: the command runs directly with DISPLAY inherited */
|
||||
job_cmd = strdup((char*)info->cmd);
|
||||
} else {
|
||||
job_cmd = generate_cmd((char*)info->cmd, port_info,
|
||||
mydata.transfer_info->interactive_mode);
|
||||
}
|
||||
|
||||
/* bug-1936: run pre-exec and the job in the submission's cwd, with HOME
|
||||
* = its exec home; before apply_env_vars() so an explicit --env HOME
|
||||
|
|
@ -1139,7 +1656,33 @@ static void worker_job_callback(job_assignment_t* info)
|
|||
* for the next job (previously exit(1) here took down the whole
|
||||
* per-job worker with no user-visible reason). */
|
||||
if (XJOB == mydata.transfer_info->interactive_mode) {
|
||||
if (xjob_start_vnc() != 0) {
|
||||
if (1 == mydata.transfer_info->x11_mode) {
|
||||
/* X11 forwarding backend (xcecli -I -X): listen for the app's X
|
||||
* connections and point DISPLAY at the listener instead of
|
||||
* running VNC. */
|
||||
if (x11_start_forwarding() != 0) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||
"Failed to start X11 forwarding for job %s",
|
||||
jobidStr);
|
||||
x11_stop_forwarding(); /* best-effort cleanup of partial start */
|
||||
|
||||
check_and_reconnect();
|
||||
|
||||
job_runtime_info.job_pid = 0;
|
||||
job_runtime_info.job_status = JOB_FAILED;
|
||||
job_runtime_info.worker_status = WORKER_EXITED;
|
||||
job_runtime_info.exit_code = 1;
|
||||
|
||||
ipc_worker_send_status(
|
||||
&job_runtime_info,
|
||||
"Failed to start X11 forwarding (check worker log)");
|
||||
worker_exit_flag = 1;
|
||||
return_code = 1;
|
||||
FREEUP(job_cmd);
|
||||
ipc_worker_run(UV_RUN_NOWAIT);
|
||||
return;
|
||||
}
|
||||
} else if (xjob_start_vnc() != 0) {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||
"Failed to start VNC server for job %s",
|
||||
jobidStr);
|
||||
|
|
@ -1204,6 +1747,13 @@ static void worker_job_callback(job_assignment_t* info)
|
|||
tty_attach_watchdog_start();
|
||||
}
|
||||
|
||||
/* X11-forwarding jobs: reap when no X connection ever arrives (the
|
||||
* app blocks inside XOpenDisplay and never exits on its own) */
|
||||
if (XJOB == mydata.transfer_info->interactive_mode
|
||||
&& 1 == mydata.transfer_info->x11_mode) {
|
||||
x11_no_client_timer_start();
|
||||
}
|
||||
|
||||
if (mydata.transfer_info->interactive_mode > 0) {
|
||||
url = generate_url(port_info,
|
||||
mydata.transfer_info->interactive_mode);
|
||||
|
|
@ -1213,6 +1763,23 @@ static void worker_job_callback(job_assignment_t* info)
|
|||
url);
|
||||
}
|
||||
|
||||
if (x11_backend && url) {
|
||||
/* rewrite the XJOB URL ("ip::port") to the forwarding form
|
||||
* the apiserver's xattach relay and the xcecli -X client
|
||||
* expect ("x11://ip:port") -- the double colon is the VNC
|
||||
* websockify convention and not a dialable address */
|
||||
char* dcolon = strstr(url, "::");
|
||||
char* x11url = calloc(1, URL_MAX_LENGTH);
|
||||
if (x11url && dcolon) {
|
||||
snprintf(x11url, URL_MAX_LENGTH, "x11://%.*s:%s",
|
||||
(int)(dcolon - url), url, dcolon + 2);
|
||||
FREEUP(url);
|
||||
url = x11url;
|
||||
} else {
|
||||
FREEUP(x11url);
|
||||
}
|
||||
}
|
||||
|
||||
start_exdata = calloc(1, sizeof(start_exdata_t));
|
||||
if (!start_exdata) {
|
||||
XCECD_WORKER_LOG(
|
||||
|
|
@ -1221,6 +1788,13 @@ static void worker_job_callback(job_assignment_t* info)
|
|||
exit(1);
|
||||
}
|
||||
start_exdata->url = url;
|
||||
/* -X jobs report the per-job X11 auth cookie next to the URL so
|
||||
* the user's client can register it with their Xauthority */
|
||||
if (XJOB == mydata.transfer_info->interactive_mode
|
||||
&& 1 == mydata.transfer_info->x11_mode
|
||||
&& x11_fwd.cookie_hex[0] != '\0') {
|
||||
start_exdata->x11cookie = strdup(x11_fwd.cookie_hex);
|
||||
}
|
||||
|
||||
start_exdata_info = generate_exdata(start_exdata);
|
||||
if (!start_exdata_info) {
|
||||
|
|
@ -1252,7 +1826,10 @@ static void worker_job_callback(job_assignment_t* info)
|
|||
|
||||
FREEUP(job_cmd);
|
||||
FREEUP(url);
|
||||
FREEUP(start_exdata);
|
||||
if (start_exdata) {
|
||||
FREEUP(start_exdata->x11cookie);
|
||||
FREEUP(start_exdata);
|
||||
}
|
||||
|
||||
} else {
|
||||
XCECD_WORKER_LOG(XCE_LOG_ERROR,
|
||||
|
|
@ -1881,10 +2458,18 @@ int main(int argc, char* argv[])
|
|||
uv_timer_init(uv_loop, &tty_attach_timer);
|
||||
tty_attach_timer_ready = true;
|
||||
|
||||
/* X11-forwarding no-client reaper (x11_no_client_callback) */
|
||||
uv_timer_init(uv_loop, &x11_fwd.no_client_timer);
|
||||
x11_fwd.timer_ready = true;
|
||||
|
||||
do {
|
||||
|
||||
ipc_worker_run(UV_RUN_NOWAIT);
|
||||
|
||||
/* X11 forwarding: accept the app's X connection and the apiserver
|
||||
* relay's connection, then bridge them (non-blocking step) */
|
||||
x11_accept_and_pump();
|
||||
|
||||
usleep(1000);
|
||||
|
||||
} while (worker_exit_flag == 0);
|
||||
|
|
|
|||
|
|
@ -149,6 +149,62 @@ static void test_apiserver_nofile_limit_parsed(void)
|
|||
TEST_ASSERT_EQUAL_UINT64(60, limits.run_limit);
|
||||
}
|
||||
|
||||
/* x11=1 (exdata key "x11", xcecli -I -X): selects the X11-forwarding
|
||||
* backend for an XJOB job -- DISPLAY goes to the worker's forwarding port,
|
||||
* no VNC. Must tolerate the string form like attach/interactive do. */
|
||||
static void test_x11_key_parsed(void)
|
||||
{
|
||||
TEST_ASSERT_TRUE(parse_exdata(
|
||||
"{\"interactive\":2,\"x11\":1}", &env, &tinfo, 1));
|
||||
TEST_ASSERT_EQUAL_INT(XJOB, tinfo.interactive_mode);
|
||||
TEST_ASSERT_EQUAL_INT(1, tinfo.x11_mode);
|
||||
}
|
||||
|
||||
static void test_x11_key_string_form_parsed(void)
|
||||
{
|
||||
TEST_ASSERT_TRUE(parse_exdata(
|
||||
"{\"interactive\":\"2\",\"x11\":\"1\"}", &env, &tinfo, 1));
|
||||
TEST_ASSERT_EQUAL_INT(XJOB, tinfo.interactive_mode);
|
||||
TEST_ASSERT_EQUAL_INT(1, tinfo.x11_mode);
|
||||
}
|
||||
|
||||
/* x11 on a non-XJOB job is recorded as parsed (the lifecycle branch keys
|
||||
* off both fields together, so it stays inert there) but the parse itself
|
||||
* must not fail the job. */
|
||||
static void test_x11_key_absent_defaults_zero(void)
|
||||
{
|
||||
TEST_ASSERT_TRUE(parse_exdata(
|
||||
"{\"interactive\":2}", &env, &tinfo, 1));
|
||||
TEST_ASSERT_EQUAL_INT(XJOB, tinfo.interactive_mode);
|
||||
TEST_ASSERT_EQUAL_INT(0, tinfo.x11_mode);
|
||||
}
|
||||
|
||||
/* generate_exdata packs the -X job's auth cookie next to the URL; without
|
||||
* a cookie (ttyd/VNC jobs) the exdata stays exactly {"url": ...}. */
|
||||
static void test_generate_exdata_x11cookie(void)
|
||||
{
|
||||
start_exdata_t start = {.url = "x11://10.0.1.5:6100",
|
||||
.x11cookie = "aabbccdd00112233"};
|
||||
char* json = generate_exdata(&start);
|
||||
|
||||
TEST_ASSERT_NOT_NULL(json);
|
||||
TEST_ASSERT_TRUE(strstr(json, "\"url\":\"x11://10.0.1.5:6100\"") != NULL);
|
||||
TEST_ASSERT_TRUE(
|
||||
strstr(json, "\"x11cookie\":\"aabbccdd00112233\"") != NULL);
|
||||
free(json);
|
||||
}
|
||||
|
||||
static void test_generate_exdata_without_cookie(void)
|
||||
{
|
||||
start_exdata_t start = {.url = "http://10.0.1.5:9000",
|
||||
.x11cookie = NULL};
|
||||
char* json = generate_exdata(&start);
|
||||
|
||||
TEST_ASSERT_NOT_NULL(json);
|
||||
TEST_ASSERT_EQUAL_STRING("{\"url\":\"http://10.0.1.5:9000\"}", json);
|
||||
free(json);
|
||||
}
|
||||
|
||||
int main(void)
|
||||
{
|
||||
UNITY_BEGIN();
|
||||
|
|
@ -158,5 +214,10 @@ int main(void)
|
|||
RUN_TEST(test_both_keys_keep_first_list);
|
||||
RUN_TEST(test_apiserver_limits_parsed);
|
||||
RUN_TEST(test_apiserver_nofile_limit_parsed);
|
||||
RUN_TEST(test_x11_key_parsed);
|
||||
RUN_TEST(test_x11_key_string_form_parsed);
|
||||
RUN_TEST(test_x11_key_absent_defaults_zero);
|
||||
RUN_TEST(test_generate_exdata_x11cookie);
|
||||
RUN_TEST(test_generate_exdata_without_cookie);
|
||||
return UNITY_END();
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue