x5/khatus.c

   1 #include <sys/select.h>
   2 #include <sys/stat.h>
   3
   4 #include <assert.h>
   5 #include <ctype.h>
   6 #include <errno.h>
   7 #include <fcntl.h>
   8 #include <signal.h>
   9 #include <stdio.h>
  10 #include <stdlib.h>
  11 #include <string.h>
  12 #include <time.h>
  13 #include <unistd.h>
  14
  15 #include <X11/Xlib.h>
  16
  17 #include "bsdtimespec.h"
  18 #include "khlib_log.h"
  19 #include "khlib_time.h"
  20
  21 #define usage(...) { \
  22         print_usage(); \
  23         fprintf(stderr, "Error:\n    " __VA_ARGS__); \
  24         exit(EXIT_FAILURE); \
  25 }
  26 #define ERRMSG "ERROR"
  27
  28
  29 static const char errmsg[] = ERRMSG;
  30 static const int  errlen   = sizeof(ERRMSG) - 1;
  31
  32 char *argv0 = NULL;
  33 int running = 1;
  34
  35 /* TODO: Convert slot list to slot array. */
  36 typedef struct Slot Slot;
  37 struct Slot {
  38         char            *in_fifo;
  39         int              in_fd;
  40         struct timespec  in_last_read;
  41         struct timespec  out_ttl;
  42         int              out_width;
  43         int              out_pos_lo;   /* Lowest  position on the output buffer. */
  44         int              out_pos_cur;  /* Current position on the output buffer. */
  45         int              out_pos_hi;   /* Highest position on the output buffer. */
  46         Slot            *next;
  47 };
  48
  49 typedef struct Config Config;
  50 struct Config {
  51         double interval;
  52         char * separator;
  53         Slot * slots;
  54         int    slot_count;
  55         int    buf_width;
  56         int    to_x_root;
  57 };
  58
  59 enum read_status {
  60         END_OF_FILE,
  61         END_OF_MESSAGE,
  62         RETRY,
  63         FAILURE
  64 };
  65
  66 char *
  67 buf_create(Config *cfg)
  68 {
  69         int seplen;
  70         char *buf;
  71         Slot *s;
  72
  73         buf = calloc(1, cfg->buf_width + 1);
  74         if (buf == NULL)
  75                 khlib_fatal(
  76                     "[memory] Failed to allocate buffer of %d bytes",
  77                     cfg->buf_width
  78                 );
  79         memset(buf, ' ', cfg->buf_width);
  80         buf[cfg->buf_width] = '\0';
  81         seplen = strlen(cfg->separator);
  82         /* Set the separators */
  83         for (s = cfg->slots; s; s = s->next) {
  84                 /* Skip the first, left-most */
  85                 if (s->out_pos_lo) {
  86                         /* Copying only seplen ensures we omit the '\0' byte. */
  87                         strncpy(
  88                             buf + (s->out_pos_lo - seplen),
  89                             cfg->separator,
  90                             seplen
  91                         );
  92                 }
  93         }
  94         return buf;
  95 }
  96
  97 Slot *
  98 slots_rev(Slot *old)
  99 {
 100         Slot *tmp = NULL;
 101         Slot *new = NULL;
 102
 103         while (old) {
 104                 tmp       = old->next;
 105                 old->next = new;
 106                 new       = old;
 107                 old       = tmp;
 108         }
 109         return new;
 110 }
 111
 112 void
 113 slot_log(Slot *s)
 114 {
 115         khlib_info("Slot "
 116             "{"
 117             " in_fifo = %s,"
 118             " in_fd = %d,"
 119             " out_width = %d,"
 120             " in_last_read = {tv_sec = %ld, tv_nsec = %ld}"
 121             " out_ttl = {tv_sec = %ld, tv_nsec = %ld},"
 122             " out_pos_lo = %d,"
 123             " out_pos_cur = %d,"
 124             " out_pos_hi = %d,"
 125             " next = %p,"
 126             " }\n",
 127             s->in_fifo,
 128             s->in_fd,
 129             s->out_width,
 130             s->in_last_read.tv_sec,
 131             s->in_last_read.tv_nsec,
 132             s->out_ttl.tv_sec,
 133             s->out_ttl.tv_nsec,
 134             s->out_pos_lo,
 135             s->out_pos_cur,
 136             s->out_pos_hi,
 137             s->next
 138         );
 139 }
 140
 141 void
 142 slots_log(Slot *head)
 143 {
 144         Slot *s = head;
 145
 146         for (; s; s = s->next) {
 147                 slot_log(s);
 148         }
 149 }
 150
 151 void
 152 slots_assert_fifos_exist(Slot *s)
 153 {
 154         struct stat st;
 155         int errors = 0;
 156
 157         for (; s; s = s->next) {
 158                 if (lstat(s->in_fifo, &st) < 0) {
 159                         khlib_error(
 160                             "Cannot stat \"%s\". Error: %s\n",
 161                             s->in_fifo,
 162                             strerror(errno)
 163                         );
 164                         errors++;
 165                         continue;
 166                 }
 167                 if (!(st.st_mode & S_IFIFO)) {
 168                         khlib_error("\"%s\" is not a FIFO\n", s->in_fifo);
 169                         errors++;
 170                         continue;
 171                 }
 172         }
 173         if (errors)
 174                 khlib_fatal(
 175                     "Encountered errors with given file paths. See log.\n"
 176                 );
 177 }
 178
 179 void
 180 slot_close(Slot *s)
 181 {
 182         close(s->in_fd);
 183         s->in_fd        = -1;
 184         s->out_pos_cur  = s->out_pos_lo;
 185 }
 186
 187 void
 188 slots_close(Slot *s)
 189 {
 190         for (; s; s = s->next)
 191                 if (s->in_fd > -1)
 192                         slot_close(s);
 193 }
 194
 195
 196 void
 197 slot_expire(Slot *s, struct timespec t, char *buf)
 198 {
 199         struct timespec td;
 200
 201         timespecsub(&t, &(s->in_last_read), &td);
 202         if (timespeccmp(&td, &(s->out_ttl), >=)) {
 203                 /* TODO: Maybe configurable expiry character. */
 204                 memset(buf + s->out_pos_lo, '_', s->out_width);
 205                 khlib_warn("Slot expired: \"%s\"\n", s->in_fifo);
 206         }
 207 }
 208
 209 void
 210 slot_set_error(Slot *s, char *buf)
 211 {
 212         char *b;
 213         int i;
 214
 215         s->in_fd = -1;
 216         b = buf + s->out_pos_lo;
 217         /* Copy as much of the error message as possible.
 218          * EXCLUDING the terminating \0. */
 219         for (i = 0; i < errlen && i < s->out_width; i++)
 220                 b[i] = errmsg[i];
 221         /* Any remaining positions: */
 222         memset(b + i, '_', s->out_width - i);
 223 }
 224
 225 enum read_status
 226 slot_read(Slot *s, char *buf)
 227 {
 228         char c;  /* Character read. */
 229         int  r;  /* Remaining unused positions in buffer range. */
 230
 231         for (;;) {
 232                 switch (read(s->in_fd, &c, 1)) {
 233                 case -1:
 234                         khlib_error(
 235                             "Failed to read: \"%s\". errno: %d, msg: %s\n",
 236                             s->in_fifo,
 237                             errno,
 238                             strerror(errno)
 239                         );
 240                         switch (errno) {
 241                         case EINTR:
 242                         case EAGAIN:
 243                                 return RETRY;
 244                         default:
 245                                 return FAILURE;
 246                         }
 247                 case  0:
 248                         khlib_debug("%s: End of FILE\n", s->in_fifo);
 249                         s->out_pos_cur = s->out_pos_lo;
 250                         return END_OF_FILE;
 251                 case  1:
 252                         /* TODO: Consider making msg term char a CLI option */
 253                         if (c == '\n' || c == '\0') {
 254                                 r = (s->out_pos_hi - s->out_pos_cur) + 1;
 255                                 if (r > 0)
 256                                         memset(buf + s->out_pos_cur, ' ', r);
 257                                 return END_OF_MESSAGE;
 258                         } else {
 259                                 if (s->out_pos_cur <= s->out_pos_hi)
 260                                         buf[s->out_pos_cur++] = c;
 261                                 else
 262                                         /*
 263                                          * Force EOM beyond available range.
 264                                          * To ensure that a rogue large message
 265                                          * doesn't trap us here needlessly
 266                                          * long.
 267                                          */
 268                                         return END_OF_MESSAGE;
 269                         }
 270                         break;
 271                 default:
 272                         assert(0);
 273                 }
 274         }
 275 }
 276
 277 void
 278 slots_read(Config *cfg, struct timespec *ti, char *buf)
 279 {
 280         fd_set fds;
 281         int maxfd = -1;
 282         int ready = 0;
 283         struct stat st;
 284         struct timespec t;
 285         Slot *s;
 286
 287         FD_ZERO(&fds);
 288         for (s = cfg->slots; s; s = s->next) {
 289                 /* TODO: Create the FIFO if it doesn't already exist. */
 290                 if (lstat(s->in_fifo, &st) < 0) {
 291                         khlib_error(
 292                             "Cannot stat \"%s\". Error: %s\n",
 293                             s->in_fifo,
 294                             strerror(errno)
 295                         );
 296                         slot_set_error(s, buf);
 297                         continue;
 298                 }
 299                 if (!(st.st_mode & S_IFIFO)) {
 300                         khlib_error("\"%s\" is not a FIFO\n", s->in_fifo);
 301                         slot_set_error(s, buf);
 302                         continue;
 303                 }
 304                 if (s->in_fd < 0) {
 305                         khlib_debug(
 306                             "%s: closed. opening. in_fd: %d\n",
 307                             s->in_fifo,
 308                             s->in_fd
 309                         );
 310                         s->in_fd = open(s->in_fifo, O_RDONLY | O_NONBLOCK);
 311                 } else {
 312                         khlib_debug(
 313                             "%s: already openned. in_fd: %d\n",
 314                             s->in_fifo,
 315                             s->in_fd
 316                         );
 317                 }
 318                 if (s->in_fd == -1) {
 319                         /* TODO Consider backing off retries for failed slots */
 320                         khlib_error("Failed to open \"%s\"\n", s->in_fifo);
 321                         slot_set_error(s, buf);
 322                         continue;
 323                 }
 324                 khlib_debug("%s: open. in_fd: %d\n", s->in_fifo, s->in_fd);
 325                 if (s->in_fd > maxfd)
 326                         maxfd = s->in_fd;
 327                 FD_SET(s->in_fd, &fds);
 328         }
 329         khlib_debug("selecting...\n");
 330         ready = pselect(maxfd + 1, &fds, NULL, NULL, ti, NULL);
 331         khlib_debug("ready: %d\n", ready);
 332         clock_gettime(CLOCK_MONOTONIC, &t);
 333         if (ready == -1) {
 334                 switch (errno) {
 335                 case EINTR:
 336                         khlib_error(
 337                             "pselect interrupted: %d, errno: %d, msg: %s\n",
 338                             ready,
 339                             errno,
 340                             strerror(errno)
 341                         );
 342                         /* TODO: Reconsider what to do here. */
 343                         return;
 344                 default:
 345                         khlib_fatal(
 346                             "pselect failed: %d, errno: %d, msg: %s\n",
 347                             ready,
 348                             errno,
 349                             strerror(errno)
 350                         );
 351                 }
 352         }
 353         /* At-least-once ensures that expiries are still checked on timeouts. */
 354         do {
 355                 for (s = cfg->slots; s; s = s->next) {
 356                         if (s->in_fd < 0)
 357                                 continue;
 358                         if (FD_ISSET(s->in_fd, &fds)) {
 359                                 khlib_debug("reading: %s\n", s->in_fifo);
 360                                 switch (slot_read(s, buf)) {
 361                                 /*
 362                                  * ### MESSAGE LOSS ###
 363                                  * is introduced by closing at EOM in addition
 364                                  * to EOF, since there may be unread messages
 365                                  * remaining in the pipe. However,
 366                                  *
 367                                  * ### INTER-MESSAGE PUSHBACK ###
 368                                  * is also gained, since pipes block at the
 369                                  * "open" call.
 370                                  *
 371                                  * This is an acceptable trade-off because we
 372                                  * are a stateless reporter of a _most-recent_
 373                                  * status, not a stateful accumulator.
 374                                  *
 375                                  * ### LOSSLESS ALTERNATIVES ###
 376                                  * - Read each pipe until EOF before reading
 377                                  *   another.
 378                                  *   PROBLEM: a fast writer can trap us in the
 379                                  *   read loop.
 380                                  *
 381                                  * - Read each pipe until EOM, but close only
 382                                  *   at EOF.
 383                                  *   PROBLEM: a fast writer can fill the pipe
 384                                  *   faster than we can read it and we end-up
 385                                  *   displaying stale data.
 386                                  *
 387                                  */
 388                                 case END_OF_MESSAGE:
 389                                 case END_OF_FILE:
 390                                 case FAILURE:
 391                                         slot_close(s);
 392                                         s->in_last_read = t;
 393                                         ready--;
 394                                         break;
 395                                 case RETRY:
 396                                         break;
 397                                 default:
 398                                         assert(0);
 399                                 }
 400                         } else {
 401                                 slot_expire(s, t, buf);
 402                         }
 403                 }
 404         } while (ready);
 405         assert(ready == 0);
 406 }
 407
 408 void
 409 config_log(Config *cfg)
 410 {
 411         khlib_info(
 412             "Config "
 413             "{"
 414             " interval = %f,"
 415             " separator = %s,"
 416             " slot_count = %d,"
 417             " buf_width = %d,"
 418             " slots = ..."
 419             " }\n",
 420             cfg->interval,
 421             cfg->separator,
 422             cfg->slot_count,
 423             cfg->buf_width
 424         );
 425         slots_log(cfg->slots);
 426 }
 427
 428 void
 429 config_stretch_for_separators(Config *cfg)
 430 {
 431         int seplen = strlen(cfg->separator);
 432         int prefix = 0;
 433         int nslots = 0;
 434         Slot *s    = cfg->slots;
 435
 436         while (s) {
 437                 s->out_pos_lo  += prefix;
 438                 s->out_pos_hi  += prefix;
 439                 s->out_pos_cur  = s->out_pos_lo;
 440                 prefix         += seplen;
 441                 nslots++;
 442                 s = s->next;
 443         }
 444         cfg->buf_width += (seplen * (nslots - 1));
 445 }
 446
 447 int
 448 is_pos_num(char *str)
 449 {
 450         while (*str != '\0')
 451                 if (!isdigit(*(str++)))
 452                         return 0;
 453         return 1;
 454 }
 455
 456 int
 457 is_decimal(char *str)
 458 {
 459         char c;
 460         int seen = 0;
 461
 462         while ((c = *(str++)) != '\0')
 463                 if (!isdigit(c)) {
 464                         if (c == '.' && !seen++)
 465                                 continue;
 466                         else
 467                                 return 0;
 468                 }
 469         return 1;
 470 }
 471
 472 void
 473 print_usage()
 474 {
 475         assert(argv0);
 476         fprintf(
 477             stderr,
 478             "\n"
 479             "Usage: %s [OPTION ...] SPEC [SPEC ...]\n"
 480             "\n"
 481             "  SPEC       = FILE_PATH DATA_WIDTH DATA_TTL\n"
 482             "  FILE_PATH  = string\n"
 483             "  DATA_WIDTH = int  (* (positive) number of characters *)\n"
 484             "  DATA_TTL   = float  (* (positive) number of seconds *)\n"
 485             "  OPTION     = -i INTERVAL\n"
 486             "             | -s SEPARATOR\n"
 487             "             | -x (* Output to X root window *)\n"
 488             "             | -l LOG_LEVEL\n"
 489             "  SEPARATOR  = string\n"
 490             "  INTERVAL   = float  (* (positive) number of seconds *)\n"
 491             "  LOG_LEVEL  = int  (* %d through %d *)\n"
 492             "\n",
 493             argv0,
 494             Nothing,
 495             Debug
 496         );
 497         fprintf(
 498             stderr,
 499             "Example: %s -i 1 /dev/shm/khatus/khatus_sensor_x 4 10\n"
 500             "\n",
 501             argv0
 502         );
 503 }
 504
 505 /* For mutually-recursive calls. */
 506 void opts_parse_any(Config *, int, char *[], int);
 507
 508 void
 509 parse_opts_opt_i(Config *cfg, int argc, char *argv[], int i)
 510 {
 511         char *param;
 512
 513         if (i >= argc)
 514                 usage("Option -i parameter is missing.\n");
 515         param = argv[i++];
 516         if (!is_decimal(param))
 517                 usage("Option -i parameter is invalid: \"%s\"\n", param);
 518         cfg->interval = atof(param);
 519         opts_parse_any(cfg, argc, argv, i);
 520 }
 521
 522 void
 523 parse_opts_opt_s(Config *cfg, int argc, char *argv[], int i)
 524 {
 525         if (i >= argc)
 526                 usage("Option -s parameter is missing.\n");
 527         cfg->separator = calloc((strlen(argv[i]) + 1), sizeof(char));
 528         strcpy(cfg->separator, argv[i]);
 529         opts_parse_any(cfg, argc, argv, ++i);
 530 }
 531
 532 void
 533 parse_opts_opt_l(Config *cfg, int argc, char *argv[], int i)
 534 {
 535         char *param;
 536         int log_level;
 537
 538         if (i >= argc)
 539                 usage("Option -l parameter is missing.\n");
 540         param = argv[i++];
 541         if (!is_pos_num(param))
 542                 usage("Option -l parameter is invalid: \"%s\"\n", param);
 543         log_level = atoi(param);
 544         if (log_level > Debug)
 545                 usage(
 546                     "Option -l value (%d) exceeds maximum (%d)\n",
 547                     log_level,
 548                     Debug
 549                 );
 550         _khlib_log_level = log_level;
 551         opts_parse_any(cfg, argc, argv, i);
 552 }
 553
 554 void
 555 parse_opts_opt(Config *cfg, int argc, char *argv[], int i)
 556 {
 557         switch (argv[i][1]) {
 558         case 'i':
 559                 /* TODO: Generic set_int */
 560                 parse_opts_opt_i(cfg, argc, argv, ++i);
 561                 break;
 562         case 's':
 563                 /* TODO: Generic set_str */
 564                 parse_opts_opt_s(cfg, argc, argv, ++i);
 565                 break;
 566         case 'x':
 567                 cfg->to_x_root = 1;
 568                 opts_parse_any(cfg, argc, argv, ++i);
 569                 break;
 570         case 'l':
 571                 /* TODO: Generic set_int */
 572                 parse_opts_opt_l(cfg, argc, argv, ++i);
 573                 break;
 574         default :
 575                 usage("Option \"%s\" is invalid\n", argv[i]);
 576         }
 577 }
 578
 579 void
 580 parse_opts_spec(Config *cfg, int argc, char *argv[], int i)
 581 {
 582         char *n;
 583         char *w;
 584         char *t;
 585         struct timespec in_last_read;
 586         Slot *s;
 587
 588         if ((i + 3) > argc)
 589                 usage(
 590                     "[spec] Parameter(s) missing for fifo \"%s\".\n",
 591                     argv[i]
 592                 );
 593
 594         n = argv[i++];
 595         w = argv[i++];
 596         t = argv[i++];
 597
 598         if (!is_pos_num(w))
 599                 usage("[spec] Invalid width: \"%s\", for fifo \"%s\"\n", w, n);
 600         if (!is_decimal(t))
 601                 usage("[spec] Invalid TTL: \"%s\", for fifo \"%s\"\n", t, n);
 602
 603         in_last_read.tv_sec  = 0;
 604         in_last_read.tv_nsec = 0;
 605         s = calloc(1, sizeof(struct Slot));
 606
 607         if (s) {
 608                 s->in_fifo      = n;
 609                 s->in_fd        = -1;
 610                 s->out_width    = atoi(w);
 611                 s->out_ttl      = khlib_timespec_of_float(atof(t));
 612                 s->in_last_read = in_last_read;
 613                 s->out_pos_lo   = cfg->buf_width;
 614                 s->out_pos_cur  = s->out_pos_lo;
 615                 s->out_pos_hi   = s->out_pos_lo + s->out_width - 1;
 616                 s->next         = cfg->slots;
 617
 618                 cfg->slots        = s;
 619                 cfg->buf_width += s->out_width;
 620                 cfg->slot_count++;
 621         } else {
 622                 khlib_fatal("[memory] Allocation failure.");
 623         }
 624         opts_parse_any(cfg, argc, argv, i);
 625 }
 626
 627 void
 628 opts_parse_any(Config *cfg, int argc, char *argv[], int i)
 629 {
 630         if (i < argc) {
 631                 switch (argv[i][0]) {
 632                 case '-':
 633                         parse_opts_opt(cfg, argc, argv, i);
 634                         break;
 635                 default :
 636                         parse_opts_spec(cfg, argc, argv, i);
 637                 }
 638         }
 639 }
 640
 641 void
 642 opts_parse(Config *cfg, int argc, char *argv[])
 643 {
 644         opts_parse_any(cfg, argc, argv, 1);
 645         cfg->slots = slots_rev(cfg->slots);
 646         config_log(cfg);
 647         if (cfg->slots == NULL)
 648                 usage("No slot specs were given!\n");
 649 }
 650
 651 void
 652 loop(Config *cfg, char *buf, Display *d)
 653 {
 654         struct timespec
 655                 t0,  /* time stamp. before reading slots */
 656                 t1,  /* time stamp. after  reading slots */
 657                 ti,  /* time interval desired    (t1 - t0) */
 658                 td,  /* time interval measured   (t1 - t0) */
 659                 tc;  /* time interval correction (ti - td) when td < ti */
 660
 661         ti = khlib_timespec_of_float(cfg->interval);
 662         while (running) {
 663                 clock_gettime(CLOCK_MONOTONIC, &t0); // FIXME: check errors
 664                 slots_read(cfg, &ti, buf);
 665                 if (cfg->to_x_root) {
 666                         if (XStoreName(d, DefaultRootWindow(d), buf) < 0)
 667                                 khlib_fatal("XStoreName failed.\n");
 668                         XFlush(d);
 669                 } else {
 670                         puts(buf);
 671                         fflush(stdout);
 672                 }
 673                 clock_gettime(CLOCK_MONOTONIC, &t1); // FIXME: check errors
 674                 timespecsub(&t1, &t0, &td);
 675                 khlib_debug(
 676                     "td {tv_sec = %ld, tv_nsec = %ld}\n",
 677                     td.tv_sec,
 678                     td.tv_nsec
 679                 );
 680                 if (timespeccmp(&td, &ti, <)) {
 681                         /*
 682                          * Pushback on data producers by refusing to read the
 683                          * pipe more frequently than the interval.
 684                          */
 685                         timespecsub(&ti, &td, &tc);
 686                         khlib_sleep(&tc);
 687                 }
 688         }
 689 }
 690
 691 void
 692 terminate(int s)
 693 {
 694         khlib_debug("terminating due to signal %d\n", s);
 695         running = 0;
 696 }
 697
 698 int
 699 main(int argc, char *argv[])
 700 {
 701         argv0 = argv[0];
 702
 703         Config cfg = {
 704                 .interval    = 1.0,
 705                 .separator   = "|",
 706                 .slots       = NULL,
 707                 .slot_count  = 0,
 708                 .buf_width   = 0,
 709                 .to_x_root   = 0,
 710         };
 711         char *buf;
 712         Display *d = NULL;
 713         struct sigaction sa;
 714
 715         memset(&sa, 0, sizeof(sa));
 716         sa.sa_handler = terminate;
 717         sigaction(SIGTERM, &sa, NULL);
 718         sigaction(SIGINT , &sa, NULL);
 719
 720         opts_parse(&cfg, argc, argv);
 721         slots_assert_fifos_exist(cfg.slots);
 722         config_stretch_for_separators(&cfg);
 723         buf = buf_create(&cfg);
 724         if (cfg.to_x_root && !(d = XOpenDisplay(NULL)))
 725                 khlib_fatal("XOpenDisplay failed with: %p\n", d);
 726         loop(&cfg, buf, d);
 727         slots_close(cfg.slots);
 728         return EXIT_SUCCESS;
 729 }