lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

improve resmgr

Browse source
This commit is contained in:
Rongsong Shen 2025-12-18 16:20:41 +08:00
parent 4db6b039c5
commit 1f1e4fdaba
10 changed files with 107 additions and 221 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/controller/cluster.c
View file

@ -500,7 +500,9 @@ int cluster_init(cluster_t* c)
c->myconf->cluster_config->jobmgr->delta);
resmgr_init(&(c->resmgr), c->myconf->cluster_config->resmgr->nhash_slots,
c->myconf->cluster_config->resmgr->delta,
c->myconf->cluster_config->resmgr->inst_delta);
(zarray_item_killer_t)res_obj_killer);
zarray_init(&(c->sharedinst), c->myconf->cluster_config->resmgr->inst_delta,
(zarray_item_killer_t)resval_obj_killer);
workermgr_init(&(c->workermgr),
c->myconf->cluster_config->workermgr->nhash_slots,
c->myconf->cluster_config->workermgr->delta);

2
src/controller/dataex.c
View file

@ -464,7 +464,7 @@ void collect_shared_resvals(cluster_t* c, collect_check_fn_t fn, int* nres,
xce_val_with_idx_t*** resvals)
{
collect_resval_with_idx(&(c->resmgr.sharedinst), nres, resvals);
collect_resval_with_idx(&(c->sharedinst), nres, resvals);
return;
}

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

@ -49,17 +49,10 @@ 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 resource_killer(void*);
extern int add_resource(void*, char*, char*, int, xce_order_t, int, int);
extern int resmgr_init(resmgr_t*, int, int, int);
extern void resmgr_fini(resmgr_t*);
extern int32_t resmgr_add_res(resmgr_t*, xce_resource_t*);
extern xce_resource_t* resmgr_lockres(resmgr_t* mgr, char*, int);
extern void res_obj_killer(xce_resource_t*);
extern void resmgr_del_res(resmgr_t*, xce_resource_t*);
extern int resmgr_iter(resmgr_t*, xlist_t*, void*);
extern void resval_obj_killer(xce_resval_t*);
extern void resource_killer(void*);
extern int add_resource(void*, char*, char*, int, xce_order_t, int, int);
extern void config_obj_killer(xce_config_t*);
extern int update_resvals(cluster_t*, worker_t*, int, xce_resval_t*);

1
src/controller/meson.build
View file

