/**
   Read from descriptors until they are empty.

   \param j the job to test
*/
static void read_try(job_t *j)
{
    io_buffer_t *buff = NULL;

    /*
      Find the last buffer, which is the one we want to read from
    */
    const io_chain_t chain = j->all_io_redirections();
    for (size_t idx = 0; idx < chain.size(); idx++)
    {
        io_data_t *d = chain.at(idx).get();
        if (d->io_mode == IO_BUFFER)
        {
            buff = static_cast<io_buffer_t *>(d);
        }
    }

    if (buff)
    {
        debug(3, L"proc::read_try('%ls')\n", j->command_wcstr());
        while (1)
        {
            char b[BUFFER_SIZE];
            long l;

            l=read_blocked(buff->pipe_fd[0],
                           b, BUFFER_SIZE);
            if (l==0)
            {
                break;
            }
            else if (l<0)
            {
                if (errno != EAGAIN)
                {
                    debug(1,
                          _(L"An error occured while reading output from code block"));
                    wperror(L"read_try");
                }
                break;
            }
            else
            {
                buff->out_buffer_append(b, l);
            }
        }
    }
}

int move_fd_to_unused(int fd, const io_chain_t &io_chain, bool cloexec) {
    if (fd < 0 || io_chain.get_io_for_fd(fd).get() == NULL) {
        return fd;
    }

    // We have fd >= 0, and it's a conflict. dup it and recurse. Note that we recurse before
    // anything is closed; this forces the kernel to give us a new one (or report fd exhaustion).
    int new_fd = fd;
    int tmp_fd;
    do {
        tmp_fd = dup(fd);
    } while (tmp_fd < 0 && errno == EINTR);

    assert(tmp_fd != fd);
    if (tmp_fd < 0) {
        // Likely fd exhaustion.
        new_fd = -1;
    } else {
        // Ok, we have a new candidate fd. Recurse. If we get a valid fd, either it's the same as
        // what we gave it, or it's a new fd and what we gave it has been closed. If we get a
        // negative value, the fd also has been closed.
        if (cloexec) set_cloexec(tmp_fd);
        new_fd = move_fd_to_unused(tmp_fd, io_chain);
    }

    // We're either returning a new fd or an error. In both cases, we promise to close the old one.
    assert(new_fd != fd);
    int saved_errno = errno;
    exec_close(fd);
    errno = saved_errno;
    return new_fd;
}

void io_print(const io_chain_t &chain)
{
    if (chain.empty())
    {
        fprintf(stderr, "Empty chain %p\n", &chain);
        return;
    }

    fprintf(stderr, "Chain %p (%ld items):\n", &chain, (long)chain.size());
    for (size_t i=0; i < chain.size(); i++)
    {
        const shared_ptr<const io_data_t> &io = chain.at(i);
        fprintf(stderr, "\t%lu: fd:%d, ", (unsigned long)i, io->fd);
        io->print();
    }
}

/** Make sure the fd used by each redirection is not used by a pipe. Note that while this does not modify the vector, it does modify the IO redirections within (gulp) */
static void free_redirected_fds_from_pipes(const io_chain_t &io_chain)
{
    size_t max = io_chain.size();
    for (size_t i = 0; i < max; i++)
    {
        int fd_to_free = io_chain.at(i)->fd;

        /* We only have to worry about fds beyond the three standard ones */
        if (fd_to_free <= 2)
            continue;

        /* Make sure the fd is not used by a pipe */
        for (size_t j = 0; j < max; j++)
        {
            /* We're only interested in pipes */
            io_data_t *io = io_chain.at(j).get();
            if (io->io_mode != IO_PIPE && io->io_mode != IO_BUFFER)
                continue;

            CAST_INIT(io_pipe_t *, possible_conflict, io);
            /* If the pipe is a conflict, dup it to some other value */
            for (int k=0; k<2; k++)
            {
                /* If it's not a conflict, we don't care */
                if (possible_conflict->pipe_fd[k] != fd_to_free)
                    continue;

                /* Repeat until we have a replacement fd */
                int replacement_fd = -1;
                while (replacement_fd < 0)
                {
                    replacement_fd = dup(fd_to_free);
                    if (replacement_fd == -1 && errno != EINTR)
                    {
                        debug_safe_int(1, FD_ERROR, fd_to_free);
                        safe_perror("dup");
                        FATAL_EXIT();
                    }
                }
                possible_conflict->pipe_fd[k] = replacement_fd;
            }
        }
    }
}

