myxce/src/lib/client.c
Rongsong Shen 0cad68507f
feat: acknowledged job start/finish/fail reports (bug-1741)
A worker reported job start, finish and failure with a fire-and-forget
OPCode_notify the controller never answers, so a report lost while the
worker stayed connected (the live incident: 6 of 7 finish reports in a
burst) left the job RUN forever; the rejoin resync (165fc158) only
recovers reports lost around a reconnect.

- controller.capnp: OPCode_notifyAcked -- the same NotifyData body,
  answered. A separate opcode because replies are matched through the
  client's single wbuf, not by sequence number: answering plain notifies
  would feed stale statuses to the next synchronous request.
- controller (cluster.c): do_notify_acked() applies the fact like
  do_notify, then waits on the commit barrier and replies 0 only once it
  is majority-committed. A barrier failure (the fact is still appended and
  retained) replies -E_XCE_NOT_READY, so the worker retries in place
  instead of rotating away from a healthy primary (cf. bug-1850). Admitted
  by the no-quorum fact-bearing gate like OPCode_notify.
- libxce: xce_notify_acked_req/xce_notify_acked (send, wait for the
  status) and xcecli_notify_acked (XCECLI_FAILOVER_CALL retry/failover).
  xcecli_notify_jobstart/jobfinish/jobfail use it; on -E_XCE_BAD_OPCODE
  (a controller predating the opcode) they fall back to the plain
  fire-and-forget notify, so workers and controllers can be upgraded in
  either order.
- Resends are safe: START/FINISH are duplicate-guarded and a repeated FAIL
  yields the same state.
- worker: logs a WARN when a report is still unacknowledged after the
  retry budget (the rejoin resync remains the backstop). These calls run
  on the worker daemon's IPC callback, so while the controller lacks
  quorum other jobs' IPC messages wait (buffered, not lost) up to the
  retry budget.

Tests: test_notify_acked_reply_only_after_commit and
test_notify_acked_withheld_replies_not_ready (no reply before the covering
ACK; -E_XCE_NOT_READY with the fact retained when ACKs are withheld).
Written with the opcode, so no separate RED run. Full /xce-build all:
111 suites pass; controller suite 1245/0.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-27 06:47:47 +00:00

1060 lines
29 KiB
C

