Skip to content

Commit

Permalink
mxqdump: Move database related functions to mxq_group and mxq_job
Browse files Browse the repository at this point in the history
  • Loading branch information
@mariux
mariux committed Jun 9, 2015
1 parent 5389e46 commit 3adf030
Show file tree
Hide file tree
Showing 6 changed files with 521 additions and 548 deletions.
2 changes: 2 additions & 0 deletions Makefile
View file
Original file line number Diff line number Diff line change
Expand Up @@ -356,6 +356,7 @@ mxqd: mxq_group.o
mxqd: mxq_job.o
mxqd: mxq_util.o
mxqd: mxq_mysql.o
mxqd: mx_mysql.o
mxqd: LDLIBS += $(LDLIBS_MYSQL)

build: mxqd
Expand Down Expand Up @@ -392,6 +393,7 @@ mxqdump: mxq_util.o
mxqdump: mx_util.o
mxqdump: mx_getopt.o
mxqdump: LDLIBS += $(LDLIBS_MYSQL)
mxqdump: CFLAGS += -Wunused-function

build: mxqdump

Expand Down
285 changes: 285 additions & 0 deletions mxq_group.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,34 @@
#include "mxq_group.h"
#include "mxq_job.h"
#include "mxq_mysql.h"
#include "mx_mysql.h"

#define GROUP_FIELDS_CNT 24
#define GROUP_FIELDS \
" group_id," \
" group_name," \
" group_status," \
" group_priority," \
" user_uid," \
" user_name," \
" user_gid," \
" user_group," \
" job_command," \
" job_threads," \
" job_memory," \
" job_time," \
" group_jobs," \
" group_jobs_inq," \
" group_jobs_running," \
" group_jobs_finished," \
" group_jobs_failed," \
" group_jobs_cancelled," \
" group_jobs_unknown," \
" group_slots_running," \
" stats_max_maxrss," \
" stats_max_utime_sec," \
" stats_max_stime_sec," \
" stats_max_real_sec"

#define MXQ_GROUP_FIELDS "group_id," \
"group_name," \
Expand Down Expand Up @@ -62,6 +90,48 @@ enum mxq_group_columns {
MXQ_GROUP_COL__END
};

static int bind_result_group_fields(struct mx_mysql_bind *result, struct mxq_group *g)
{
int res = 0;
int idx = 0;

res = mx_mysql_bind_init_result(result, GROUP_FIELDS_CNT);
assert(res >= 0);

res += mx_mysql_bind_var(result, idx++, uint64, &(g->group_id));
res += mx_mysql_bind_var(result, idx++, string, &(g->group_name));
res += mx_mysql_bind_var(result, idx++, uint8, &(g->group_status));
res += mx_mysql_bind_var(result, idx++, uint16, &(g->group_priority));

res += mx_mysql_bind_var(result, idx++, uint32, &(g->user_uid));
res += mx_mysql_bind_var(result, idx++, string, &(g->user_name));
res += mx_mysql_bind_var(result, idx++, uint32, &(g->user_gid));
res += mx_mysql_bind_var(result, idx++, string, &(g->user_group));

res += mx_mysql_bind_var(result, idx++, string, &(g->job_command));

res += mx_mysql_bind_var(result, idx++, uint16, &(g->job_threads));
res += mx_mysql_bind_var(result, idx++, uint64, &(g->job_memory));
res += mx_mysql_bind_var(result, idx++, uint32, &(g->job_time));

res += mx_mysql_bind_var(result, idx++, uint64, &(g->group_jobs));
res += mx_mysql_bind_var(result, idx++, uint64, &(g->group_jobs_inq));
res += mx_mysql_bind_var(result, idx++, uint64, &(g->group_jobs_running));
res += mx_mysql_bind_var(result, idx++, uint64, &(g->group_jobs_finished));
res += mx_mysql_bind_var(result, idx++, uint64, &(g->group_jobs_failed));
res += mx_mysql_bind_var(result, idx++, uint64, &(g->group_jobs_cancelled));
res += mx_mysql_bind_var(result, idx++, uint64, &(g->group_jobs_unknown));

res += mx_mysql_bind_var(result, idx++, uint64, &(g->group_slots_running));

res += mx_mysql_bind_var(result, idx++, uint32, &(g->stats_max_maxrss));
res += mx_mysql_bind_var(result, idx++, int64, &(g->stats_max_utime.tv_sec));
res += mx_mysql_bind_var(result, idx++, int64, &(g->stats_max_stime.tv_sec));
res += mx_mysql_bind_var(result, idx++, int64, &(g->stats_max_real.tv_sec));

return res;
}

