myxce/src/lib/utils.c
Rongsong Shen b7250c2ec6
fix(scheduler): downgrade routine EEXIST resync log, dedup hash_string
xce_core_push_events() WARNed on every JOBS_PUSH resync redelivery
(sched_add_job rc=-EEXIST), which is the expected outcome under
on_scheduler_join()'s full resync or the array-push dedup race
(bug-1734/bug-1714) and flooded operational logs. Only that case now
logs via a new DBG() macro (utils.h); any other add_rc (e.g. -ENOMEM)
still WARNs. DBG is named to avoid colliding with the build's own
-DDEBUG command-line define (debug/asan buildtypes).

Also collapsed hash_string()'s duplicated djb2 loop to delegate to
hash_block(), which already implements the same RUN_HASH loop.

Logged as bug-1809 in .wolf/buglog.json.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RLSNJhNoULvVcUhXEFafQC
2026-09-15 02:46:52 +00:00

806 lines
26 KiB
C

#include <ctype.h>
#include <errno.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#if defined(__GLIBC__) && (__GLIBC__ == 2 && __GLIBC_MINOR__ < 28)
#include <pthread.h>
#else
#include <threads.h>
#endif
#include <sys/param.h>
#include <unistd.h>
#if defined(__linux__)
#include <execinfo.h>
#endif
#include <regex.h>
#include <tlog.h>
#include "config.h"
#include "controller.capnp.h"
#include <capnp_c.h>
#include "utils.h"
#include "xce/errnox.h"
#include "xce/types.h"
#define ARGV_INCR_DELTA (16)
#define FILL_ARGV(rc, argv, argc, maxargc, delta, s) \
do { \
(rc) = 0; \
char* s1; \
if (((argc) >= (maxargc))) { \
char** p; \
int nc, i; \
nc = (maxargc) + (delta); \
p = (char**)realloc((argv), nc * sizeof(char*)); \
if (p == NULL) { \
(rc) = -ENOMEM; \
break; \
} \
for (i = (maxargc); i < nc; i++) { \
p[i] = NULL; \
} \
(argv) = p; \
(maxargc) = nc; \
} \
s1 = strdup(s); \
if (s1 == NULL) { \
(rc) = -ENOMEM; \
break; \
} \
(argv)[argc] = s1; \
(argc)++; \
} while (0)
#define FREE_ARGV(argv, argc) \
do { \
int i; \
if ((argv) == NULL) { \
break; \
} \
for (i = 0; i < (argc); i++) { \
if ((argv)[i] != NULL) { \
free((argv)[i]); \
(argv)[i] = NULL; \
} \
} \
free(argv); \
} while (0)
int parse_argv(char* input, char*** argv)
{
int max_argc = 0;
char* arg = NULL;
int i = 0;
int in_quotes = 0;
char quote_char = 0;
int argc = 0;
int rc = 0;
rc = 0;
arg = (char*)malloc(strlen(input) + 1);
if (arg == NULL) {
return -ENOMEM;
}
*argv = NULL;
while (*input) {
if (in_quotes) {
if (*input == quote_char) {
in_quotes = 0;
} else if (*input == '\\') {
input++;
if (*input) {
arg[i++] = *input;
}
} else {
arg[i++] = *input;
}
} else {
if (isspace(*input)) {
if (i > 0) {
arg[i] = '\0';
FILL_ARGV(rc, *argv, argc, max_argc, ARGV_INCR_DELTA, arg);
if (rc < 0) {
FREE_ARGV(*argv, argc);
free(arg);
return rc;
}
i = 0;
}
} else if ((*input == '"') || (*input == '\'')) {
in_quotes = 1;
quote_char = *input;
} else if (*input == '\\') {
input++;
if (*input) {
arg[i++] = *input;
}
} else {
arg[i++] = *input;
}
}
input++;
}
if (i > 0) {
arg[i] = '\0';
FILL_ARGV(rc, *argv, argc, max_argc, ARGV_INCR_DELTA, arg);
if (rc < 0) {
FREE_ARGV(*argv, argc);
free(arg);
return rc;
}
}
free(arg);
return argc;
}
void free_argv(char** argv, int argc) { FREE_ARGV(argv, argc); }
int get_myname(char* name, size_t size) { return gethostname(name, size); }
char* frame_symbol(int k)
{
#if defined(__linux__)
void* array[k + 1];
size_t size = backtrace(array, k + 1);
char** symbols = backtrace_symbols(array, size);
char* s = NULL;
if (symbols == NULL) {
return NULL;
}
s = strdup(symbols[MIN(size - 1, k)]);
free(symbols);
return s;
#else
return NULL;
#endif
}
int xce_vprintf(const char* fmt, va_list ap)
{
int rc;
if (tlog_get_root() == NULL) {
rc = vfprintf(stdout, fmt, ap);
} else {
rc = tlog_vext(TLOG_INFO, __FILE__, __LINE__, __func__, NULL, fmt, ap);
}
return rc;
}
int name_to_level(char* lvlname)
{
static struct {
char* lvlstr;
int lvl;
} levels[] = {
{ "debug", TLOG_DEBUG },
{ "debug2", TLOG_DEBUG2 },
{ "debug3", TLOG_DEBUG3 },
{ "info", TLOG_INFO },
{ "notice", TLOG_NOTICE },
{ "warn", TLOG_WARN },
{ "error", TLOG_ERROR },
{ "fatal", TLOG_FATAL },
{ NULL, -1 }
};
int i;
for (i = 0; levels[i].lvlstr != NULL; i++) {
if (strcasecmp(levels[i].lvlstr, lvlname) == 0) {
return levels[i].lvl;
}
}
return TLOG_INFO;
}
static int my_tlog_format(char* buff, int maxlen, struct tlog_loginfo* info,
void* userptr, const char* fmt, va_list ap)
{
int len = 0;
int total_len = 0;
struct tlog_time* tm = &info->time;
xce_use(userptr);
/* format prefix */
if (info->level < TLOG_INFO) {
len = snprintf(
buff, maxlen,
"[%.4d-%.2d-%.2d %.2d:%.2d:%.2d,%.3d] [%5d] [%6s] [%17s:%-4d] ",
tm->year, tm->mon, tm->mday, tm->hour, tm->min, tm->sec,
tm->usec / 1000, getpid(), tlog_get_level_string(info->level),
info->file, info->line);
} else {
len = snprintf(
buff, maxlen,
"[%.4d-%.2d-%.2d %.2d:%.2d:%.2d,%.3d] [%5d] [%6s] [%22s] ",
tm->year, tm->mon, tm->mday, tm->hour, tm->min, tm->sec,
tm->usec / 1000, getpid(), tlog_get_level_string(info->level),
info->func);
}
if (len < 0 || len >= maxlen) {
return -1;
}
buff += len;
total_len += len;
maxlen -= len;
/* format log message */
len = vsnprintf(buff, maxlen, fmt, ap);
if (len < 0 || len == maxlen) {
return -1;
}
buff += len;
total_len += len;
/* return total length */
return total_len;
}
int mylog_init(char* logfile, char* loglvl, int max_log_size, int max_log_count)
{
int flag = 0;
int rc = 0;
if (isatty(STDERR_FILENO)) {
flag = TLOG_SCREEN_COLOR;
}
rc = tlog_init(logfile, max_log_size, max_log_count, 0, flag);
if (rc >= 0) {
rc = tlog_setlevel(name_to_level(loglvl));
}
tlog_reg_format_func(my_tlog_format);
return rc;
}
char* verbose_to_loglevel(int verbose)
{
char* loglevel = "info";
switch (verbose) {
case 1:
loglevel = "debug";
break;
case 2:
loglevel = "debug2";
break;
case 3:
loglevel = "debug3";
break;
default:
if (verbose > 3) {
loglevel = "debug3";
}
break;
}
return loglevel;
}
int mylog_setlevel(char* level) { return tlog_setlevel(name_to_level(level)); }
void mylog_fini() { tlog_exit(); }
/* build-time-versioning task 3.1: see the SUPERSEDED comment above
* MK_VERSION in utils.h. raw is typically
* STR(VERSION_COMMIT_ID)/STR(VERSION_BUILD_TIME), which is "" (never a
* missing/NULL token) when the corresponding meson option was left unset
* -- NULL is also accepted defensively for any other caller. */
const char* xce_version_field(const char* raw)
{
if (raw == NULL || raw[0] == '\0') {
return "unknown";
}
return raw;
}
/* NOTE FOR CALLERS with a file-scope `argp_program_version`-style global
* (controller/main.c, scheduler/src/main.c both currently have one, built
* today via the now-superseded MK_VERSION macro -- see utils.h): this
* function is a regular runtime call, NOT a constant expression, so its
* result cannot initialize a file-scope variable
* (`const char* argp_program_version = xce_compose_version(...);` at file
* scope will not compile). Compute the version string into a
* `static char buf[...]` (sized generously enough for prefix + commit +
* " built " + timestamp, e.g. 128 bytes) inside main() -- or any function
* that runs before argp_parse() is called, since argp reads
* argp_program_version by pointer when --version/-V is processed -- then
* point the (non-const-initialized, runtime-assigned) global at that
* buffer. buf must outlive the call (static or otherwise long-lived
* storage, never a stack buffer that goes out of scope before argp_parse()
* runs). */
int xce_compose_version(char* buf, size_t bufsz, const char* prefix,
const char* commit, const char* build_time)
{
if (buf == NULL || bufsz == 0) {
return -1;
}
return snprintf(buf, bufsz, "%s%s built %s",
(prefix != NULL) ? prefix : "",
xce_version_field(commit), xce_version_field(build_time));
}
/* dirty-content-fingerprint task 3.1: buffer for the commit-id portion
* after an optional "-dirty" strip + "-<dirty_fingerprint>" append, built
* by xce_compose_version_ex() below before handing it to
* xce_compose_version()'s own "%s%s built %s" formatting. Sized generously
* for a full 40-hex-char SHA-1 commit id + "-dirty" (never both present at
* once, but sized as if they were) + "-" + a generous fingerprint length. */
#define XCE_COMPOSE_COMMIT_BUF_SIZE 128
static const char XCE_DIRTY_SUFFIX[] = "-dirty";
#define XCE_DIRTY_SUFFIX_LEN (sizeof(XCE_DIRTY_SUFFIX) - 1)
/* dirty-content-fingerprint task 3.1: --version-only sibling of
* xce_compose_version() above -- see the doc comment on its extern
* declaration in utils.h for the full contract. commit_id's own
* computation and its existing "-dirty" suffix (both from the
* already-shipped build-time-versioning feature) are NOT modified by this
* function or by anything it calls -- it only changes what gets composed
* into the returned --version display string. The startup-log composition
* call sites must keep calling xce_compose_version() directly and never
* this function. */
int xce_compose_version_ex(char* buf, size_t bufsz, const char* prefix,
const char* commit, const char* build_time,
const char* dirty_fingerprint)
{
const char* commit_field;
size_t commit_len;
char commit_buf[XCE_COMPOSE_COMMIT_BUF_SIZE];
if (buf == NULL || bufsz == 0) {
return -1;
}
if (dirty_fingerprint == NULL || dirty_fingerprint[0] == '\0') {
/* No fingerprint available -- delegate straight to the untouched,
* already-shipped function. Byte-identical output, including its
* own "unknown" fallback for a NULL/empty commit, and including
* commit's own "-dirty" suffix (if any) left completely as-is --
* this is Requirement 5.2's fallback, satisfied for free. */
return xce_compose_version(buf, bufsz, prefix, commit, build_time);
}
/* dirty_fingerprint is non-empty: strip a literal trailing "-dirty"
* from the commit field (anchored to the actual suffix position, not a
* blind substring removal -- see design.md's Risks section) and append
* "-<dirty_fingerprint>" in its place. */
commit_field = xce_version_field(commit);
commit_len = strlen(commit_field);
if (commit_len >= XCE_DIRTY_SUFFIX_LEN
&& strcmp(commit_field + (commit_len - XCE_DIRTY_SUFFIX_LEN),
XCE_DIRTY_SUFFIX)
== 0) {
snprintf(commit_buf, sizeof(commit_buf), "%.*s-%s",
(int)(commit_len - XCE_DIRTY_SUFFIX_LEN), commit_field,
dirty_fingerprint);
} else {
snprintf(commit_buf, sizeof(commit_buf), "%s-%s", commit_field,
dirty_fingerprint);
}
return snprintf(buf, bufsz, "%s%s built %s",
(prefix != NULL) ? prefix : "", commit_buf,
xce_version_field(build_time));
}
static _Thread_local ectx_t ectx;
void set_error(char* file, char* func, int line, int code)
{
ectx.file = file;
ectx.func = func;
ectx.line = line;
ectx.code = code;
errno = code;
}
ectx_t* get_ectx() { return &ectx; }
/* One string per E_XCE_* code, in enum order (errnox.h). Index = code -
* XCE_ERRNO_BASE. MUST stay in lockstep with the enum — the _Static_assert
* below fails the build if a code is added to errnox.h without a string here
* (that drift caused a global-buffer-overflow in xce_strerror_r: bug-NOT_READY,
* the HA codes 10011-10014 had no entries). */
static char* xce_err_str[] = { "invalid job", /* E_XCE_BAD_JOB */
"invalid worker", /* E_XCE_BAD_WORKER */
"duplicated jobs", /* E_XCE_DUP_JOB */
"duplicated workers", /* E_XCE_DUP_WORKER */
"bad opcode", /* E_XCE_BAD_OPCODE */
"unknown role", /* E_XCE_BAD_ROLE */
"unknown resource", /* E_XCE_BAD_RES */
"unknown queue", /* E_XCE_BAD_QUEUE */
"unknown group", /* E_XCE_BAD_GROUP */
"internal error", /* E_XCE_INTERNAL */
"no known primary", /* E_XCE_NOMASTER */
"redirect to primary", /* E_XCE_REDIRECT */
"controller not ready", /* E_XCE_NOT_READY */
"controller is standby" };/* E_XCE_STANDBY */
_Static_assert(sizeof(xce_err_str) / sizeof(xce_err_str[0])
== (size_t)(E_XCE_MAX - XCE_ERRNO_BASE),
"xce_err_str[] is out of sync with the E_XCE_* enum (errnox.h): "
"add a string for every new error code");
int xce_strerror_r(int errnum, char* buf, size_t bufsize)
{
if (bufsize == 0) {
return -1;
}
if (errnum < XCE_ERRNO_BASE) {
return strerror_r(errnum, buf, bufsize);
}
/* Defense in depth: bound against the ACTUAL table size (not just E_XCE_MAX)
* and reject NULL slots, so a future enum/table drift degrades to "unknown
* error" instead of reading past xce_err_str[]. */
size_t idx = (size_t)(errnum - XCE_ERRNO_BASE);
size_t nstr = sizeof(xce_err_str) / sizeof(xce_err_str[0]);
if (errnum >= E_XCE_MAX || idx >= nstr || xce_err_str[idx] == NULL) {
strncpy(buf, "unknown error", bufsize);
buf[bufsize - 1] = '\0';
return -1;
}
strncpy(buf, xce_err_str[idx], bufsize);
buf[bufsize - 1] = '\0';
return 0;
}
int copy_int_list(int** dst, int* src, int count)
{
int i;
FREEUP(*dst);
if (count == 0) {
return 0;
}
*dst = (int*)calloc(count, sizeof(int));
if ((*dst) == NULL) {
return -ENOMEM;
}
for (i = 0; i < count; i++) {
(*dst)[i] = (int)src[i];
}
return count;
}
int copy_strlist(xce_strlist_t* dst, xce_strlist_t src)
{
if (!dst) {
return -1;
}
for (int i = 0; i < dst->count; i++) {
FREEUP(dst->data[i]);
}
FREEUP(dst->data);
dst->count = src.count;
if (dst->count == 0) {
return 0;
}
dst->data = (char**)calloc(dst->count, sizeof(char*));
if (dst->data == NULL) {
return -ENOMEM;
}
for (int i = 0; i < src.count; i++) {
if (src.data[i]) {
dst->data[i] = strndup(src.data[i], strlen(src.data[i]));
if (dst->data[i] == NULL) {
return -ENOMEM;
}
}
}
return 0;
}
void free_name_val_list(xce_name_val_t*** vals, int *count_p)
{
xce_name_val_t** resvals = *vals;
int count = *count_p;
int i;
if (resvals == NULL || count <= 0) {
*count_p = 0;
return;
}
for (i = 0; i < count; i++) {
xce_name_val_t* nv = resvals[i];
if (nv == NULL) {
continue;
}
FREEUP(nv->name);
if (nv->val != NULL) {
if (nv->val->which == Value_u_svalue) {
FREEUP(nv->val->u.svalue);
}
FREEUP(nv->val);
}
FREEUP(nv);
}
FREEUP(resvals);
*vals = NULL;
*count_p = 0;
}
void free_strlist(xce_strlist_t* l)
{
int i;
if (l == NULL) {
return;
}
if ((l->count <= 0) || (l->data == NULL)) {
return;
}
for (i = 0; i < l->count; i++) {
FREEUP(l->data[i]);
}
FREEUP(l->data);
}
int fmt_jobid_ex(char* buf, size_t sz, uint64_t jobid, char* lc, char* rc)
{
uint32_t tag, id_cluster, id_apiserver, id_timestamp;
char data[128];
int len = 0;
tag = EXTRACT_ID_TAG(jobid);
id_cluster = EXTRACT_ID_CLUSTER(jobid);
id_apiserver = EXTRACT_ID_APISERVER(jobid);
id_timestamp = EXTRACT_ID_TIMESTAMP(jobid);
memset(data, 0x0, sizeof(data));
if ((tag & TAG_JOB_ARRAY) != 0) {
uint32_t id_job, id_idx;
id_job = EXTRACT_ARRAY_JOB(jobid);
id_idx = EXTRACT_ARRAY_IDX(jobid);
snprintf(data, sizeof(data) - 1, "%d-%d-%d-%d%s%d%s", id_cluster,
id_apiserver, id_timestamp, id_job, lc, id_idx, rc);
} else {
uint32_t id_seq;
id_seq = EXTRACT_ID_SEQUENCE(jobid);
snprintf(data, sizeof(data) - 1, "%d-%d-%d-%d", id_cluster,
id_apiserver, id_timestamp, id_seq);
}
len = strlen(data);
if (len > sz - 1) {
return -1;
}
strncpy(buf, data, sz);
return len;
}
int fmt_jobid(char* buf, size_t sz, uint64_t jobid)
{
return fmt_jobid_ex(buf, sz, jobid, "[", "]");
}
int parse_jobid(uint64_t* jobid, char* buf)
{
regex_t regex;
regmatch_t pmatch[N_MAX_FIELDS_JOBID + 1];
int rc;
int len = strlen(buf);
if ((jobid == NULL) || (buf == NULL)) {
return -EINVAL;
}
rc = regcomp(&regex, REGEX_JOBARRAY_ID, REG_EXTENDED);
if (rc != 0) {
return -errno;
}
*jobid = 0;
rc = regexec(&regex, buf, N_EL_ARRAY(pmatch), pmatch, 0);
if (rc == 0) {
regfree(&regex);
if ((pmatch[0].rm_so == 0) && (pmatch[0].rm_eo == len)) {
uint32_t tag, id_cluster, id_apiserver, id_timestamp, id_job,
id_idx;
tag = 0x01; /* job array */
id_cluster = atoi(&buf[pmatch[1].rm_so]);
id_apiserver = atoi(&buf[pmatch[2].rm_so]);
id_timestamp = atoi(&buf[pmatch[3].rm_so]);
id_job = atoi(&buf[pmatch[4].rm_so]);
id_idx = atoi(&buf[pmatch[5].rm_so]);
*jobid = MAKE_JOBID(id_cluster, id_apiserver, id_timestamp, tag,
id_job + id_idx);
return 1;
}
}
regfree(&regex);
rc = regcomp(&regex, REGEX_JOBID, REG_EXTENDED);
if (rc != 0) {
return -errno;
}
rc = regexec(&regex, buf, N_EL_ARRAY(pmatch), pmatch, 0);
if (rc == 0) {
uint32_t id_cluster, id_apiserver, id_timestamp, id_job;
if ((pmatch[0].rm_so != 0) || (pmatch[0].rm_eo != len)) {
regfree(&regex);
return -EINVAL;
}
id_cluster = atoi(&buf[pmatch[1].rm_so]);
id_apiserver = atoi(&buf[pmatch[2].rm_so]);
id_timestamp = atoi(&buf[pmatch[3].rm_so]);
id_job = atoi(&buf[pmatch[4].rm_so]);
*jobid
= MAKE_JOBID(id_cluster, id_apiserver, id_timestamp, 0x0, id_job);
regfree(&regex);
return 1;
}
regfree(&regex);
return -EINVAL;
}
uint64_t array_jobid(uint64_t jobid)
{
uint32_t tag = EXTRACT_ID_TAG(jobid);
if (tag == 0x00) {
return jobid;
}
return EXTRACT_ARRAY_BASEID(jobid);
}
uint64_t array_idx(uint64_t jobid)
{
uint32_t tag = EXTRACT_ID_TAG(jobid);
if (tag == 0x00) {
return 0;
}
return EXTRACT_ARRAY_IDX(jobid);
}
void unblock_sig(int signo)
{
sigset_t mask;
sigemptyset(&mask);
sigaddset(&mask, signo);
sigprocmask(SIG_UNBLOCK, &mask, NULL);
}
void block_sig(int signo)
{
sigset_t mask;
sigemptyset(&mask);
sigaddset(&mask, signo);
sigprocmask(SIG_BLOCK, &mask, NULL);
}
#define rot(x, k) (((x) << (k)) | ((x) >> (32 - (k))))
#define mix(a, b, c) \
{ \
a -= c; \
a ^= rot(c, 4); \
c += b; \
b -= a; \
b ^= rot(a, 6); \
a += c; \
c -= b; \
c ^= rot(b, 8); \
b += a; \
a -= c; \
a ^= rot(c, 16); \
c += b; \
b -= a; \
b ^= rot(a, 19); \
a += c; \
c -= b; \
c ^= rot(b, 4); \
b += a; \
}
#define final(a, b, c) \
{ \
c ^= b; \
c -= rot(b, 14); \
a ^= c; \
a -= rot(c, 11); \
b ^= a; \
b -= rot(a, 25); \
c ^= b; \
c -= rot(b, 16); \
a ^= c; \
a -= rot(c, 4); \
b ^= a; \
b -= rot(a, 14); \
c ^= b; \
c -= rot(b, 24); \
}
uint32_t hash4(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);
return i2;
}
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 hash4(a, b, c, d);
}
#define RUN_HASH(hc, c) (((hc) << 5) + (hc) + (c))
uint32_t hash_block(char* buf, size_t len)
{
uint32_t hc = 5381;
int i;
for (i = 0; i < len; i++) {
int c = (int)buf[i];
hc = RUN_HASH(hc, c);
}
return hc;
}
uint32_t hash_string(const char* key)
{
return hash_block((char*)key, strlen(key));
}