X-Git-Url: https://git.xandkar.net/?p=khatus.git;a=blobdiff_plain;f=x5%2Fkhatus.c;h=4dd42da54409bc9d19eb712982c535ec1ac85ccd;hp=4c8a6656fa8f702c15b5cefe81f1ee7f8120f606;hb=16e0239d76a7de132b20a564cc2c14ff2358ecf2;hpb=8ce5d971312e8bb27817e359e9712b3edfe011b5 diff --git a/x5/khatus.c b/x5/khatus.c index 4c8a665..4dd42da 100644 --- a/x5/khatus.c +++ b/x5/khatus.c @@ -17,7 +17,11 @@ #include "khatus_lib_log.h" #include "khatus_lib_time.h" -#define usage(...) {print_usage(); fprintf(stderr, "Error:\n " __VA_ARGS__); exit(EXIT_FAILURE);} +#define usage(...) { \ + print_usage(); \ + fprintf(stderr, "Error:\n " __VA_ARGS__); \ + exit(EXIT_FAILURE); \ +} #define ERRMSG "ERROR" static const char errmsg[] = ERRMSG; @@ -31,8 +35,8 @@ struct Fifo { char *name; int fd; int width; - int last_read; - int ttl; + struct timespec last_read; + struct timespec ttl; int pos_init; /* Initial position on the output buffer. */ int pos_curr; /* Current position on the output buffer. */ int pos_final; /* Final position on the output buffer. */ @@ -47,13 +51,6 @@ struct Config { int fifo_count; int total_width; int output_to_x_root_window; -} defaults = { - .interval = 1.0, - .separator = "|", - .fifos = NULL, - .fifo_count = 0, - .total_width = 0, - .output_to_x_root_window = 0, }; enum read_status { @@ -71,8 +68,8 @@ fifo_print_one(Fifo *f) " name = %s," " fd = %d," " width = %d," - " last_read = %d," - " ttl = %d," + " last_read = {tv_sec = %ld, tv_nsec = %ld}" + " ttl = {tv_sec = %ld, tv_nsec = %ld}," " pos_init = %d," " pos_curr = %d," " pos_final = %d," @@ -81,8 +78,10 @@ fifo_print_one(Fifo *f) f->name, f->fd, f->width, - f->last_read, - f->ttl, + f->last_read.tv_sec, + f->last_read.tv_nsec, + f->ttl.tv_sec, + f->ttl.tv_nsec, f->pos_init, f->pos_curr, f->pos_final, @@ -155,13 +154,13 @@ print_usage() " SPEC = FILE_PATH DATA_WIDTH DATA_TTL\n" " FILE_PATH = string\n" " DATA_WIDTH = int (* (positive) number of characters *)\n" - " DATA_TTL = int (* (positive) number of seconds *)\n" + " DATA_TTL = float (* (positive) number of seconds *)\n" " OPTION = -i INTERVAL\n" " | -s SEPARATOR\n" " | -x (* Output to X root window *)\n" " | -l LOG_LEVEL\n" " SEPARATOR = string\n" - " INTERVAL = int (* (positive) number of seconds *)\n" + " INTERVAL = float (* (positive) number of seconds *)\n" " LOG_LEVEL = int (* %d through %d *)\n" "\n", argv0, @@ -176,7 +175,8 @@ print_usage() ); } -void opts_parse_any(Config *, int, char *[], int); /* For mutually-recursive calls. */ +/* For mutually-recursive calls. */ +void opts_parse_any(Config *, int, char *[], int); void parse_opts_opt_i(Config *cfg, int argc, char *argv[], int i) @@ -215,7 +215,11 @@ parse_opts_opt_l(Config *cfg, int argc, char *argv[], int i) usage("Option -l parameter is invalid: \"%s\"\n", param); log_level = atoi(param); if (log_level > Debug) - usage("Option -l value (%d) exceeds maximum (%d)\n", log_level, Debug); + usage( + "Option -l value (%d) exceeds maximum (%d)\n", + log_level, + Debug + ); _khatus_lib_log_level = log_level; opts_parse_any(cfg, argc, argv, i); } @@ -249,23 +253,30 @@ void parse_opts_spec(Config *cfg, int argc, char *argv[], int i) { if ((i + 3) > argc) - usage("[spec] Parameter(s) missing for fifo \"%s\".\n", argv[i]); + usage( + "[spec] Parameter(s) missing for fifo \"%s\".\n", + argv[i] + ); char *n = argv[i++]; char *w = argv[i++]; char *t = argv[i++]; + struct timespec last_read; + if (!is_pos_num(w)) usage("[spec] Invalid width: \"%s\", for fifo \"%s\"\n", w, n); - if (!is_pos_num(t)) + if (!is_decimal(t)) usage("[spec] Invalid TTL: \"%s\", for fifo \"%s\"\n", t, n); + last_read.tv_sec = 0; + last_read.tv_nsec = 0; Fifo *f = calloc(1, sizeof(struct Fifo)); if (f) { f->name = n; f->fd = -1; f->width = atoi(w); - f->ttl = atoi(t); - f->last_read = 0; + f->ttl = timespec_of_float(atof(t)); + f->last_read = last_read; f->pos_init = cfg->total_width; f->pos_curr = f->pos_init; f->pos_final = f->pos_init + f->width - 1; @@ -309,6 +320,19 @@ opts_parse(Config *cfg, int argc, char *argv[]) } } +void +fifo_expire(Fifo *f, struct timespec t, char *buf) +{ + struct timespec td; + + timespecsub(&t, &(f->last_read), &td); + if (timespeccmp(&td, &(f->ttl), >=)) { + /* TODO: Maybe configurable expiry character. */ + memset(buf + f->pos_init, '_', f->pos_final - f->pos_init); + warn("Data source expired: \"%s\"\n", f->name); + } +} + void fifo_read_error(Fifo *f, char *buf) { @@ -317,7 +341,7 @@ fifo_read_error(Fifo *f, char *buf) b = buf + f->pos_init; /* Copy as much of the error message as possible. - * EXCLUDING the reminating \0. */ + * EXCLUDING the terminating \0. */ for (i = 0; i < errlen && i < f->width; i++) b[i] = errmsg[i]; /* Any remaining slots: */ @@ -326,17 +350,13 @@ fifo_read_error(Fifo *f, char *buf) } enum read_status -fifo_read_one(Fifo *f, char *buf) +fifo_read_one(Fifo *f, struct timespec t, char *buf) { - /* Initialize all to an impossible value: */ - ssize_t n = -5; /* Number of bytes read. */ - char c = -1; /* Character read. */ - int r = -1; /* Remaining unused slots in buffer range. */ + char c; /* Character read. */ + int r; /* Remaining unused slots in buffer range. */ for (;;) { - n = read(f->fd, &c, 1); - assert(n >= -1 && n <= 1); - switch (n) { + switch (read(f->fd, &c, 1)) { case -1: error("Failed to read: \"%s\". errno: %d, msg: %s\n", f->name, errno, strerror(errno)); @@ -358,33 +378,43 @@ fifo_read_one(Fifo *f, char *buf) if (r > 0) memset(buf + f->pos_curr, ' ', r); f->pos_curr = f->pos_init; + f->last_read = t; return END_OF_MESSAGE; } else { if (f->pos_curr <= f->pos_final) buf[f->pos_curr++] = c; /* Drop beyond available range. */ + /* + * TODO Define max after which we stop reading. + * To ensure that a rogue large message + * doesn't trap us here. + */ } break; default: assert(0); } } - /* TODO Record timestamp read */ } void -fifo_read_all(Config *cfg, char *buf) +fifo_read_all(Config *cfg, struct timespec *ti, char *buf) { fd_set fds; int maxfd = -1; - int ready; + int ready = 0; struct stat st; + struct timespec t; FD_ZERO(&fds); for (Fifo *f = cfg->fifos; f; f = f->next) { /* TODO: Create the FIFO if it doesn't already exist. */ if (lstat(f->name, &st) < 0) { - error("Cannot stat \"%s\". Error: %s\n", f->name, strerror(errno)); + error( + "Cannot stat \"%s\". Error: %s\n", + f->name, + strerror(errno) + ); fifo_read_error(f, buf); continue; } @@ -400,7 +430,7 @@ fifo_read_all(Config *cfg, char *buf) debug("%s: already openned. fd: %d\n", f->name, f->fd); } if (f->fd == -1) { - /* TODO: Consider backing off retries for failed fifos. */ + /* TODO Consider backing off retries for failed fifos */ error("Failed to open \"%s\"\n", f->name); fifo_read_error(f, buf); continue; @@ -411,42 +441,80 @@ fifo_read_all(Config *cfg, char *buf) FD_SET(f->fd, &fds); } debug("selecting...\n"); - ready = select(maxfd + 1, &fds, NULL, NULL, NULL); + ready = pselect(maxfd + 1, &fds, NULL, NULL, ti, NULL); debug("ready: %d\n", ready); - assert(ready != 0); - if (ready < 0) - fatal("%s", strerror(errno)); - for (Fifo *f = cfg->fifos; f; f = f->next) { - if (FD_ISSET(f->fd, &fds)) { - debug("reading: %s\n", f->name); - switch (fifo_read_one(f, buf)) { - case END_OF_FILE: - case FAILURE: - close(f->fd); - f->fd = -1; - break; - case END_OF_MESSAGE: - case RETRY: - break; - default: - assert(0); - } + clock_gettime(CLOCK_MONOTONIC, &t); + if (ready == -1) { + switch (errno) { + case EINTR: + error("pselect temp failure: %d, errno: %d, msg: %s\n", + ready, errno, strerror(errno)); + /* TODO: Reconsider what to do here. */ + return; + default: + fatal("pselect failed: %d, errno: %d, msg: %s\n", + ready, errno, strerror(errno)); } } + /* At-least-once ensures that expiries are still checked on timeouts. */ + do { + for (Fifo *f = cfg->fifos; f; f = f->next) { + if (FD_ISSET(f->fd, &fds)) { + debug("reading: %s\n", f->name); + switch (fifo_read_one(f, t, buf)) { + /* + * ### MESSAGE LOSS ### + * is introduced by closing at EOM in addition + * to EOF, since there may be unread messages + * remaining in the pipe. However, + * + * ### INTER-MESSAGE PUSHBACK ### + * is also gained, since pipes block at the + * "open" call. + * + * This is an acceptable trade-off because we + * are a stateless reporter of a _most-recent_ + * status, not a stateful accumulator. + */ + case END_OF_MESSAGE: + case END_OF_FILE: + case FAILURE: + close(f->fd); + f->fd = -1; + ready--; + break; + case RETRY: + break; + default: + assert(0); + } + } else { + fifo_expire(f, t, buf); + } + } + } while (ready); + assert(ready == 0); } int main(int argc, char *argv[]) { + Config cfg = { + .interval = 1.0, + .separator = "|", + .fifos = NULL, + .fifo_count = 0, + .total_width = 0, + .output_to_x_root_window = 0, + }; + int width = 0; int nfifos = 0; int seplen = 0; int prefix = 0; int errors = 0; char *buf; - Config cfg0 = defaults; - Config *cfg = &cfg0; - Display *display = NULL; + Display *d = NULL; struct stat st; struct timespec t0, /* time stamp. before reading fifos */ @@ -457,19 +525,23 @@ main(int argc, char *argv[]) argv0 = argv[0]; - opts_parse(cfg, argc, argv); + opts_parse(&cfg, argc, argv); debug("argv0 = %s\n", argv0); - config_print(cfg); + config_print(&cfg); - ti = timespec_of_float(cfg->interval); + ti = timespec_of_float(cfg.interval); - if (cfg->fifos == NULL) + if (cfg.fifos == NULL) usage("No fifo specs were given!\n"); /* 1st pass to check file existence and type */ - for (Fifo *f = cfg->fifos; f; f = f->next) { + for (Fifo *f = cfg.fifos; f; f = f->next) { if (lstat(f->name, &st) < 0) { - error("Cannot stat \"%s\". Error: %s\n", f->name, strerror(errno)); + error( + "Cannot stat \"%s\". Error: %s\n", + f->name, + strerror(errno) + ); errors++; continue; } @@ -480,13 +552,13 @@ main(int argc, char *argv[]) } } if (errors) - fatal("Encountered errors with the given file paths. See log.\n"); + fatal("Encountered errors with given file paths. See log.\n"); - width = cfg->total_width; - seplen = strlen(cfg->separator); + width = cfg.total_width; + seplen = strlen(cfg.separator); /* 2nd pass to make space for separators */ - for (Fifo *f = cfg->fifos; f; f = f->next) { + for (Fifo *f = cfg.fifos; f; f = f->next) { f->pos_init += prefix; f->pos_final += prefix; f->pos_curr = f->pos_init; @@ -500,36 +572,38 @@ main(int argc, char *argv[]) memset(buf, ' ', width); buf[width] = '\0'; /* 3rd pass to set the separators */ - for (Fifo *f = cfg->fifos; f; f = f->next) { + for (Fifo *f = cfg.fifos; f; f = f->next) { if (f->pos_init) { /* Skip the first, left-most */ /* Copying only seplen ensures we omit the '\0' byte. */ - strncpy(buf + (f->pos_init - seplen), cfg->separator, seplen); + strncpy( + buf + (f->pos_init - seplen), + cfg.separator, + seplen + ); } } - if (cfg->output_to_x_root_window && !(display = XOpenDisplay(NULL))) - fatal("XOpenDisplay failed with: %p\n", display); + if (cfg.output_to_x_root_window && !(d = XOpenDisplay(NULL))) + fatal("XOpenDisplay failed with: %p\n", d); /* TODO: Handle signals */ for (;;) { clock_gettime(CLOCK_MONOTONIC, &t0); // FIXME: check errors - /* TODO: Cache expiration. i.e. use the TTL */ - /* TODO: How to trigger TTL check? On select? Alarm signal? */ - /* TODO: Set timeout on fifo_read_all based on diff of last time of - * fifo_read_all and desired time of next TTL check. - */ - /* TODO: How long to wait on IO? Max TTL? */ - fifo_read_all(cfg, buf); - if (cfg->output_to_x_root_window) { - if (XStoreName(display, DefaultRootWindow(display), buf) < 0) + fifo_read_all(&cfg, &ti, buf); + if (cfg.output_to_x_root_window) { + if (XStoreName(d, DefaultRootWindow(d), buf) < 0) fatal("XStoreName failed.\n"); - XFlush(display); + XFlush(d); } else { puts(buf); fflush(stdout); } clock_gettime(CLOCK_MONOTONIC, &t1); // FIXME: check errors timespecsub(&t1, &t0, &td); - debug("td {tv_sec = %ld, tv_nsec = %ld}\n", td.tv_sec, td.tv_nsec); + debug( + "td {tv_sec = %ld, tv_nsec = %ld}\n", + td.tv_sec, + td.tv_nsec + ); if (timespeccmp(&td, &ti, <)) { /* Pushback on data producers by refusing to read the * pipe more frequently than the interval.