/**
   Check if there are buffers associated with the job, and select on
   them for a while if available.

   \param j the job to test

   \return 1 if buffers were available, zero otherwise
*/
static int select_try(job_t *j)
{
    fd_set fds;
    int maxfd=-1;

    FD_ZERO(&fds);

    const io_chain_t chain = j->all_io_redirections();
    for (size_t idx = 0; idx < chain.size(); idx++)
    {
        const io_data_t *io = chain.at(idx).get();
        if (io->io_mode == IO_BUFFER)
        {
            CAST_INIT(const io_pipe_t *, io_pipe, io);
            int fd = io_pipe->pipe_fd[0];
//			fwprintf( stderr, L"fd %d on job %ls\n", fd, j->command );
            FD_SET(fd, &fds);
            maxfd = maxi(maxfd, fd);
            debug(3, L"select_try on %d\n", fd);
        }
    }

    if (maxfd >= 0)
    {
        int retval;
        struct timeval tv;

        tv.tv_sec=0;
        tv.tv_usec=10000;

        retval =select(maxfd+1, &fds, 0, 0, &tv);
        if (retval == 0) {
            debug(3, L"select_try hit timeout\n");
        }
        return retval > 0;
    }

    return -1;
}

// This isn't used so the lint tools were complaining about its presence. I'm keeping it in the
// source because it could be useful for debugging.
void io_print(const io_chain_t &chain)
{
    if (chain.empty())
    {
        std::fwprintf(stderr, L"Empty chain %p\n", &chain);
        return;
    }

    std::fwprintf(stderr, L"Chain %p (%ld items):\n", &chain, (long)chain.size());
    for (size_t i=0; i < chain.size(); i++)
    {
        const shared_ptr<io_data_t> &io = chain.at(i);
        if (io.get() == NULL)
        {
            std::fwprintf(stderr, L"\t(null)\n");
        }
        else
        {
            std::fwprintf(stderr, L"\t%lu: fd:%d, ", (unsigned long)i, io->fd);
            io->print();
        }
    }
}

void io_chain_destroy(io_chain_t &chain)
{
    chain.destroy();
}

void io_duplicate_prepend(const io_chain_t &src, io_chain_t &dst)
{
    return dst.duplicate_prepend(src);
}

io_chain_t io_duplicate(const io_chain_t &chain)
{
    return chain.duplicate();
}

void io_remove(io_chain_t &list, const io_data_t *element)
{
    list.remove(element);
}

