From c12fcf690579a577734a5f760925da75722d2cd9 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Mon, 22 Dec 2025 11:58:04 +0800 Subject: [PATCH] improve ordq --- src/include/ordq.h | 7 ++-- src/lib/ordq.c | 72 ++++++++++++++++++++++++--------- src/lib/testsuite/test_ordq.c | 5 ++- src/scheduler/src/plugin_fcfs.c | 21 +++++----- 4 files changed, 70 insertions(+), 35 deletions(-) diff --git a/src/include/ordq.h b/src/include/ordq.h index 24077158..09336365 100644 --- a/src/include/ordq.h +++ b/src/include/ordq.h @@ -2,6 +2,7 @@ #define _ORDQ_H_ 1 +#include #include #include "mylock.h" @@ -13,7 +14,7 @@ typedef struct _ordq_item { typedef int (*ordq_cmp_fn_t)(void*, void*, void*); typedef void (*ordq_data_killer_t)(void*); -typedef int (*ordq_skip_fn_t)(void*, void*); +typedef int (*ordq_skip_fn_t)(void*, va_list); typedef struct { mylock_t lock; @@ -26,8 +27,8 @@ typedef struct { extern int ordq_init(ordq_t*, ordq_data_killer_t, ordq_cmp_fn_t); extern void ordq_fini(ordq_t* q); extern int ordq_put(ordq_t*, void*); -extern ordq_item_t* ordq_get(ordq_t*, ordq_skip_fn_t); -extern ordq_item_t* ordq_peek(ordq_t*, ordq_skip_fn_t); +extern ordq_item_t* ordq_get(ordq_t*, ordq_skip_fn_t, ...); +extern ordq_item_t* ordq_peek(ordq_t*, ordq_skip_fn_t, ...); extern void ordq_set_ctx(ordq_t*, void*); #endif diff --git a/src/lib/ordq.c b/src/lib/ordq.c index 96e3b0a5..2dc4868d 100644 --- a/src/lib/ordq.c +++ b/src/lib/ordq.c @@ -79,49 +79,83 @@ void ordq_fini(ordq_t* q) mylock_fini(&(q->lock)); } -static ordq_item_t* ordq_get_(ordq_t* q, ordq_skip_fn_t skipfn) +static ordq_item_t* ordq_get_(ordq_t* q, ordq_skip_fn_t skipfn, va_list ap) { - ordq_item_t** p = &(q->head); + ordq_item_t** p = NULL; + ordq_item_t* it = NULL; + + p = &(q->head); while ((*p) != NULL) { - if ((skipfn == NULL) || !skipfn(q->ctx, (*p)->data)) { - ordq_item_t* it = (*p); - *p = it->next; - it->next = NULL; - return it; + va_list ap1; + int rc = 0; + + if (skipfn == NULL) { + break; + } + + va_copy(ap1, ap); + rc = skipfn((*p)->data, ap1); + va_end(ap1); + + if (!rc) { + break; } p = &((*p)->next); } - return NULL; -} - -ordq_item_t* ordq_get(ordq_t* q, ordq_skip_fn_t skipfn) -{ - ordq_item_t* it; - - mylock_lock(&(q->lock)); - it = ordq_get_(q, skipfn); - mylock_unlock(&(q->lock)); + it = *p; + if (it != NULL) { + *p = it->next; + it->next = NULL; + } return it; } -ordq_item_t* ordq_peek(ordq_t* q, ordq_skip_fn_t skipfn) +ordq_item_t* ordq_get(ordq_t* q, ordq_skip_fn_t skipfn, ...) +{ + ordq_item_t* it; + va_list ap; + + va_start(ap, skipfn); + mylock_lock(&(q->lock)); + it = ordq_get_(q, skipfn, ap); + mylock_unlock(&(q->lock)); + va_end(ap); + + return it; +} + +ordq_item_t* ordq_peek(ordq_t* q, ordq_skip_fn_t skipfn, ...) { ordq_item_t* it = NULL; + va_list ap; + va_start(ap, skipfn); mylock_lock(&(q->lock)); it = q->head; while (it != NULL) { - if ((skipfn == NULL) || !skipfn(q->ctx, it->data)) { + int rc = 0; + va_list ap1; + + if (skipfn == NULL) { + break; + } + + va_copy(ap1, ap); + rc = skipfn(it->data, ap1); + va_end(ap1); + + if (!rc) { break; } it = it->next; } mylock_unlock(&(q->lock)); + va_end(ap); return it; } diff --git a/src/lib/testsuite/test_ordq.c b/src/lib/testsuite/test_ordq.c index cc4ab1a2..1dbb8c06 100644 --- a/src/lib/testsuite/test_ordq.c +++ b/src/lib/testsuite/test_ordq.c @@ -1,4 +1,5 @@ #include +#include #include #include #include @@ -46,7 +47,7 @@ static void test_ordq_get(void) return; } -static int test_skip(void* ctx, void* data) +static int test_skip(void* data, va_list ap) { int val = (int)(long)data; @@ -124,7 +125,7 @@ static void test_ordq_peek(void) return; } -static int test_skip2(void* ctx, void* data) +static int test_skip2(void* data, va_list ap) { int val = (int)(long)data; diff --git a/src/scheduler/src/plugin_fcfs.c b/src/scheduler/src/plugin_fcfs.c index 3dde74ce..de81f89d 100644 --- a/src/scheduler/src/plugin_fcfs.c +++ b/src/scheduler/src/plugin_fcfs.c @@ -105,10 +105,10 @@ static int fcfs_update_config(void* arg, xce_conf4sched_t* config) return 0; } -static int skip_not_matched_queue(void* arg1, void* arg2) +static int skip_not_matched_queue(void* arg1, va_list ap) { - char* name = (char*)arg1; - fcfs_queue_t* q = (fcfs_queue_t*)arg2; + fcfs_queue_t* q = (fcfs_queue_t*)arg1; + char* name = va_arg(ap, char*); if ((name == NULL) || (q == NULL)) { return 1; @@ -126,8 +126,8 @@ static int fcfs_update_queue(void* arg, xce_sched_queue_t* q) TRACE("%s queue %s...", __func__, q->name); - privdata->queues.ctx = q->name; - it = ordq_get(&(privdata->queues), (ordq_skip_fn_t)skip_not_matched_queue); + it = ordq_get(&(privdata->queues), (ordq_skip_fn_t)skip_not_matched_queue, + q->name); if (it == NULL) { queue = (fcfs_queue_t*)calloc(1, sizeof(*queue)); if (queue == NULL) { @@ -151,8 +151,7 @@ static fcfs_queue_t* fcfs_get_queue(fcfs_plugin_data_t* data, char* name) { ordq_item_t* it = NULL; - data->queues.ctx = name; - it = ordq_peek(&(data->queues), (ordq_skip_fn_t)skip_not_matched_queue); + it = ordq_peek(&(data->queues), (ordq_skip_fn_t)skip_not_matched_queue, name); if (it == NULL) { return NULL; } @@ -194,9 +193,9 @@ static int fcfs_update_job(void* arg, xce_sched_job_t* j) return 0; } -static int skip_no_pend_job_queue(void* arg1, void* arg2) +static int skip_no_pend_job_queue(void* arg1, va_list ap) { - fcfs_queue_t* q = (fcfs_queue_t*)arg2; + fcfs_queue_t* q = (fcfs_queue_t*)arg1; if (q == NULL) { return 1; @@ -205,9 +204,9 @@ static int skip_no_pend_job_queue(void* arg1, void* arg2) return (ordq_peek(&(q->pendJobs), NULL) == NULL); } -static int skip_not_ready_job(void* arg1, void* arg2) +static int skip_not_ready_job(void* arg1, va_list ap) { - fcfs_job_t* job = (fcfs_job_t*)arg2; + fcfs_job_t* job = (fcfs_job_t*)arg1; return !sched_job_is_ready(job->job); }