diff options
Diffstat (limited to 'run-command.c')
| -rw-r--r-- | run-command.c | 110 |
1 files changed, 96 insertions, 14 deletions
diff --git a/run-command.c b/run-command.c index ed9575bd6a..f217adcad6 100644 --- a/run-command.c +++ b/run-command.c @@ -1578,7 +1578,10 @@ static void pp_cleanup(struct parallel_processes *pp, * When get_next_task added messages to the buffer in its last * iteration, the buffered output is non empty. */ - strbuf_write(&pp->buffered_output, stderr); + if (opts->consume_sideband) + opts->consume_sideband(&pp->buffered_output, opts->data); + else + strbuf_write(&pp->buffered_output, stderr); strbuf_release(&pp->buffered_output); sigchain_pop_common(); @@ -1652,6 +1655,44 @@ static int pp_start_one(struct parallel_processes *pp, return 0; } +static void pp_buffer_stdin(struct parallel_processes *pp, + const struct run_process_parallel_opts *opts) +{ + /* Buffer stdin for each pipe. */ + for (ssize_t i = 0; i < opts->processes; i++) { + struct child_process *proc = &pp->children[i].process; + int ret; + + if (pp->children[i].state != GIT_CP_WORKING || proc->in <= 0) + continue; + + /* + * child input is provided via path_to_stdin when the feed_pipe cb is + * missing, so we just signal an EOF. + */ + if (!opts->feed_pipe) { + close(proc->in); + proc->in = 0; + continue; + } + + /** + * Feed the pipe: + * ret < 0 means error + * ret == 0 means there is more data to be fed + * ret > 0 means feeding finished + */ + ret = opts->feed_pipe(proc->in, opts->data, pp->children[i].data); + if (ret < 0) + die_errno("feed_pipe"); + + if (ret) { + close(proc->in); + proc->in = 0; + } + } +} + static void pp_buffer_stderr(struct parallel_processes *pp, const struct run_process_parallel_opts *opts, int output_timeout) @@ -1679,13 +1720,17 @@ static void pp_buffer_stderr(struct parallel_processes *pp, } } -static void pp_output(const struct parallel_processes *pp) +static void pp_output(const struct parallel_processes *pp, + const struct run_process_parallel_opts *opts) { size_t i = pp->output_owner; if (pp->children[i].state == GIT_CP_WORKING && pp->children[i].err.len) { - strbuf_write(&pp->children[i].err, stderr); + if (opts->consume_sideband) + opts->consume_sideband(&pp->children[i].err, opts->data); + else + strbuf_write(&pp->children[i].err, stderr); strbuf_reset(&pp->children[i].err); } } @@ -1722,6 +1767,7 @@ static int pp_collect_finished(struct parallel_processes *pp, pp->children[i].state = GIT_CP_FREE; if (pp->pfd) pp->pfd[i].fd = -1; + pp->children[i].process.in = 0; child_process_init(&pp->children[i].process); if (opts->ungroup) { @@ -1732,11 +1778,15 @@ static int pp_collect_finished(struct parallel_processes *pp, } else { const size_t n = opts->processes; - strbuf_write(&pp->children[i].err, stderr); + /* Output errors, then all other finished child processes */ + if (opts->consume_sideband) { + opts->consume_sideband(&pp->children[i].err, opts->data); + opts->consume_sideband(&pp->buffered_output, opts->data); + } else { + strbuf_write(&pp->children[i].err, stderr); + strbuf_write(&pp->buffered_output, stderr); + } strbuf_reset(&pp->children[i].err); - - /* Output all other finished child processes */ - strbuf_write(&pp->buffered_output, stderr); strbuf_reset(&pp->buffered_output); /* @@ -1756,6 +1806,32 @@ static int pp_collect_finished(struct parallel_processes *pp, return result; } +static void pp_handle_child_IO(struct parallel_processes *pp, + const struct run_process_parallel_opts *opts, + int output_timeout) +{ + /* + * First push input, if any (it might no-op), to child tasks to avoid them blocking + * after input. This also prevents deadlocks when ungrouping below, if a child blocks + * while the parent also waits for them to finish. + */ + pp_buffer_stdin(pp, opts); + + if (opts->ungroup) { + for (size_t i = 0; i < opts->processes; i++) { + int child_ready_for_cleanup = + pp->children[i].state == GIT_CP_WORKING && + pp->children[i].process.in == 0; + + if (child_ready_for_cleanup) + pp->children[i].state = GIT_CP_WAIT_CLEANUP; + } + } else { + pp_buffer_stderr(pp, opts, output_timeout); + pp_output(pp, opts); + } +} + void run_processes_parallel(const struct run_process_parallel_opts *opts) { int i, code; @@ -1775,6 +1851,16 @@ void run_processes_parallel(const struct run_process_parallel_opts *opts) "max:%"PRIuMAX, (uintmax_t)opts->processes); + if (opts->ungroup && opts->consume_sideband) + BUG("ungroup and reading sideband are mutualy exclusive"); + + /* + * Child tasks might receive input via stdin, terminating early (or not), so + * ignore the default SIGPIPE which gets handled by each feed_pipe_fn which + * actually writes the data to children stdin fds. + */ + sigchain_push(SIGPIPE, SIG_IGN); + pp_init(&pp, opts, &pp_sig); while (1) { for (i = 0; @@ -1792,13 +1878,7 @@ void run_processes_parallel(const struct run_process_parallel_opts *opts) } if (!pp.nr_processes) break; - if (opts->ungroup) { - for (size_t i = 0; i < opts->processes; i++) - pp.children[i].state = GIT_CP_WAIT_CLEANUP; - } else { - pp_buffer_stderr(&pp, opts, output_timeout); - pp_output(&pp); - } + pp_handle_child_IO(&pp, opts, output_timeout); code = pp_collect_finished(&pp, opts); if (code) { pp.shutdown = 1; @@ -1809,6 +1889,8 @@ void run_processes_parallel(const struct run_process_parallel_opts *opts) pp_cleanup(&pp, opts); + sigchain_pop(SIGPIPE); + if (do_trace2) trace2_region_leave(tr2_category, tr2_label, NULL); } |