Leave TODO to define a max after which to stop reading
[khatus.git] / 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 "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 }
This page took 0.097496 seconds and 4 git commands to generate.