From 592820b7a4300cfdc4f85ecd9548f7c2321689fc Mon Sep 17 00:00:00 2001

From: Tomas Bzatek <tbzatek@redhat.com>

Date: Wed, 16 Sep 2020 17:45:07 +0200

Subject: [PATCH 1/5] exec: Fix polling for stdout and stderr

The condition of having both the stdout and the stderr fds ready

may never be satisfied in the case of a full stdout buffer waiting

to be read with no output on stderr yet while both fds being still

open. In such case the old code got stuck in an endless loop and

the spawned process being stuck on writing to stdout/stderr. Let's

read data from any fd once available and only react on EOF/HUP.

This change also makes use of POSIX poll() instead of g_poll()

as it's more clear what the return values mean - Glib docs are

vague in this regard and one can only guess.

---

 src/utils/exec.c | 32 ++++++++++++++++++--------------

 1 file changed, 18 insertions(+), 14 deletions(-)

diff --git a/src/utils/exec.c b/src/utils/exec.c

index 37bd960..ebbcaf2 100644

--- a/src/utils/exec.c

+++ b/src/utils/exec.c

@@ -22,6 +22,7 @@

 #include "extra_arg.h"

 #include <syslog.h>

 #include <stdlib.h>

+#include <poll.h>

 #include <errno.h>

 #include <sys/types.h>

 #include <sys/wait.h>

