X-Git-Url: https://git.xandkar.net/?p=khatus.git;a=blobdiff_plain;f=x5%2Fkhatus.c;h=5407acde0b145131f038515d773e0a8d17bf2526;hp=382c8731aaeee3c6494955567ecbd806468bb783;hb=HEAD;hpb=900b9a6b1cf64ec6fc67bca260784ab93aec4fa1 diff --git a/x5/khatus.c b/x5/khatus.c index 382c873..5407acd 100644 --- a/x5/khatus.c +++ b/x5/khatus.c @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -24,10 +25,12 @@ } #define ERRMSG "ERROR" + static const char errmsg[] = ERRMSG; static const int errlen = sizeof(ERRMSG) - 1; -char *argv0; +char *argv0 = NULL; +int running = 1; /* TODO: Convert slot list to slot array. */ typedef struct Slot Slot; @@ -47,10 +50,11 @@ typedef struct Config Config; struct Config { double interval; char * separator; + char expiry_character; Slot * slots; int slot_count; - int total_width; - int output_to_x_root_window; + int buf_width; + int to_x_root; }; enum read_status { @@ -60,8 +64,54 @@ enum read_status { FAILURE }; +char * +buf_create(Config *cfg) +{ + int seplen; + char *buf; + Slot *s; + + buf = calloc(1, cfg->buf_width + 1); + if (buf == NULL) + khlib_fatal( + "[memory] Failed to allocate buffer of %d bytes", + cfg->buf_width + ); + memset(buf, ' ', cfg->buf_width); + buf[cfg->buf_width] = '\0'; + seplen = strlen(cfg->separator); + /* Set the separators */ + for (s = cfg->slots; s; s = s->next) { + /* Skip the first, left-most */ + if (s->out_pos_lo) { + /* Copying only seplen ensures we omit the '\0' byte. */ + strncpy( + buf + (s->out_pos_lo - seplen), + cfg->separator, + seplen + ); + } + } + return buf; +} + +Slot * +slots_rev(Slot *old) +{ + Slot *tmp = NULL; + Slot *new = NULL; + + while (old) { + tmp = old->next; + old->next = new; + new = old; + old = tmp; + } + return new; +} + void -slot_print_one(Slot *s) +slot_log(Slot *s) { khlib_info("Slot " "{" @@ -90,15 +140,277 @@ slot_print_one(Slot *s) } void -slot_print_all(Slot *head) +slots_log(Slot *head) +{ + Slot *s = head; + + for (; s; s = s->next) { + slot_log(s); + } +} + +void +slots_assert_fifos_exist(Slot *s) +{ + struct stat st; + int errors = 0; + + for (; s; s = s->next) { + if (lstat(s->in_fifo, &st) < 0) { + khlib_error( + "Cannot stat \"%s\". Error: %s\n", + s->in_fifo, + strerror(errno) + ); + errors++; + continue; + } + if (!(st.st_mode & S_IFIFO)) { + khlib_error("\"%s\" is not a FIFO\n", s->in_fifo); + errors++; + continue; + } + } + if (errors) + khlib_fatal( + "Encountered errors with given file paths. See log.\n" + ); +} + +void +slot_close(Slot *s) +{ + close(s->in_fd); + s->in_fd = -1; + s->out_pos_cur = s->out_pos_lo; +} + +void +slots_close(Slot *s) +{ + for (; s; s = s->next) + if (s->in_fd > -1) + slot_close(s); +} + + +void +slot_expire(Slot *s, struct timespec t, char expiry_character, char *buf) { - for (Slot *s = head; s; s = s->next) { - slot_print_one(s); + struct timespec td; + + timespecsub(&t, &(s->in_last_read), &td); + if (timespeccmp(&td, &(s->out_ttl), >=)) { + memset( + buf + s->out_pos_lo, + expiry_character, + s->out_width + ); + khlib_warn("Slot expired: \"%s\"\n", s->in_fifo); + } +} + +void +slot_set_error(Slot *s, char *buf) +{ + char *b; + int i; + + s->in_fd = -1; + b = buf + s->out_pos_lo; + /* Copy as much of the error message as possible. + * EXCLUDING the terminating \0. */ + for (i = 0; i < errlen && i < s->out_width; i++) + b[i] = errmsg[i]; + /* Any remaining positions: */ + memset(b + i, '_', s->out_width - i); +} + +enum read_status +slot_read(Slot *s, char *buf) +{ + char c; /* Character read. */ + int r; /* Remaining unused positions in buffer range. */ + + for (;;) { + switch (read(s->in_fd, &c, 1)) { + case -1: + khlib_error( + "Failed to read: \"%s\". errno: %d, msg: %s\n", + s->in_fifo, + errno, + strerror(errno) + ); + switch (errno) { + case EINTR: + case EAGAIN: + return RETRY; + default: + return FAILURE; + } + case 0: + khlib_debug("%s: End of FILE\n", s->in_fifo); + s->out_pos_cur = s->out_pos_lo; + return END_OF_FILE; + case 1: + /* TODO: Consider making msg term char a CLI option */ + if (c == '\n' || c == '\0') { + r = (s->out_pos_hi - s->out_pos_cur) + 1; + if (r > 0) + memset(buf + s->out_pos_cur, ' ', r); + return END_OF_MESSAGE; + } else { + if (s->out_pos_cur <= s->out_pos_hi) + buf[s->out_pos_cur++] = c; + else + /* + * Force EOM beyond available range. + * To ensure that a rogue large message + * doesn't trap us here needlessly + * long. + */ + return END_OF_MESSAGE; + } + break; + default: + assert(0); + } + } +} + +void +slots_read(Config *cfg, struct timespec *ti, char *buf) +{ + fd_set fds; + int maxfd = -1; + int ready = 0; + struct stat st; + struct timespec t; + Slot *s; + + FD_ZERO(&fds); + for (s = cfg->slots; s; s = s->next) { + /* TODO: Create the FIFO if it doesn't already exist. */ + if (lstat(s->in_fifo, &st) < 0) { + khlib_error( + "Cannot stat \"%s\". Error: %s\n", + s->in_fifo, + strerror(errno) + ); + slot_set_error(s, buf); + continue; + } + if (!(st.st_mode & S_IFIFO)) { + khlib_error("\"%s\" is not a FIFO\n", s->in_fifo); + slot_set_error(s, buf); + continue; + } + if (s->in_fd < 0) { + khlib_debug( + "%s: closed. opening. in_fd: %d\n", + s->in_fifo, + s->in_fd + ); + s->in_fd = open(s->in_fifo, O_RDONLY | O_NONBLOCK); + } else { + khlib_debug( + "%s: already openned. in_fd: %d\n", + s->in_fifo, + s->in_fd + ); + } + if (s->in_fd == -1) { + /* TODO Consider backing off retries for failed slots */ + khlib_error("Failed to open \"%s\"\n", s->in_fifo); + slot_set_error(s, buf); + continue; + } + khlib_debug("%s: open. in_fd: %d\n", s->in_fifo, s->in_fd); + if (s->in_fd > maxfd) + maxfd = s->in_fd; + FD_SET(s->in_fd, &fds); } + khlib_debug("selecting...\n"); + ready = pselect(maxfd + 1, &fds, NULL, NULL, ti, NULL); + khlib_debug("ready: %d\n", ready); + clock_gettime(CLOCK_MONOTONIC, &t); + if (ready == -1) { + switch (errno) { + case EINTR: + khlib_error( + "pselect interrupted: %d, errno: %d, msg: %s\n", + ready, + errno, + strerror(errno) + ); + /* TODO: Reconsider what to do here. */ + return; + default: + khlib_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 (s = cfg->slots; s; s = s->next) { + if (s->in_fd < 0) + continue; + if (FD_ISSET(s->in_fd, &fds)) { + khlib_debug("reading: %s\n", s->in_fifo); + switch (slot_read(s, 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. + * + * ### LOSSLESS ALTERNATIVES ### + * - Read each pipe until EOF before reading + * another. + * PROBLEM: a fast writer can trap us in the + * read loop. + * + * - Read each pipe until EOM, but close only + * at EOF. + * PROBLEM: a fast writer can fill the pipe + * faster than we can read it and we end-up + * displaying stale data. + * + */ + case END_OF_MESSAGE: + case END_OF_FILE: + case FAILURE: + slot_close(s); + s->in_last_read = t; + ready--; + break; + case RETRY: + break; + default: + assert(0); + } + } else { + slot_expire(s, t, cfg->expiry_character, buf); + } + } + } while (ready); + assert(ready == 0); } void -config_print(Config *cfg) +config_log(Config *cfg) { khlib_info( "Config " @@ -106,15 +418,34 @@ config_print(Config *cfg) " interval = %f," " separator = %s," " slot_count = %d," - " total_width = %d," + " buf_width = %d," " slots = ..." " }\n", cfg->interval, cfg->separator, cfg->slot_count, - cfg->total_width + cfg->buf_width ); - slot_print_all(cfg->slots); + slots_log(cfg->slots); +} + +void +config_stretch_for_separators(Config *cfg) +{ + int seplen = strlen(cfg->separator); + int prefix = 0; + int nslots = 0; + Slot *s = cfg->slots; + + while (s) { + s->out_pos_lo += prefix; + s->out_pos_hi += prefix; + s->out_pos_cur = s->out_pos_lo; + prefix += seplen; + nslots++; + s = s->next; + } + cfg->buf_width += (seplen * (nslots - 1)); } int @@ -159,9 +490,12 @@ print_usage() " | -s SEPARATOR\n" " | -x (* Output to X root window *)\n" " | -l LOG_LEVEL\n" + " | -e EXPIRY_CHARACTER\n" " SEPARATOR = string\n" " INTERVAL = float (* (positive) number of seconds *)\n" " LOG_LEVEL = int (* %d through %d *)\n" + " EXPIRY_CHARACTER = string " + "(* Character with which to fill the slot upon expiration. *)\n" "\n", argv0, Nothing, @@ -224,6 +558,15 @@ parse_opts_opt_l(Config *cfg, int argc, char *argv[], int i) opts_parse_any(cfg, argc, argv, i); } +void +parse_opts_opt_e(Config *cfg, int argc, char *argv[], int i) +{ + if (i >= argc) + usage("Option -e parameter is missing.\n"); + cfg->expiry_character = argv[i++][0]; + opts_parse_any(cfg, argc, argv, i); +} + void parse_opts_opt(Config *cfg, int argc, char *argv[], int i) { @@ -237,13 +580,17 @@ parse_opts_opt(Config *cfg, int argc, char *argv[], int i) parse_opts_opt_s(cfg, argc, argv, ++i); break; case 'x': - cfg->output_to_x_root_window = 1; + cfg->to_x_root = 1; opts_parse_any(cfg, argc, argv, ++i); break; case 'l': /* TODO: Generic set_int */ parse_opts_opt_l(cfg, argc, argv, ++i); break; + case 'e': + /* TODO: Generic set_str */ + parse_opts_opt_e(cfg, argc, argv, ++i); + break; default : usage("Option \"%s\" is invalid\n", argv[i]); } @@ -252,38 +599,44 @@ parse_opts_opt(Config *cfg, int argc, char *argv[], int i) void parse_opts_spec(Config *cfg, int argc, char *argv[], int i) { + char *n; + char *w; + char *t; + struct timespec in_last_read; + Slot *s; + if ((i + 3) > argc) 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 in_last_read; + n = argv[i++]; + w = argv[i++]; + t = argv[i++]; if (!is_pos_num(w)) usage("[spec] Invalid width: \"%s\", for fifo \"%s\"\n", w, n); if (!is_decimal(t)) usage("[spec] Invalid TTL: \"%s\", for fifo \"%s\"\n", t, n); + in_last_read.tv_sec = 0; in_last_read.tv_nsec = 0; - Slot *s = calloc(1, sizeof(struct Slot)); + s = calloc(1, sizeof(struct Slot)); + if (s) { s->in_fifo = n; s->in_fd = -1; - s->out_width = atoi(w); - s->out_ttl = khlib_timespec_of_float(atof(t)); + s->out_width = atoi(w); + s->out_ttl = khlib_timespec_of_float(atof(t)); s->in_last_read = in_last_read; - s->out_pos_lo = cfg->total_width; + s->out_pos_lo = cfg->buf_width; s->out_pos_cur = s->out_pos_lo; - s->out_pos_hi = s->out_pos_lo + s->out_width - 1; - s->next = cfg->slots; + s->out_pos_hi = s->out_pos_lo + s->out_width - 1; + s->next = cfg->slots; cfg->slots = s; - cfg->total_width += s->out_width; + cfg->buf_width += s->out_width; cfg->slot_count++; } else { khlib_fatal("[memory] Allocation failure."); @@ -309,233 +662,15 @@ void opts_parse(Config *cfg, int argc, char *argv[]) { opts_parse_any(cfg, argc, argv, 1); - - Slot *last = cfg->slots; - cfg->slots = NULL; - for (Slot *s = last; s; ) { - Slot *next = s->next; - s->next = cfg->slots; - cfg->slots = s; - s = next; - } -} - -void -slot_expire(Slot *s, struct timespec t, char *buf) -{ - struct timespec td; - - timespecsub(&t, &(s->in_last_read), &td); - if (timespeccmp(&td, &(s->out_ttl), >=)) { - /* TODO: Maybe configurable expiry character. */ - memset(buf + s->out_pos_lo, '_', s->out_pos_hi - s->out_pos_lo); - khlib_warn("Slot expired: \"%s\"\n", s->in_fifo); - } -} - -void -slot_read_error(Slot *s, char *buf) -{ - char *b; - int i; - - b = buf + s->out_pos_lo; - /* Copy as much of the error message as possible. - * EXCLUDING the terminating \0. */ - for (i = 0; i < errlen && i < s->out_width; i++) - b[i] = errmsg[i]; - /* Any remaining slots: */ - for (; i < s->out_width; i++) - b[i] = '_'; -} - -enum read_status -slot_read_one(Slot *s, struct timespec t, char *buf) -{ - char c; /* Character read. */ - int r; /* Remaining unused slots in buffer range. */ - - for (;;) { - switch (read(s->in_fd, &c, 1)) { - case -1: - khlib_error( - "Failed to read: \"%s\". errno: %d, msg: %s\n", - s->in_fifo, - errno, - strerror(errno) - ); - switch (errno) { - case EINTR: - case EAGAIN: - return RETRY; - default: - return FAILURE; - } - case 0: - khlib_debug("%s: End of FILE\n", s->in_fifo); - s->out_pos_cur = s->out_pos_lo; - return END_OF_FILE; - case 1: - /* TODO: Consider making msg term char a CLI option */ - if (c == '\n' || c == '\0') { - r = s->out_pos_hi - s->out_pos_cur; - if (r > 0) - memset(buf + s->out_pos_cur, ' ', r); - s->out_pos_cur = s->out_pos_lo; - s->in_last_read = t; - return END_OF_MESSAGE; - } else { - if (s->out_pos_cur <= s->out_pos_hi) - buf[s->out_pos_cur++] = 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); - } - } + cfg->slots = slots_rev(cfg->slots); + config_log(cfg); + if (cfg->slots == NULL) + usage("No slot specs were given!\n"); } void -slot_read_all(Config *cfg, struct timespec *ti, char *buf) -{ - fd_set fds; - int maxfd = -1; - int ready = 0; - struct stat st; - struct timespec t; - - FD_ZERO(&fds); - for (Slot *s = cfg->slots; s; s = s->next) { - /* TODO: Create the FIFO if it doesn't already exist. */ - if (lstat(s->in_fifo, &st) < 0) { - khlib_error( - "Cannot stat \"%s\". Error: %s\n", - s->in_fifo, - strerror(errno) - ); - slot_read_error(s, buf); - continue; - } - if (!(st.st_mode & S_IFIFO)) { - khlib_error("\"%s\" is not a FIFO\n", s->in_fifo); - slot_read_error(s, buf); - continue; - } - if (s->in_fd < 0) { - khlib_debug( - "%s: closed. opening. in_fd: %d\n", - s->in_fifo, - s->in_fd - ); - s->in_fd = open(s->in_fifo, O_RDONLY | O_NONBLOCK); - } else { - khlib_debug( - "%s: already openned. in_fd: %d\n", - s->in_fifo, - s->in_fd - ); - } - if (s->in_fd == -1) { - /* TODO Consider backing off retries for failed slots */ - khlib_error("Failed to open \"%s\"\n", s->in_fifo); - slot_read_error(s, buf); - continue; - } - khlib_debug("%s: open. in_fd: %d\n", s->in_fifo, s->in_fd); - if (s->in_fd > maxfd) - maxfd = s->in_fd; - FD_SET(s->in_fd, &fds); - } - khlib_debug("selecting...\n"); - ready = pselect(maxfd + 1, &fds, NULL, NULL, ti, NULL); - khlib_debug("ready: %d\n", ready); - clock_gettime(CLOCK_MONOTONIC, &t); - if (ready == -1) { - switch (errno) { - case EINTR: - khlib_error( - "pselect temp failure: %d, errno: %d, msg: %s\n", - ready, - errno, - strerror(errno) - ); - /* TODO: Reconsider what to do here. */ - return; - default: - khlib_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 (Slot *s = cfg->slots; s; s = s->next) { - if (FD_ISSET(s->in_fd, &fds)) { - khlib_debug("reading: %s\n", s->in_fifo); - switch (slot_read_one(s, 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(s->in_fd); - s->in_fd = -1; - ready--; - break; - case RETRY: - break; - default: - assert(0); - } - } else { - slot_expire(s, t, buf); - } - } - } while (ready); - assert(ready == 0); -} - -int -main(int argc, char *argv[]) +loop(Config *cfg, char *buf, Display *d) { - Config cfg = { - .interval = 1.0, - .separator = "|", - .slots = NULL, - .slot_count = 0, - .total_width = 0, - .output_to_x_root_window = 0, - }; - - int width = 0; - int nslots = 0; - int seplen = 0; - int prefix = 0; - int errors = 0; - char *buf; - Display *d = NULL; - struct stat st; struct timespec t0, /* time stamp. before reading slots */ t1, /* time stamp. after reading slots */ @@ -543,78 +678,11 @@ main(int argc, char *argv[]) td, /* time interval measured (t1 - t0) */ tc; /* time interval correction (ti - td) when td < ti */ - argv0 = argv[0]; - - opts_parse(&cfg, argc, argv); - khlib_debug("argv0 = %s\n", argv0); - config_print(&cfg); - - ti = khlib_timespec_of_float(cfg.interval); - - if (cfg.slots == NULL) - usage("No slot specs were given!\n"); - - /* 1st pass to check file existence and type */ - for (Slot *s = cfg.slots; s; s = s->next) { - if (lstat(s->in_fifo, &st) < 0) { - khlib_error( - "Cannot stat \"%s\". Error: %s\n", - s->in_fifo, - strerror(errno) - ); - errors++; - continue; - } - if (!(st.st_mode & S_IFIFO)) { - khlib_error("\"%s\" is not a FIFO\n", s->in_fifo); - errors++; - continue; - } - } - if (errors) - khlib_fatal( - "Encountered errors with given file paths. See log.\n" - ); - - width = cfg.total_width; - seplen = strlen(cfg.separator); - - /* 2nd pass to make space for separators */ - for (Slot *s = cfg.slots; s; s = s->next) { - s->out_pos_lo += prefix; - s->out_pos_hi += prefix; - s->out_pos_cur = s->out_pos_lo; - prefix += seplen; - nslots++; - } - width += (seplen * (nslots - 1)); - buf = calloc(1, width + 1); - if (buf == NULL) - khlib_fatal( - "[memory] Failed to allocate buffer of %d bytes", - width - ); - memset(buf, ' ', width); - buf[width] = '\0'; - /* 3rd pass to set the separators */ - for (Slot *s = cfg.slots; s; s = s->next) { - if (s->out_pos_lo) { /* Skip the first, left-most */ - /* Copying only seplen ensures we omit the '\0' byte. */ - strncpy( - buf + (s->out_pos_lo - seplen), - cfg.separator, - seplen - ); - } - } - - if (cfg.output_to_x_root_window && !(d = XOpenDisplay(NULL))) - khlib_fatal("XOpenDisplay failed with: %p\n", d); - /* TODO: Handle signals */ - for (;;) { + ti = khlib_timespec_of_float(cfg->interval); + while (running) { clock_gettime(CLOCK_MONOTONIC, &t0); // FIXME: check errors - slot_read_all(&cfg, &ti, buf); - if (cfg.output_to_x_root_window) { + slots_read(cfg, &ti, buf); + if (cfg->to_x_root) { if (XStoreName(d, DefaultRootWindow(d), buf) < 0) khlib_fatal("XStoreName failed.\n"); XFlush(d); @@ -630,16 +698,53 @@ main(int argc, char *argv[]) td.tv_nsec ); if (timespeccmp(&td, &ti, <)) { - /* Pushback on data producers by refusing to read the + /* + * Pushback on data producers by refusing to read the * pipe more frequently than the interval. */ timespecsub(&ti, &td, &tc); - khlib_debug("khlib_sleep YES\n"); khlib_sleep(&tc); - } else { - khlib_debug("khlib_sleep NO\n"); } } +} + +void +terminate(int s) +{ + khlib_debug("terminating due to signal %d\n", s); + running = 0; +} +int +main(int argc, char *argv[]) +{ + argv0 = argv[0]; + + Config cfg = { + .interval = 1.0, + .separator = "|", + .expiry_character = '_', + .slots = NULL, + .slot_count = 0, + .buf_width = 0, + .to_x_root = 0, + }; + char *buf; + Display *d = NULL; + struct sigaction sa; + + memset(&sa, 0, sizeof(sa)); + sa.sa_handler = terminate; + sigaction(SIGTERM, &sa, NULL); + sigaction(SIGINT , &sa, NULL); + + opts_parse(&cfg, argc, argv); + slots_assert_fifos_exist(cfg.slots); + config_stretch_for_separators(&cfg); + buf = buf_create(&cfg); + if (cfg.to_x_root && !(d = XOpenDisplay(NULL))) + khlib_fatal("XOpenDisplay failed with: %p\n", d); + loop(&cfg, buf, d); + slots_close(cfg.slots); return EXIT_SUCCESS; }