mxq_daemon.c

#include <stdio.h>

#include <assert.h>
#include <errno.h>

#include <mysql.h>

#include "mx_log.h"

#include "mx_util.h"
#include "mx_mysql.h"

#include "mxq_daemon.h"

#define DAEMON_FIELDS_CNT 21
#define DAEMON_FIELDS \
            " daemon_id," \
            " daemon_name," \
            " status," \
            " hostname," \
            " mxq_version," \
            " boot_id," \
            " pid_starttime," \
            " daemon_pid," \
            " daemon_slots," \
            " daemon_memory," \
            " daemon_maxtime" \
            " daemon_memory_limit_slot_soft," \
            " daemon_memory_limit_slot_hard," \
            " daemon_jobs_running," \
            " daemon_slots_running," \
            " daemon_threads_running," \
            " daemon_memory_used," \
            " daemon_gpus_max," \
            " daemon_gpus_used," \
            " UNIX_TIMESTAMP(mtime)        as mtime," \
            " UNIX_TIMESTAMP(daemon_start) as daemon_start," \
            " UNIX_TIMESTAMP(daemon_stop)  as daemon_stop," \
            " daemon_flags," \
            " tags," \
            " prerequisites"

#undef _to_string
#undef status_str
#define _to_string(s) #s
#define status_str(x) _to_string(x)

int mxq_daemon_register(struct mx_mysql *mysql, struct mxq_daemon *daemon)
{
    struct mx_mysql_stmt *stmt = NULL;
    unsigned long long num_rows = 0;
    unsigned long long insert_id = 0;
    int res;
    int idx;

    assert(daemon);

    assert(daemon->daemon_name);
    assert(daemon->hostname);
    assert(daemon->mxq_version);
    assert(daemon->boot_id);

    assert(*daemon->daemon_name);
    assert(*daemon->hostname);
    assert(*daemon->mxq_version);
    assert(*daemon->boot_id);

    assert(daemon->pid_starttime);

    assert(daemon->daemon_pid);
    assert(daemon->daemon_slots);
    assert(daemon->daemon_memory);
    assert(daemon->daemon_memory_limit_slot_soft <= daemon->daemon_memory_limit_slot_hard);
    assert(daemon->daemon_memory_limit_slot_hard <= daemon->daemon_memory);

    stmt = mx_mysql_statement_prepare(mysql,
            "INSERT INTO"
                " mxq_daemon"
            " SET"
                " daemon_name   = ?,"
                " status        = ?,"
                " hostname      = ?,"
                " mxq_version   = ?,"
                " boot_id       = ? ,"
                " pid_starttime = ?,"
                " daemon_pid    = ?,"
                " daemon_slots  = ?,"
                " daemon_memory = ?,"
                " daemon_maxtime= ?,"
                " daemon_memory_limit_slot_soft = ?,"
                " daemon_memory_limit_slot_hard = ?,"
                " daemon_jobs_running    = 0,"
                " daemon_slots_running   = 0,"
                " daemon_threads_running = 0,"
                " daemon_memory_used     = 0,"
                " daemon_gpus_max        = ?,"
                " daemon_gpus_used       = 0,"
                " mtime        = NULL,"
                " daemon_start = CURRENT_TIMESTAMP(),"
                " daemon_stop  = 0,"
                " daemon_flags = ?,"
                " tags = ?,"
                " prerequisites = ?"
                );
    if (!stmt) {
        mx_log_err("mx_mysql_statement_prepare(): %s", mx_mysql_error());
        mx_mysql_statement_close(&stmt);
        return -EINVAL;
    }

    idx  = 0;

    mx_mysql_statement_param_bind(stmt, idx++, string, &(daemon->daemon_name));
    mx_mysql_statement_param_bind(stmt, idx++,  uint8, &(daemon->status));
    mx_mysql_statement_param_bind(stmt, idx++, string, &(daemon->hostname));
    mx_mysql_statement_param_bind(stmt, idx++, string, &(daemon->mxq_version));
    mx_mysql_statement_param_bind(stmt, idx++, string, &(daemon->boot_id));

    mx_mysql_statement_param_bind(stmt, idx++, uint64, &(daemon->pid_starttime));
    mx_mysql_statement_param_bind(stmt, idx++, uint32, &(daemon->daemon_pid));

    mx_mysql_statement_param_bind(stmt, idx++, uint32, &(daemon->daemon_slots));
    mx_mysql_statement_param_bind(stmt, idx++, uint64, &(daemon->daemon_memory));
    mx_mysql_statement_param_bind(stmt, idx++, uint64, &(daemon->daemon_maxtime));

    mx_mysql_statement_param_bind(stmt, idx++, uint64, &(daemon->daemon_memory_limit_slot_soft));
    mx_mysql_statement_param_bind(stmt, idx++, uint64, &(daemon->daemon_memory_limit_slot_hard));
    mx_mysql_statement_param_bind(stmt, idx++, uint16, &(daemon->gpus_max));
    mx_mysql_statement_param_bind(stmt, idx++, int32,  &(daemon->daemon_flags));
    mx_mysql_statement_param_bind(stmt, idx++, string, &(daemon->tags));
    mx_mysql_statement_param_bind(stmt, idx++, string, &(daemon->prerequisites));

    res = mx_mysql_statement_execute(stmt, &num_rows);
    if (res < 0) {
        mx_log_err("mx_mysql_statement_execute(): %s", mx_mysql_error());
        mx_mysql_statement_close(&stmt);
        return res;
    }

    assert(num_rows == 1);
    mx_mysql_statement_insert_id(stmt, &insert_id);
    assert(insert_id > 0);

    daemon->daemon_id = insert_id;

    mx_mysql_statement_close(&stmt);

    return (int)num_rows;
}