@ -12,7 +12,6 @@ controller_src = [
'qmgr.c',
'queue.c',
'reply.c',
'resmgr.c',
'resources.c',
'sched.c',
'task.c',

191
src/controller/resmgr.c
View file

@ -1,191 +0,0 @@
#include <errno.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>
#include "cluster.h"
#include "htab.h"
#include "utils.h"
#include "xce/cluster_def.h"
#include "xce/errnox.h"
#include "xstring.h"
extern void res_obj_killer(xce_resource_t*);
extern void resval_obj_killer(xce_resval_t*);
int resmgr_init(resmgr_t* mgr, int nslots, int delta, int inst_delta)
{
TRACE("%s nslots %d delta %d inst_delta %d", __func__, nslots, delta,
inst_delta);
mylock_init(&(mgr->lock));
ht_init(&(mgr->htab), nslots, string_hash_key);
zarray_init(&(mgr->defs), delta, (zarray_item_killer_t)res_obj_killer);
zarray_init(&(mgr->sharedinst), inst_delta,
(zarray_item_killer_t)resval_obj_killer);
mgr->nshared = 0;
mgr->nunshared = 0;
return 0;
}
void resmgr_fini(resmgr_t* mgr)
{
zarray_fini(&(mgr->sharedinst));
zarray_fini(&(mgr->defs));
ht_fini(&(mgr->htab));
mylock_fini(&(mgr->lock));
return;
}
static int32_t resmgr_add_res_(resmgr_t* mgr, xce_resource_t* res)
{
hent_t* e = NULL;
int idx;
e = ht_get(&(mgr->htab), (void*)res->name);
if (e != NULL) {
return -EEXIST;
}
idx = zarray_store(&(mgr->defs), res);
if (idx < 0) {
return idx;
}
e = ht_add(&(mgr->htab), (void*)res->name);
if (e == NULL) {
zarray_del(&(mgr->defs), idx);
return -errno;
}
e->data = idx;
if (res->flags & XCE_RESF_SHARED) {
res->idx = mgr->nshared;
mgr->nshared++;
} else {
res->idx = mgr->nunshared;
mgr->nunshared++;
}
return idx;
}
int32_t resmgr_add_res(resmgr_t* mgr, xce_resource_t* res)
{
int32_t idx;
if ((mgr == NULL) || (res == NULL)) {
return -EINVAL;
}
mylock_lock(&(mgr->lock));
idx = resmgr_add_res_(mgr, res);
mylock_unlock(&(mgr->lock));
return idx;
}
static xce_resource_t* resmgr_lockres_(resmgr_t* mgr, char* resname, int op)
{
hent_t* e = NULL;
int idx;
xce_resource_t* r = NULL;
e = ht_get(&(mgr->htab), (void*)resname);
if (e == NULL) {
SET_ERROR(E_XCE_BAD_RES);
return NULL;
}
idx = e->data;
if (idx < 0) {
SET_ERROR(E_XCE_INTERNAL);
return NULL;
}
r = (xce_resource_t*)zarray_get(&(mgr->defs), idx);
if (r == NULL) {
SET_ERROR(E_XCE_INTERNAL);
return NULL;
}
return obj_ref(r, op);
}
xce_resource_t* resmgr_lockres(resmgr_t* mgr, char* resname, int op)
{
xce_resource_t* r = NULL;
if (mgr == NULL) {
SET_ERROR(EINVAL);
return NULL;
}
if (resname == 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));
r = resmgr_lockres_(mgr, resname, op);
mylock_unlock(&(mgr->lock));
return r;
}
static void resmgr_del_res_(resmgr_t* mgr, xce_resource_t* r)
{
hent_t* e = NULL;
int idx;
e = ht_get(&(mgr->htab), (void*)r->name);
if (e == NULL) {
SET_ERROR(E_XCE_BAD_RES);
return;
}
idx = e->data;
if (idx < 0) {
SET_ERROR(E_XCE_INTERNAL);
return;
}
zarray_del(&(mgr->defs), idx);
}
void resmgr_del_res(resmgr_t* mgr, xce_resource_t* r)
{
if ((mgr == NULL) || (r == NULL)) {
SET_ERROR(EINVAL);
return;
}
mylock_lock(&(mgr->lock));
resmgr_del_res_(mgr, r);
mylock_unlock(&(mgr->lock));
}
int resmgr_iter(resmgr_t* mgr, xlist_t* l, void* arg)
{
int rc = 0;
mylock_lock(&(mgr->lock));
rc = zarray_iter_ex(&(mgr->defs), NULL, l, arg);
mylock_unlock(&(mgr->lock));
return rc;
}

4
src/controller/worker.c
View file

