builtin_eval to direct output to its iostreams

Prior to this fix, builtin_eval would direct output to the io_chain of the
job. The problem is with pipes: `builtin_eval` might happily attempt to
write unlimited output to the write end of a pipe, but the corresponding
reading process has not yet been launched. This results in deadlock.

The fix is to buffer all the output from `builtin_eval`. This is not fun
but the best that can be done until we have real concurrent processes.

cherry-pick of a1f1b9c2d9

Fixes #6806

src/builtin_eval.cpp

@@ -17,6 +17,9 @@
 /// Implementation of eval builtin.
 int builtin_eval(parser_t &parser, io_streams_t &streams, wchar_t **argv) {
     int argc = builtin_count_args(argv);
+    if (argc <= 1) {
+        return STATUS_CMD_OK;
+    }
 
     wcstring new_cmd;
     for (int i = 1; i < argc; ++i) {
@@ -24,20 +27,42 @@ int builtin_eval(parser_t &parser, io_streams_t &streams, wchar_t **argv) {
         new_cmd += argv[i];
     }
 
+    // The output and errput of eval must go to our streams, not to the io_chain in our streams,
+    // because they may contain a pipe which is intended to be consumed by a process which is not
+    // yet launched (#6806). Make bufferfills to capture it.
+    // TODO: we are sloppy and do not honor other redirections; eval'd code won't see for example a
+    // redirection of fd 3.
+    shared_ptr<io_bufferfill_t> stdout_fill =
+        io_bufferfill_t::create(fd_set_t{}, parser.libdata().read_limit, STDOUT_FILENO);
+    shared_ptr<io_bufferfill_t> stderr_fill =
+        io_bufferfill_t::create(fd_set_t{}, parser.libdata().read_limit, STDERR_FILENO);
+    if (!stdout_fill || !stderr_fill) {
+        // This can happen if we are unable to create a pipe.
+        return STATUS_CMD_ERROR;
+    }
+
     const auto cached_exec_count = parser.libdata().exec_count;
     int status = STATUS_CMD_OK;
-    if (argc > 1) {
-        if (parser.eval(new_cmd, *streams.io_chain, block_type_t::top, streams.parent_pgid) !=
-            eval_result_t::ok) {
-            status = STATUS_CMD_ERROR;
-        } else if (cached_exec_count == parser.libdata().exec_count) {
-            // Issue #5692, in particular, to catch `eval ""`, `eval "begin; end;"`, etc.
-            // where we have an argument but nothing is executed.
-            status = STATUS_CMD_OK;
-        } else {
-            status = parser.get_last_status();
-        }
+    if (parser.eval(new_cmd, io_chain_t{stdout_fill, stderr_fill}, block_type_t::top,
+                    streams.parent_pgid) != eval_result_t::ok) {
+        status = STATUS_CMD_ERROR;
+    } else if (cached_exec_count == parser.libdata().exec_count) {
+        // Issue #5692, in particular, to catch `eval ""`, `eval "begin; end;"`, etc.
+        // where we have an argument but nothing is executed.
+        status = STATUS_CMD_OK;
+    } else {
+        status = parser.get_last_status();
     }
 
+    // Finish the bufferfills - exhaust and close our pipes.
+    // Note it is important that we hold no other references to the bufferfills here - they need to
+    // deallocate to close.
+    std::shared_ptr<io_buffer_t> output = io_bufferfill_t::finish(std::move(stdout_fill));
+    std::shared_ptr<io_buffer_t> errput = io_bufferfill_t::finish(std::move(stderr_fill));
+
+    // Copy the output from the bufferfill back to the streams.
+    streams.out.append_narrow_buffer(output->buffer());
+    streams.err.append_narrow_buffer(errput->buffer());
+
     return status;
 }

src/io.cpp

@@ -174,8 +174,10 @@ void io_buffer_t::complete_background_fillthread() {
     fillthread_waiter_ = {};
 }
 
-shared_ptr<io_bufferfill_t> io_bufferfill_t::create(const fd_set_t &conflicts,
-                                                    size_t buffer_limit) {
+shared_ptr<io_bufferfill_t> io_bufferfill_t::create(const fd_set_t &conflicts, size_t buffer_limit,
+                                                    int target) {
+    assert(target >= 0 && "Invalid target fd");
+
     // Construct our pipes.
     auto pipes = make_autoclose_pipes(conflicts);
     if (!pipes) {
@@ -192,7 +194,7 @@ shared_ptr<io_bufferfill_t> io_bufferfill_t::create(const fd_set_t &conflicts,
     // Our fillthread gets the read end of the pipe; out_pipe gets the write end.
     auto buffer = std::make_shared<io_buffer_t>(buffer_limit);
     buffer->begin_background_fillthread(std::move(pipes->read));
-    return std::make_shared<io_bufferfill_t>(std::move(pipes->write), buffer);
+    return std::make_shared<io_bufferfill_t>(target, std::move(pipes->write), buffer);
 }
 
 std::shared_ptr<io_buffer_t> io_bufferfill_t::finish(std::shared_ptr<io_bufferfill_t> &&filler) {
@@ -346,3 +348,9 @@ shared_ptr<const io_data_t> io_chain_t::io_for_fd(int fd) const {
     }
     return nullptr;
 }
+
+void output_stream_t::append_narrow_buffer(const separated_buffer_t<std::string> &buffer) {
+    for (const auto &rhs_elem : buffer.elements()) {
+        buffer_.append(str2wcstring(rhs_elem.contents), rhs_elem.separation);
+    }
+}

src/io.h

@@ -252,7 +252,6 @@ class io_buffer_t;
 class io_chain_t;
 
 /// Represents filling an io_buffer_t. Very similar to io_pipe_t.
-/// Bufferfills always target stdout.
 class io_bufferfill_t : public io_data_t {
     /// Write end. The other end is connected to an io_buffer_t.
     const autoclose_fd_t write_fd_;
@@ -265,8 +264,8 @@ class io_bufferfill_t : public io_data_t {
 
     // The ctor is public to support make_shared() in the static create function below.
     // Do not invoke this directly.
-    io_bufferfill_t(autoclose_fd_t write_fd, std::shared_ptr<io_buffer_t> buffer)
-        : io_data_t(io_mode_t::bufferfill, STDOUT_FILENO, write_fd.fd()),
+    io_bufferfill_t(int target, autoclose_fd_t write_fd, std::shared_ptr<io_buffer_t> buffer)
+        : io_data_t(io_mode_t::bufferfill, target, write_fd.fd()),
           write_fd_(std::move(write_fd)),
           buffer_(std::move(buffer)) {
         assert(write_fd_.valid() && "fd is not valid");
@@ -279,9 +278,11 @@ class io_bufferfill_t : public io_data_t {
     /// Create an io_bufferfill_t which, when written from, fills a buffer with the contents.
     /// \returns nullptr on failure, e.g. too many open fds.
     ///
+    /// \param target the fd which this will be dup2'd to - typically stdout.
     /// \param conflicts A set of fds. The function ensures that any pipe it makes does
     /// not conflict with an fd redirection in this list.
-    static shared_ptr<io_bufferfill_t> create(const fd_set_t &conflicts, size_t buffer_limit = 0);
+    static shared_ptr<io_bufferfill_t> create(const fd_set_t &conflicts, size_t buffer_limit = 0,
+                                              int target = STDOUT_FILENO);
 
     /// Reset the receiver (possibly closing the write end of the pipe), and complete the fillthread
     /// of the buffer. \return the buffer.
@@ -421,6 +422,9 @@ class output_stream_t {
 
     void append(const wchar_t *s, size_t amt) { buffer_.append(s, s + amt); }
 
+    // Append data from a narrow buffer, widening it.
+    void append_narrow_buffer(const separated_buffer_t<std::string> &buffer);
+
     void push_back(wchar_t c) { append(c); }
 
     void append_format(const wchar_t *format, ...) {