int mxq_daemon_shutdown(struct mx_mysql *mysql, struct mxq_daemon *daemon)
{
    assert(daemon);
    assert(daemon->daemon_id);

    struct mx_mysql_bind param = {0};
    char *query;
    int idx;
    int res;

    query = "UPDATE"
                " mxq_daemon"
            " SET"
                " mtime        = NULL,"
                " daemon_stop  = CURRENT_TIMESTAMP(),"
                " status       = " status_str(MXQ_DAEMON_STATUS_EXITED)
            " WHERE daemon_id = ?";

    mx_mysql_bind_init_param(&param, 1);

    idx  = 0;
    mx_mysql_bind_var(&param, idx++, uint32, &(daemon->daemon_id));

    res = mx_mysql_do_statement_noresult_retry_on_fail(mysql, query, &param);
    if (res < 0) {
        mx_log_err("mx_mysql_do_statement(): %s", mx_mysql_error());
        return res;
    }

    daemon->status = MXQ_DAEMON_STATUS_EXITED;

    return res;
}

int mxq_daemon_mark_crashed(struct mx_mysql *mysql, struct mxq_daemon *daemon)
{
    assert(daemon);
    assert(daemon->daemon_id);

    struct mx_mysql_bind param = {0};
    char *query;
    int idx;
    int res;

    query = "UPDATE"
                " mxq_daemon"
            " SET"
                " daemon_stop  = CURRENT_TIMESTAMP(),"
                " status       = " status_str(MXQ_DAEMON_STATUS_CRASHED)
            " WHERE status NOT IN ("
                    status_str(MXQ_DAEMON_STATUS_EXITED) ","
                    status_str(MXQ_DAEMON_STATUS_CRASHED) ")"
              " AND daemon_id  != ?"
              " AND hostname    = ?"
              " AND daemon_name = ?";

    mx_mysql_bind_init_param(&param, 3);

    idx  = 0;
    mx_mysql_bind_var(&param, idx++, uint32, &daemon->daemon_id);
    mx_mysql_bind_var(&param, idx++, string, &daemon->hostname);
    mx_mysql_bind_var(&param, idx++, string, &daemon->daemon_name);

    res = mx_mysql_do_statement_noresult_retry_on_fail(mysql, query, &param);
    if (res < 0) {
        mx_log_err("mx_mysql_do_statement(): %s", mx_mysql_error());
        return res;
    }

    return res;
}

int mxq_daemon_set_status(struct mx_mysql *mysql, struct mxq_daemon *daemon, uint8_t status)
{
    assert(daemon);
    assert(daemon->daemon_id);

    struct mx_mysql_bind param = {0};
    char *query;
    int idx;
    int res;

    /* set the same status only once every >= 5 minutes or return 0 */
    if (daemon->status == status)
        mx_within_rate_limit_or_return(5*60, 0);

    query = "UPDATE"
                " mxq_daemon"
            " SET"
                " mtime        = NULL,"
                " status       = ?"
            " WHERE daemon_id = ?";

    mx_mysql_bind_init_param(&param, 2);

    idx  = 0;
    mx_mysql_bind_var(&param, idx++,  uint8, &(status));
    mx_mysql_bind_var(&param, idx++, uint32, &(daemon->daemon_id));

    res = mx_mysql_do_statement_noresult_retry_on_fail(mysql, query, &param);
    if (res < 0) {
        mx_log_err("mx_mysql_do_statement(): %s", mx_mysql_error());
        return res;
    }

    daemon->status = status;

    return res;
}

int mxq_daemon_update_statistics(struct mx_mysql *mysql, struct mxq_daemon *daemon)
{
    assert(daemon);
    assert(daemon->daemon_id);

    struct mx_mysql_bind param = {0};
    char *query;
    int idx;
    int res;

    query = "UPDATE"
                " mxq_daemon"
            " SET"
                " mtime = NULL,"
                " daemon_jobs_running    = ?,"
                " daemon_slots_running   = ?,"
                " daemon_threads_running = ?,"
                " daemon_memory_used     = ?,"
                " daemon_gpus_used       = ?"
            " WHERE daemon_id = ?";

    mx_mysql_bind_init_param(&param, 6);

    idx  = 0;
    mx_mysql_bind_var(&param, idx++, uint32, &(daemon->daemon_jobs_running));
    mx_mysql_bind_var(&param, idx++, uint32, &(daemon->daemon_slots_running));
    mx_mysql_bind_var(&param, idx++, uint32, &(daemon->daemon_threads_running));
    mx_mysql_bind_var(&param, idx++, uint64, &(daemon->daemon_memory_used));
    mx_mysql_bind_var(&param, idx++, uint16, &(daemon->gpus_used));
    mx_mysql_bind_var(&param, idx++, uint32, &(daemon->daemon_id));

    res = mx_mysql_do_statement_noresult_retry_on_fail(mysql, query, &param);
    if (res < 0) {
        mx_log_err("mx_mysql_do_statement(): %s", mx_mysql_error());
        return res;
    }

    return res;
}