From 5abec8f71177edb330e79b16a62920238c2462fd Mon Sep 17 00:00:00 2001 From: Emily Shaffer Date: Fri, 17 Oct 2025 17:15:35 +0300 Subject: run-command: add stdin callback for parallelization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit If a user of the run_processes_parallel() API wants to pipe a large amount of information to the stdin of each parallel command, that data could exceed the pipe buffer of the process's stdin and can be too big to store in-memory via strbuf & friends or to slurp to a file. Generally this is solved by repeatedly writing to child_process.in between calls to start_command() and finish_command(). For a specific pre-existing example of this, see transport.c:run_pre_push_hook(). This adds a generic callback API to run_processes_parallel() to do exactly that in a unified manner, similar to the existing callback APIs, which can then be used by hooks.h to convert the remaining hooks to the new, simpler parallel interface. Signed-off-by: Emily Shaffer Signed-off-by: Ævar Arnfjörð Bjarmason Signed-off-by: Adrian Ratiu Signed-off-by: Junio C Hamano --- t/helper/test-run-command.c | 52 +++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 50 insertions(+), 2 deletions(-) (limited to 't/helper/test-run-command.c') diff --git a/t/helper/test-run-command.c b/t/helper/test-run-command.c index 3719f23cc2..dfdb03b3ab 100644 --- a/t/helper/test-run-command.c +++ b/t/helper/test-run-command.c @@ -23,19 +23,26 @@ static int number_callbacks; static int parallel_next(struct child_process *cp, struct strbuf *err, void *cb, - void **task_cb UNUSED) + void **task_cb) { struct child_process *d = cb; if (number_callbacks >= 4) return 0; strvec_pushv(&cp->args, d->args.v); + cp->in = d->in; + cp->no_stdin = d->no_stdin; if (err) strbuf_addstr(err, "preloaded output of a child\n"); else fprintf(stderr, "preloaded output of a child\n"); number_callbacks++; + + /* test_stdin callback will use this to count remaining lines */ + *task_cb = xmalloc(sizeof(int)); + *(int*)(*task_cb) = 2; + return 1; } @@ -54,15 +61,48 @@ static int no_job(struct child_process *cp UNUSED, static int task_finished(int result UNUSED, struct strbuf *err, void *pp_cb UNUSED, - void *pp_task_cb UNUSED) + void *pp_task_cb) { if (err) strbuf_addstr(err, "asking for a quick stop\n"); else fprintf(stderr, "asking for a quick stop\n"); + if (pp_task_cb) + FREE_AND_NULL(pp_task_cb); return 1; } +static int task_finished_quiet(int result UNUSED, + struct strbuf *err UNUSED, + void *pp_cb UNUSED, + void *pp_task_cb) +{ + if (pp_task_cb) + FREE_AND_NULL(pp_task_cb); + return 0; +} + +static int test_stdin_pipe_feed(int hook_stdin_fd, void *cb UNUSED, void *task_cb) +{ + int *lines_remaining = task_cb; + + if (*lines_remaining) { + struct strbuf buf = STRBUF_INIT; + strbuf_addf(&buf, "sample stdin %d\n", --(*lines_remaining)); + if (write_in_full(hook_stdin_fd, buf.buf, buf.len) < 0) { + if (errno == EPIPE) { + /* child closed stdin, nothing more to do */ + strbuf_release(&buf); + return 1; + } + die_errno("write"); + } + strbuf_release(&buf); + } + + return !(*lines_remaining); +} + struct testsuite { struct string_list tests, failed; int next; @@ -157,6 +197,7 @@ static int testsuite(int argc, const char **argv) struct run_process_parallel_opts opts = { .get_next_task = next_test, .start_failure = test_failed, + .feed_pipe = test_stdin_pipe_feed, .task_finished = test_finished, .data = &suite, }; @@ -460,12 +501,19 @@ int cmd__run_command(int argc, const char **argv) if (!strcmp(argv[1], "run-command-parallel")) { opts.get_next_task = parallel_next; + opts.task_finished = task_finished_quiet; } else if (!strcmp(argv[1], "run-command-abort")) { opts.get_next_task = parallel_next; opts.task_finished = task_finished; } else if (!strcmp(argv[1], "run-command-no-jobs")) { opts.get_next_task = no_job; opts.task_finished = task_finished; + } else if (!strcmp(argv[1], "run-command-stdin")) { + proc.in = -1; + proc.no_stdin = 0; + opts.get_next_task = parallel_next; + opts.task_finished = task_finished_quiet; + opts.feed_pipe = test_stdin_pipe_feed; } else { ret = 1; fprintf(stderr, "check usage\n"); -- cgit v1.2.3 From bdf49ba9aa4090d1b3784eaa0e9e364e340202fa Mon Sep 17 00:00:00 2001 From: Emily Shaffer Date: Fri, 17 Oct 2025 17:15:41 +0300 Subject: run-command: allow capturing of collated output MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Some callers, for example server-side hooks which wish to relay hook output to clients across a transport, want to capture what would normally print to stderr and do something else with it. Allow that via a callback. By calling the callback regardless of whether there's output available, we allow clients to send e.g. a keepalive if necessary. Because we expose a strbuf, not a fd or FILE*, there's no need to create a temporary pipe or similar - we can just skip the print to stderr and instead hand it to the caller. Signed-off-by: Emily Shaffer Signed-off-by: Ævar Arnfjörð Bjarmason Signed-off-by: Adrian Ratiu Signed-off-by: Junio C Hamano --- run-command.c | 30 ++++++++++++++++++++++-------- run-command.h | 20 ++++++++++++++++++++ t/helper/test-run-command.c | 15 +++++++++++++++ t/t0061-run-command.sh | 7 +++++++ 4 files changed, 64 insertions(+), 8 deletions(-) (limited to 't/helper/test-run-command.c') diff --git a/run-command.c b/run-command.c index 5bc6db5bb1..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(); @@ -1717,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); } } @@ -1771,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); /* @@ -1817,7 +1828,7 @@ static void pp_handle_child_IO(struct parallel_processes *pp, } } else { pp_buffer_stderr(pp, opts, output_timeout); - pp_output(pp); + pp_output(pp, opts); } } @@ -1840,6 +1851,9 @@ 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 diff --git a/run-command.h b/run-command.h index e536ed7544..2c2484478b 100644 --- a/run-command.h +++ b/run-command.h @@ -436,6 +436,20 @@ typedef int (*feed_pipe_fn)(int child_in, void *pp_cb, void *pp_task_cb); +/** + * If this callback is provided, instead of collating process output to stderr, + * they will be collated into a new pipe. consume_sideband_fn will be called + * repeatedly. When output is available on that pipe, it will be contained in + * 'output'. But it will be called with an empty 'output' too, to allow for + * keepalives or similar operations if necessary. + * + * pp_cb is the callback cookie as passed into run_processes_parallel. + * + * Since this callback is provided with the collated output, no task cookie is + * provided. + */ +typedef void (*consume_sideband_fn)(struct strbuf *output, void *pp_cb); + /** * This callback is called on every child process that finished processing. * @@ -495,6 +509,12 @@ struct run_process_parallel_opts */ feed_pipe_fn feed_pipe; + /* + * consume_sideband: see consume_sideband_fn() above. This can be NULL + * to omit any special handling. + */ + consume_sideband_fn consume_sideband; + /** * task_finished: See task_finished_fn() above. This can be * NULL to omit any special handling. diff --git a/t/helper/test-run-command.c b/t/helper/test-run-command.c index dfdb03b3ab..95152a0395 100644 --- a/t/helper/test-run-command.c +++ b/t/helper/test-run-command.c @@ -58,6 +58,16 @@ static int no_job(struct child_process *cp UNUSED, return 0; } +static void test_consume_sideband(struct strbuf *output, void *cb UNUSED) +{ + FILE *sideband; + + sideband = fopen("./sideband", "a"); + + strbuf_write(output, sideband); + fclose(sideband); +} + static int task_finished(int result UNUSED, struct strbuf *err, void *pp_cb UNUSED, @@ -198,6 +208,7 @@ static int testsuite(int argc, const char **argv) .get_next_task = next_test, .start_failure = test_failed, .feed_pipe = test_stdin_pipe_feed, + .consume_sideband = test_consume_sideband, .task_finished = test_finished, .data = &suite, }; @@ -514,6 +525,10 @@ int cmd__run_command(int argc, const char **argv) opts.get_next_task = parallel_next; opts.task_finished = task_finished_quiet; opts.feed_pipe = test_stdin_pipe_feed; + } else if (!strcmp(argv[1], "run-command-sideband")) { + opts.get_next_task = parallel_next; + opts.consume_sideband = test_consume_sideband; + opts.task_finished = task_finished_quiet; } else { ret = 1; fprintf(stderr, "check usage\n"); diff --git a/t/t0061-run-command.sh b/t/t0061-run-command.sh index 2f77fde0d9..f133d71783 100755 --- a/t/t0061-run-command.sh +++ b/t/t0061-run-command.sh @@ -164,6 +164,13 @@ test_expect_success 'run_command runs ungrouped in parallel with more tasks than test_line_count = 4 err ' +test_expect_success 'run_command can divert output' ' + test_when_finished rm sideband && + test-tool run-command run-command-sideband 3 sh -c "printf \"%s\n%s\n\" Hello World" 2>actual && + test_must_be_empty actual && + test_cmp expect sideband +' + test_expect_success 'run_command listens to stdin' ' cat >expect <<-\EOF && preloaded output of a child -- cgit v1.2.3