void exec_job(parser_t &parser, shared_ptr<job_t> j) {
    assert(j && "null job_t passed to exec_job!");

    // Set to true if something goes wrong while exec:ing the job, in which case the cleanup code
    // will kick in.
    bool exec_error = false;

    // If fish was invoked with -n or --no-execute, then no_exec will be set and we do nothing.
    if (no_exec) {
        return;
    }

    // Unfortunately `exec_job()` is called recursively when functions are encountered, with a new
    // job id (and therefore pgrp) each time, but always from the main thread. This breaks terminal
    // control since new pgrps take terminal control away from commands upstream in a different pgrp.
    // We try to work around this with a heuristic to determine whether to reuse the same pgrp as the
    // last-spawned pgrp if part of an existing job pipeline (keeping in mind that new jobs are
    // recursively started for both foreground and background jobs, and that a function can expand
    // to more than one external command, one (or more?) of which may want to read from upstream or
    // write to downstream of a pipe.
    // By keeping track of (select) "jobs in flight" we can try to marshall newly-created external
    // processes into existing pgrps. Fixes #3952.
    // This is a HACK and the correct solution would be to pass the active job through the pipeline
    // to the newly established parser context so that the funtion as parsed and evaluated can be
    // directly associated with this job and not a new one, BUT sometimes functions NEED to start a
    // new job. This HACK seeks a compromise by letting functions trigger the unilateral creation of
    // a new job, but reusing the "parent" job's existing pgrp in case of terminal control.
    static std::stack<decltype(j)> active_jobs;
    // There's an assumption that there's a one-to-one mapping between jobs under job control and
    // pgrps. When we share a parent job's pgrp, we risk reaping its processes before it is fully
    // constructed, causing later setpgrp(2) calls to fails (#5219). While the parent job is still
    // under construction, child jobs have job_flag_t::WAIT_BY_PROCESS set to prevent early repaing.
    // We store them here until the parent job is constructed, at which point it unsets this flag.
    static std::stack<decltype(j)> child_jobs;

    auto parent_job = active_jobs.empty() ? nullptr : active_jobs.top();
    bool job_pushed = false;
    if (j->get_flag(job_flag_t::TERMINAL) && j->get_flag(job_flag_t::JOB_CONTROL)) {
        // This will be popped before this job leaves exec_job
        active_jobs.push(j);
        job_pushed = true;
    }

    if (parent_job && j->processes.front()->type == EXTERNAL) {
        if (parent_job->pgid != INVALID_PID) {
            j->pgid = parent_job->pgid;
            j->set_flag(job_flag_t::JOB_CONTROL, true);
            j->set_flag(job_flag_t::NESTED, true);
            j->set_flag(job_flag_t::WAIT_BY_PROCESS, true);
            child_jobs.push(j);
        }
    }

    // Verify that all IO_BUFFERs are output. We used to support a (single, hacked-in) magical input
    // IO_BUFFER used by fish_pager, but now the claim is that there are no more clients and it is
    // removed. This assertion double-checks that.
    size_t stdout_read_limit = 0;
    const io_chain_t all_ios = j->all_io_redirections();
    for (size_t idx = 0; idx < all_ios.size(); idx++) {
        const shared_ptr<io_data_t> &io = all_ios.at(idx);

        if ((io->io_mode == IO_BUFFER)) {
            io_buffer_t *io_buffer = static_cast<io_buffer_t *>(io.get());
            assert(!io_buffer->is_input);
            stdout_read_limit = io_buffer->buffer().limit();
        }
    }

    if (j->processes.front()->type == INTERNAL_EXEC) {
        internal_exec(j.get(), std::move(all_ios));
        DIE("this should be unreachable");
    }

    // We may have block IOs that conflict with fd redirections. For example, we may have a command
    // with a redireciton like <&3; we may also have chosen 3 as the fd for our pipe. Ensure we have
    // no conflicts.
    for (const auto io : all_ios) {
        if (io->io_mode == IO_BUFFER) {
            auto *io_buffer = static_cast<io_buffer_t *>(io.get());
            if (!io_buffer->avoid_conflicts_with_io_chain(all_ios)) {
                // We could not avoid conflicts, probably due to fd exhaustion. Mark an error.
                exec_error = true;
                job_mark_process_as_failed(j.get(), j->processes.front().get());
                break;
            }
        }
    }

    // This loop loops over every process_t in the job, starting it as appropriate. This turns out
    // to be rather complex, since a process_t can be one of many rather different things.
    //
    // The loop also has to handle pipelining between the jobs.
    //
    // We can have up to three pipes "in flight" at a time:
    //
    // 1. The pipe the current process should read from (courtesy of the previous process)
    // 2. The pipe that the current process should write to
    // 3. The pipe that the next process should read from (courtesy of us)
    //
    autoclose_fd_t pipe_next_read;
//.........这里部分代码省略.........

/**
   Set up a childs io redirections. Should only be called by
   setup_child_process(). Does the following: First it closes any open
   file descriptors not related to the child by calling
   close_unused_internal_pipes() and closing the universal variable
   server file descriptor. It then goes on to perform all the
   redirections described by \c io.

   \param io the list of IO redirections for the child

   \return 0 on sucess, -1 on failiure
*/
static int handle_child_io(const io_chain_t &io_chain)
{
    for (size_t idx = 0; idx < io_chain.size(); idx++)
    {
        const io_data_t *io = io_chain.at(idx).get();
        int tmp;

        if (io->io_mode == IO_FD && io->fd == static_cast<const io_fd_t*>(io)->old_fd)
        {
            continue;
        }

        switch (io->io_mode)
        {
            case IO_CLOSE:
            {
                if (log_redirections) fprintf(stderr, "%d: close %d\n", getpid(), io->fd);
                if (close(io->fd))
                {
                    debug_safe_int(0, "Failed to close file descriptor %s", io->fd);
                    safe_perror("close");
                }
                break;
            }

            case IO_FILE:
            {
                // Here we definitely do not want to set CLO_EXEC because our child needs access
                CAST_INIT(const io_file_t *, io_file, io);
                if ((tmp=open(io_file->filename_cstr,
                              io_file->flags, OPEN_MASK))==-1)
                {
                    if ((io_file->flags & O_EXCL) &&
                            (errno ==EEXIST))
                    {
                        debug_safe(1, NOCLOB_ERROR, io_file->filename_cstr);
                    }
                    else
                    {
                        debug_safe(1, FILE_ERROR, io_file->filename_cstr);
                        safe_perror("open");
                    }

                    return -1;
                }
                else if (tmp != io->fd)
                {
                    /*
                      This call will sometimes fail, but that is ok,
                      this is just a precausion.
                    */
                    close(io->fd);

                    if (dup2(tmp, io->fd) == -1)
                    {
                        debug_safe_int(1,  FD_ERROR, io->fd);
                        safe_perror("dup2");
                        return -1;
                    }
                    exec_close(tmp);
                }
                break;
            }

            case IO_FD:
            {
                int old_fd = static_cast<const io_fd_t *>(io)->old_fd;
                if (log_redirections) fprintf(stderr, "%d: fd dup %d to %d\n", getpid(), old_fd, io->fd);

                /*
                  This call will sometimes fail, but that is ok,
                  this is just a precausion.
                */
                close(io->fd);


                if (dup2(old_fd, io->fd) == -1)
                {
                    debug_safe_int(1, FD_ERROR, io->fd);
                    safe_perror("dup2");
                    return -1;
                }
                break;
            }

            case IO_BUFFER:
            case IO_PIPE:
            {
//.........这里部分代码省略.........

/// Make a copy of the specified io redirection chain, but change file redirection into fd
/// redirection. This makes the redirection chain suitable for use as block-level io, since the file
/// won't be repeatedly reopened for every command in the block, which would reset the cursor
/// position.
///
/// \return true on success, false on failure. Returns the output chain and opened_fds by reference.
static bool io_transmogrify(const io_chain_t &in_chain, io_chain_t *out_chain,
                            std::vector<int> *out_opened_fds) {
    ASSERT_IS_MAIN_THREAD();
    assert(out_chain != NULL && out_opened_fds != NULL);
    assert(out_chain->empty());

    // Just to be clear what we do for an empty chain.
    if (in_chain.empty()) {
        return true;
    }

    bool success = true;

    // Make our chain of redirections.
    io_chain_t result_chain;

    // In the event we can't finish transmorgrifying, we'll have to close all the files we opened.
    std::vector<int> opened_fds;

    for (size_t idx = 0; idx < in_chain.size(); idx++) {
        const shared_ptr<io_data_t> &in = in_chain.at(idx);
        shared_ptr<io_data_t> out;  // gets allocated via new

        switch (in->io_mode) {
            case IO_PIPE:
            case IO_FD:
            case IO_BUFFER:
            case IO_CLOSE: {
                // These redirections don't need transmogrification. They can be passed through.
                out = in;
                break;
            }
            case IO_FILE: {
                // Transmogrify file redirections.
                int fd;
                io_file_t *in_file = static_cast<io_file_t *>(in.get());
                if ((fd = open(in_file->filename_cstr, in_file->flags, OPEN_MASK)) == -1) {
                    debug(1, FILE_ERROR, in_file->filename_cstr);

                    wperror(L"open");
                    success = false;
                    break;
                }

                opened_fds.push_back(fd);
                out.reset(new io_fd_t(in->fd, fd, false));
                break;
            }
        }

        if (out.get() != NULL) result_chain.push_back(out);

        // Don't go any further if we failed.
        if (!success) {
            break;
        }
    }

    // Now either return success, or clean up.
    if (success) {
        *out_chain = std::move(result_chain);
        *out_opened_fds = std::move(opened_fds);
    } else {
        result_chain.clear();
        io_cleanup_fds(opened_fds);
    }
    return success;
}

void io_remove(io_chain_t &list, const shared_ptr<const io_data_t> &element)
{
    list.remove(element);
}

void io_chain_t::append(const io_chain_t &chain)
{
    this->insert(this->end(), chain.begin(), chain.end());
}

shared_ptr<io_data_t> io_chain_get(io_chain_t &src, int fd)
{
    return src.get_io_for_fd(fd);
}

/// Execute an internal builtin. Given a parser, a job within that parser, and a process within that
/// job corresponding to a builtin, execute the builtin with the given streams. If pipe_read is set,
/// assign stdin to it; otherwise infer stdin from the IO chain.
/// \return true on success, false if there is an exec error.
static bool exec_internal_builtin_proc(parser_t &parser, job_t *j, process_t *p,
                                       const io_pipe_t *pipe_read, const io_chain_t &proc_io_chain,
                                       io_streams_t &streams) {
    assert(p->type == INTERNAL_BUILTIN && "Process must be a builtin");
    int local_builtin_stdin = STDIN_FILENO;
    bool close_stdin = false;

    // If this is the first process, check the io redirections and see where we should
    // be reading from.
    if (pipe_read) {
        local_builtin_stdin = pipe_read->pipe_fd[0];
    } else if (const auto in = proc_io_chain.get_io_for_fd(STDIN_FILENO)) {
        switch (in->io_mode) {
            case IO_FD: {
                const io_fd_t *in_fd = static_cast<const io_fd_t *>(in.get());
                // Ignore user-supplied fd redirections from an fd other than the
                // standard ones. e.g. in source <&3 don't actually read from fd 3,
                // which is internal to fish. We still respect this redirection in
                // that we pass it on as a block IO to the code that source runs,
                // and therefore this is not an error. Non-user supplied fd
                // redirections come about through transmogrification, and we need
                // to respect those here.
                if (!in_fd->user_supplied || (in_fd->old_fd >= 0 && in_fd->old_fd < 3)) {
                    local_builtin_stdin = in_fd->old_fd;
                }
                break;
            }
            case IO_PIPE: {
                const io_pipe_t *in_pipe = static_cast<const io_pipe_t *>(in.get());
                local_builtin_stdin = in_pipe->pipe_fd[0];
                break;
            }
            case IO_FILE: {
                // Do not set CLO_EXEC because child needs access.
                const io_file_t *in_file = static_cast<const io_file_t *>(in.get());
                local_builtin_stdin = open(in_file->filename_cstr, in_file->flags, OPEN_MASK);
                if (local_builtin_stdin == -1) {
                    debug(1, FILE_ERROR, in_file->filename_cstr);
                    wperror(L"open");
                } else {
                    close_stdin = true;
                }

                break;
            }
            case IO_CLOSE: {
                // FIXME: When requesting that stdin be closed, we really don't do
                // anything. How should this be handled?
                local_builtin_stdin = -1;

                break;
            }
            default: {
                local_builtin_stdin = -1;
                debug(1, _(L"Unknown input redirection type %d"), in->io_mode);
                break;
            }
        }
    }

    if (local_builtin_stdin == -1) return false;

    // Determine if we have a "direct" redirection for stdin.
    bool stdin_is_directly_redirected;
    if (!p->is_first_in_job) {
        // We must have a pipe
        stdin_is_directly_redirected = true;
    } else {
        // We are not a pipe. Check if there is a redirection local to the process
        // that's not IO_CLOSE.
        const shared_ptr<const io_data_t> stdin_io = io_chain_get(p->io_chain(), STDIN_FILENO);
        stdin_is_directly_redirected = stdin_io && stdin_io->io_mode != IO_CLOSE;
    }

    streams.stdin_fd = local_builtin_stdin;
    streams.out_is_redirected = has_fd(proc_io_chain, STDOUT_FILENO);
    streams.err_is_redirected = has_fd(proc_io_chain, STDERR_FILENO);
    streams.stdin_is_directly_redirected = stdin_is_directly_redirected;
    streams.io_chain = &proc_io_chain;

    // Since this may be the foreground job, and since a builtin may execute another
    // foreground job, we need to pretend to suspend this job while running the
    // builtin, in order to avoid a situation where two jobs are running at once.
    //
    // The reason this is done here, and not by the relevant builtins, is that this
    // way, the builtin does not need to know what job it is part of. It could
    // probably figure that out by walking the job list, but it seems more robust to
    // make exec handle things.
    const int fg = j->is_foreground();
    j->set_flag(job_flag_t::FOREGROUND, false);

    // Note this call may block for a long time, while the builtin performs I/O.
    p->status = builtin_run(parser, j->pgid, p->get_argv(), streams);

    // Restore the fg flag, which is temporarily set to false during builtin
    // execution so as not to confuse some job-handling builtins.
//.........这里部分代码省略.........

bool fork_actions_make_spawn_properties(posix_spawnattr_t *attr, posix_spawn_file_actions_t *actions, job_t *j, process_t *p, const io_chain_t &io_chain)
{
    /* Initialize the output */
    if (posix_spawnattr_init(attr) != 0)
    {
        return false;
    }

    if (posix_spawn_file_actions_init(actions) != 0)
    {
        posix_spawnattr_destroy(attr);
        return false;
    }

    bool should_set_parent_group_id = false;
    int desired_parent_group_id = 0;
    if (job_get_flag(j, JOB_CONTROL))
    {
        should_set_parent_group_id = true;

        // PCA: I'm quite fuzzy on process groups,
        // but I believe that the default value of 0
        // means that the process becomes its own
        // group leader, which is what set_child_group did
        // in this case. So we want this to be 0 if j->pgid is 0.
        desired_parent_group_id = j->pgid;
    }

    /* Set the handling for job control signals back to the default.  */
    bool reset_signal_handlers = true;

    /* Remove all signal blocks */
    bool reset_sigmask = true;

    /* Set our flags */
    short flags = 0;
    if (reset_signal_handlers)
        flags |= POSIX_SPAWN_SETSIGDEF;
    if (reset_sigmask)
        flags |= POSIX_SPAWN_SETSIGMASK;
    if (should_set_parent_group_id)
        flags |= POSIX_SPAWN_SETPGROUP;

    int err = 0;
    if (! err)
        err = posix_spawnattr_setflags(attr, flags);

    if (! err && should_set_parent_group_id)
        err = posix_spawnattr_setpgroup(attr, desired_parent_group_id);

    /* Everybody gets default handlers */
    if (! err && reset_signal_handlers)
    {
        sigset_t sigdefault;
        get_signals_with_handlers(&sigdefault);
        err = posix_spawnattr_setsigdefault(attr, &sigdefault);
    }

    /* No signals blocked */
    sigset_t sigmask;
    sigemptyset(&sigmask);
    if (! err && reset_sigmask)
        err = posix_spawnattr_setsigmask(attr, &sigmask);
    
    for (size_t idx = 0; idx < io_chain.size(); idx++)
    {
        const shared_ptr<const io_data_t> io = io_chain.at(idx);

        if (io->io_mode == IO_FD)
        {
            CAST_INIT(const io_fd_t *, io_fd, io.get());
            if (io->fd == io_fd->old_fd)
                continue;
        }

        switch (io->io_mode)
        {
            case IO_CLOSE:
            {
                if (! err)
                    err = posix_spawn_file_actions_addclose(actions, io->fd);
                break;
            }

            case IO_FILE:
            {
                CAST_INIT(const io_file_t *, io_file, io.get());
                if (! err)
                    err = posix_spawn_file_actions_addopen(actions, io->fd, io_file->filename_cstr, io_file->flags /* mode */, OPEN_MASK);
                break;
            }

            case IO_FD:
            {
                CAST_INIT(const io_fd_t *, io_fd, io.get());
                if (! err)
                    err = posix_spawn_file_actions_adddup2(actions, io_fd->old_fd /* from */, io->fd /* to */);
                break;
            }

//.........这里部分代码省略.........

/// Execute a block node or function "process".
/// \p user_ios contains the list of user-specified ios, used so we can avoid stomping on them with
/// our pipes. \return true on success, false on error.
static bool exec_block_or_func_process(parser_t &parser, job_t *j, process_t *p,
                                       const io_chain_t &user_ios, io_chain_t io_chain) {
    assert((p->type == INTERNAL_FUNCTION || p->type == INTERNAL_BLOCK_NODE) &&
           "Unexpected process type");

    // Create an output buffer if we're piping to another process.
    shared_ptr<io_buffer_t> block_output_io_buffer{};
    if (!p->is_last_in_job) {
        // Be careful to handle failure, e.g. too many open fds.
        block_output_io_buffer = io_buffer_t::create(STDOUT_FILENO, user_ios);
        if (!block_output_io_buffer) {
            job_mark_process_as_failed(j, p);
            return false;
        } else {
            // This looks sketchy, because we're adding this io buffer locally - they
            // aren't in the process or job redirection list. Therefore select_try won't
            // be able to read them. However we call block_output_io_buffer->read()
            // below, which reads until EOF. So there's no need to select on this.
            io_chain.push_back(block_output_io_buffer);
        }
    }

    if (p->type == INTERNAL_FUNCTION) {
        const wcstring func_name = p->argv0();
        auto props = function_get_properties(func_name);
        if (!props) {
            debug(0, _(L"Unknown function '%ls'"), p->argv0());
            return false;
        }

        const std::map<wcstring, env_var_t> inherit_vars = function_get_inherit_vars(func_name);

        function_block_t *fb =
            parser.push_block<function_block_t>(p, func_name, props->shadow_scope);
        function_prepare_environment(func_name, p->get_argv() + 1, inherit_vars);
        parser.forbid_function(func_name);

        internal_exec_helper(parser, props->parsed_source, props->body_node, io_chain);

        parser.allow_function();
        parser.pop_block(fb);
    } else {
        assert(p->type == INTERNAL_BLOCK_NODE);
        assert(p->block_node_source && p->internal_block_node && "Process is missing node info");
        internal_exec_helper(parser, p->block_node_source, p->internal_block_node, io_chain);
    }

    int status = proc_get_last_status();

    // Handle output from a block or function. This usually means do nothing, but in the
    // case of pipes, we have to buffer such io, since otherwise the internal pipe
    // buffer might overflow.
    if (!block_output_io_buffer.get()) {
        // No buffer, so we exit directly. This means we have to manually set the exit
        // status.
        if (p->is_last_in_job) {
            proc_set_last_status(j->get_flag(job_flag_t::NEGATE) ? (!status) : status);
        }
        p->completed = 1;
        return true;
    }

    // Here we must have a non-NULL block_output_io_buffer.
    assert(block_output_io_buffer.get() != NULL);
    io_chain.remove(block_output_io_buffer);
    block_output_io_buffer->read();

    const std::string buffer_contents = block_output_io_buffer->buffer().newline_serialized();
    const char *buffer = buffer_contents.data();
    size_t count = buffer_contents.size();
    if (count > 0) {
        // We don't have to drain threads here because our child process is simple.
        const char *fork_reason =
            p->type == INTERNAL_BLOCK_NODE ? "internal block io" : "internal function io";
        if (!fork_child_for_process(j, p, io_chain, false, fork_reason, [&] {
                exec_write_and_exit(block_output_io_buffer->fd, buffer, count, status);
            })) {
            return false;
        }
    } else {
        if (p->is_last_in_job) {
            proc_set_last_status(j->get_flag(job_flag_t::NEGATE) ? (!status) : status);
        }
        p->completed = 1;
    }
    return true;
}