static inline int mxq_group_bind_results(MYSQL_BIND *bind, struct mxq_group *g)
{
memset(bind, 0, sizeof(*bind)*MXQ_GROUP_COL__END);
Expand Down Expand Up @@ -173,7 +243,222 @@ inline uint64_t mxq_group_jobs_inq(struct mxq_group *g)
return inq;
}

int mxq_load_group(struct mx_mysql *mysql, struct mxq_group **mxq_groups, uint64_t group_id)
{
int res;
struct mxq_group *groups = NULL;
struct mxq_group g = {0};
struct mx_mysql_bind param = {0};
struct mx_mysql_bind result = {0};

assert(mysql);
assert(mxq_groups);
assert(!(*mxq_groups));

char *query =
"SELECT"
GROUP_FIELDS
" FROM mxq_group"
" WHERE group_id = ?"
" LIMIT 1";

res = mx_mysql_bind_init_param(&param, 1);
assert(res == 0);
res = mx_mysql_bind_var(&param, 0, uint64, &group_id);
assert(res == 0);

res = bind_result_group_fields(&result, &g);
assert(res == 0);

res = mx_mysql_do_statement(mysql, query, &param, &result, &g, (void **)&groups, sizeof(*groups));
if (res < 0) {
mx_log_err("mx_mysql_do_statement(): %m");
return res;
}

*mxq_groups = groups;
return res;
}

int mxq_load_all_groups(struct mx_mysql *mysql, struct mxq_group **mxq_groups)
{
int res;
struct mxq_group *groups = NULL;
struct mxq_group g = {0};
struct mx_mysql_bind result = {0};

assert(mysql);
assert(mxq_groups);
assert(!(*mxq_groups));

char *query =
"SELECT"
GROUP_FIELDS
" FROM mxq_group"
" ORDER BY user_name, group_mtime"
" LIMIT 1000";

res = bind_result_group_fields(&result, &g);
assert(res == 0);

res = mx_mysql_do_statement(mysql, query, NULL, &result, &g, (void **)&groups, sizeof(*groups));
if (res < 0) {
mx_log_err("mx_mysql_do_statement(): %m");
return res;
}

*mxq_groups = groups;
return res;
}

int mxq_load_all_groups_for_user(struct mx_mysql *mysql, struct mxq_group **mxq_groups, uint64_t user_uid)
{
int res;
struct mxq_group *groups = NULL;
struct mxq_group g = {0};
struct mx_mysql_bind param = {0};
struct mx_mysql_bind result = {0};

assert(mysql);
assert(mxq_groups);
assert(!(*mxq_groups));

char *query =
"SELECT"
GROUP_FIELDS
" FROM mxq_group"
" WHERE user_uid = ?"
" ORDER BY user_name, group_mtime"
" LIMIT 1000";

res = mx_mysql_bind_init_param(&param, 1);
assert(res == 0);
res = mx_mysql_bind_var(&param, 0, uint64, &user_uid);
assert(res == 0);

res = bind_result_group_fields(&result, &g);
assert(res == 0);

res = mx_mysql_do_statement(mysql, query, &param, &result, &g, (void **)&groups, sizeof(*groups));
if (res < 0) {
mx_log_err("mx_mysql_do_statement(): %m");
return res;
}

*mxq_groups = groups;
return res;
}

int mxq_load_active_groups_for_user(struct mx_mysql *mysql, struct mxq_group **mxq_groups, uint64_t user_uid)
{
int res;
struct mxq_group *groups = NULL;
struct mxq_group g = {0};
struct mx_mysql_bind result = {0};
struct mx_mysql_bind param = {0};

assert(mysql);
assert(mxq_groups);
assert(!(*mxq_groups));

char *query =
"SELECT"
GROUP_FIELDS
" FROM mxq_group"
" WHERE ((group_jobs_inq > 0 OR group_jobs_running > 0)"
" OR (NOW()-group_mtime < 604800))"
" AND user_uid = ?"
" ORDER BY user_name, group_mtime"
" LIMIT 1000";

res = mx_mysql_bind_init_param(&param, 1);
assert(res == 0);
res = mx_mysql_bind_var(&param, 0, uint64, &user_uid);
assert(res == 0);

res = bind_result_group_fields(&result, &g);
assert(res == 0);

res = mx_mysql_do_statement(mysql, query, &param, &result, &g, (void **)&groups, sizeof(*groups));
if (res < 0) {
mx_log_err("mx_mysql_do_statement(): %m");
return res;
}

*mxq_groups = groups;
return res;
}

