From 486b38d679046d0fe8f516ee87a050d2847a0166 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 11 Dec 2025 17:07:51 +0800 Subject: [PATCH] support passing args to request handle for elimination of global var --- src/controller/testsuite/test.c | 6 +- src/include/client.h | 7 +- src/include/xcectx.h | 5 +- src/lib/client.c | 5 +- src/lib/xcectx.c | 9 ++- src/scheduler/src/handler.c | 95 ++++++++++++++++++++++++++- src/scheduler/src/main.c | 4 +- src/worker/src/controller_connector.c | 8 +-- 8 files changed, 120 insertions(+), 19 deletions(-) diff --git a/src/controller/testsuite/test.c b/src/controller/testsuite/test.c index 68b78f5b..b2769033 100644 --- a/src/controller/testsuite/test.c +++ b/src/controller/testsuite/test.c @@ -83,7 +83,7 @@ typedef struct { static run_job_t run_jobs[MAX_RUN_JOBS]; static run_task_t run_tasks[MAX_RUN_TASKS]; -static int myreq_handler(xce_request_t*); +static int myreq_handler(xce_request_t*, void*); extern int cmd_help(int, char*[]); extern int cmd_up(int, char*[]); @@ -265,7 +265,7 @@ int enter_ctx(command_ctx_t* ctx) config.key = get_setting("secret"); xcecli_ctx_init(&(ctx->ctx), &config); - xcecli_set_handler(ctx->ctx, &myreq_handler); + xcecli_set_handler(ctx->ctx, &myreq_handler, NULL); xcecli_start(ctx->ctx); return 0; @@ -2100,7 +2100,7 @@ static int handle_runtasks(xce_request_t* request) return 0; } -static int myreq_handler(xce_request_t* request) +static int myreq_handler(xce_request_t* request, void* args) { int rc = 0; diff --git a/src/include/client.h b/src/include/client.h index d5dff668..cfbb6857 100644 --- a/src/include/client.h +++ b/src/include/client.h @@ -9,7 +9,10 @@ #include "xce/types.h" #include "xceapi.h" -typedef int (*xce_request_handler_t)(xce_request_t*); +#if 0 +typedef int (*xce_request_handler_t)(xce_request_t*, void*); +#endif + typedef void* xcecli_ctx_t; #define N_MAX_MASTERS (3) @@ -26,7 +29,7 @@ typedef struct { 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); +extern int xcecli_set_handler(xcecli_ctx_t, xce_request_handler_t, void*); extern int xcecli_start(xcecli_ctx_t); extern int xcecli_stop(xcecli_ctx_t); extern int xcecli_wakeup(xcecli_ctx_t, int); diff --git a/src/include/xcectx.h b/src/include/xcectx.h index 8d95e3f2..a80d7f74 100644 --- a/src/include/xcectx.h +++ b/src/include/xcectx.h @@ -15,7 +15,7 @@ typedef enum { XCE_CTX_ERR } xce_ctx_state_t; -typedef int (*xce_request_handler_t)(xce_request_t*); +typedef int (*xce_request_handler_t)(xce_request_t*, void*); typedef struct { xce_ctx_state_t state; @@ -24,6 +24,7 @@ typedef struct { vtaskmgr_t vtaskmgr; endpoint_t* ep; xce_request_handler_t handler; + void* handler_args; cbuf_t wbuf; dataset_t* dataset; @@ -39,7 +40,7 @@ typedef struct { extern int xce_ctx_init(xce_ctx_t*, char*, char*); extern void xce_ctx_fini(xce_ctx_t*); -extern int xce_ctx_set_handler(xce_ctx_t*, xce_request_handler_t); +extern int xce_ctx_set_handler(xce_ctx_t*, xce_request_handler_t, void*); extern int xce_ctx_start(xce_ctx_t*); extern int xce_ctx_stop(xce_ctx_t*); extern int xce_wakeup(xce_ctx_t*, int); diff --git a/src/lib/client.c b/src/lib/client.c index f993de36..4a58ec55 100644 --- a/src/lib/client.c +++ b/src/lib/client.c @@ -76,7 +76,8 @@ void xcecli_ctx_fini(xcecli_ctx_t c) free(ctx); } -int xcecli_set_handler(xcecli_ctx_t c, xce_request_handler_t handler) +int xcecli_set_handler(xcecli_ctx_t c, xce_request_handler_t handler, + void* args) { api_ctx_t* ctx = (api_ctx_t*)c; @@ -84,7 +85,7 @@ int xcecli_set_handler(xcecli_ctx_t c, xce_request_handler_t handler) return -EINVAL; } - return xce_ctx_set_handler(&(ctx->ctx), handler); + return xce_ctx_set_handler(&(ctx->ctx), handler, args); } int xcecli_wakeup(xcecli_ctx_t c, int status) diff --git a/src/lib/xcectx.c b/src/lib/xcectx.c index 502357cb..7aa3f267 100644 --- a/src/lib/xcectx.c +++ b/src/lib/xcectx.c @@ -107,7 +107,7 @@ static int vtask_entry(vtask_t* task) } if (callnext) { - rc = (ctx->handler)(&request); + rc = (ctx->handler)(&request, ctx->handler_args); } } @@ -198,13 +198,16 @@ void xce_ctx_fini(xce_ctx_t* ctx) return; } -int xce_ctx_set_handler(xce_ctx_t* ctx, xce_request_handler_t handler) +int xce_ctx_set_handler(xce_ctx_t* ctx, xce_request_handler_t handler, + void* args) { if ((ctx == NULL) || (handler == NULL)) { return -EINVAL; } - ctx->handler = handler; + ctx->handler = handler; + ctx->handler_args = args; + return 0; } diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index 8c5844c1..85ca075e 100644 --- a/src/scheduler/src/handler.c +++ b/src/scheduler/src/handler.c @@ -1,3 +1,4 @@ +#include #include #include #include @@ -6,5 +7,97 @@ #include "xce/types.h" #include "scheduler.h" +#include "utils.h" -int scheduler_req_handler(xce_request_t* request) { return 0; } +static int sched_handle_config(xce_conf4sched_t* config) +{ + TRACE("%s...", __func__); + TRACE("my cluster name %s", config->clustername); + + return 0; +} + +static int sched_handle_jobpush(xce_jobs_data_t* jobs) +{ + TRACE("%s...", __func__); + + return 0; +} + +static int sched_handle_workerpush(xce_workers_data_t* workers) +{ + TRACE("%s...", __func__); + + return 0; +} + +static int sched_on_jobs(int njobs, xce_job_t **jobs) { + return 0; +} + +static int sched_on_workers(int nworkers, xce_worker_t **workers) { + return 0; +} + +static int sched_handle_dataset(dataset_t* dataset) +{ + int rc = 0; + + TRACE("%s table = %d", __func__, dataset->table); + + switch (dataset->table) { + case Data_data_schedjobs: + rc = sched_on_jobs(dataset->nrows, (xce_job_t **)dataset->rows); + break; + case Data_data_schedworkers: + rc = sched_on_workers(dataset->nrows, (xce_worker_t **)dataset->rows); + break; + case Data_data_resources: + break; + case Data_data_resvals: + break; + } + + return rc; +} + +int scheduler_req_handler(xce_request_t* request) +{ + int rc = 0; + + switch (request->opcode) { + case OPCode_push: + switch (request->body_which) { + case Request_body_configpushmsg: + /* configuration arriving */ + rc = sched_handle_config( + request->body.configpushmsg->data.conf4sched); + break; + case Request_body_pushmsg: + /* data arriving */ + switch (request->body.pushmsg->table) { + case Data_data_schedjobs: + /* job arriving */ + rc = sched_handle_jobpush(request->body.pushmsg->data.jobs); + break; + case Data_data_schedworkers: + /* workers arriving */ + rc = sched_handle_workerpush( + request->body.pushmsg->data.workers); + break; + } + break; + case Request_body_datamsg: + rc = sched_handle_dataset(request->body.datamsg); + break; + default: + rc = -EINVAL; + break; + } + break; + default: + rc = -EINVAL; + break; + } + return rc; +} diff --git a/src/scheduler/src/main.c b/src/scheduler/src/main.c index b529201a..247683e4 100644 --- a/src/scheduler/src/main.c +++ b/src/scheduler/src/main.c @@ -11,7 +11,7 @@ #include "utils.h" -extern int scheduler_req_handler(xce_request_t*); +extern int scheduler_req_handler(xce_request_t*, void*); static char doc[] = "xce_scheduler"; static char args_doc[] = ""; @@ -70,7 +70,7 @@ static int scheduler_init(xcecli_ctx_t* ctx, xce_client_config_t* config) return rc; } - xcecli_set_handler(*ctx, scheduler_req_handler); + xcecli_set_handler(*ctx, scheduler_req_handler, NULL); rc = xcecli_start(*ctx); if (rc < 0) { return rc; diff --git a/src/worker/src/controller_connector.c b/src/worker/src/controller_connector.c index 9a1db02a..6d5484af 100644 --- a/src/worker/src/controller_connector.c +++ b/src/worker/src/controller_connector.c @@ -29,7 +29,7 @@ xcecli_ctx_t xcecli_ctx = { 0 }; xce_client_config_t xce_client_cfg = { 0 }; controller_connector_ctx_t* cc_ctx = NULL; -static int myreq_handler(xce_request_t*); +static int myreq_handler(xce_request_t*, void*); /* make compiler happy */ typedef struct { @@ -240,7 +240,7 @@ int xcecd_join_cluster(hostattrs_t* attrs) int role = HostRole_worker; xcecli_ctx_init(&xcecli_ctx, &xce_client_cfg); - xcecli_set_handler(xcecli_ctx, &myreq_handler); + xcecli_set_handler(xcecli_ctx, &myreq_handler, NULL); xcecli_start(xcecli_ctx); rc = xcecli_join(xcecli_ctx, role, attrs); @@ -1218,7 +1218,7 @@ static int handle_jobcontrol(xce_request_t* request) * @param ctx Pointer to the I/O context structure. * @return int 0 on success, -1 on failure. */ -static int myreq_handler(xce_request_t* request) +static int myreq_handler(xce_request_t* request, void* args) { int ret = 0; @@ -1513,4 +1513,4 @@ int xcecd_jobpost_done(uint64_t jobid) XCECD_LOG(XCE_LOG_DEBUG, "Job post done and calling xcecli_notify_jobpostdone..."); return xcecli_notify_jobpostdone(xcecli_ctx, jobid); -} \ No newline at end of file +}