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