feat(component-failover-adoption): add apiserver's rejoin logic with correctly-placed cgo registration
Registers a rejoin flag-setter (go_backend_rejoin_handler) as the client library's reconnect callback: zero I/O, sets an atomic flag, returns immediately -- avoiding self-deadlock on the library's per-ctx failover lock the same way worker's and scheduler's rejoin callbacks already do. The C shim, its extern declaration, and the xcecli_set_reconnect_handler() registration call all live together in clib.go's xce_backend_init(), placed after xcecli_start()'s success check (xcecli_start() wipes any earlier-registered handler as part of its own init -- the same ordering bug found and fixed in worker's/ scheduler's task 2.3/3.1 work). A shared checkAndPerformRejoin() helper wraps every gCtx.ctx-based backend call (SubmitJob, ControlJob, ControlQueue, ControlHost, FetchFile): after the call returns, if a rejoin is pending, issue xcecli_join() as a separate step and clear the flag only on success -- on failure it stays set so the next backend call's return retries it, no separate retry loop. needsRejoin is a proper sync/atomic flag rather than a plain bool, since apiserver (unlike the single-threaded C daemons) serves concurrent HTTP requests that can all reach the shared helper at once. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NNB4zGPCsHyDUCDJyMbiQN
This commit is contained in:
parent
e635903996
commit
2a205291c0
4 changed files with 497 additions and 30 deletions
|
|
@ -26,14 +26,33 @@ static int xce_backend_handler(xce_request_t *req, void *arg) {
|
|||
return go_push_handler(req, arg);
|
||||
}
|
||||
|
||||
// go_backend_rejoin_handler is exported from Go (reconnect.go) and
|
||||
// registered below as the client library's xcecli_reconnect_cb
|
||||
// (component-failover-adoption task 4.3). It only sets a flag and returns
|
||||
// immediately -- see reconnect.go's own doc comment for why (the library
|
||||
// invokes it while still holding ac->failover.lock).
|
||||
extern int go_backend_rejoin_handler(xcecli_ctx_t, void*);
|
||||
|
||||
static int xce_backend_rejoin_handler(xcecli_ctx_t ctx, void *arg) {
|
||||
return go_backend_rejoin_handler(ctx, arg);
|
||||
}
|
||||
|
||||
// xce_backend_init initialises the client context, sets the handler,
|
||||
// starts the network loop, then joins as HostRole_apiserver.
|
||||
// starts the network loop, registers the rejoin callback, then joins as
|
||||
// HostRole_apiserver.
|
||||
static int xce_backend_init(xce_client_config_t *config, backend_ctx_t *ctx) {
|
||||
int rc = xcecli_ctx_init(&(ctx->ctx), config);
|
||||
if (rc < 0) return rc;
|
||||
xcecli_set_handler(ctx->ctx, (xce_request_handler_t)xce_backend_handler, NULL);
|
||||
rc = xcecli_start(ctx->ctx);
|
||||
if (rc < 0) return rc;
|
||||
// Registered AFTER xcecli_start() succeeds, not before
|
||||
// (component-failover-adoption task 4.3; matches worker's/scheduler's
|
||||
// task 2.3/3.1 fix): xcecli_start() unconditionally resets
|
||||
// ctx->failover.rejoin_cb to NULL as part of initializing per-context
|
||||
// failover state (client.c) -- registering before this call would be
|
||||
// silently wiped out immediately afterward.
|
||||
xcecli_set_reconnect_handler(ctx->ctx, xce_backend_rejoin_handler, NULL);
|
||||
return xcecli_join(ctx->ctx, HostRole_apiserver, NULL);
|
||||
}
|
||||
|
||||
|
|
@ -520,6 +539,7 @@ func SubmitJob(job *models.JobInfo, arraySize int) error {
|
|||
|
||||
var outJobID C.uint64_t
|
||||
rc := C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID)
|
||||
checkAndPerformRejoin()
|
||||
if rc < 0 {
|
||||
return fmt.Errorf("xcecli_jobsubmit failed: %d", int(rc))
|
||||
}
|
||||
|
|
@ -555,6 +575,7 @@ func ControlJob(jobid uint64, action int, content string) error {
|
|||
}
|
||||
|
||||
rc := C.xcecli_jobctrl(gCtx.ctx, &ctrl)
|
||||
checkAndPerformRejoin()
|
||||
if rc < 0 {
|
||||
return fmt.Errorf("xcecli_jobctrl failed: %d", int(rc))
|
||||
}
|
||||
|
|
@ -573,6 +594,7 @@ func ControlQueue(queuename string, action int) error {
|
|||
}
|
||||
|
||||
rc := C.xcecli_queuectrl(gCtx.ctx, &ctrl)
|
||||
checkAndPerformRejoin()
|
||||
if rc < 0 {
|
||||
return fmt.Errorf("xcecli_queuectrl failed: %d", int(rc))
|
||||
}
|
||||
|
|
@ -590,6 +612,7 @@ func ControlHost(hostname string, action int) error {
|
|||
}
|
||||
|
||||
rc := C.xcecli_hostctrl(gCtx.ctx, &ctrl)
|
||||
checkAndPerformRejoin()
|
||||
if rc < 0 {
|
||||
return fmt.Errorf("xcecli_hostctrl failed: %d", int(rc))
|
||||
}
|
||||
|
|
@ -619,36 +642,37 @@ func signalNumber(signame string, action int) int {
|
|||
// controller→worker chain. The filepath must be the fully resolved absolute
|
||||
// path on the exec host.
|
||||
func FetchFile(jobid uint64, filepath string, offset uint64, maxBytes uint32) (*FetchFileResult, error) {
|
||||
cFilepath := C.CString(filepath)
|
||||
defer C.free(unsafe.Pointer(cFilepath))
|
||||
cFilepath := C.CString(filepath)
|
||||
defer C.free(unsafe.Pointer(cFilepath))
|
||||
|
||||
msg := C.xce_fetchfile_t{
|
||||
jobid: C.uint64_t(jobid),
|
||||
filepath: cFilepath,
|
||||
offset: C.uint64_t(offset),
|
||||
max_bytes: C.uint32_t(maxBytes),
|
||||
}
|
||||
msg := C.xce_fetchfile_t{
|
||||
jobid: C.uint64_t(jobid),
|
||||
filepath: cFilepath,
|
||||
offset: C.uint64_t(offset),
|
||||
max_bytes: C.uint32_t(maxBytes),
|
||||
}
|
||||
|
||||
var reply *C.xce_fetchfile_reply_t
|
||||
rc := C.xcecli_fetchfile(gCtx.ctx, &msg, &reply)
|
||||
if rc < 0 {
|
||||
return nil, fmt.Errorf("xcecli_fetchfile failed: %d", int(rc))
|
||||
}
|
||||
if reply == nil {
|
||||
return nil, fmt.Errorf("xcecli_fetchfile returned nil reply")
|
||||
}
|
||||
defer C.free(unsafe.Pointer(reply))
|
||||
var reply *C.xce_fetchfile_reply_t
|
||||
rc := C.xcecli_fetchfile(gCtx.ctx, &msg, &reply)
|
||||
checkAndPerformRejoin()
|
||||
if rc < 0 {
|
||||
return nil, fmt.Errorf("xcecli_fetchfile failed: %d", int(rc))
|
||||
}
|
||||
if reply == nil {
|
||||
return nil, fmt.Errorf("xcecli_fetchfile returned nil reply")
|
||||
}
|
||||
defer C.free(unsafe.Pointer(reply))
|
||||
|
||||
if reply.error != 0 {
|
||||
return nil, fmt.Errorf("fetchfile remote error: %d", int(reply.error))
|
||||
}
|
||||
if reply.error != 0 {
|
||||
return nil, fmt.Errorf("fetchfile remote error: %d", int(reply.error))
|
||||
}
|
||||
|
||||
result := &FetchFileResult{
|
||||
FileSize: uint64(reply.file_size),
|
||||
BytesRead: uint32(reply.bytes_read),
|
||||
}
|
||||
if reply.bytes_read > 0 && reply.content != nil {
|
||||
result.Content = C.GoBytes(unsafe.Pointer(reply.content), C.int(reply.bytes_read))
|
||||
}
|
||||
return result, nil
|
||||
result := &FetchFileResult{
|
||||
FileSize: uint64(reply.file_size),
|
||||
BytesRead: uint32(reply.bytes_read),
|
||||
}
|
||||
if reply.bytes_read > 0 && reply.content != nil {
|
||||
result.Content = C.GoBytes(unsafe.Pointer(reply.content), C.int(reply.bytes_read))
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue