wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

improve resmgr/jobmgr/qmgr/workermgr

Browse source
This commit is contained in:
Rongsong Shen 2025-12-18 17:37:34 +08:00
parent 2362b32825
commit 8d78f0cbeb
12 changed files with 161 additions and 575 deletions
Download patch file Download diff file Expand all files Collapse all files

9
src/controller/cluster.c
View file

@ -497,7 +497,8 @@ int cluster_init(cluster_t* c)
objmgr_init(&(c->objmgr));
jobmgr_init(&(c->jobmgr), c->myconf->cluster_config->jobmgr->nhash_slots,
c->myconf->cluster_config->jobmgr->delta);
c->myconf->cluster_config->jobmgr->delta,
(zarray_item_killer_t)job_obj_killer);
resmgr_init(&(c->resmgr), c->myconf->cluster_config->resmgr->nhash_slots,
c->myconf->cluster_config->resmgr->delta,
(zarray_item_killer_t)res_obj_killer);
@ -505,9 +506,11 @@ int cluster_init(cluster_t* c)
(zarray_item_killer_t)resval_obj_killer);
workermgr_init(&(c->workermgr),
c->myconf->cluster_config->workermgr->nhash_slots,
c->myconf->cluster_config->workermgr->delta);
c->myconf->cluster_config->workermgr->delta,
(zarray_item_killer_t)worker_obj_killer);
qmgr_init(&(c->qmgr), c->myconf->cluster_config->qmgr->nhash_slots,
c->myconf->cluster_config->qmgr->delta);
c->myconf->cluster_config->qmgr->delta,
(zarray_item_killer_t)queue_obj_killer);
iomgr_init(&(c->iomgr));
iomgr_set_ssl_ctx(&(c->iomgr), ssl_ctx);

31
src/controller/include/cluster.h
View file

@ -22,33 +22,19 @@
typedef int (*collect_check_fn_t)(void*);
extern int check_controller(void*);
extern int cluster_init(cluster_t*);
extern int cluster_start(cluster_t*);
extern int cluster_run(cluster_t*);
extern void cluster_fini(cluster_t*);
extern int register_types(cluster_t*);
extern int jobmgr_init(jobmgr_t*, int, int);
extern void jobmgr_fini(jobmgr_t*);
extern int workermgr_init(workermgr_t*, int, int);
extern void workermgr_fini(workermgr_t*);
extern int workermgr_add(workermgr_t*, worker_t*);
extern worker_t* workermgr_lock(workermgr_t*, char*, int);
extern worker_t* workermgr_lock_byid(workermgr_t*, int, int);
extern void workermgr_del(workermgr_t*, worker_t*);
extern int worker_ctrl(cluster_t*, xce_hostctrl_t*);
extern void qmgr_fini(qmgr_t*);
extern int qmgr_init(qmgr_t*, int, int);
extern int qmgr_add(qmgr_t*, queue_t*);
extern queue_t* qmgr_lockq(qmgr_t*, char*, int);
extern void qmgr_delq(qmgr_t*, queue_t*);
extern int qmgr_iter(qmgr_t*, xlist_t*, void*);
extern int check_controller(void*);
extern int cluster_init(cluster_t*);
extern int cluster_start(cluster_t*);
extern int cluster_run(cluster_t*);
extern void cluster_fini(cluster_t*);
extern int register_types(cluster_t*);
extern int worker_ctrl(cluster_t*, xce_hostctrl_t*);
extern xce_resource_t* get_resource(cluster_t*, char*);
extern int add_to_xlist(xlist_t*, void*);
extern int add_worker_to_xlist(xlist_t*, void*);
extern int add_queue_to_xlist(xlist_t*, void*);
extern int add_job_to_xlist(xlist_t*, void*);
extern void queue_obj_killer(queue_t*);
extern void res_obj_killer(xce_resource_t*);
extern void resval_obj_killer(xce_resval_t*);
extern void resource_killer(void*);
@ -83,6 +69,7 @@ extern job_t* jobmgr_lockjob(jobmgr_t*, uint64_t, int);
extern int32_t jobmgr_add_job(jobmgr_t*, job_t*);
extern void jobmgr_del_job(jobmgr_t*, job_t*);
extern void job_obj_killer(job_t*);
extern void worker_obj_killer(worker_t*);
extern int handle_push(struct capn*, cluster_t*, endpoint_t*, struct Request*);
extern int do_get_op(cluster_t*, endpoint_t*, xce_request_t*);
extern int do_update_op(cluster_t*, endpoint_t*, xce_request_t*);

14
src/controller/include/worker.h
View file

@ -7,20 +7,6 @@
#include "xce/types.h"
#include "zarray.h"
typedef struct {
uint64_t jobid;
xce_alloc_t alloc;
double used;
} res_usage_t;
typedef struct {
xce_worker_t host;
char hostkey[XCE_MAXKEY_LEN];
endpoint_t* ep;
int idx;
zarray_t usages;
} worker_t;
extern int add_worker(void*, endpoint_t*, const char*, hostattrs_t*);
extern int restore_worker(void*, xce_worker_t);
#endif

155
src/controller/jobmgr.c
View file

@ -1,155 +0,0 @@
#include <errno.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>
#include "cluster.h"
#include "htab.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xstring.h"
int jobmgr_init(jobmgr_t* mgr, int nhslots, int delta)
{
ht_init(&(mgr->htab), nhslots, int_hash_key);
zarray_init(&(mgr->jobs), delta, (zarray_item_killer_t)job_obj_killer);
mylock_init(&(mgr->lock));
return 0;
}
void jobmgr_fini(jobmgr_t* mgr)
{
ht_fini(&(mgr->htab));
zarray_fini(&(mgr->jobs));
mylock_fini(&(mgr->lock));
}
job_t* jobmgr_lockjob_(jobmgr_t* mgr, uint64_t jobid, int op)
{
hent_t* e = NULL;
int idx;
job_t* job = NULL;
e = ht_get(&(mgr->htab), (void*)jobid);
if (e == NULL) {
SET_ERROR(E_XCE_BAD_JOB);
return NULL;
}
idx = e->data;
if (idx < 0) {
SET_ERROR(E_XCE_BAD_JOB);
return NULL;
}
job = (job_t*)zarray_get(&(mgr->jobs), idx);
return obj_ref(job, op);
}
job_t* jobmgr_lockjob(jobmgr_t* mgr, uint64_t jobid, int op)
{
job_t* job = NULL;
if (mgr == NULL) {
SET_ERROR(EINVAL);
return NULL;
}
/* clang-format off */
if ((op != OBJREF_NOP) &&
(op != OBJREF_READ) &&
(op != OBJREF_UPDATE)) {
SET_ERROR(EINVAL);
return NULL;
}
/* clang-format on */
mylock_lock(&(mgr->lock));
job = jobmgr_lockjob_(mgr, jobid, op);
mylock_unlock(&(mgr->lock));
return job;
}
static int32_t jobmgr_add_job_(jobmgr_t* mgr, job_t* job)
{
hent_t* e = NULL;
int idx;
e = ht_get(&(mgr->htab), (void*)job->job.jobid);
if (e != NULL) {
return -EEXIST;
}
idx = zarray_store(&(mgr->jobs), job);
if (idx < 0) {
return idx;
}
e = ht_add(&(mgr->htab), (void*)job->job.jobid);
if (e == NULL) {
zarray_del(&(mgr->jobs), idx);
return -errno;
}
e->data = idx;
return idx;
}
int32_t jobmgr_add_job(jobmgr_t* mgr, job_t* job)
{
int32_t idx;
if ((mgr == NULL) || (job == NULL)) {
return -EINVAL;
}
mylock_lock(&(mgr->lock));
idx = jobmgr_add_job_(mgr, job);
mylock_unlock(&(mgr->lock));
return idx;
}
static void jobmgr_del_job_(jobmgr_t* mgr, job_t* job)
{
hent_t* e = NULL;
int idx;
e = ht_get(&(mgr->htab), (void*)job->job.jobid);
if (e == NULL) {
return;
}
idx = e->data;
if (idx < 0) {
return;
}
zarray_del(&(mgr->jobs), idx);
return;
}
void jobmgr_del_job(jobmgr_t* mgr, job_t* job)
{
if ((mgr == NULL) || (job == NULL)) {
return;
}
mylock_lock(&(mgr->lock));
jobmgr_del_job_(mgr, job);
mylock_unlock(&(mgr->lock));
}
int jobmgr_iter(jobmgr_t* mgr, xlist_t* l, void* arg)
{
int rc = 0;
mylock_lock(&(mgr->lock));
rc = zarray_iter_ex(&(mgr->jobs), NULL, l, arg);
mylock_unlock(&(mgr->lock));
return rc;
}

3
src/controller/meson.build
View file

@ -2,21 +2,18 @@ controller_src = [
'cluster.c',
'dataex.c',
'job.c',
'jobmgr.c',
'join.c',
'main.c',
'misc.c',
'myapp.c',
'opget.c',
'opupd.c',
'qmgr.c',
'queue.c',
'reply.c',
'resources.c',
'sched.c',
'task.c',
'worker.c',
'workermgr.c',
'xceobjs.c',
'myconf.c'
]

156
src/controller/qmgr.c
View file

@ -1,156 +0,0 @@
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include "cluster.h"
#include "htab.h"
#include "xce/cluster_def.h"
#include "xstring.h"
extern void queue_obj_killer(queue_t*);
int qmgr_init(qmgr_t* mgr, int nslots, int delta)
{
mylock_init(&(mgr->lock));
ht_init(&(mgr->htab), nslots, string_hash_key);
zarray_init(&(mgr->queues), delta, (zarray_item_killer_t)queue_obj_killer);
return 0;
}
void qmgr_fini(qmgr_t* mgr)
{
zarray_fini(&(mgr->queues));
ht_fini(&(mgr->htab));
mylock_fini(&(mgr->lock));
}
static int qmgr_add_(qmgr_t* mgr, queue_t* q)
{
hent_t* e = NULL;
int idx;
e = ht_get(&(mgr->htab), (void*)q->queue.queue);
if (e != NULL) {
return -EEXIST;
}
idx = zarray_store(&(mgr->queues), q);
if (idx < 0) {
return idx;
}
e = ht_add(&(mgr->htab), (void*)q->queue.queue);
if (e == NULL) {
zarray_del(&(mgr->queues), idx);
return -errno;
}
e->data = idx;
return idx;
}
int qmgr_add(qmgr_t* mgr, queue_t* q)
{
int idx;
if ((mgr == NULL) || (q == NULL)) {
return -EINVAL;
}
mylock_lock(&(mgr->lock));
idx = qmgr_add_(mgr, q);
mylock_unlock(&(mgr->lock));
return idx;
}
static queue_t* qmgr_lockq_(qmgr_t* mgr, char* name, int op)
{
hent_t* e = NULL;
int idx;
queue_t* q = NULL;
e = ht_get(&(mgr->htab), (void*)name);
if (e == NULL) {
SET_ERROR(ENOENT);
return NULL;
}
idx = e->data;
if (idx < 0) {
SET_ERROR(ENOENT);
return NULL;
}
q = (queue_t*)zarray_get(&(mgr->queues), idx);
return obj_ref(q, op);
}
queue_t* qmgr_lockq(qmgr_t* mgr, char* name, int op)
{
queue_t* q = NULL;
if (mgr == NULL) {
SET_ERROR(EINVAL);
return NULL;
}
if (name == NULL) {
SET_ERROR(EINVAL);
return NULL;
}
/* clang-format off */
if ((op != OBJREF_NOP) &&
(op != OBJREF_READ) &&
(op != OBJREF_UPDATE)) {
SET_ERROR(EINVAL);
return NULL;
}
/* clang-format on */
mylock_lock(&(mgr->lock));
q = qmgr_lockq_(mgr, name, op);
mylock_unlock(&(mgr->lock));
return q;
}
static void qmgr_delq_(qmgr_t* mgr, queue_t* q)
{
hent_t* e = NULL;
int idx;
e = ht_get(&(mgr->htab), (void*)q->queue.queue);
if (e == NULL) {
return;
}
idx = e->data;
if (idx < 0) {
return;
}
zarray_del(&(mgr->queues), idx);
}
void qmgr_delq(qmgr_t* mgr, queue_t* q)
{
if ((mgr == NULL) || (q == NULL)) {
return;
}
mylock_lock(&(mgr->lock));
qmgr_delq_(mgr, q);
mylock_unlock(&(mgr->lock));
}
int qmgr_iter(qmgr_t* mgr, xlist_t* l, void* arg)
{
int rc = 0;
mylock_lock(&(mgr->lock));
rc = zarray_iter_ex(&(mgr->queues), NULL, l, arg);
mylock_unlock(&(mgr->lock));
return rc;
}

190
src/controller/workermgr.c
View file