@@ -293,7 +294,7 @@ static gboolean _utils_exec_and_report_progress (const gchar **argv, const BDExt

     gint poll_status = 0;

     guint i = 0;

     guint8 completion = 0;

-    GPollFD fds[2] = {ZERO_INIT, ZERO_INIT};

+    struct pollfd fds[2] = { ZERO_INIT, ZERO_INIT };

     gboolean out_done = FALSE;

     gboolean err_done = FALSE;

     GString *stdout_data = g_string_new (NULL);

@@ -360,13 +361,16 @@ static gboolean _utils_exec_and_report_progress (const gchar **argv, const BDExt

     fds[0].fd = out_fd;

     fds[1].fd = err_fd;

-    fds[0].events = G_IO_IN | G_IO_HUP | G_IO_ERR;

-    fds[1].events = G_IO_IN | G_IO_HUP | G_IO_ERR;

+    fds[0].events = POLLIN | POLLHUP | POLLERR;

+    fds[1].events = POLLIN | POLLHUP | POLLERR;

     while (!out_done || !err_done) {

-        poll_status = g_poll (fds, 2, -1);

+        poll_status = poll (fds, 2, -1);

+        g_warn_if_fail (poll_status != 0);  /* no timeout specified, zero should never be returned */

         if (poll_status < 0) {

+            if (errno == EAGAIN || errno == EINTR)

+                continue;

             g_set_error (error, BD_UTILS_EXEC_ERROR, BD_UTILS_EXEC_ERROR_FAILED,

-                         "Failed to poll output FDs.");

+                         "Failed to poll output FDs: %m");

             bd_utils_report_finished (progress_id, (*error)->message);

             g_io_channel_shutdown (out_pipe, FALSE, NULL);

             g_io_channel_unref (out_pipe);

@@ -375,12 +379,9 @@ static gboolean _utils_exec_and_report_progress (const gchar **argv, const BDExt

             g_string_free (stdout_data, TRUE);

             g_string_free (stderr_data, TRUE);

             return FALSE;

-        } else if (poll_status != 2)

-            /* both revents fields were not filled yet */

-            continue;

-        if (!(fds[0].revents & G_IO_IN))

-            out_done = TRUE;

-        while (!out_done) {

+        }

+

+        while (!out_done && (fds[0].revents & POLLIN)) {

             io_status = g_io_channel_read_line (out_pipe, &line, NULL, NULL, error);

             if (io_status == G_IO_STATUS_NORMAL) {

                 if (prog_extract && prog_extract (line, &completion))

@@ -401,9 +402,10 @@ static gboolean _utils_exec_and_report_progress (const gchar **argv, const BDExt

                 return FALSE;

             }

         }

-        if (!(fds[1].revents & G_IO_IN))

-            err_done = TRUE;

-        while (!err_done) {

+        if (fds[0].revents & POLLHUP || fds[0].revents & POLLERR || fds[0].revents & POLLNVAL)

+            out_done = TRUE;

+

+        while (!err_done && (fds[1].revents & POLLIN)) {

             io_status = g_io_channel_read_line (err_pipe, &line, NULL, NULL, error);

             if (io_status == G_IO_STATUS_NORMAL) {

                 g_string_append (stderr_data, line);

@@ -421,6 +423,8 @@ static gboolean _utils_exec_and_report_progress (const gchar **argv, const BDExt

                 return FALSE;

             }

         }

+        if (fds[1].revents & POLLHUP || fds[1].revents & POLLERR || fds[1].revents & POLLNVAL)

+            err_done = TRUE;

     }

     child_ret = waitpid (pid, &status, 0);

-- 

2.26.2

From 3025deda9ab670a454bfe373166e49f2acd1c151 Mon Sep 17 00:00:00 2001

From: Tomas Bzatek <tbzatek@redhat.com>

Date: Fri, 25 Sep 2020 18:19:46 +0200

Subject: [PATCH 2/5] exec: Use non-blocking read and process the buffer

 manually

Waiting for data or a newline character on one fd may create a deadlock

while the other fd is being filled with data, exhausting the pipe buffer.

Setting both stdout and stderr fds non-blocking allows us to get indication

of an empty pipe, continuing with the read cycle over remaining watched fds.

This also gets rid of GIOChannel as no extended functionality like GSource

notifications were used, degrading GIOChannel in a simple GObject wrapper

over an fd with just a convenience read_line method that we had to get rid of

anyway. Let's use standard POSIX calls and split the read buffer manually

by matching the newline character. NULL bytes should be handled gracefully

however the stack higher up is not ready for that anyway.

---

 src/utils/exec.c | 277 +++++++++++++++++++++++++++--------------------

 1 file changed, 159 insertions(+), 118 deletions(-)

diff --git a/src/utils/exec.c b/src/utils/exec.c

index ebbcaf2..317fb55 100644

--- a/src/utils/exec.c

+++ b/src/utils/exec.c

@@ -23,6 +23,7 @@

 #include <syslog.h>

 #include <stdlib.h>

 #include <poll.h>

+#include <fcntl.h>

 #include <errno.h>

 #include <sys/types.h>

 #include <sys/wait.h>

@@ -272,6 +273,87 @@ gboolean bd_utils_exec_and_report_status_error (const gchar **argv, const BDExtr

     return TRUE;

 }

+/* buffer size in bytes used to read from stdout and stderr */

+#define _EXEC_BUF_SIZE 64*1024

+

+/* similar to g_strstr_len() yet treats 'null' byte as @needle. */

+static gchar *bd_strchr_len_null (const gchar *haystack, gssize haystack_len, const gchar needle) {

+    gchar *ret;

+    gchar *ret_null;

+

+    ret = memchr (haystack, needle, haystack_len);

+    ret_null = memchr (haystack, 0, haystack_len);

+    if (ret && ret_null)

+        return MIN (ret, ret_null);

+    else

+        return MAX (ret, ret_null);

+}

+

+static gboolean

+_process_fd_event (gint fd, struct pollfd *poll_fd, GString *read_buffer, GString *filtered_buffer, gsize *read_buffer_pos, gboolean *done,

+                   guint64 progress_id, guint8 *progress, BDUtilsProgExtract prog_extract, GError **error) {

+    gchar buf[_EXEC_BUF_SIZE] = { 0 };

+    ssize_t num_read;

+    gchar *line;

+    gchar *newline_pos;

+    int errno_saved;

+    gboolean eof = FALSE;

+

+    if (! *done && (poll_fd->revents & POLLIN)) {

+        /* read until we get EOF (0) or error (-1), expecting EAGAIN */

+        while ((num_read = read (fd, buf, _EXEC_BUF_SIZE)) > 0)

+            g_string_append_len (read_buffer, buf, num_read);

+        errno_saved = errno;

+

+        /* process the fresh data by lines */

+        if (read_buffer->len > *read_buffer_pos) {

+            gchar *buf_ptr;

+            gsize buf_len;

+

+            while ((buf_ptr = read_buffer->str + *read_buffer_pos,

+                    buf_len = read_buffer->len - *read_buffer_pos,

+                    newline_pos = bd_strchr_len_null (buf_ptr, buf_len, '\n'))) {

+                line = g_strndup (buf_ptr, newline_pos - buf_ptr + 1);

+                if (prog_extract && prog_extract (line, progress))

+                    bd_utils_report_progress (progress_id, *progress, NULL);

+                else

+                    g_string_append (filtered_buffer, line);

+                g_free (line);

+                *read_buffer_pos = newline_pos - read_buffer->str + 1;

+            }

+        }

+

+        /* read error */

+        if (num_read < 0 && errno_saved != EAGAIN && errno_saved != EINTR) {

+            g_set_error (error, BD_UTILS_EXEC_ERROR, BD_UTILS_EXEC_ERROR_FAILED,

+                         "Error reading from pipe: %s", g_strerror (errno_saved));

+            return FALSE;

+        }

+

+        /* EOF */

+        if (num_read == 0)

+            eof = TRUE;

+    }

+

+    if (poll_fd->revents & POLLHUP || poll_fd->revents & POLLERR || poll_fd->revents & POLLNVAL)

+        eof = TRUE;

+

+    if (eof) {

+        *done = TRUE;

+        /* process the remaining buffer */

+        line = read_buffer->str + *read_buffer_pos;

+        /* GString guarantees the buffer is always NULL-terminated. */

+        if (strlen (line) > 0) {

+            if (prog_extract && prog_extract (line, progress))

+                bd_utils_report_progress (progress_id, *progress, NULL);

+            else

+                g_string_append (filtered_buffer, line);

+        }

+    }

+

+    return TRUE;

+}

+

 static gboolean _utils_exec_and_report_progress (const gchar **argv, const BDExtraArg **extra, BDUtilsProgExtract prog_extract, gint *proc_status, gchar **stdout, gchar **stderr, GError **error) {

     const gchar **args = NULL;

     guint args_len = 0;

@@ -283,24 +365,26 @@ static gboolean _utils_exec_and_report_progress (const gchar **argv, const BDExt

     gchar *msg = NULL;

     GPid pid = 0;

     gint out_fd = 0;

-    GIOChannel *out_pipe = NULL;

     gint err_fd = 0;

-    GIOChannel *err_pipe = NULL;

-    gchar *line = NULL;

     gint child_ret = -1;

     gint status = 0;

     gboolean ret = FALSE;

-    GIOStatus io_status = G_IO_STATUS_NORMAL;

     gint poll_status = 0;

     guint i = 0;

     guint8 completion = 0;

     struct pollfd fds[2] = { ZERO_INIT, ZERO_INIT };

+    int flags;

     gboolean out_done = FALSE;

     gboolean err_done = FALSE;

-    GString *stdout_data = g_string_new (NULL);

-    GString *stderr_data = g_string_new (NULL);

+    GString *stdout_data;

+    GString *stdout_buffer;

+    GString *stderr_data;

+    GString *stderr_buffer;

+    gsize stdout_buffer_pos = 0;

+    gsize stderr_buffer_pos = 0;

     gchar **old_env = NULL;

     gchar **new_env = NULL;

+    gboolean success = TRUE;

     /* TODO: share this code between functions */

     if (extra) {

@@ -336,15 +420,13 @@ static gboolean _utils_exec_and_report_progress (const gchar **argv, const BDExt

                                     G_SPAWN_DEFAULT|G_SPAWN_SEARCH_PATH|G_SPAWN_DO_NOT_REAP_CHILD,

                                     NULL, NULL, &pid, NULL, &out_fd, &err_fd, error);

+    g_strfreev (new_env);

+

     if (!ret) {

         /* error is already populated */

-        g_string_free (stdout_data, TRUE);

-        g_string_free (stderr_data, TRUE);

-        g_strfreev (new_env);

         g_free (args);

         return FALSE;

     }

-    g_strfreev (new_env);

     args_str = g_strjoinv (" ", args ? (gchar **) args : (gchar **) argv);

     msg = g_strdup_printf ("Started '%s'", args_str);

@@ -353,18 +435,25 @@ static gboolean _utils_exec_and_report_progress (const gchar **argv, const BDExt

     g_free (args);

     g_free (msg);

-    out_pipe = g_io_channel_unix_new (out_fd);

-    err_pipe = g_io_channel_unix_new (err_fd);

+    /* set both fds for non-blocking read */

+    flags = fcntl (out_fd, F_GETFL, 0);

+    if (fcntl (out_fd, F_SETFL, flags | O_NONBLOCK))

+        g_warning ("_utils_exec_and_report_progress: Failed to set out_fd non-blocking: %m");

+    flags = fcntl (err_fd, F_GETFL, 0);

+    if (fcntl (err_fd, F_SETFL, flags | O_NONBLOCK))

+        g_warning ("_utils_exec_and_report_progress: Failed to set err_fd non-blocking: %m");

-    g_io_channel_set_encoding (out_pipe, NULL, NULL);

-    g_io_channel_set_encoding (err_pipe, NULL, NULL);

+    stdout_data = g_string_new (NULL);

+    stdout_buffer = g_string_new (NULL);

+    stderr_data = g_string_new (NULL);

+    stderr_buffer = g_string_new (NULL);

     fds[0].fd = out_fd;

     fds[1].fd = err_fd;

     fds[0].events = POLLIN | POLLHUP | POLLERR;

     fds[1].events = POLLIN | POLLHUP | POLLERR;

-    while (!out_done || !err_done) {

-        poll_status = poll (fds, 2, -1);

+    while (! (out_done && err_done)) {

+        poll_status = poll (fds, 2, -1 /* timeout */);

         g_warn_if_fail (poll_status != 0);  /* no timeout specified, zero should never be returned */

         if (poll_status < 0) {

             if (errno == EAGAIN || errno == EINTR)

@@ -372,140 +461,90 @@ static gboolean _utils_exec_and_report_progress (const gchar **argv, const BDExt

             g_set_error (error, BD_UTILS_EXEC_ERROR, BD_UTILS_EXEC_ERROR_FAILED,

                          "Failed to poll output FDs: %m");

             bd_utils_report_finished (progress_id, (*error)->message);

-            g_io_channel_shutdown (out_pipe, FALSE, NULL);

-            g_io_channel_unref (out_pipe);

-            g_io_channel_shutdown (err_pipe, FALSE, NULL);

-            g_io_channel_unref (err_pipe);

-            g_string_free (stdout_data, TRUE);

-            g_string_free (stderr_data, TRUE);

-            return FALSE;

+            success = FALSE;

+            break;

         }

-        while (!out_done && (fds[0].revents & POLLIN)) {

-            io_status = g_io_channel_read_line (out_pipe, &line, NULL, NULL, error);

-            if (io_status == G_IO_STATUS_NORMAL) {

-                if (prog_extract && prog_extract (line, &completion))

-                    bd_utils_report_progress (progress_id, completion, NULL);

-                else

-                    g_string_append (stdout_data, line);

-                g_free (line);

-            } else if (io_status == G_IO_STATUS_EOF) {

-                out_done = TRUE;

-            } else if (error && (*error)) {

+        if (!out_done) {

+            if (! _process_fd_event (out_fd, &fds[0], stdout_buffer, stdout_data, &stdout_buffer_pos, &out_done, progress_id, &completion, prog_extract, error)) {

                 bd_utils_report_finished (progress_id, (*error)->message);

-                g_io_channel_shutdown (out_pipe, FALSE, NULL);

-                g_io_channel_unref (out_pipe);

-                g_io_channel_shutdown (err_pipe, FALSE, NULL);

-                g_io_channel_unref (err_pipe);

-                g_string_free (stdout_data, TRUE);

-                g_string_free (stderr_data, TRUE);

-                return FALSE;

+                success = FALSE;

+                break;

             }

         }

-        if (fds[0].revents & POLLHUP || fds[0].revents & POLLERR || fds[0].revents & POLLNVAL)

-            out_done = TRUE;

-        while (!err_done && (fds[1].revents & POLLIN)) {

-            io_status = g_io_channel_read_line (err_pipe, &line, NULL, NULL, error);

-            if (io_status == G_IO_STATUS_NORMAL) {

-                g_string_append (stderr_data, line);

-                g_free (line);

-            } else if (io_status == G_IO_STATUS_EOF) {

-                err_done = TRUE;

-            } else if (error && (*error)) {

+        if (!err_done) {

+            if (! _process_fd_event (err_fd, &fds[1], stderr_buffer, stderr_data, &stderr_buffer_pos, &err_done, progress_id, &completion, prog_extract, error)) {

                 bd_utils_report_finished (progress_id, (*error)->message);

-                g_io_channel_shutdown (out_pipe, FALSE, NULL);

-                g_io_channel_unref (out_pipe);

-                g_io_channel_shutdown (err_pipe, FALSE, NULL);

-                g_io_channel_unref (err_pipe);

-                g_string_free (stdout_data, TRUE);

-                g_string_free (stderr_data, TRUE);

-                return FALSE;

+                success = FALSE;

+                break;

             }

         }

-        if (fds[1].revents & POLLHUP || fds[1].revents & POLLERR || fds[1].revents & POLLNVAL)

-            err_done = TRUE;

     }

+    g_string_free (stdout_buffer, TRUE);

+    g_string_free (stderr_buffer, TRUE);

+    close (out_fd);

+    close (err_fd);

+

     child_ret = waitpid (pid, &status, 0);

-    *proc_status = WEXITSTATUS(status);

-    if (child_ret > 0) {

-        if (*proc_status != 0) {

-            if (stderr_data->str && (g_strcmp0 ("", stderr_data->str) != 0))

-                msg = stderr_data->str;

-            else

-                msg = stdout_data->str;

-            g_set_error (error, BD_UTILS_EXEC_ERROR, BD_UTILS_EXEC_ERROR_FAILED,

-                         "Process reported exit code %d: %s", *proc_status, msg);

-            bd_utils_report_finished (progress_id, (*error)->message);

-            g_io_channel_shutdown (out_pipe, FALSE, NULL);

-            g_io_channel_unref (out_pipe);

-            g_io_channel_shutdown (err_pipe, FALSE, NULL);

-            g_io_channel_unref (err_pipe);

-            g_string_free (stdout_data, TRUE);

-            g_string_free (stderr_data, TRUE);

-            return FALSE;

-        }

-        if (WIFSIGNALED(status)) {

-            g_set_error (error, BD_UTILS_EXEC_ERROR, BD_UTILS_EXEC_ERROR_FAILED,

-                         "Process killed with a signal");

-            bd_utils_report_finished (progress_id, (*error)->message);

-            g_io_channel_shutdown (out_pipe, FALSE, NULL);

-            g_io_channel_unref (out_pipe);

-            g_io_channel_shutdown (err_pipe, FALSE, NULL);

-            g_io_channel_unref (err_pipe);

-            g_string_free (stdout_data, TRUE);

-            g_string_free (stderr_data, TRUE);

-            return FALSE;

-        }

-    } else if (child_ret == -1) {

-        if (errno != ECHILD) {

-            errno = 0;

-            g_set_error (error, BD_UTILS_EXEC_ERROR, BD_UTILS_EXEC_ERROR_FAILED,

-                         "Failed to wait for the process");

-            bd_utils_report_finished (progress_id, (*error)->message);

-            g_io_channel_shutdown (out_pipe, FALSE, NULL);

-            g_io_channel_unref (out_pipe);

-            g_io_channel_shutdown (err_pipe, FALSE, NULL);

-            g_io_channel_unref (err_pipe);

-            g_string_free (stdout_data, TRUE);

-            g_string_free (stderr_data, TRUE);

-            return FALSE;

+    *proc_status = WEXITSTATUS (status);

+    if (success) {

+        if (child_ret > 0) {

+            if (*proc_status != 0) {

+                msg = stderr_data->len > 0 ? stderr_data->str : stdout_data->str;

+                g_set_error (error, BD_UTILS_EXEC_ERROR, BD_UTILS_EXEC_ERROR_FAILED,

+                             "Process reported exit code %d: %s", *proc_status, msg);

+                bd_utils_report_finished (progress_id, (*error)->message);

+                success = FALSE;

+            } else if (WIFSIGNALED (status)) {

+                g_set_error (error, BD_UTILS_EXEC_ERROR, BD_UTILS_EXEC_ERROR_FAILED,

+                             "Process killed with a signal");

+                bd_utils_report_finished (progress_id, (*error)->message);

+                success = FALSE;

+            }

+        } else if (child_ret == -1) {

+            if (errno != ECHILD) {

+                errno = 0;

+                g_set_error (error, BD_UTILS_EXEC_ERROR, BD_UTILS_EXEC_ERROR_FAILED,

+                             "Failed to wait for the process");

+                bd_utils_report_finished (progress_id, (*error)->message);

+                success = FALSE;

+            } else {

+                /* no such process (the child exited before we tried to wait for it) */

+                errno = 0;

+            }

         }

-        /* no such process (the child exited before we tried to wait for it) */

-        errno = 0;

+        if (success)

+            bd_utils_report_finished (progress_id, "Completed");

     }

-

-    bd_utils_report_finished (progress_id, "Completed");

     log_out (task_id, stdout_data->str, stderr_data->str);

     log_done (task_id, *proc_status);

-    /* we don't care about the status here */

-    g_io_channel_shutdown (out_pipe, FALSE, error);

-    g_io_channel_unref (out_pipe);

-    g_io_channel_shutdown (err_pipe, FALSE, error);

-    g_io_channel_unref (err_pipe);

-

-    if (stdout)

+    if (success && stdout)

         *stdout = g_string_free (stdout_data, FALSE);

     else

         g_string_free (stdout_data, TRUE);

-    if (stderr)

+    if (success && stderr)

         *stderr = g_string_free (stderr_data, FALSE);

     else

         g_string_free (stderr_data, TRUE);

-    return TRUE;

+    return success;

 }

 /**

  * bd_utils_exec_and_report_progress:

  * @argv: (array zero-terminated=1): the argv array for the call

  * @extra: (allow-none) (array zero-terminated=1): extra arguments

- * @prog_extract: (scope notified): function for extracting progress information

+ * @prog_extract: (scope notified) (nullable): function for extracting progress information

  * @proc_status: (out): place to store the process exit status

  * @error: (out): place to store error (if any)

  *

+ * Note that any NULL bytes read from standard output and standard error

+ * output are treated as separators similar to newlines and @prog_extract

+ * will be called with the respective chunk.

+ *

  * Returns: whether the @argv was successfully executed (no error and exit code 0) or not

  */

 gboolean bd_utils_exec_and_report_progress (const gchar **argv, const BDExtraArg **extra, BDUtilsProgExtract prog_extract, gint *proc_status, GError **error) {

@@ -519,6 +558,9 @@ gboolean bd_utils_exec_and_report_progress (const gchar **argv, const BDExtraArg

  * @output: (out): variable to store output to

  * @error: (out): place to store error (if any)

  *

+ * Note that any NULL bytes read from standard output and standard error

+ * output will be discarded.

+ *

  * Returns: whether the @argv was successfully executed capturing the output or not

  */

 gboolean bd_utils_exec_and_capture_output (const gchar **argv, const BDExtraArg **extra, gchar **output, GError **error) {

@@ -549,7 +591,6 @@ gboolean bd_utils_exec_and_capture_output (const gchar **argv, const BDExtraArg

         g_free (stderr);

         return TRUE;

     }

-

 }

 /**

-- 

2.26.2

From f5eb61c91ffc6b0d1fd709076a9579105655ff17 Mon Sep 17 00:00:00 2001

From: Tomas Bzatek <tbzatek@redhat.com>

Date: Fri, 25 Sep 2020 18:27:02 +0200

Subject: [PATCH 3/5] exec: Clarify the BDUtilsProgExtract callback

 documentation

---

 src/utils/exec.h | 24 ++++++++++++++++++++++--

 1 file changed, 22 insertions(+), 2 deletions(-)

diff --git a/src/utils/exec.h b/src/utils/exec.h

index ad169e4..0e262a2 100644

--- a/src/utils/exec.h

+++ b/src/utils/exec.h

@@ -31,10 +31,30 @@ typedef void (*BDUtilsProgFunc) (guint64 task_id, BDUtilsProgStatus status, guin

 /**

  * BDUtilsProgExtract:

- * @line: line from extract progress from

+ * @line: line to extract progress from

  * @completion: (out): percentage of completion

  *

- * Returns: whether the line was a progress reporting line or not

+ * Callback function used to process a line captured from spawned command's standard

+ * output and standard error output. Typically used to extract completion percentage

+ * of a long-running job.

+ *

+ * Note that both outputs are read simultaneously with no guarantees of message order

+ * this function is called with.

+ *

+ * The value the @completion points to may contain value previously returned from

+ * this callback or zero when called for the first time. This is useful for extractors

+ * where only some kind of a tick mark is printed out as a progress and previous value

+ * is needed to compute an incremented value. It's important to keep in mind that this

+ * function is only called over lines, i.e. progress reporting printing out tick marks

+ * (e.g. dots) without a newline character might not work properly.

+ *

+ * The @line string usually contains trailing newline character, which may be absent

+ * however in case the spawned command exits without printing one. It's guaranteed

+ * this function is called over remaining buffer no matter what the trailing

+ * character is.

+ *

+ * Returns: whether the line was a progress reporting line and should be excluded

+ *          from the collected standard output string or not.

  */

 typedef gboolean (*BDUtilsProgExtract) (const gchar *line, guint8 *completion);

-- 

2.26.2

From 8a7f0de5f63099a3e8bcaca005c4de04df959113 Mon Sep 17 00:00:00 2001

From: Tomas Bzatek <tbzatek@redhat.com>

Date: Fri, 25 Sep 2020 18:27:41 +0200

Subject: [PATCH 4/5] tests: Add bufferbloat exec tests

This should test pipe buffer exhaustion as well as potential pipe

read starvation while filling the other fd.

---

 tests/utils_test.py | 105 +++++++++++++++++++++++++++++++++++++++++++-

 1 file changed, 104 insertions(+), 1 deletion(-)

diff --git a/tests/utils_test.py b/tests/utils_test.py

index 2bec5ed..ed13611 100644

--- a/tests/utils_test.py

+++ b/tests/utils_test.py

@@ -1,8 +1,9 @@

 import unittest

 import re

 import os

+import six

 import overrides_hack

-from utils import fake_utils, create_sparse_tempfile, create_lio_device, delete_lio_device, run_command, TestTags, tag_test

+from utils import fake_utils, create_sparse_tempfile, create_lio_device, delete_lio_device, run_command, TestTags, tag_test, read_file

 from gi.repository import BlockDev, GLib

@@ -65,6 +66,9 @@ class UtilsExecLoggingTest(UtilsTestCase):

         succ = BlockDev.utils_exec_and_report_error(["true"])

         self.assertTrue(succ)

+        with six.assertRaisesRegex(self, GLib.GError, r"Process reported exit code 1"):

+            succ = BlockDev.utils_exec_and_report_error(["/bin/false"])

+

         succ, out = BlockDev.utils_exec_and_capture_output(["echo", "hi"])

         self.assertTrue(succ)

         self.assertEqual(out, "hi\n")

@@ -178,6 +182,105 @@ class UtilsExecLoggingTest(UtilsTestCase):

         self.assertTrue(succ)

         self.assertIn("LC_ALL=C", out)

+    @tag_test(TestTags.NOSTORAGE, TestTags.CORE)

+    def test_exec_buffer_bloat(self):

+        """Verify that very large output from a command is handled properly"""

+

+        # easy 64kB of data

+        cnt = 65536

+        succ, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", "for i in {1..%d}; do echo -n .; done" % cnt])

+        self.assertTrue(succ)

+        self.assertEquals(len(out), cnt)

+

+        succ, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", "for i in {1..%d}; do echo -n .; echo -n \# >&2; done" % cnt])

+        self.assertTrue(succ)

+        self.assertEquals(len(out), cnt)

+

+        # now exceed the system pipe buffer size

+        # pipe(7): The maximum size (in bytes) of individual pipes that can be set by users without the CAP_SYS_RESOURCE capability.

+        cnt = int(read_file("/proc/sys/fs/pipe-max-size")) + 100

+        self.assertGreater(cnt, 0)

+

+        succ, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", "for i in {1..%d}; do echo -n .; done" % cnt])

+        self.assertTrue(succ)

+        self.assertEquals(len(out), cnt)

+

+        succ, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", "for i in {1..%d}; do echo -n .; echo -n \# >&2; done" % cnt])

+        self.assertTrue(succ)

+        self.assertEquals(len(out), cnt)

+

+        # make use of some newlines

+        succ, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", "for i in {1..%d}; do echo -n .; if [ $(($i%%500)) -eq 0 ]; then echo ''; fi; done" % cnt])

+        self.assertTrue(succ)

+        self.assertGreater(len(out), cnt)

+

+        succ, out = BlockDev.utils_exec_and_capture_output(["bash", "-c", "for i in {1..%d}; do echo -n .; echo -n \# >&2; if [ $(($i%%500)) -eq 0 ]; then echo ''; echo '' >&2; fi; done" % cnt])

+        self.assertTrue(succ)

+        self.assertGreater(len(out), cnt)

+

+

+    EXEC_PROGRESS_MSG = "Aloha, I'm the progress line you should match."

+

+    def my_exec_progress_func_concat(self, line):

+        """Expect an concatenated string"""

+        s = ""

+        for i in range(10):

+            s += self.EXEC_PROGRESS_MSG

+        self.assertEqual(line, s)

+        self.num_matches += 1

+        return 0

+

+    def my_exec_progress_func(self, line):

+        self.assertTrue(re.match(r".*%s.*" % self.EXEC_PROGRESS_MSG, line))

+        self.num_matches += 1

+        return 0

+

+    def test_exec_buffer_bloat_progress(self):

+        """Verify that progress reporting can handle large output"""

+

+        # no newlines, should match just a single occurrence on EOF

+        cnt = 10

+        self.num_matches = 0

+        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..10}; do echo -n \"%s\"; done" % self.EXEC_PROGRESS_MSG], None, self.my_exec_progress_func_concat)

+        self.assertTrue(status)

+        self.assertEqual(self.num_matches, 1)

+

+        # ten matches

+        self.num_matches = 0

+        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo \"%s\"; done" % (cnt, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)

+        self.assertTrue(status)

+        self.assertEqual(self.num_matches, cnt)

+

+        # 100k matches

+        cnt = 100000

+        self.num_matches = 0

+        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo \"%s\"; done" % (cnt, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)

+        self.assertTrue(status)

+        self.assertEqual(self.num_matches, cnt)

+

+        # 100k matches on stderr

+        self.num_matches = 0

+        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo \"%s\" >&2; done" % (cnt, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)

+        self.assertTrue(status)

+        self.assertEqual(self.num_matches, cnt)

+

+        # 100k matches on stderr and stdout

+        self.num_matches = 0

+        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo \"%s\"; echo \"%s\" >&2; done" % (cnt, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)

+        self.assertTrue(status)

+        self.assertEqual(self.num_matches, cnt * 2)

+

+        # stdout and stderr output, non-zero return from the command and very long exception message

+        self.num_matches = 0

+        with six.assertRaisesRegex(self, GLib.GError, r"Process reported exit code 66"):

+            status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo \"%s\"; echo \"%s\" >&2; done; exit 66" % (cnt, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG)], None, self.my_exec_progress_func)

+        self.assertEqual(self.num_matches, cnt * 2)

+

+        # no progress reporting callback given, tests slightly different code path

+        status = BlockDev.utils_exec_and_report_progress(["bash", "-c", "for i in {1..%d}; do echo \"%s\"; echo \"%s\" >&2; done" % (cnt, self.EXEC_PROGRESS_MSG, self.EXEC_PROGRESS_MSG)], None, None)

+        self.assertTrue(status)

+

+

 class UtilsDevUtilsTestCase(UtilsTestCase):

     @tag_test(TestTags.NOSTORAGE, TestTags.CORE)

     def test_resolve_device(self):

-- 

2.26.2