@ -65,7 +65,7 @@ static int update_val_res(cluster_t* c, worker_t* w, xce_resource_t* r,
if (r->flags & XCE_RESF_SHARED) {
/* update shared resource */
array = &(c->resmgr.sharedinst);
array = &(c->sharedinst);
} else {
/* update host based resource */
array = &(w->host.resvals);
@ -457,7 +457,7 @@ int get_resval(cluster_t* c, char* resname, char* host, xce_val_t* val)
array = &(w->host.resvals);
} else {
array = &(c->resmgr.sharedinst);
array = &(c->sharedinst);
}
if (r->idx < array->caps) {

14
src/include/kvmgr.h
View file

@ -4,6 +4,7 @@
#include "htab.h"
#include "mylock.h"
#include "xce/types.h"
#include "xlist.h"
#include "zarray.h"
@ -13,6 +14,12 @@ typedef struct {
zarray_t data;
} kvmgr_t;
typedef struct {
kvmgr_t resdef;
int nshared;
int nunshared;
} resmgr_t;
extern int kvmgr_init(kvmgr_t*, hash_key_fn_t, int, int, zarray_item_killer_t);
extern void kvmgr_fini(kvmgr_t*);
extern void* kvmgr_lock(kvmgr_t*, void*, int);
@ -21,4 +28,11 @@ extern int32_t kvmgr_add(kvmgr_t*, void*, void*);
extern void kvmgr_del(kvmgr_t*, void*);
extern int kvmgr_iter(kvmgr_t*, zarray_iter_skip_fn_t, xlist_t*, void*);
extern int resmgr_init(resmgr_t*, int, int, zarray_item_killer_t);
extern void resmgr_fini(resmgr_t*);
extern int32_t resmgr_add_res(resmgr_t*, xce_resource_t*);
extern xce_resource_t* resmgr_lockres(resmgr_t*, char*, int);
extern void resmgr_del_res(resmgr_t*, xce_resource_t*);
extern int resmgr_iter(resmgr_t*, xlist_t*, void*);
#endif

3
src/include/xce/cluster_def.h
View file

@ -90,6 +90,7 @@ typedef struct {
zarray_t jobs;
} jobmgr_t;
#if 0
typedef struct {
mylock_t lock;
htab_t htab;
@ -98,6 +99,7 @@ typedef struct {
zarray_t defs;
zarray_t sharedinst;
} resmgr_t;
#endif
typedef struct {
mylock_t lock;
@ -125,6 +127,7 @@ typedef struct {
zarray_t endpoints;
endpoint_t ep;
zarray_t configs;
zarray_t sharedinst;
controller_t controllers[N_CONTROLLER_MAX];
int myidx;
int master;

63
src/lib/kvmgr.c
View file

@ -9,6 +9,8 @@
#include "xce/types.h"
#include "zarray.h"
#include "controller.capnp.h"
int kvmgr_init(kvmgr_t* mgr, hash_key_fn_t keyfn, int nslots, int delta,
zarray_item_killer_t killer)
{
@ -183,3 +185,64 @@ int kvmgr_iter(kvmgr_t* mgr, zarray_iter_skip_fn_t skipfn, xlist_t* l,
return rc;
}
int resmgr_init(resmgr_t* mgr, int nslots, int delta,
zarray_item_killer_t killer)
{
int rc = 0;
rc = kvmgr_init(&(mgr->resdef), string_hash_key, nslots, delta, killer);
if (rc < 0) {
return rc;
}
mgr->nshared = 0;
mgr->nunshared = 0;
return 0;
}
void resmgr_fini(resmgr_t* mgr)
{
kvmgr_fini(&(mgr->resdef));
mgr->nshared = 0;
mgr->nunshared = 0;
}
int32_t resmgr_add_res(resmgr_t* mgr, xce_resource_t* res)
{
int rc = 0;
rc = kvmgr_add(&(mgr->resdef), (void*)res->name, (void*)res);
if (rc < 0) {
return rc;
}
if (res->flags & XCE_RESF_SHARED) {
res->idx = mgr->nshared;
mgr->nshared++;
} else {
res->idx = mgr->nunshared;
mgr->nunshared++;
}
return rc;
}
xce_resource_t* resmgr_lockres(resmgr_t* mgr, char* resname, int op)
{
xce_resource_t* r = NULL;
r = (xce_resource_t*)kvmgr_lock(&(mgr->resdef), (void*)resname, op);
return r;
}
void resmgr_del_res(resmgr_t* mgr, xce_resource_t* r)
{
kvmgr_del(&(mgr->resdef), (void*)r->name);
}
int resmgr_iter(resmgr_t* mgr, xlist_t* l, void* arg)
{
return kvmgr_iter(&(mgr->resdef), NULL, l, arg);
}

33
src/lib/utils.c
View file

@ -1,9 +1,9 @@
#include <ctype.h>
#include <errno.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#if defined(__GLIBC__) && (__GLIBC__ == 2 && __GLIBC_MINOR__ < 28)
#include <pthread.h>
#else
@ -629,31 +629,34 @@ void block_sig(int signo)
c -= rot(b, 24); \
}
uint32_t hash(uint32_t a, uint32_t b,uint32_t c, uint32_t d) {
uint32_t hash(uint32_t a, uint32_t b, uint32_t c, uint32_t d)
{
uint32_t i1 = a, i2 = b, i3 = c, i4 = d;
mix(i1,i2,i3);
mix(i2,i3,i4);
mix(i3,i4,i1);
final(i4,i1,i2);
mix(i1, i2, i3);
mix(i2, i3, i4);
mix(i3, i4, i1);
final(i4, i1, i2);
return i2;
}
uint32_t hash64(uint64_t x) {
uint32_t a = (uint32_t)(x & UMAX_BITS(32));
uint32_t b = (uint32_t)(rot(x, 16) & UMAX_BITS(32));
uint32_t c = (uint32_t)(rot(x,32) & UMAX_BITS(32));
uint32_t d = (uint32_t)(rot(x, 48) & UMAX_BITS(32));
uint32_t hash64(uint64_t x)
{
uint32_t a = (uint32_t)(x >> 32);
uint32_t b = (uint32_t)(x & UMAX_BITS(32));
uint32_t c = ((a & UMAX_BITS(16)) << 16) | (b >> 16);
uint32_t d = ((b & UMAX_BITS(16)) << 16) | (a >> 16);
return hash(a,b,c,d);
return hash(a, b, c, d);
}
uint32_t hash_string(const char *key) {
uint32_t hash_string(const char* key)
{
uint32_t hc = 5381;
int c;
int c;
while ((c = *key ++)) {
while ((c = *key++)) {
hc = ((hc << 5) + hc) + c;
}