From: Siraaj Khandkar <siraaj@khandkar.net>
Date: Tue, 24 Mar 2020 21:12:01 +0000 (-0400)
Subject: Fix queueing-up unhandled messages without pushback
X-Git-Url: https://git.xandkar.net/?p=khatus.git;a=commitdiff_plain;h=6c4ca91c4760f69d559ab7be98846c57daffbe19

Fix queueing-up unhandled messages without pushback

by skipping sleep when there're open FDs (which can be written to during
sleep).

While this doesn't grow the queue, this still prevents pushback by
allowing data to be written before reader can reach EOF.
---

diff --git a/x5/khatus.c b/x5/khatus.c
index fad3994..d1574a5 100644
--- a/x5/khatus.c
+++ b/x5/khatus.c
@@ -368,11 +368,12 @@ fifo_read_one(Fifo *f, char *buf)
 	/* TODO Record timestamp read */
 }
 
-void
+int
 fifo_read_all(Config *cfg, char *buf)
 {
 	fd_set fds;
 	int maxfd = -1;
+	int reading = 0;  /* Number of FDs with unfinished reads */
 	int ready;
 	struct stat st;
 
@@ -414,12 +415,14 @@ fifo_read_all(Config *cfg, char *buf)
 		fatal("%s", strerror(errno));
 	for (Fifo *f = cfg->fifos; f; f = f->next) {
 		if (FD_ISSET(f->fd, &fds)) {
+			reading++;
 			debug("reading: %s\n", f->name);
 			switch (fifo_read_one(f, buf)) {
 			case END_OF_FILE:
 			case FAILURE:
 				close(f->fd);
 				f->fd = -1;
+				reading--;
 				break;
 			case END_OF_MESSAGE:
 			case RETRY:
@@ -429,6 +432,8 @@ fifo_read_all(Config *cfg, char *buf)
 			}
 		}
 	}
+	assert(reading >= 0);
+	return reading;
 }
 
 int
@@ -439,6 +444,7 @@ main(int argc, char *argv[])
 	int seplen = 0;
 	int prefix = 0;
 	int errors = 0;
+	int reading = 0;
 	char *buf;
 	Config cfg0 = defaults;
 	Config *cfg = &cfg0;
@@ -514,7 +520,8 @@ main(int argc, char *argv[])
 		 *       fifo_read_all and desired time of next TTL check.
 		 */
 		/* TODO: How long to wait on IO? Max TTL? */
-		fifo_read_all(cfg, buf);
+		reading = fifo_read_all(cfg, buf);
+		debug("reading: %d\n", reading);
 		if (cfg->output_to_x_root_window) {
 			if (XStoreName(display, DefaultRootWindow(display), buf) < 0)
 				fatal("XStoreName failed.\n");
@@ -526,10 +533,11 @@ main(int argc, char *argv[])
 		clock_gettime(CLOCK_MONOTONIC, &t1); // FIXME: check errors
 		timespecsub(&t1, &t0, &td);
 		debug("td {tv_sec = %ld, tv_nsec = %ld}\n", td.tv_sec, td.tv_nsec);
-		if (timespeccmp(&td, &ti, <)) {
+		if (!reading && timespeccmp(&td, &ti, <)) {
 			/* Pushback on data producers by refusing to read the
 			 * pipe more frequently than the interval.
 			 */
+			/* FIXME: Client is never blocked after initial open. */
 			timespecsub(&ti, &td, &tc);
 			debug("snooze YES\n");
 			snooze(&tc);