lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/include/client.h
Rongsong Shen 88b0ac9937
fix(scheduler): close post-failover rejoin detection gap (bug-1760) 
An idle scheduler (nothing queued to publish) never noticed a dead
controller connection: sched_needs_rejoin was only ever set as a side
effect of an active client-library call, and an idle round makes none.
After any controller failover with nothing already in flight, jobs
submitted to the new primary stayed PEND indefinitely.

Adds xcecli_is_connection_broken() (libxce) -- a pure, socket-free read
of the transport's already-async-maintained dead/alive flag, set for
free by the existing RDHUP disconnect handler with zero application
calls. sched_check_pending_rejoin() now retries xcecli_join() on either
the existing reactive flag or this new passive signal, so a healthy
idle scheduler still makes zero extra network calls while a genuinely
dead connection self-heals within one ~1s round -- live-verified
against a 3-node HA cluster (rejoin fired ~300ms after new-primary
election with nothing submitted in between).

Also closes out bug-1735 (job memory/CPU usage reporting): confirmed
already fixed on this branch since 2026-08-30, just never marked in
the bug log.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XYdtrEWWZfKRRGED7raEYg
2026-09-04 04:37:11 +00:00

173 lines
9.1 KiB
C
Raw Blame History

#if !defined(_CLIENT_H_)
#define _CLIENT_H_ 1
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include "config.h"
#include "netio.h"
#include "request.h"
#include "xce/types.h"
#include "xceapi.h"
typedef void* xcecli_ctx_t;
/*
* xcecli_reconnect_cb: consumer rejoin hook invoked once per transparent
* reconnect/failover, AFTER the reconnect/recovery and BEFORE the interrupted
* op resumes. The callback CONTENT (join handshake, worker job re-sync) belongs
* to the consumer; libxce owns only the mechanism + ordering. Runs on the
* calling thread. Return 0 to proceed with the interrupted op; <0 to treat the
* rejoin as failed (the new primary rejected the join) so the failover loop
* rotates/backs off again instead of resuming. Pure clients register NO hook
* (NULL) — the loop then resumes directly.
*/
typedef int (*xcecli_reconnect_cb)(xcecli_ctx_t ctx, void* arg);
/* N_MAX_MASTERS derives from config.h's MAX_HA_CANDIDATES so the two limits
* can never silently diverge (Req 1 AC7, component-failover-adoption). */
#define N_MAX_MASTERS MAX_HA_CANDIDATES
typedef struct {
int nmasters;
char* masters[N_MAX_MASTERS];
int master_ports[N_MAX_MASTERS];
char* from;
char* key;
} xce_client_config_t;
#define xcecli_jobfinish(ctx, finish) xcecli_notify_jobfinish(ctx, finish)
extern int xcecli_ctx_init(xcecli_ctx_t*, xce_client_config_t*);
extern void xcecli_ctx_fini(xcecli_ctx_t);
extern int xcecli_set_handler(xcecli_ctx_t, xce_request_handler_t, void*);
extern int xcecli_set_reconnect_handler(xcecli_ctx_t, xcecli_reconnect_cb,
void*);
extern int xcecli_start(xcecli_ctx_t);
extern int xcecli_stop(xcecli_ctx_t);
extern int xcecli_wakeup(xcecli_ctx_t, int);
extern int xcecli_hello(xcecli_ctx_t, char*);
extern int xcecli_nop(xcecli_ctx_t);
extern int xcecli_heartbeat(xcecli_ctx_t, int);
/*
* xcecli_is_connection_broken (bug-1760) - PURE, socket-free, side-effect-free
* query: has the transport been detected dead since the last successful
* (re)connect? Mirrors the private xcecli_ff_needs_reconnect() predicate
* (client_internal.h) exactly, exposed publicly so a consumer's own idle
* loop can passively check for a dead connection -- with ZERO network I/O
* and zero blocking risk -- instead of periodically probing a possibly-
* healthy one. The underlying signal (ctx->state == XCE_CTX_ERR) is set
* asynchronously by the transport's own RDHUP/EPOLLERR handler
* (handle_disconnect(), xcectx.c) the instant the connection drops, with no
* application call involved -- this function only ever reads that
* already-computed flag. Returns false for a NULL ctx.
*/
extern bool xcecli_is_connection_broken(xcecli_ctx_t);
extern int xcecli_shutdown(xcecli_ctx_t);
extern int xcecli_reslist(xcecli_ctx_t, xce_resource_t***);
extern int xcecli_join(xcecli_ctx_t, int, hostattrs_t*);
extern int xcecli_workerlist(xcecli_ctx_t, xce_worker_t***);
extern int xcecli_jobdispatch(xcecli_ctx_t, xce_job_dispatch_t*);
extern int xcecli_joblist(xcecli_ctx_t, xce_job_t***);
extern int xcecli_clusterconf(xcecli_ctx_t, xce_clusterconf_t**);
extern int xcecli_ctrlstatus(xcecli_ctx_t, xce_ctrlstatus_t**);
/*
* xcecli_ctrlstatus_primary (controller-status task 3.2): AUTHORITATIVE
* controller status read. Returns ONLY a reply whose responder role is the
* primary. When the first answer is from a non-primary, it follows the redirect
* (targets the reported primary via the reply's peer list, else rotates the
* configured controller list) until the primary answers or the failover budget
* is exhausted. On budget exhaustion / no reachable primary it returns a typed
* negative and NEVER hands back a non-primary hint reply (Req 9.1, 9.2). The
* any-node hint view is still available via xcecli_ctrlstatus above.
*/
extern int xcecli_ctrlstatus_primary(xcecli_ctx_t, xce_ctrlstatus_t**);
/*
* Pure decision helpers backing the authoritative read (no I/O — unit-testable
* and exposed for tests). xcecli_ctrlstatus_is_authoritative returns true iff
* the reply's responderRole is the primary role (ctrl_role_t PRIMARY == 0).
* xcecli_ctrlstatus_primary_addr maps the reply's primaryName to its host:port
* via the reply's peers list: returns 0 and fills host/port on a hit, -ENOENT
* when the primary is unknown/absent from peers, -EINVAL on a bad argument.
*/
extern bool xcecli_ctrlstatus_is_authoritative(const xce_ctrlstatus_t*);
extern int xcecli_ctrlstatus_primary_addr(const xce_ctrlstatus_t*,
char* host_out, size_t host_sz,
uint16_t* port_out);
extern int xcecli_notify_jobfinish(xcecli_ctx_t, xce_jobfinish_t*);
extern int xcecli_notify_jobstart(xcecli_ctx_t, job_pidinfo_t*);
extern int xcecli_notify_jobfail(xcecli_ctx_t c, job_failinfo_t*);
extern int xcecli_notify_jobstop(xcecli_ctx_t, uint64_t);
extern int xcecli_notify_jobresume(xcecli_ctx_t, uint64_t);
extern int xcecli_notify_jobpostdone(xcecli_ctx_t, uint64_t);
extern int xcecli_jobsubmit(xcecli_ctx_t, xce_job_submit_t*, uint64_t*);
extern int xcecli_update_resvals(xcecli_ctx_t, int, xce_resval_t**);
extern int xcecli_resval(xcecli_ctx_t, char*, char*, xce_resval_t***);
extern int xcecli_update_jusage(xcecli_ctx_t, uint64_t, int, int*, int, int*,
int, xce_rusage_t**);
extern int xcecli_notify(xcecli_ctx_t, xce_notify_data_t*);
extern int xcecli_jobctrl(xcecli_ctx_t, xce_jobctrl_t*);
extern int xcecli_notify_jobchkpt(xcecli_ctx_t, uint64_t, char*);
extern int xcecli_queuectrl(xcecli_ctx_t, xce_queuectrl_t*);
extern int xcecli_hostctrl(xcecli_ctx_t, xce_hostctrl_t*);
/* xcecli_upgradeconfig: submit a configuration upgrade (sha256 + serialised
* ClusterConfig blob) to the primary via OP_upgradeConfig, transparently
* surviving a primary change mid-call via XCECLI_FAILOVER_CALL (same
* write-op failover contract as xcecli_hostctrl/xcecli_queuectrl). Returns
* the server's reply status (0 on success). */
extern int xcecli_upgradeconfig(xcecli_ctx_t, const uint8_t *sha256,
size_t sha256_len, const uint8_t *bytes,
size_t bytes_len);
extern int xcecli_runtask(xcecli_ctx_t, xce_taskrun_t*);
extern int xcecli_notify_taskstart(xcecli_ctx_t, xce_task_pidinfo_t*);
extern int xcecli_notify_taskfail(xcecli_ctx_t, xce_task_failinfo_t*);
extern int xcecli_notify_taskfinish(xcecli_ctx_t, xce_task_finish_t*);
extern endpoint_t* xcecli_get_my_endpoint(xcecli_ctx_t);
extern xce_peerinfo_t* xcecli_get_myinfo(xcecli_ctx_t);
extern void xcecli_block_tasks(xcecli_ctx_t);
extern void xcecli_unblock_tasks(xcecli_ctx_t);
extern int xcecli_set_pend_reason(xcecli_ctx_t, xce_preason_set_t*);
extern int xcecli_clear_reason(xcecli_ctx_t, int, uint64_t*);
extern int xcecli_jobinfo(xcecli_ctx_t, uint64_t, xce_job_t**);
#define xcecli_free_ress(nres, ress) xce_free_ress(nres, ress)
#define xcecli_free_workers(nworkers, workers) \
xce_free_workers(nworkers, workers)
#define xcecli_free_jobs(njobs, jobs) xce_free_jobs(njobs, jobs)
#define xcecli_free_queues(nqueues, queues) xce_free_queues(nqueues, queues)
#define xcecli_free_groups(ngroups, groups) xce_free_groups(ngroups, groups)
#define xcecli_free_clusterconf(conf) xce_free_clusterconf(conf)
/* ctrlstatus reply is freed by the generated codec free. free_CtrlStatus()
* only frees the fields INSIDE the struct; the xce_ctrlstatus_t itself is
* calloc'd by decode_CtrlStatus_ptr() and must be freed via free_CtrlStatus_ptr()
* (which also nulls the caller's pointer) or every call leaks sizeof(xce_ctrlstatus_t)
* (bug-1401). */
#define xce_free_ctrlstatus(s) free_CtrlStatus_ptr(&(s))
#define xcecli_free_ctrlstatus(s) xce_free_ctrlstatus(s)
extern int xcecli_reply_fetchfile(xcecli_ctx_t, xce_request_t*, xce_fetchfile_reply_t*);
extern int xcecli_fetchfile(xcecli_ctx_t, xce_fetchfile_t*, xce_fetchfile_reply_t**);
/*
* xcecli_reply_schedadmin: send a SchedAdminReply back over `c`'s
* connection in response to a previously-received OPCode_schedAdmin
* request (scheduler-core spec, task 9.4, Req 5 AC 10). Mirrors
* xcecli_reply_fetchfile() above exactly, with one difference: it takes
* the original request's `seq` directly rather than an `xce_request_t*`,
* since `recordStart`/`recordStop` defer this reply to the driver thread
* (see driver.c's xce_sched_driver_translate_pending_events()), by which
* point the original xce_request_t has already been freed by the
* request-handling thread (xcectx.c's vtask_entry(), immediately after the
* registered handler returns) -- seq is the only field of it the reply
* needs (opcode is always OPCode_schedAdmin). `recordStatus` (answered
* synchronously from the handler thread, real request still in scope)
* uses this same function, simply passing `request->seq` directly.
* Returns xce_reply()'s own return value; -EINVAL if `c` or `areply` is
* NULL.
*/
extern int xcecli_reply_schedadmin(xcecli_ctx_t, uint64_t seq,
sched_admin_reply_t* areply);
#endif
Reference in a new issue View git blame Copy permalink