run-command: add stdin callback for parallelization
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 <emilyshaffer@google.com> Signed-off-by: Ævar Arnfjörð Bjarmason <avarab@gmail.com> Signed-off-by: Adrian Ratiu <adrian.ratiu@collabora.com> Signed-off-by: Junio C Hamano <gitster@pobox.com>
Emily Shaffer committed
Jan 28, 2026 at 23:39 UTC
ec0becacc9847406f2b0147a81f62e023b006351
4 files changed
+182
-9
run-command.c
+80
-7
@@ -1490,6 +1490,16 @@ static int child_is_working(const struct parallel_child *pp_child)
1490
return pp_child->state == GIT_CP_WORKING;
1491
}
1492
1493
+static int child_is_ready_for_cleanup(const struct parallel_child *pp_child)
1494
+{
1495
+ return child_is_working(pp_child) && !pp_child->process.in;
1496
+}
1497
+
1498
+static int child_is_receiving_input(const struct parallel_child *pp_child)
1499
+{
1500
+ return child_is_working(pp_child) && pp_child->process.in > 0;
1501
+}
1502
+
1503
struct parallel_processes {
1504
size_t nr_processes;
1505
@@ -1659,6 +1669,44 @@ static int pp_start_one(struct parallel_processes *pp,
1669
return 0;
1670
}
1671
1672
+static void pp_buffer_stdin(struct parallel_processes *pp,
1673
+ const struct run_process_parallel_opts *opts)
1674
+{
1675
+ /* Buffer stdin for each pipe. */
1676
+ for (size_t i = 0; i < opts->processes; i++) {
1677
+ struct child_process *proc = &pp->children[i].process;
1678
+ int ret;
1679
+
1680
+ if (!child_is_receiving_input(&pp->children[i]))
1681
+ continue;
1682
+
1683
+ /*
1684
+ * child input is provided via path_to_stdin when the feed_pipe cb is
1685
+ * missing, so we just signal an EOF.
1686
+ */
1687
+ if (!opts->feed_pipe) {
1688
+ close(proc->in);
1689
+ proc->in = 0;
1690
+ continue;
1691
+ }
1692
+
1693
+ /**
1694
+ * Feed the pipe:
1695
+ * ret < 0 means error
1696
+ * ret == 0 means there is more data to be fed
1697
+ * ret > 0 means feeding finished
1698
+ */
1699
+ ret = opts->feed_pipe(proc->in, opts->data, pp->children[i].data);
1700
+ if (ret < 0)
1701
+ die_errno("feed_pipe");
1702
+
1703
+ if (ret) {
1704
+ close(proc->in);
1705
+ proc->in = 0;
1706
+ }
1707
+ }
1708
+}
1709
+
1710
static void pp_buffer_stderr(struct parallel_processes *pp,
1711
const struct run_process_parallel_opts *opts,
1712
int output_timeout)
@@ -1729,6 +1777,7 @@ static int pp_collect_finished(struct parallel_processes *pp,
1777
pp->children[i].state = GIT_CP_FREE;
1778
if (pp->pfd)
1779
pp->pfd[i].fd = -1;
1780
+ pp->children[i].process.in = 0;
1781
child_process_init(&pp->children[i].process);
1782
1783
if (opts->ungroup) {
@@ -1763,6 +1812,27 @@ static int pp_collect_finished(struct parallel_processes *pp,
1812
return result;
1813
}
1814
1815
+static void pp_handle_child_IO(struct parallel_processes *pp,
1816
+ const struct run_process_parallel_opts *opts,
1817
+ int output_timeout)
1818
+{
1819
+ /*
1820
+ * First push input, if any (it might no-op), to child tasks to avoid them blocking
1821
+ * after input. This also prevents deadlocks when ungrouping below, if a child blocks
1822
+ * while the parent also waits for them to finish.
1823
+ */
1824
+ pp_buffer_stdin(pp, opts);
1825
+
1826
+ if (opts->ungroup) {
1827
+ for (size_t i = 0; i < opts->processes; i++)
1828
+ if (child_is_ready_for_cleanup(&pp->children[i]))
1829
+ pp->children[i].state = GIT_CP_WAIT_CLEANUP;
1830
+ } else {
1831
+ pp_buffer_stderr(pp, opts, output_timeout);
1832
+ pp_output(pp);
1833
+ }
1834
+}
1835
+
1836
void run_processes_parallel(const struct run_process_parallel_opts *opts)
1837
{
1838
int i, code;
@@ -1782,6 +1852,13 @@ void run_processes_parallel(const struct run_process_parallel_opts *opts)
1852
"max:%"PRIuMAX,
1853
(uintmax_t)opts->processes);
1854
1855
+ /*
1856
+ * Child tasks might receive input via stdin, terminating early (or not), so
1857
+ * ignore the default SIGPIPE which gets handled by each feed_pipe_fn which
1858
+ * actually writes the data to children stdin fds.
1859
+ */
1860
+ sigchain_push(SIGPIPE, SIG_IGN);
1861
+
1862
pp_init(&pp, opts, &pp_sig);
1863
while (1) {
1864
for (i = 0;
@@ -1799,13 +1876,7 @@ void run_processes_parallel(const struct run_process_parallel_opts *opts)
1876
}
1877
if (!pp.nr_processes)
1878
break;
1802
- if (opts->ungroup) {
1803
- for (size_t i = 0; i < opts->processes; i++)
1804
- pp.children[i].state = GIT_CP_WAIT_CLEANUP;
1805
- } else {
1806
- pp_buffer_stderr(&pp, opts, output_timeout);
1807
- pp_output(&pp);
1808
- }
1879
+ pp_handle_child_IO(&pp, opts, output_timeout);
1880
code = pp_collect_finished(&pp, opts);
1881
if (code) {
1882
pp.shutdown = 1;
@@ -1816,6 +1887,8 @@ void run_processes_parallel(const struct run_process_parallel_opts *opts)
1887
1888
pp_cleanup(&pp, opts);
1889
1890
+ sigchain_pop(SIGPIPE);
1891
+
1892
if (do_trace2)
1893
trace2_region_leave(tr2_category, tr2_label, NULL);
1894
}
run-command.h
+21
@@ -420,6 +420,21 @@ typedef int (*start_failure_fn)(struct strbuf *out,
420
void *pp_cb,
421
void *pp_task_cb);
422
423
+/**
424
+ * This callback is repeatedly called on every child process who requests
425
+ * start_command() to create a pipe by setting child_process.in < 0.
426
+ *
427
+ * pp_cb is the callback cookie as passed into run_processes_parallel, and
428
+ * pp_task_cb is the callback cookie as passed into get_next_task_fn.
429
+ *
430
+ * Returns < 0 for error
431
+ * Returns == 0 when there is more data to be fed (will be called again)
432
+ * Returns > 0 when finished (child closed fd or no more data to be fed)
433
+ */
434
+typedef int (*feed_pipe_fn)(int child_in,
435
+ void *pp_cb,
436
+ void *pp_task_cb);
437
+
438
/**
439
* This callback is called on every child process that finished processing.
440
*
@@ -473,6 +488,12 @@ struct run_process_parallel_opts
488
*/
489
start_failure_fn start_failure;
490
491
+ /*
492
+ * feed_pipe: see feed_pipe_fn() above. This can be NULL to omit any
493
+ * special handling.
494
+ */
495
+ feed_pipe_fn feed_pipe;
496
+
497
/**
498
* task_finished: See task_finished_fn() above. This can be
499
* NULL to omit any special handling.
t/helper/test-run-command.c
+50
-2
@@ -23,19 +23,26 @@ static int number_callbacks;
23
static int parallel_next(struct child_process *cp,
24
struct strbuf *err,
25
void *cb,
26
- void **task_cb UNUSED)
26
+ void **task_cb)
27
{
28
struct child_process *d = cb;
29
if (number_callbacks >= 4)
30
return 0;
31
32
strvec_pushv(&cp->args, d->args.v);
33
+ cp->in = d->in;
34
+ cp->no_stdin = d->no_stdin;
35
if (err)
36
strbuf_addstr(err, "preloaded output of a child\n");
37
else
38
fprintf(stderr, "preloaded output of a child\n");
39
40
number_callbacks++;
41
+
42
+ /* test_stdin callback will use this to count remaining lines */
43
+ *task_cb = xmalloc(sizeof(int));
44
+ *(int*)(*task_cb) = 2;
45
+
46
return 1;
47
}
48
@@ -54,15 +61,48 @@ static int no_job(struct child_process *cp UNUSED,
61
static int task_finished(int result UNUSED,
62
struct strbuf *err,
63
void *pp_cb UNUSED,
57
- void *pp_task_cb UNUSED)
64
+ void *pp_task_cb)
65
{
66
if (err)
67
strbuf_addstr(err, "asking for a quick stop\n");
68
else
69
fprintf(stderr, "asking for a quick stop\n");
70
+
71
+ FREE_AND_NULL(pp_task_cb);
72
+
73
return 1;
74
}
75
76
+static int task_finished_quiet(int result UNUSED,
77
+ struct strbuf *err UNUSED,
78
+ void *pp_cb UNUSED,
79
+ void *pp_task_cb)
80
+{
81
+ FREE_AND_NULL(pp_task_cb);
82
+ return 0;
83
+}
84
+
85
+static int test_stdin_pipe_feed(int hook_stdin_fd, void *cb UNUSED, void *task_cb)
86
+{
87
+ int *lines_remaining = task_cb;
88
+
89
+ if (*lines_remaining) {
90
+ struct strbuf buf = STRBUF_INIT;
91
+ strbuf_addf(&buf, "sample stdin %d\n", --(*lines_remaining));
92
+ if (write_in_full(hook_stdin_fd, buf.buf, buf.len) < 0) {
93
+ if (errno == EPIPE) {
94
+ /* child closed stdin, nothing more to do */
95
+ strbuf_release(&buf);
96
+ return 1;
97
+ }
98
+ die_errno("write");
99
+ }
100
+ strbuf_release(&buf);
101
+ }
102
+
103
+ return !(*lines_remaining);
104
+}
105
+
106
struct testsuite {
107
struct string_list tests, failed;
108
int next;
@@ -157,6 +197,7 @@ static int testsuite(int argc, const char **argv)
197
struct run_process_parallel_opts opts = {
198
.get_next_task = next_test,
199
.start_failure = test_failed,
200
+ .feed_pipe = test_stdin_pipe_feed,
201
.task_finished = test_finished,
202
.data = &suite,
203
};
@@ -460,12 +501,19 @@ int cmd__run_command(int argc, const char **argv)
501
502
if (!strcmp(argv[1], "run-command-parallel")) {
503
opts.get_next_task = parallel_next;
504
+ opts.task_finished = task_finished_quiet;
505
} else if (!strcmp(argv[1], "run-command-abort")) {
506
opts.get_next_task = parallel_next;
507
opts.task_finished = task_finished;
508
} else if (!strcmp(argv[1], "run-command-no-jobs")) {
509
opts.get_next_task = no_job;
510
opts.task_finished = task_finished;
511
+ } else if (!strcmp(argv[1], "run-command-stdin")) {
512
+ proc.in = -1;
513
+ proc.no_stdin = 0;
514
+ opts.get_next_task = parallel_next;
515
+ opts.task_finished = task_finished_quiet;
516
+ opts.feed_pipe = test_stdin_pipe_feed;
517
} else {
518
ret = 1;
519
fprintf(stderr, "check usage\n");
t/t0061-run-command.sh
+31
@@ -164,6 +164,37 @@ test_expect_success 'run_command runs ungrouped in parallel with more tasks than
164
test_line_count = 4 err
165
'
166
167
+test_expect_success 'run_command listens to stdin' '
168
+ cat >expect <<-\EOF &&
169
+ preloaded output of a child
170
+ listening for stdin:
171
+ sample stdin 1
172
+ sample stdin 0
173
+ preloaded output of a child
174
+ listening for stdin:
175
+ sample stdin 1
176
+ sample stdin 0
177
+ preloaded output of a child
178
+ listening for stdin:
179
+ sample stdin 1
180
+ sample stdin 0
181
+ preloaded output of a child
182
+ listening for stdin:
183
+ sample stdin 1
184
+ sample stdin 0
185
+ EOF
186
+
187
+ write_script stdin-script <<-\EOF &&
188
+ echo "listening for stdin:"
189
+ while read line
190
+ do
191
+ echo "$line"
192
+ done
193
+ EOF
194
+ test-tool run-command run-command-stdin 2 ./stdin-script 2>actual &&
195
+ test_cmp expect actual
196
+'
197
+
198
cat >expect <<-EOF
199
preloaded output of a child
200
asking for a quick stop