journal-remote: process events without delay

journal-remote buffers input, and then parses it handling one journal entry at a time.
It was possible for useful data to be left in the buffer after some entries were
processesed. But all data would be already read from the fd, so there would be
no reason for the event loop to call the handler again. After some new data came in,
the handler would be called again, and would then process the "old" data in the buffer.

Fix this by enabling a handler wherever we process input data and do not exhaust data
from the input buffer (i.e. when EAGAIN was not encountered). The handler runs until
we encounter EAGAIN.

Looping over the input data is done in this roundabout way to allow the event loop
to dispatch other events in the meanwhile. If the loop was inside the handler, a
source which produced data fast enough could completely monopolize the process.

https://bugs.freedesktop.org/show_bug.cgi?id=89516

diff --git a/src/journal-remote/journal-remote-parse.c b/src/journal-remote/journal-remote-parse.c
index 6c096de03aa30e110046f805ee3ec2bdc3e84b0b..7e6295435113d6035d48725e97ef93d528cd322e 100644 (file)
--- a/src/journal-remote/journal-remote-parse.c
+++ b/src/journal-remote/journal-remote-parse.c
@@ -41,6 +41,7 @@ void source_free(RemoteSource *source) {
         writer_unref(source->writer);
 
         sd_event_source_unref(source->event);
+        sd_event_source_unref(source->buffer_event);
 
         free(source);
 }

diff --git a/src/journal-remote/journal-remote-parse.h b/src/journal-remote/journal-remote-parse.h
index 22db550913c50c481cc2361ef4a4023f25e230d1..06a50296a184077646ad7332da9d97fd82a621a6 100644 (file)
--- a/src/journal-remote/journal-remote-parse.h
+++ b/src/journal-remote/journal-remote-parse.h
@@ -54,6 +54,7 @@ typedef struct RemoteSource {
         Writer *writer;
 
         sd_event_source *event;
+        sd_event_source *buffer_event;
 } RemoteSource;
 
 RemoteSource* source_new(int fd, bool passive_fd, char *name, Writer *writer);

diff --git a/src/journal-remote/journal-remote.c b/src/journal-remote/journal-remote.c
index e6aa07243e488820990d99127f784d3bae37a47b..1a2c1368a3f18b8308c8d6cbb8cfba09f9f877c8 100644 (file)
--- a/src/journal-remote/journal-remote.c
+++ b/src/journal-remote/journal-remote.c
@@ -285,6 +285,8 @@ static int dispatch_raw_source_event(sd_event_source *event,
                                      int fd,
                                      uint32_t revents,
                                      void *userdata);
+static int dispatch_raw_source_until_block(sd_event_source *event,
+                                           void *userdata);
 static int dispatch_blocking_source_event(sd_event_source *event,
                                           void *userdata);
 static int dispatch_raw_connection_event(sd_event_source *event,
@@ -372,8 +374,15 @@ static int add_source(RemoteServer *s, int fd, char* name, bool own_name) {
 
         r = sd_event_add_io(s->events, &source->event,
                             fd, EPOLLIN|EPOLLRDHUP|EPOLLPRI,
-                            dispatch_raw_source_event, s);
-        if (r == -EPERM) {
+                            dispatch_raw_source_event, source);
+        if (r == 0) {
+                /* Add additional source for buffer processing. It will be
+                 * enabled later. */
+                r = sd_event_add_defer(s->events, &source->buffer_event,
+                                       dispatch_raw_source_until_block, source);
+                if (r == 0)
+                        sd_event_source_set_enabled(source->buffer_event, SD_EVENT_OFF);
+        } else if (r == -EPERM) {
                 log_debug("Falling back to sd_event_add_defer for fd:%d (%s)", fd, name);
                 r = sd_event_add_defer(s->events, &source->event,
                                        dispatch_blocking_source_event, source);
@@ -993,15 +1002,18 @@ static void server_destroy(RemoteServer *s) {
  **********************************************************************
  **********************************************************************/
 
-static int dispatch_raw_source_event(sd_event_source *event,
-                                     int fd,
-                                     uint32_t revents,
-                                     void *userdata) {
+static int handle_raw_source(sd_event_source *event,
+                             int fd,
+                             uint32_t revents,
+                             RemoteServer *s) {
 
-        RemoteServer *s = userdata;
         RemoteSource *source;
         int r;
 
+        /* Returns 1 if there might be more data pending,
+         * 0 if data is currently exhausted, negative on error.
+         */
+
         assert(fd >= 0 && fd < (ssize_t) s->sources_size);
         source = s->sources[fd];
         assert(source->fd == fd);
@@ -1032,11 +1044,48 @@ static int dispatch_raw_source_event(sd_event_source *event,
                 return 1;
 }
 
+static int dispatch_raw_source_until_block(sd_event_source *event,
+                                           void *userdata) {
+        RemoteSource *source = userdata;
+        int r;
+
+        /* Make sure event stays around even if source is destroyed */
+        sd_event_source_ref(event);
+
+        r = handle_raw_source(event, source->fd, EPOLLIN, server);
+        if (r != 1)
+                /* No more data for now */
+                sd_event_source_set_enabled(event, SD_EVENT_OFF);
+
+        sd_event_source_unref(event);
+
+        return r;
+}
+
+static int dispatch_raw_source_event(sd_event_source *event,
+                                     int fd,
+                                     uint32_t revents,
+                                     void *userdata) {
+        RemoteSource *source = userdata;
+        int r;
+
+        assert(source->event);
+        assert(source->buffer_event);
+
+        r = handle_raw_source(event, fd, EPOLLIN, server);
+        if (r == 1)
+                /* Might have more data. We need to rerun the handler
+                 * until we are sure the buffer is exhausted. */
+                sd_event_source_set_enabled(source->buffer_event, SD_EVENT_ON);
+
+        return r;
+}
+
 static int dispatch_blocking_source_event(sd_event_source *event,
                                           void *userdata) {
         RemoteSource *source = userdata;
 
-        return dispatch_raw_source_event(event, source->fd, EPOLLIN, server);
+        return handle_raw_source(event, source->fd, EPOLLIN, server);
 }
 
 static int accept_connection(const char* type, int fd,