1/* log.rs: a bounded queue drained by one thread, a ring of recent lines, counters. */
2#include "mesh/log.h"
3#include "mesh/text.h"
4
5#include <errno.h>
6#include <pthread.h>
7#include <stdarg.h>
8#include <stdatomic.h>
9#include <stdio.h>
10#include <stdbool.h>
11#include <stdlib.h>
12#include <string.h>
13#include <time.h>
14#include <unistd.h>
15
16static pthread_mutex_t mu = PTHREAD_MUTEX_INITIALIZER;
17static pthread_cond_t cv = PTHREAD_COND_INITIALIZER;
18static char *queue[MESH_LOG_QUEUE];
19static size_t qhead, qlen;
20static char *recent[MESH_LOG_KEEP];
21static size_t rhead, rlen;
22static bool started, thread_failed;
23static atomic_uint_fast64_t dropped, queued, done;
24static atomic_int out_fd = 2;
25
26void mesh_log_set_fd(int fd) { atomic_store(&out_fd, fd); }
27
28void mesh_log_stamp(char buf[16]) {
29    struct timespec ts;
30    clock_gettime(CLOCK_REALTIME, &ts);
31    unsigned s = (unsigned)(ts.tv_sec % 86400), ms = (unsigned)(ts.tv_nsec / 1000000) % 1000;
32    snprintf(buf, 16, "%02u:%02u:%02u.%03u", s / 3600 % 24, s / 60 % 60, s % 60, ms);
33}
34
35/* The whole line on the fd, or false on an error (EPIPE included; SIGPIPE is the program's to ignore). */
36static bool write_all(int fd, const char *b, size_t n) {
37    size_t off = 0;
38    while (off < n) {
39        ssize_t k = write(fd, b + off, n - off);
40        if (k > 0) {
41            off += (size_t)k;
42            continue;
43        }
44        if (k < 0 && errno == EINTR) continue;
45        return false;
46    }
47    return true;
48}
49
50static void *drain(void *arg) {
51    (void)arg;
52    for (;;) {
53        pthread_mutex_lock(&mu);
54        while (qlen == 0) pthread_cond_wait(&cv, &mu);
55        char *l = queue[qhead];
56        qhead = (qhead + 1) % MESH_LOG_QUEUE;
57        qlen--;
58        pthread_mutex_unlock(&mu);
59        size_t n = strlen(l);
60        l[n] = '\n'; /* the slot has room: see emit */
61        if (!write_all(atomic_load(&out_fd), l, n + 1)) atomic_fetch_add(&dropped, 1);
62        free(l);
63        atomic_fetch_add(&done, 1);
64    }
65    return NULL;
66}
67
68void mesh_log_emit(const char *line) {
69    size_t n = strlen(line);
70    char *keep = mesh_strndup(line, n);
71    char *q = mesh_alloc(n + 2); /* the line, its '\n', NUL */
72    memcpy(q, line, n);
73    pthread_mutex_lock(&mu);
74    if (rlen == MESH_LOG_KEEP) {
75        free(recent[rhead]);
76        recent[rhead] = keep;
77        rhead = (rhead + 1) % MESH_LOG_KEEP;
78    } else {
79        recent[(rhead + rlen++) % MESH_LOG_KEEP] = keep;
80    }
81    if (!started) {
82        started = true;
83        pthread_t t;
84        pthread_attr_t at;
85        pthread_attr_init(&at);
86        pthread_attr_setdetachstate(&at, PTHREAD_CREATE_DETACHED);
87        thread_failed = pthread_create(&t, &at, drain, NULL) != 0;
88        pthread_attr_destroy(&at);
89    }
90    bool ok = !thread_failed && qlen < MESH_LOG_QUEUE;
91    if (ok) {
92        queue[(qhead + qlen++) % MESH_LOG_QUEUE] = q;
93        pthread_cond_signal(&cv);
94    }
95    pthread_mutex_unlock(&mu);
96    if (ok) atomic_fetch_add(&queued, 1);
97    else {
98        free(q);
99        atomic_fetch_add(&dropped, 1);
100    }
101}
102
103static void emitv(const char *level, const char *fmt, va_list ap) {
104    char st[16];
105    mesh_log_stamp(st);
106    mesh_text t = {0};
107    mesh_text_printf(&t, "%s %s ", st, level);
108    mesh_text_vprintf(&t, fmt, ap);
109    mesh_log_emit(mesh_text_str(&t));
110    mesh_text_free(&t);
111}
112
113void mesh_info(const char *fmt, ...) {
114    va_list ap;
115    va_start(ap, fmt);
116    emitv("INFO ", fmt, ap);
117    va_end(ap);
118}
119
120void mesh_error(const char *fmt, ...) {
121    va_list ap;
122    va_start(ap, fmt);
123    emitv("ERROR", fmt, ap);
124    va_end(ap);
125}
126
127uint64_t mesh_log_dropped(void) { return (uint64_t)atomic_load(&dropped); }
128
129void mesh_log_flush(unsigned limit_ms) {
130    struct timespec t0, now, nap = {0, 1000000};
131    clock_gettime(CLOCK_MONOTONIC, &t0);
132    for (;;) {
133        if (atomic_load(&done) >= atomic_load(&queued)) return;
134        clock_gettime(CLOCK_MONOTONIC, &now);
135        long ms = (long)(now.tv_sec - t0.tv_sec) * 1000 + (now.tv_nsec - t0.tv_nsec) / 1000000;
136        if (ms >= (long)limit_ms) return;
137        nanosleep(&nap, NULL);
138    }
139}
140
141_Noreturn void mesh_log_exit(int code) {
142    mesh_log_flush(500);
143    exit(code);
144}
145
146size_t mesh_log_recent(size_t n, char ***lines) {
147    pthread_mutex_lock(&mu);
148    size_t k = n < rlen ? n : rlen;
149    *lines = mesh_alloc((k ? k : 1) * sizeof **lines);
150    for (size_t i = 0; i < k; i++) (*lines)[i] = mesh_strdup(recent[(rhead + rlen - k + i) % MESH_LOG_KEEP]);
151    pthread_mutex_unlock(&mu);
152    return k;
153}