libsubprocess/test: cover output buffer overflow

Problem: There are no unit tests for when the output buffer is
filled to the max.

Add unit tests.

src/common/libsubprocess/test/iostress.c

@@ -76,6 +76,12 @@ static void iostress_output_cb (flux_subprocess_t *p, const char *stream)
     ctx->outputcount++;
 }
 
+static void iostress_output_noread_cb (flux_subprocess_t *p, const char *stream)
+{
+    struct iostress_ctx *ctx = flux_subprocess_aux_get (p, "ctx");
+    ctx->outputcount++;
+}
+
 static void iostress_completion_cb (flux_subprocess_t *p)
 {
     struct iostress_ctx *ctx = flux_subprocess_aux_get (p, "ctx");
@@ -294,11 +300,18 @@ int main (int argc, char *argv[])
     ok (iostress_run_check (h, "balanced", false, 0, 0, 8, 8, 80, NULL),
         "balanced worked");
 
-    // (remote?) stdout buffer is overrun
-    // Needs further investigation as no errors are thrown and completion is
-    // not called called after subprocess exit.  The doomsday timer stops
-    // the test.
-    ok (!iostress_run_check (h, "tinystdout", false, 0, 128, 1, 1, 256, NULL),
+    // stdout buffer is overrun
+
+    // libsubprocess will attempt to get the user to read from the buffer that
+    // is overrun.  So generally speaking, stdout buffer overrun should still
+    // work.
+    ok (iostress_run_check (h, "tinystdout", false, 0, 128, 1, 1, 256, NULL),
+        "tinystdout works");
+
+    // if the user does not read from the output buffer when it is full, this
+    // should lead to an error.
+    ok (!iostress_run_check (h, "tinystdout", false, 0, 128, 1, 1, 256,
+                             iostress_output_noread_cb),
         "tinystdout failed as expected");
 
     // local stdin buffer is overrun (immediately)

src/common/libsubprocess/test/stdio.c

@@ -1336,6 +1336,126 @@ void test_stream_start_stop_mid_stop (flux_reactor_t *r)
     flux_watcher_destroy (tw);
 }
 
+void overflow_output_cb (flux_subprocess_t *p, const char *stream)
+{
+    const char *buf = NULL;
+    int len;
+
+    if (strcasecmp (stream, "stdout") != 0) {
+        ok (false, "unexpected stream %s", stream);
+        return;
+    }
+
+    /* first callback should return "0123" for 4 byte buffer.
+     * second callback should return "456\n" in 4 byte buffer
+     */
+    if (stdout_output_cb_count == 0) {
+        len = flux_subprocess_read_line (p, stream, &buf);
+        ok (len > 0
+            && buf != NULL,
+            "flux_subprocess_read_line on %s success", stream);
+
+        ok (streq (buf, "0123"),
+            "flux_subprocess_read_line returned correct data");
+        ok (len == 4,
+            "flux_subprocess_read_line returned correct data len");
+    }
+    else if (stdout_output_cb_count == 1) {
+        len = flux_subprocess_read_line (p, stream, &buf);
+        ok (len > 0
+            && buf != NULL,
+            "flux_subprocess_read_line on %s success", stream);
+
+        ok (streq (buf, "456\n"),
+            "flux_subprocess_read_line returned correct data");
+        ok (len == 4,
+            "flux_subprocess_read_line returned correct data len");
+    }
+    else {
+        ok (flux_subprocess_read_stream_closed (p, stream),
+            "flux_subprocess_read_stream_closed saw EOF on %s", stream);
+
+        len = flux_subprocess_read (p, stream, &buf);
+        ok (len == 0,
+            "flux_subprocess_read on %s read EOF", stream);
+    }
+    stdout_output_cb_count++;
+}
+
+/* Set buffer size to 4 and have 7 bytes of output (8 including newline) */
+void test_overflow_output_buffer (flux_reactor_t *r)
+{
+    char *av[] = { TEST_SUBPROCESS_DIR "test_echo", "-O", "0123456", NULL };
+    flux_cmd_t *cmd;
+    flux_subprocess_t *p = NULL;
+
+    ok ((cmd = flux_cmd_create (3, av, environ)) != NULL, "flux_cmd_create");
+
+    ok (flux_cmd_setopt (cmd, "stdout_BUFSIZE", "4") == 0,
+        "flux_cmd_setopt set stdout_BUFSIZE success");
+
+    flux_subprocess_ops_t ops = {
+        .on_completion = completion_cb,
+        .on_stdout = overflow_output_cb
+    };
+    completion_cb_count = 0;
+    stdout_output_cb_count = 0;
+    p = flux_local_exec (r, 0, cmd, &ops);
+    ok (p != NULL, "flux_local_exec");
+
+    ok (flux_subprocess_state (p) == FLUX_SUBPROCESS_RUNNING,
+        "subprocess state == RUNNING after flux_local_exec");
+
+    int rc = flux_reactor_run (r, 0);
+    ok (rc == 0, "flux_reactor_run returned zero status");
+    ok (completion_cb_count == 1, "completion callback called 1 time");
+    ok (stdout_output_cb_count == 3, "stdout output callback called 3 times");
+    flux_subprocess_destroy (p);
+    flux_cmd_destroy (cmd);
+}
+
+void overflow_output_noread_cb (flux_subprocess_t *p, const char *stream)
+{
+    if (strcasecmp (stream, "stdout") != 0) {
+        ok (false, "unexpected stream %s", stream);
+        return;
+    }
+    /* do not read from the buffer, we should fail b/c it overflows */
+    stdout_output_cb_count++;
+}
+
+/* Set buffer size to 4 and have 7 bytes of output (8 including newline) */
+void test_overflow_output_buffer_noread (flux_reactor_t *r)
+{
+    char *av[] = { TEST_SUBPROCESS_DIR "test_echo", "-O", "0123456", NULL };
+    flux_cmd_t *cmd;
+    flux_subprocess_t *p = NULL;
+
+    ok ((cmd = flux_cmd_create (3, av, environ)) != NULL, "flux_cmd_create");
+
+    ok (flux_cmd_setopt (cmd, "stdout_BUFSIZE", "4") == 0,
+        "flux_cmd_setopt set stdout_BUFSIZE success");
+
+    flux_subprocess_ops_t ops = {
+        .on_completion = completion_cb,
+        .on_stdout = overflow_output_noread_cb
+    };
+    completion_cb_count = 0;
+    stdout_output_cb_count = 0;
+    p = flux_local_exec (r, 0, cmd, &ops);
+    ok (p != NULL, "flux_local_exec");
+
+    ok (flux_subprocess_state (p) == FLUX_SUBPROCESS_RUNNING,
+        "subprocess state == RUNNING after flux_local_exec");
+
+    int rc = flux_reactor_run (r, 0);
+    ok (rc == 0, "flux_reactor_run returned zero status");
+    ok (completion_cb_count == 1, "completion callback called 1 time");
+    ok (stdout_output_cb_count == 3, "stdout output callback called 3 times");
+    flux_subprocess_destroy (p);
+    flux_cmd_destroy (cmd);
+}
+
 int main (int argc, char *argv[])
 {
     flux_reactor_t *r;
@@ -1395,6 +1515,10 @@ int main (int argc, char *argv[])
     test_stream_start_stop_initial_stop (r);
     diag ("stream_start_stop_mid_stop");
     test_stream_start_stop_mid_stop (r);
+    diag ("overflow_output_buffer");
+    test_overflow_output_buffer (r);
+    diag ("overflow_output_buffer_noread");
+    test_overflow_output_buffer_noread (r);
 
     end_fdcount = fdcount ();