int mxq_load_running_groups(struct mx_mysql *mysql, struct mxq_group **mxq_groups)
{
int res;
struct mxq_group *groups = NULL;
struct mxq_group g = {0};
struct mx_mysql_bind result = {0};

assert(mysql);
assert(mxq_groups);
assert(!(*mxq_groups));

char *query =
"SELECT"
GROUP_FIELDS
" FROM mxq_group"
" WHERE (group_jobs_inq > 0 OR group_jobs_running > 0)"
" ORDER BY user_name, group_mtime"
" LIMIT 1000";

res = bind_result_group_fields(&result, &g);
assert(res == 0);

res = mx_mysql_do_statement(mysql, query, NULL, &result, &g, (void **)&groups, sizeof(*groups));
if (res < 0) {
mx_log_err("mx_mysql_do_statement(): %m");
return res;
}

*mxq_groups = groups;
return res;
}

int mxq_load_running_groups_for_user(struct mx_mysql *mysql, struct mxq_group **mxq_groups, uint64_t user_uid)
{
int res;
struct mxq_group *groups = NULL;
struct mxq_group g = {0};
struct mx_mysql_bind param = {0};
struct mx_mysql_bind result = {0};

assert(mysql);
assert(mxq_groups);
assert(!(*mxq_groups));

char *query =
"SELECT"
GROUP_FIELDS
" FROM mxq_group"
" WHERE (group_jobs_inq > 0 OR group_jobs_running > 0)"
" AND user_uid = ?"
" ORDER BY user_name, group_mtime"
" LIMIT 1000";

res = mx_mysql_bind_init_param(&param, 1);
assert(res == 0);
res = mx_mysql_bind_var(&param, 0, uint64, &user_uid);
assert(res == 0);

res = bind_result_group_fields(&result, &g);
assert(res == 0);

res = mx_mysql_do_statement(mysql, query, &param, &result, &g, (void **)&groups, sizeof(*groups));
if (res < 0) {
mx_log_err("mx_mysql_do_statement(): %m");
return res;
}

*mxq_groups = groups;
return res;
}

int mxq_group_load_active_groups(MYSQL *mysql, struct mxq_group **mxq_group)
{
Expand Down
9 changes: 9 additions & 0 deletions mxq_group.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@
#include <stdint.h>
#include <mysql.h>

#include "mx_mysql.h"

struct mxq_group {
uint64_t group_id;

Expand Down Expand Up @@ -52,6 +54,13 @@ inline uint64_t mxq_group_jobs_done(struct mxq_group *g);
inline uint64_t mxq_group_jobs_active(struct mxq_group *g);
inline uint64_t mxq_group_jobs_inq(struct mxq_group *g);

int mxq_load_group(struct mx_mysql *mysql, struct mxq_group **mxq_groups, uint64_t group_id);
int mxq_load_all_groups(struct mx_mysql *mysql, struct mxq_group **mxq_groups);
int mxq_load_all_groups_for_user(struct mx_mysql *mysql, struct mxq_group **mxq_groups, uint64_t user_uid);
int mxq_load_active_groups_for_user(struct mx_mysql *mysql, struct mxq_group **mxq_groups, uint64_t user_uid);
int mxq_load_running_groups(struct mx_mysql *mysql, struct mxq_group **mxq_groups);
int mxq_load_running_groups_for_user(struct mx_mysql *mysql, struct mxq_group **mxq_groups, uint64_t user_uid);

int mxq_group_load_active_groups(MYSQL *mysql, struct mxq_group **mxq_group);
void mxq_group_free_content(struct mxq_group *g);
int mxq_group_update_status_cancelled(MYSQL *mysql, struct mxq_group *group);
Expand Down
Loading

0 comments on commit 3adf030

Please sign in to comment.