feat: P1-1 handle resource value push via OPCode_notify

Add OPCode_notify handler in go_push_handler top-level switch
(not inside handlePush) to receive resource value updates from
the controller via xce_notify_data_t/schedUpdateResvals.

- Add 8 CGO union accessors for xce_val_t (which, i32, i64,
  u32, u64, fval, str, bool) since CGO cannot access union fields
- Add handleNotifyData() dispatching NotifyData_data_schedUpdateResvals
  events, iterating xce_resval_list_t and calling cResvalToDistribution
- Add cResvalToDistribution in convert.go with 7 separate switch
  cases for Value_u_* union discriminants (no combining i32/i64)
- Add rawResvalData + rawResvalToDistribution pure-Go version
  in convert_pure.go for testability without CGO
- Add 10 TDD tests covering all 7 value types, nil val guard,
  unknown discriminant, and boolean false
- All 42 backend tests passing
This commit is contained in:
Rongsong Shen 2026-04-17 13:54:03 +00:00
parent dc029a6d14
commit bc3103a6d5
5 changed files with 365 additions and 23 deletions

View file

@ -48,6 +48,16 @@ static void jobctrl_set_siginfo(xce_jobctrl_t *ctrl, xce_signal_t *sig) {
static void jobctrl_set_ckptoption(xce_jobctrl_t *ctrl, xce_ckpt_options_t *ckpt) {
ctrl->u.ckptoption = ckpt;
}
// xce_val_t union accessors — CGO cannot access C union fields by name.
static int xce_val_which(xce_val_t *v) { return v->which; }
static int32_t xce_val_i32(xce_val_t *v) { return v->u.ivalue32; }
static int64_t xce_val_i64(xce_val_t *v) { return v->u.ivalue64; }
static uint32_t xce_val_u32(xce_val_t *v) { return v->u.uvalue32; }
static uint64_t xce_val_u64(xce_val_t *v) { return v->u.uvalue64; }
static double xce_val_fval(xce_val_t *v) { return v->u.fvalue; }
static char* xce_val_str(xce_val_t *v) { return v->u.svalue; }
static int xce_val_bool(xce_val_t *v) { return v->u.bvalue; }
*/
import "C"
@ -160,6 +170,15 @@ func go_push_handler(req *C.xce_request_t, _ unsafe.Pointer) C.int {
switch req.opcode {
case C.OPCode_push:
return handlePush(req)
case C.OPCode_notify:
if req.body_which != C.Request_body_notifymsg {
return 0
}
nd := (*C.xce_notify_data_t)(unsafe.Pointer(C.xce_request_body(req)))
if nd == nil {
return 0
}
return handleNotifyData(nd)
}
return 0
}
@ -238,6 +257,36 @@ func handleConfigPush(cfgdata *C.xce_configdata_t) C.int {
return 0
}
// handleNotifyData processes an OPCode_notify request.
// xce_notify_data_t uses 'event' (not 'data_which') as its union discriminant.
func handleNotifyData(nd *C.xce_notify_data_t) C.int {
if nd == nil {
return 0
}
switch nd.event {
case C.NotifyData_data_schedUpdateResvals:
rvl := (*C.xce_resval_list_t)(C.xce_notify_data_ptr(nd))
if rvl == nil {
return 0
}
n := int(rvl.count)
if n == 0 || rvl.vals == nil {
return 0
}
vals := unsafe.Slice(rvl.vals, n)
for _, vp := range vals {
if vp == nil {
continue
}
rd := cResvalToDistribution(vp)
if rd != nil {
gHandler.OnResourceLocationUpdated(rd)
}
}
}
return 0
}
// SubmitJob sends a job submission to the backend via xcecli_jobsubmit.
// arraySize=1 for normal jobs, >1 for array jobs.
// The store is updated only when the backend pushes the resulting OnJobCreated