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