@ -1,190 +0,0 @@
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include "cluster.h"
#include "htab.h"
#include "mylock.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xstring.h"
extern void worker_obj_killer(worker_t*);
int workermgr_init(workermgr_t* mgr, int nslots, int delta)
{
TRACE("%s nslots %d delta %d", __func__, nslots, delta);
mylock_init(&(mgr->lock));
ht_init(&(mgr->htab), nslots, string_hash_key);
zarray_init(&(mgr->workers), delta,
(zarray_item_killer_t)worker_obj_killer);
return 0;
}
void workermgr_fini(workermgr_t* mgr)
{
zarray_fini(&(mgr->workers));
ht_fini(&(mgr->htab));
mylock_fini(&(mgr->lock));
}
static int32_t workermgr_add_(workermgr_t* mgr, worker_t* w)
{
hent_t* e = NULL;
int idx;
e = ht_get(&(mgr->htab), (void*)w->host.name);
if (e != NULL) {
return (int)e->data;
}
idx = zarray_store(&(mgr->workers), w);
if (idx < 0) {
return idx;
}
e = ht_add(&(mgr->htab), (void*)w->host.name);
if (e == NULL) {
zarray_del(&(mgr->workers), idx);
return -errno;
}
e->data = idx;
return idx;
}
int32_t workermgr_add(workermgr_t* mgr, worker_t* w)
{
int idx;
if ((mgr == NULL) || (w == NULL)) {
return -EINVAL;
}
mylock_lock(&(mgr->lock));
idx = workermgr_add_(mgr, w);
mylock_unlock(&(mgr->lock));
return idx;
}
static worker_t* workermgr_lock_(workermgr_t* mgr, char* name, int op)
{
hent_t* e = NULL;
int idx;
worker_t* w = NULL;
e = ht_get(&(mgr->htab), (void*)name);
if (e == NULL) {
SET_ERROR(E_XCE_BAD_WORKER);
return NULL;
}
idx = e->data;
if (idx < 0) {
SET_ERROR(E_XCE_BAD_WORKER);
return NULL;
}
w = zarray_get(&(mgr->workers), idx);
return obj_ref(w, op);
}
worker_t* workermgr_lock(workermgr_t* mgr, char* name, int op)
{
worker_t* w = NULL;
if (mgr == NULL) {
SET_ERROR(EINVAL);
return NULL;
}
if (name == NULL) {
SET_ERROR(EINVAL);
return NULL;
}
if ((op != OBJREF_NOP) && (op != OBJREF_READ) && (op != OBJREF_UPDATE)) {
SET_ERROR(EINVAL);
return NULL;
}
mylock_lock(&(mgr->lock));
w = workermgr_lock_(mgr, name, op);
mylock_unlock(&(mgr->lock));
return w;
}
worker_t* workermgr_lock_byid(workermgr_t* mgr, int idx, int op)
{
worker_t* w = NULL;
if (mgr == NULL) {
SET_ERROR(EINVAL);
return NULL;
}
if ((idx < 0) || (idx >= mgr->workers.caps)) {
SET_ERROR(EINVAL);
return NULL;
}
if ((op != OBJREF_NOP) && (op != OBJREF_READ) && (op != OBJREF_UPDATE)) {
SET_ERROR(EINVAL);
return NULL;
}
mylock_lock(&(mgr->lock));
w = (worker_t*)zarray_get(&(mgr->workers), idx);
if (w != NULL) {
w = obj_ref(w, op);
}
mylock_unlock(&(mgr->lock));
return w;
}
static void workermgr_del_(workermgr_t* mgr, worker_t* w)
{
hent_t* e = NULL;
int idx;
e = ht_get(&(mgr->htab), (void*)w->host.name);
if (e == NULL) {
return;
}
idx = e->data;
if (idx < 0) {
return;
}
zarray_del(&(mgr->workers), idx);
}
void workermgr_del(workermgr_t* mgr, worker_t* w)
{
if ((mgr == NULL) || (w == NULL)) {
SET_ERROR(EINVAL);
return;
}
mylock_lock(&(mgr->lock));
workermgr_del_(mgr, w);
mylock_unlock(&(mgr->lock));
}
int workermgr_iter(workermgr_t* mgr, xlist_t* l, void* arg)
{
int rc = 0;
mylock_lock(&(mgr->lock));
rc = zarray_iter_ex(&(mgr->workers), NULL, l, arg);
mylock_unlock(&(mgr->lock));
return rc;
}