#include <errno.h>
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include "client.h"
#include "client_internal.h"
#include "config.h" /* DEFAULT_CONNECT_TIMEOUT_MS */
#include "controller.capnp.h"
#include "mylock.h"
#include "request.h"
#include "xce/errnox.h" /* E_XCE_NOT_READY */
#include "xce/types.h"
#include "xceapi.h"
#include "xcectx.h"
#include "xlist.h"
int xcecli_ctx_init(xcecli_ctx_t* c, xce_client_config_t* cfg)
{
api_ctx_t* ctx;
int rc;
if ((c == NULL) || (cfg == NULL)) {
return -EINVAL;
}
if (cfg->nmasters < 1) {
return -EINVAL;
}
ctx = (api_ctx_t*)calloc(1, sizeof(*ctx));
if (ctx == NULL) {
return -ENOMEM;
}
ctx->conf = cfg;
rc = xce_ctx_init(&(ctx->ctx), ctx->conf->from, ctx->conf->key);
if (rc < 0) {
free(ctx);
return rc;
}
endpoint_init(&(ctx->ep), AF_INET, SOCK_STREAM, 0, 0);
/* set information for endpoint */
endpoint_set_peer(&(ctx->ep), &(ctx->ctx.myinfo));
xce_ctx_set_endpoint(&(ctx->ctx), &(ctx->ep));
*c = (void*)ctx;
return 0;
}
void xcecli_ctx_fini(xcecli_ctx_t c)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (ctx == NULL) {
return;
}
xce_ctx_fini(&(ctx->ctx));
endpoint_fini(&(ctx->ep));
free(ctx);
}
int xcecli_set_handler(xcecli_ctx_t c, xce_request_handler_t handler,
void* args)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if ((c == NULL) || (handler == NULL)) {
return -EINVAL;
}
return xce_ctx_set_handler(&(ctx->ctx), handler, args);
}
/*
* xcecli_set_reconnect_handler (task 4.1) — register the consumer rejoin hook
* the failover loop invokes after a transparent reconnect, before resuming the
* interrupted op. A NULL cb clears any previously registered hook (back to the
* pure-client no-op path). Stores cb+arg on the per-ctx failover state.
*/
int xcecli_set_reconnect_handler(xcecli_ctx_t c, xcecli_reconnect_cb cb,
void* arg)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
ctx->failover.rejoin_cb = cb;
ctx->failover.rejoin_arg = arg;
return 0;
}
int xcecli_wakeup(xcecli_ctx_t c, int status)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
return xce_wakeup(&(ctx->ctx), status);
}
int xcecli_try_wakeup(xcecli_ctx_t c, int status)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
return xce_try_wakeup(&(ctx->ctx), status);
}
int xcecli_start(xcecli_ctx_t c)
{
int rc = -ECONNREFUSED;
api_ctx_t* ctx = (api_ctx_t*)c;
int i;
int connected_idx = -1;
if (c == NULL) {
return -EINVAL;
}
/* Initial connect rotates through ALL configured masters, not just
* masters[0] (bug-854 dead-first-master): if the first master is down
* (connection-refused after a failover that killed it), a hardcoded
* masters[0] connect fails the whole `client` session before the failover
* loop — which only runs on an ALREADY-connected ctx — ever gets a chance
* to rotate. Try each master in order; the first that connects becomes the
* sticky cursor start. The subsequent ctrlstatus/op failover loop then
* rotates onward to the authoritative primary as needed.
*
* ctx->ep was endpoint_init'd + bound to the iomgr in xcecli_init. A failed
* endpoint_connect leaves the socket fd unusable, so before retrying the
* NEXT master we must rebuild the endpoint exactly as xce_ctx_reconnect
* does: endpoint_fini → endpoint_init (fresh fd) → re-set peer + re-bind the
* iomgr (endpoint_connect needs ep->iomgr->ssl_ctx). */
for (i = 0; i < ctx->conf->nmasters; i++) {
if (i > 0) {
endpoint_fini(&(ctx->ep));
if (endpoint_init(&(ctx->ep), AF_INET, SOCK_STREAM, 0, 0) < 0)
continue;
endpoint_set_peer(&(ctx->ep), &(ctx->ctx.myinfo));
ctx->ep.iomgr = &(ctx->ctx.iomgr);
}
rc = endpoint_connect(&(ctx->ep), ctx->conf->masters[i],
ctx->conf->master_ports[i],
DEFAULT_CONNECT_TIMEOUT_MS);
if (rc >= 0) {
connected_idx = i;
break;
}
}
if (connected_idx < 0) {
/* No master reachable at all — surface the last transport error. */
endpoint_fini(&(ctx->ep));
return rc;
}
/* Initialize per-context failover state: sticky primary starts at the master
* we actually connected to (not always index 0), retry tunables default to
* XCE_RETRY_DEFAULTS, and no rejoin hook is installed for pure clients. */
rc = mylock_init(&(ctx->failover.lock));
if (rc < 0) {
endpoint_fini(&(ctx->ep));
return rc;
}
ctx->failover.cur_master = connected_idx;
ctx->failover.retry = XCE_RETRY_DEFAULTS;
ctx->failover.rejoin_cb = NULL;
ctx->failover.rejoin_arg = NULL;
rc = xce_ctx_start(&(ctx->ctx));
return rc;
}
int xcecli_stop(xcecli_ctx_t c)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
xce_ctx_stop(&(ctx->ctx));
endpoint_fini(&(ctx->ep));
mylock_fini(&(ctx->failover.lock));
return 0;
}
endpoint_t* xcecli_get_my_endpoint(xcecli_ctx_t c)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return NULL;
}
return &(ctx->ep);
}
xce_peerinfo_t* xcecli_get_myinfo(xcecli_ctx_t c)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return NULL;
}
return &(ctx->ctx.myinfo);
}
int xcecli_hello(xcecli_ctx_t c, char* msg)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
xce_hello(&(ctx->ctx), msg));
}
int xcecli_nop(xcecli_ctx_t c)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
/* Fire-and-forget: best-effort transport reconnect on a known-dead
* connection, then a single-shot send. No status classification, no
* no-primary backoff, no retry budget (task 5.2). */
return XCECLI_FF_CALL(ctx, xce_nop(&(ctx->ctx)));
}
int xcecli_heartbeat(xcecli_ctx_t c, int hb_interval)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
/* Fire-and-forget: best-effort transport reconnect on a known-dead
* connection, then a single-shot send (task 5.2). A heartbeat must NEVER go
* through the no-primary backoff retry budget. */
return XCECLI_FF_CALL(ctx, xce_heartbeat(&(ctx->ctx), hb_interval));
}
bool xcecli_is_connection_broken(xcecli_ctx_t c)
{
/* Delegates to the exact predicate XCECLI_FF_CALL already uses
* internally (client_internal.h) -- same NULL-safe, socket-free,
* side-effect-free read of ctx->ctx.state, just exposed publicly (see
* client.h's doc comment for the full rationale, bug-1760). */
return xcecli_ff_needs_reconnect((api_ctx_t*)c);
}
int xcecli_shutdown(xcecli_ctx_t c)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
return xce_shutdown(&(ctx->ctx));
}
int xcecli_reslist(xcecli_ctx_t c, xce_resource_t*** res)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
/* xce_reslist returns a non-negative ROW COUNT on success, not a 0/status
* code — must go through the count-aware macro (bug-1436). */
return XCECLI_FAILOVER_CALL_COUNT(ctx, xce_reslist(&(ctx->ctx), res));
}
int xcecli_join(xcecli_ctx_t c, int role, hostattrs_t* attrs)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (ctx == NULL) {
return -EINVAL;
}
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_join(&(ctx->ctx), role, attrs));
}
int xcecli_workerlist(xcecli_ctx_t c, xce_worker_t*** workers)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (ctx == NULL) {
return -EINVAL;
}
/* xce_workerlist returns a non-negative ROW COUNT on success, not a
* 0/status code — must go through the count-aware macro (bug-1436). */
return XCECLI_FAILOVER_CALL_COUNT(ctx, xce_workerlist(&(ctx->ctx), workers));
}
int xcecli_jobdispatch(xcecli_ctx_t c, xce_job_dispatch_t* info)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_jobdispatch(&(ctx->ctx), info));
}
/* bug-1425: see the doc comment on the declaration in client_internal.h for
* the full mechanism this closes. */
void jobsubmit_finalize_jobid(uint64_t* jobid, uint64_t submitted_jobid,
int overall_rc)
{
if (jobid != NULL && overall_rc >= 0)
*jobid = submitted_jobid;
}
int xcecli_jobsubmit(xcecli_ctx_t c, xce_job_submit_t* submit, uint64_t* jobid)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
int rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_jobsubmit(&(ctx->ctx), submit, jobid));
jobsubmit_finalize_jobid(jobid, submit->jobid, rc);
return rc;
}
int xcecli_joblist(xcecli_ctx_t c, xce_job_t*** jobs)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
/* xce_joblist returns a non-negative ROW COUNT on success, not a 0/status
* code — a count that happens to equal a retriable errno's magnitude
* (e.g. exactly 111 jobs == ECONNREFUSED) would be misclassified as a
* transport failure by the generic macro, forcing a spurious reconnect
* storm that eventually surfaces a real "Connection reset by peer" once
* it coincides with a genuine transient (bug-1436). Must go through the
* count-aware macro instead. */
return XCECLI_FAILOVER_CALL_COUNT(ctx, xce_joblist(&(ctx->ctx), jobs));
}
int xcecli_clusterconf(xcecli_ctx_t c, xce_clusterconf_t** conf)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
xce_clusterconf(&(ctx->ctx), conf));
}
int xcecli_ctrlstatus(xcecli_ctx_t c, xce_ctrlstatus_t** status)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
xce_ctrlstatus(&(ctx->ctx), status));
}
/*
* The primary's responder role on the wire is ctrl_role_t PRIMARY. That enum
* lives in the controller-private ha.h (CTRL_PRIMARY == 0) and is intentionally
* not exposed to libxce; the wire ordinal is part of the ctrlstatus reply
* contract (controller/ctrlstatus.c stamps responderRole = ha->role). We pin it
* here as a named constant so the comparison reads intentionally rather than as
* a bare 0. Any reorder of ctrl_role_t is a Revalidation Trigger for this spec.
*/
#define XCECLI_CTRLSTATUS_ROLE_PRIMARY ((uint8_t)0)
bool xcecli_ctrlstatus_is_authoritative(const xce_ctrlstatus_t* status)
{
if (status == NULL) {
return false;
}
return status->responderRole == XCECLI_CTRLSTATUS_ROLE_PRIMARY;
}
int xcecli_ctrlstatus_primary_addr(const xce_ctrlstatus_t* status,
char* host_out, size_t host_sz,
uint16_t* port_out)
{
int i;
if ((status == NULL) || (host_out == NULL) || (host_sz == 0)
|| (port_out == NULL)) {
return -EINVAL;
}
/* No known primary (empty/NULL primaryName) ⇒ nothing to target; the caller
* falls back to plain master rotation (Req 2.3 unknown-primary). */
if ((status->primaryName == NULL) || (status->primaryName[0] == '\0')) {
return -ENOENT;
}
/* Map the reported primaryName to its configured address via the reply's
* peers list. The list is a small bounded set (configured controllers). */
for (i = 0; i < status->peers.count; i++) {
xce_ctrlpeer_t* p = status->peers.list[i];
if ((p == NULL) || (p->name == NULL) || (p->host == NULL)) {
continue;
}
if (strcmp(p->name, status->primaryName) != 0) {
continue;
}
if ((p->host[0] == '\0') || (p->port == 0)) {
/* named primary has no usable address — treat as absent */
break;
}
strncpy(host_out, p->host, host_sz - 1);
host_out[host_sz - 1] = '\0';
*port_out = (uint16_t)p->port;
return 0;
}
/* primaryName not present in peers (or no usable address): fall back to
* rotation rather than targeting a bogus address. */
return -ENOENT;
}
int xcecli_ctrlstatus_primary(xcecli_ctx_t c, xce_ctrlstatus_t** status)
{
api_ctx_t* ctx = (api_ctx_t*)c;
xce_ctrlstatus_t* hint;
int rc;
int attempt;
int max_attempts;
if ((c == NULL) || (status == NULL)) {
return -EINVAL;
}
/* Bound the redirect loop by the configured failover retry budget so we
* never spin unbounded chasing a never-appearing primary (each inner
* xcecli_ctrlstatus already runs its own full reach-some-node failover).
* Floor it at the master-ring size so a standby-first / dead-first master
* list is always traversable in full before giving up (bug-854): one extra
* attempt lets the rotation that lands on the primary actually query it. */
max_attempts = ctx->failover.retry.max_retries;
if (max_attempts < 1) {
max_attempts = 1;
}
if (ctx->conf != NULL && ctx->conf->nmasters + 1 > max_attempts) {
max_attempts = ctx->conf->nmasters + 1;
}
for (attempt = 0; attempt < max_attempts; attempt++) {
hint = NULL;
rc = xcecli_ctrlstatus(c, &hint);
if (rc < 0) {
/* no reachable controller within the inner budget ⇒ typed failure;
* *status untouched (never a hint reply as the result, Req 9.2). */
return rc;
}
if (xcecli_ctrlstatus_is_authoritative(hint)) {
/* the primary itself answered — authoritative result */
*status = hint;
return 0;
}
/* Non-authoritative reply (standby/electing/init). The inner
* xcecli_ctrlstatus SUCCEEDS against ANY reachable node — ctrlstatus is
* answerable in every role — so it classifies the standby's status-0
* reply as success and NEVER rotates off it on its own. We must advance
* the endpoint here, or the loop re-queries the same non-primary every
* attempt and exhausts the budget without ever reaching the primary
* (bug-854). Capture a redirect hint when the responder named the primary
* (1-hop direct redirect), then FORCE rotation: xce_failover_rotate
* consumes the hint when set, else advances to the next configured
* master; recover repairs the ctx so the next attempt's send proceeds.
* Mirror xcecli_retry_next: only recover when the rotate succeeded — a
* failed rotate leaves the ctx for the next inner call's own transport-
* error failover to heal. All under failover.lock (the inner call takes
* and releases it itself, so it is not held here). */
{
char phost[XCE_MAXHOSTNAME_LEN];
uint16_t pport = 0;
mylock_lock(&ctx->failover.lock);
if (xcecli_ctrlstatus_primary_addr(hint, phost, sizeof(phost),
&pport)
== 0) {
xce_ctx_capture_redirect_hint(&ctx->ctx, phost, pport);
}
if (xce_failover_rotate(ctx) == 0) {
xcecli_ctx_recover(ctx);
}
mylock_unlock(&ctx->failover.lock);
}
/* Free the non-authoritative interim reply before the next attempt — no
* leak, and we never return a hint as the authoritative result. */
xcecli_free_ctrlstatus(hint);
}
/* Budget exhausted without the primary answering: typed failure, *status
* untouched (Req 9.2 — never return a non-primary reply as authoritative).
*/
return -E_XCE_NOT_READY;
}
int xcecli_notify_jobfinish(xcecli_ctx_t c, xce_jobfinish_t* jfinish)
{
api_ctx_t* ctx = (api_ctx_t*)c;
/* clang-format off */
xce_notify_data_t data = {
.event = NotifyData_data_jobFinishInfo,
.data = {
.finishinfo = jfinish
}
};
/* clang-format on */
if (c == NULL) {
return -EINVAL;
}
if (jfinish == NULL) {
return -EINVAL;
}
return xcecli_notify_acked(ctx, &data); /* bug-1741 */
}
int xcecli_update_resvals(xcecli_ctx_t c, int nres, xce_resval_t** vals)
{
api_ctx_t* ctx = (api_ctx_t*)c;
int rc;
if ((ctx == NULL) || (vals == NULL)) {
return -EINVAL;
}
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_update_resvals(&(ctx->ctx), nres, vals));
return rc;
}
int xcecli_update_jusage(xcecli_ctx_t c, uint64_t jobid, int npids, int* pids,
int npgids, int* pgids, int nusage,
xce_rusage_t** usages)
{
api_ctx_t* ctx = (api_ctx_t*)c;
int rc;
if ((ctx == NULL) || (jobid == 0)) {
return -EINVAL;
}
rc = XCECLI_FAILOVER_CALL(
ctx, /*is_write=*/true,
xce_update_jusage(&(ctx->ctx), jobid, npids, pids, npgids, pgids, nusage,
usages));
return rc;
}
int xcecli_resval(xcecli_ctx_t c, char* host, char* res, xce_resval_t*** val)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
if (res == NULL) {
return -EINVAL;
}
/* xce_resval returns a non-negative ROW COUNT on success, not a 0/status
* code — must go through the count-aware macro (bug-1436). */
return XCECLI_FAILOVER_CALL_COUNT(ctx, xce_resval(&(ctx->ctx), host, res, val));
}
int xcecli_notify(xcecli_ctx_t c, xce_notify_data_t* data)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
if (data == NULL) {
return -EINVAL;
}
/* Fire-and-forget chokepoint: every thin notify_* wrapper delegates
* here, so applying the best-effort reconnect at this single point covers
* them all without double-wrapping/double-locking. Best-effort transport
* reconnect on a known-dead connection, then a single-shot send (task
* 5.2). */
return XCECLI_FF_CALL(ctx, xce_notify(&(ctx->ctx), data));
}
/* bug-1741: see client.h. A lost job start/finish/fail report used to be
* gone for good (fire-and-forget, no reply), leaving the job RUN forever
* unless the worker happened to rejoin. This resends until the controller
* confirms the fact committed (duplicates are harmless controller-side). An
* old controller answers the unknown opcode with -E_XCE_BAD_OPCODE: send
* the plain fire-and-forget notify instead, so worker and controller can be
* upgraded in either order. */
int xcecli_notify_acked(xcecli_ctx_t c, xce_notify_data_t* data)
{
api_ctx_t* ctx = (api_ctx_t*)c;
int rc;
if ((c == NULL) || (data == NULL)) {
return -EINVAL;
}
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_notify_acked(&(ctx->ctx), data));
if (rc == -E_XCE_BAD_OPCODE) {
rc = xcecli_notify(ctx, data);
}
return rc;
}
int xcecli_notify_jobstart(xcecli_ctx_t c, job_pidinfo_t* pidinfo)
{
api_ctx_t* ctx = (api_ctx_t*)c;
/* clang-format off */
xce_notify_data_t data = {
.event = NotifyData_data_jobPidInfo,
.data = {
.pidinfo = pidinfo
}
};
/* clang-format on */
if ((c == NULL) || (pidinfo == NULL)) {
return -EINVAL;
}
return xcecli_notify_acked(ctx, &data); /* bug-1741 */
}
int xcecli_notify_jobfail(xcecli_ctx_t c, job_failinfo_t* failinfo)
{
api_ctx_t* ctx = (api_ctx_t*)c;
/* clang-format off */
xce_notify_data_t data = {
.event = NotifyData_data_jobFailInfo,
.data = {
.failinfo = failinfo
}
};
/* clang-format on */
if ((c == NULL) || (failinfo == NULL)) {
return -EINVAL;
}
return xcecli_notify_acked(ctx, &data); /* bug-1741 */
}
int xcecli_jobctrl(xcecli_ctx_t c, xce_jobctrl_t* ctrlinfo)
{
api_ctx_t* ctx = (api_ctx_t*)c;
int rc;
if ((ctx == NULL) || (ctrlinfo == NULL)) {
return -EINVAL;
}
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_jobctrl(&(ctx->ctx), ctrlinfo));
return rc;
}
int xcecli_notify_jobstop(xcecli_ctx_t c, uint64_t jobid)
{
api_ctx_t* ctx = (api_ctx_t*)c;
/* clang-format off */
xce_notify_data_t data = {
.event = NotifyData_data_stoppedJob,
.data = {
.stopped_job = jobid
}
};
/* clang-format on */
if ((c == NULL) || (jobid == 0)) {
return -EINVAL;
}
return xcecli_notify(ctx, &data);
}
int xcecli_notify_jobresume(xcecli_ctx_t c, uint64_t jobid)
{
api_ctx_t* ctx = (api_ctx_t*)c;
/* clang-format off */
xce_notify_data_t data = {
.event = NotifyData_data_resumedJob,
.data = {
.resumed_job = jobid
}
};
/* clang-format on */
if ((c == NULL) || (jobid == 0)) {
return -EINVAL;
}
return xcecli_notify(ctx, &data);
}
int xcecli_notify_jobchkpt(xcecli_ctx_t c, uint64_t jobid, char* info)
{
api_ctx_t* ctx = (api_ctx_t*)c;
/* clang-format off */
xce_ckptinfo_t ckptinfo = {
.jobid = jobid,
.ckptinfo = info
};
xce_notify_data_t data = {
.event = NotifyData_data_ckptinfo,
.data = {
.ckptinfo = &ckptinfo
}
};
/* clang-format on */
if ((c == NULL) || (jobid == 0) || (info == NULL)) {
return -EINVAL;
}
return xcecli_notify(ctx, &data);
}
int xcecli_notify_jobpostdone(xcecli_ctx_t c, uint64_t jobid)
{
api_ctx_t* ctx = (api_ctx_t*)c;
/* clang-format off */
xce_notify_data_t data = {
.event = NotifyData_data_postDoneJob,
.data = {
.pdone_job = jobid
}
};
/* clang-format on */
if ((c == NULL) || (jobid == 0)) {
return -EINVAL;
}
return xcecli_notify(ctx, &data);
}
int xcecli_queuectrl(xcecli_ctx_t c, xce_queuectrl_t* ctrlinfo)
{
api_ctx_t* ctx = (api_ctx_t*)c;
int rc;
if ((ctx == NULL) || (ctrlinfo == NULL)) {
return -EINVAL;
}
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_queuectrl(&(ctx->ctx), ctrlinfo));
return rc;
}
int xcecli_hostctrl(xcecli_ctx_t c, xce_hostctrl_t* ctrlinfo)
{
api_ctx_t* ctx = (api_ctx_t*)c;
int rc;
if ((ctx == NULL) || (ctrlinfo == NULL)) {
return -EINVAL;
}
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_hostctrl(&(ctx->ctx), ctrlinfo));
return rc;
}
int xcecli_upgradeconfig(xcecli_ctx_t c, const uint8_t *sha256,
size_t sha256_len, const uint8_t *bytes,
size_t bytes_len)
{
api_ctx_t* ctx = (api_ctx_t*)c;
int rc;
if ((ctx == NULL) || (sha256 == NULL) || (sha256_len == 0)
|| (bytes == NULL) || (bytes_len == 0)) {
return -EINVAL;
}
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_upgradeconfig(&(ctx->ctx), sha256,
sha256_len, bytes,
bytes_len));
return rc;
}
int xcecli_runtask(xcecli_ctx_t c, xce_taskrun_t* info)
{
api_ctx_t* ctx = (api_ctx_t*)c;
int rc;
if ((ctx == NULL) || (info == NULL)) {
return -EINVAL;
}
info->clientidx = -1; /* this field will be used by
* controller to trace back
* client
*/
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_runtask(&(ctx->ctx), info));
return rc;
}
int xcecli_notify_taskstart(xcecli_ctx_t c, xce_task_pidinfo_t* pidinfo)
{
api_ctx_t* ctx = (api_ctx_t*)c;
/* clang-format off */
xce_notify_data_t data = {
.event = NotifyData_data_taskPidInfo,
.data = {
.taskinfo = pidinfo
}
};
/* clang-format on */
if ((c == NULL) || (pidinfo == NULL)) {
return -EINVAL;
}
return xcecli_notify(ctx, &data);
}
int xcecli_notify_taskfail(xcecli_ctx_t c, xce_task_failinfo_t* failinfo)
{
api_ctx_t* ctx = (api_ctx_t*)c;
/* clang-format off */
xce_notify_data_t data = {
.event = NotifyData_data_taskFailInfo,
.data = {
.taskfailinfo = failinfo
}
};
/* clang-format on */
if ((c == NULL) || (failinfo == NULL)) {
return -EINVAL;
}
return xcecli_notify(ctx, &data);
}
int xcecli_notify_taskfinish(xcecli_ctx_t c, xce_task_finish_t* info)
{
api_ctx_t* ctx = (api_ctx_t*)c;
/* clang-format off */
xce_notify_data_t data = {
.event = NotifyData_data_taskFinish,
.data = {
.taskfinish = info
}
};
/* clang-format on */
if ((c == NULL) || (info == NULL)) {
return -EINVAL;
}
return xcecli_notify(ctx, &data);
}
void xcecli_block_tasks(xcecli_ctx_t c)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return;
}
xce_block_tasks(&(ctx->ctx));
}
void xcecli_unblock_tasks(xcecli_ctx_t c)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return;
}
xce_unblock_tasks(&(ctx->ctx));
}
int xcecli_set_pend_reason(xcecli_ctx_t c, xce_preason_set_t* reasons)
{
api_ctx_t* ctx = (api_ctx_t*)c;
int rc;
if ((ctx == NULL) || (reasons == NULL)) {
return -EINVAL;
}
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_set_pend_reason(&(ctx->ctx), reasons));
return rc;
}
int xcecli_clear_reason(xcecli_ctx_t c, int njobs, uint64_t* jobs)
{
api_ctx_t* ctx = (api_ctx_t*)c;
int rc;
/* clang-format off */
xce_clear_reason_t reason = {
.njobs = njobs,
.jobs = jobs
};
/* clang-format on */
if ((ctx == NULL) || ((njobs > 0) && (jobs == NULL))) {
return -EINVAL;
}
if (njobs == 0) {
return 0;
}
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_clear_reason(&(ctx->ctx), &reason));
return rc;
}
int xcecli_jobinfo(xcecli_ctx_t c, uint64_t jobid, xce_job_t** job)
{
api_ctx_t* ctx = (api_ctx_t*)c;
int rc;
if ((ctx == NULL) || (job == NULL)) {
return -EINVAL;
}
/* xce_jobinfo returns 1 (found) / 0 (not found) / negative (error) — a
* fixed non-zero success value, not a 0/status code. 1 never happens to
* collide with a recognized retriable magnitude today, but it is the same
* class of bug as bug-1436 (xcecli_joblist) — go through the count-aware
* macro for correctness and consistency rather than relying on that
* coincidence. */
rc = XCECLI_FAILOVER_CALL_COUNT(ctx, xce_jobinfo(&(ctx->ctx), jobid, job));
return rc;
}
int xcecli_reply_fetchfile(xcecli_ctx_t c, xce_request_t* request,
xce_fetchfile_reply_t* freply)
{
api_ctx_t* ctx = (api_ctx_t*)c;
xce_reply_t reply;
if ((ctx == NULL) || (request == NULL) || (freply == NULL)) {
return -EINVAL;
}
init_reply_by_req(&reply, request);
reply.status = freply->error;
reply.data_which = RequestReply_data_fetchfilereply;
reply.data.fetchfilereply = freply;
return xce_reply(&(ctx->ep), &(ctx->ctx.myinfo), &reply);
}
/*
* xcecli_reply_schedadmin - see client.h's declaration comment for the full
* contract (scheduler-core spec, task 9.4, Req 5 AC 10). Builds the
* RequestReply envelope directly (no init_reply_by_req(): unlike
* xcecli_reply_fetchfile() above, the caller may not have a live
* xce_request_t any more -- recordStart/recordStop replies are sent from
* the driver thread, after the original request has already been freed).
* `reply.status` is always 0: a successfully-built SchedAdminReply is
* always a valid RPC response, whether its own `.ok` field is true or
* false (business-level success/failure lives inside the SchedAdminReply
* payload itself, per its own `ok`/`error` fields -- same convention
* handle_config_request() (controller/config_repl.c) uses for
* ReplConfigReply's "up to date" case).
*/
int xcecli_reply_schedadmin(xcecli_ctx_t c, uint64_t seq,
sched_admin_reply_t* areply)
{
api_ctx_t* ctx = (api_ctx_t*)c;
xce_reply_t reply;
if ((ctx == NULL) || (areply == NULL)) {
return -EINVAL;
}
memset(&reply, 0, sizeof(reply));
reply.seq = seq;
reply.opcode = OPCode_schedAdmin;
reply.status = 0;
reply.data_which = RequestReply_data_schedadminreplymsg;
reply.data.schedadminreplymsg = areply;
return xce_reply(&(ctx->ep), &(ctx->ctx.myinfo), &reply);
}
int xcecli_fetchfile(xcecli_ctx_t c, xce_fetchfile_t* msg,
xce_fetchfile_reply_t** reply)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if ((ctx == NULL) || (msg == NULL) || (reply == NULL)) {
return -EINVAL;
}
return xce_fetchfile(&(ctx->ctx), msg, reply);
}