run-command: add an "ungroup" option to run_process_parallel()

Extend the parallel execution API added in c553c72eed6 (run-command: add an asynchronous parallel child processor, 2015-12-15) to support a mode where the stdout and stderr of the processes isn't captured and output in a deterministic order, instead we'll leave it to the kernel and stdio to sort it out. This gives the API same functionality as GNU parallel's --ungroup option. As we'll see in a subsequent commit the main reason to want this is to support stdout and stderr being connected to the TTY in the case of jobs=1, demonstrated here with GNU parallel: $ parallel --ungroup 'test -t {} && echo TTY || echo NTTY' ::: 1 2 TTY TTY $ parallel 'test -t {} && echo TTY || echo NTTY' ::: 1 2 NTTY NTTY Another is as GNU parallel's documentation notes a potential for optimization. As demonstrated in next commit our results with "git hook run" will be similar, but generally speaking this shows that if you want to run processes in parallel where the exact order isn't important this can be a lot faster: $ hyperfine -r 3 -L o ,--ungroup 'parallel {o} seq ::: 10000000 >/dev/null ' Benchmark 1: parallel seq ::: 10000000 >/dev/null Time (mean ± σ): 220.2 ms ± 9.3 ms [User: 124.9 ms, System: 96.1 ms] Range (min … max): 212.3 ms … 230.5 ms 3 runs Benchmark 2: parallel --ungroup seq ::: 10000000 >/dev/null Time (mean ± σ): 154.7 ms ± 0.9 ms [User: 136.2 ms, System: 25.1 ms] Range (min … max): 153.9 ms … 155.7 ms 3 runs Summary 'parallel --ungroup seq ::: 10000000 >/dev/null ' ran 1.42 ± 0.06 times faster than 'parallel seq ::: 10000000 >/dev/null ' A large part of the juggling in the API is to make the API safer for its maintenance and consumers alike. For the maintenance of the API we e.g. avoid malloc()-ing the "pp->pfd", ensuring that SANITIZE=address and other similar tools will catch any unexpected misuse. For API consumers we take pains to never pass the non-NULL "out" buffer to an API user that provided the "ungroup" option. The resulting code in t/helper/test-run-command.c isn't typical of such a user, i.e. they'd typically use one mode or the other, and would know whether they'd provided "ungroup" or not. We could also avoid the strbuf_init() for "buffered_output" by having "struct parallel_processes" use a static PARALLEL_PROCESSES_INIT initializer, but let's leave that cleanup for later. Using a global "run_processes_parallel_ungroup" variable to enable this option is rather nasty, but is being done here to produce as minimal of a change as possible for a subsequent regression fix. This change is extracted from a larger initial version[1] which ends up with a better end-state for the API, but in doing so needed to modify all existing callers of the API. Let's defer that for now, and narrowly focus on what we need for fixing the regression in the subsequent commit. It's safe to do this with a global variable because: A) hook.c is the only user of it that sets it to non-zero, and before we'll get any other API users we'll refactor away this method of passing in the option, i.e. re-roll [1]. B) Even if hook.c wasn't the only user we don't have callers of this API that concurrently invoke this parallel process starting API itself in parallel. As noted above "A" && "B" are rather nasty, and we don't want to live with those caveats long-term, but for now they should be an acceptable compromise. 1. https://lore.kernel.org/git/cover-v2-0.8-00000000000-20220518T195858Z-avarab@gmail.com/ Signed-off-by: Ævar Arnfjörð Bjarmason <avarab@gmail.com> Signed-off-by: Junio C Hamano <gitster@pobox.com>

Ævar Arnfjörð Bjarmason committed Jun 7, 2022 at 10:48 UTC fd3aaf53f713d424d3c08cffa7e76e29b31638ba
4 files changed +123 -29
run-command.c
+51 -19
@@ -1471,6 +1471,7 @@ enum child_state {
1471 GIT_CP_WAIT_CLEANUP,
1472 };
1473
1474 +int run_processes_parallel_ungroup;
1475 struct parallel_processes {
1476 void *data;
1477
@@ -1494,6 +1495,7 @@ struct parallel_processes {
1495 struct pollfd *pfd;
1496
1497 unsigned shutdown : 1;
1498 + unsigned ungroup : 1;
1499
1500 int output_owner;
1501 struct strbuf buffered_output; /* of finished children */
@@ -1537,7 +1539,7 @@ static void pp_init(struct parallel_processes *pp,
1539 get_next_task_fn get_next_task,
1540 start_failure_fn start_failure,
1541 task_finished_fn task_finished,
1540 - void *data)
1542 + void *data, int ungroup)
1543 {
1544 int i;
1545
@@ -1559,15 +1561,21 @@ static void pp_init(struct parallel_processes *pp,
1561 pp->nr_processes = 0;
1562 pp->output_owner = 0;
1563 pp->shutdown = 0;
1564 + pp->ungroup = ungroup;
1565 CALLOC_ARRAY(pp->children, n);
1563 - CALLOC_ARRAY(pp->pfd, n);
1566 + if (pp->ungroup)
1567 + pp->pfd = NULL;
1568 + else
1569 + CALLOC_ARRAY(pp->pfd, n);
1570 strbuf_init(&pp->buffered_output, 0);
1571
1572 for (i = 0; i < n; i++) {
1573 strbuf_init(&pp->children[i].err, 0);
1574 child_process_init(&pp->children[i].process);
1569 - pp->pfd[i].events = POLLIN | POLLHUP;
1570 - pp->pfd[i].fd = -1;
1575 + if (pp->pfd) {
1576 + pp->pfd[i].events = POLLIN | POLLHUP;
1577 + pp->pfd[i].fd = -1;
1578 + }
1579 }
1580
1581 pp_for_signal = pp;
@@ -1615,24 +1623,31 @@ static int pp_start_one(struct parallel_processes *pp)
1623 BUG("bookkeeping is hard");
1624
1625 code = pp->get_next_task(&pp->children[i].process,
1618 - &pp->children[i].err,
1626 + pp->ungroup ? NULL : &pp->children[i].err,
1627 pp->data,
1628 &pp->children[i].data);
1629 if (!code) {
1622 - strbuf_addbuf(&pp->buffered_output, &pp->children[i].err);
1623 - strbuf_reset(&pp->children[i].err);
1630 + if (!pp->ungroup) {
1631 + strbuf_addbuf(&pp->buffered_output, &pp->children[i].err);
1632 + strbuf_reset(&pp->children[i].err);
1633 + }
1634 return 1;
1635 }
1626 - pp->children[i].process.err = -1;
1627 - pp->children[i].process.stdout_to_stderr = 1;
1636 + if (!pp->ungroup) {
1637 + pp->children[i].process.err = -1;
1638 + pp->children[i].process.stdout_to_stderr = 1;
1639 + }
1640 pp->children[i].process.no_stdin = 1;
1641
1642 if (start_command(&pp->children[i].process)) {
1631 - code = pp->start_failure(&pp->children[i].err,
1643 + code = pp->start_failure(pp->ungroup ? NULL :
1644 + &pp->children[i].err,
1645 pp->data,
1646 pp->children[i].data);
1634 - strbuf_addbuf(&pp->buffered_output, &pp->children[i].err);
1635 - strbuf_reset(&pp->children[i].err);
1647 + if (!pp->ungroup) {
1648 + strbuf_addbuf(&pp->buffered_output, &pp->children[i].err);
1649 + strbuf_reset(&pp->children[i].err);
1650 + }
1651 if (code)
1652 pp->shutdown = 1;
1653 return code;
@@ -1640,7 +1655,8 @@ static int pp_start_one(struct parallel_processes *pp)
1655
1656 pp->nr_processes++;
1657 pp->children[i].state = GIT_CP_WORKING;
1643 - pp->pfd[i].fd = pp->children[i].process.err;
1658 + if (pp->pfd)
1659 + pp->pfd[i].fd = pp->children[i].process.err;
1660 return 0;
1661 }
1662
@@ -1674,6 +1690,7 @@ static void pp_buffer_stderr(struct parallel_processes *pp, int output_timeout)
1690 static void pp_output(struct parallel_processes *pp)
1691 {
1692 int i = pp->output_owner;
1693 +
1694 if (pp->children[i].state == GIT_CP_WORKING &&
1695 pp->children[i].err.len) {
1696 strbuf_write(&pp->children[i].err, stderr);
@@ -1696,7 +1713,7 @@ static int pp_collect_finished(struct parallel_processes *pp)
1713
1714 code = finish_command(&pp->children[i].process);
1715
1699 - code = pp->task_finished(code,
1716 + code = pp->task_finished(code, pp->ungroup ? NULL :
1717 &pp->children[i].err, pp->data,
1718 pp->children[i].data);
1719
@@ -1707,10 +1724,13 @@ static int pp_collect_finished(struct parallel_processes *pp)
1724
1725 pp->nr_processes--;
1726 pp->children[i].state = GIT_CP_FREE;
1710 - pp->pfd[i].fd = -1;
1727 + if (pp->pfd)
1728 + pp->pfd[i].fd = -1;
1729 child_process_init(&pp->children[i].process);
1730
1713 - if (i != pp->output_owner) {
1731 + if (pp->ungroup) {
1732 + ; /* no strbuf_*() work to do here */
1733 + } else if (i != pp->output_owner) {
1734 strbuf_addbuf(&pp->buffered_output, &pp->children[i].err);
1735 strbuf_reset(&pp->children[i].err);
1736 } else {
@@ -1747,9 +1767,14 @@ int run_processes_parallel(int n,
1767 int i, code;
1768 int output_timeout = 100;
1769 int spawn_cap = 4;
1770 + int ungroup = run_processes_parallel_ungroup;
1771 struct parallel_processes pp;
1772
1752 - pp_init(&pp, n, get_next_task, start_failure, task_finished, pp_cb);
1773 + /* unset for the next API user */
1774 + run_processes_parallel_ungroup = 0;
1775 +
1776 + pp_init(&pp, n, get_next_task, start_failure, task_finished, pp_cb,
1777 + ungroup);
1778 while (1) {
1779 for (i = 0;
1780 i < spawn_cap && !pp.shutdown &&
@@ -1766,8 +1791,15 @@ int run_processes_parallel(int n,
1791 }
1792 if (!pp.nr_processes)
1793 break;
1769 - pp_buffer_stderr(&pp, output_timeout);
1770 - pp_output(&pp);
1794 + if (ungroup) {
1795 + int i;
1796 +
1797 + for (i = 0; i < pp.max_processes; i++)
1798 + pp.children[i].state = GIT_CP_WAIT_CLEANUP;
1799 + } else {
1800 + pp_buffer_stderr(&pp, output_timeout);
1801 + pp_output(&pp);
1802 + }
1803 code = pp_collect_finished(&pp);
1804 if (code) {
1805 pp.shutdown = 1;
run-command.h
+23 -7
@@ -406,6 +406,9 @@ void check_pipe(int err);
406 * pp_cb is the callback cookie as passed to run_processes_parallel.
407 * You can store a child process specific callback cookie in pp_task_cb.
408 *
409 + * See run_processes_parallel() below for a discussion of the "struct
410 + * strbuf *out" parameter.
411 + *
412 * Even after returning 0 to indicate that there are no more processes,
413 * this function will be called again until there are no more running
414 * child processes.
@@ -424,9 +427,8 @@ typedef int (*get_next_task_fn)(struct child_process *cp,
427 * This callback is called whenever there are problems starting
428 * a new process.
429 *
427 - * You must not write to stdout or stderr in this function. Add your
428 - * message to the strbuf out instead, which will be printed without
429 - * messing up the output of the other parallel processes.
430 + * See run_processes_parallel() below for a discussion of the "struct
431 + * strbuf *out" parameter.
432 *
433 * pp_cb is the callback cookie as passed into run_processes_parallel,
434 * pp_task_cb is the callback cookie as passed into get_next_task_fn.
@@ -442,9 +444,8 @@ typedef int (*start_failure_fn)(struct strbuf *out,
444 /**
445 * This callback is called on every child process that finished processing.
446 *
445 - * You must not write to stdout or stderr in this function. Add your
446 - * message to the strbuf out instead, which will be printed without
447 - * messing up the output of the other parallel processes.
447 + * See run_processes_parallel() below for a discussion of the "struct
448 + * strbuf *out" parameter.
449 *
450 * pp_cb is the callback cookie as passed into run_processes_parallel,
451 * pp_task_cb is the callback cookie as passed into get_next_task_fn.
@@ -465,11 +466,26 @@ typedef int (*task_finished_fn)(int result,
466 *
467 * The children started via this function run in parallel. Their output
468 * (both stdout and stderr) is routed to stderr in a manner that output
468 - * from different tasks does not interleave.
469 + * from different tasks does not interleave (but see "ungroup" below).
470 *
471 * start_failure_fn and task_finished_fn can be NULL to omit any
472 * special handling.
473 + *
474 + * If the "ungroup" option isn't specified, the API will set the
475 + * "stdout_to_stderr" parameter in "struct child_process" and provide
476 + * the callbacks with a "struct strbuf *out" parameter to write output
477 + * to. In this case the callbacks must not write to stdout or
478 + * stderr as such output will mess up the output of the other parallel
479 + * processes. If "ungroup" option is specified callbacks will get a
480 + * NULL "struct strbuf *out" parameter, and are responsible for
481 + * emitting their own output, including dealing with any race
482 + * conditions due to writing in parallel to stdout and stderr.
483 + * The "ungroup" option can be enabled by setting the global
484 + * "run_processes_parallel_ungroup" to "1" before invoking
485 + * run_processes_parallel(), it will be set back to "0" as soon as the
486 + * API reads that setting.
487 */
488 +extern int run_processes_parallel_ungroup;
489 int run_processes_parallel(int n,
490 get_next_task_fn,
491 start_failure_fn,
t/helper/test-run-command.c
+19 -3
@@ -31,7 +31,11 @@ static int parallel_next(struct child_process *cp,
31 return 0;
32
33 strvec_pushv(&cp->args, d->args.v);
34 - strbuf_addstr(err, "preloaded output of a child\n");
34 + if (err)
35 + strbuf_addstr(err, "preloaded output of a child\n");
36 + else
37 + fprintf(stderr, "preloaded output of a child\n");
38 +
39 number_callbacks++;
40 return 1;
41 }
@@ -41,7 +45,10 @@ static int no_job(struct child_process *cp,
45 void *cb,
46 void **task_cb)
47 {
44 - strbuf_addstr(err, "no further jobs available\n");
48 + if (err)
49 + strbuf_addstr(err, "no further jobs available\n");
50 + else
51 + fprintf(stderr, "no further jobs available\n");
52 return 0;
53 }
54
@@ -50,7 +57,10 @@ static int task_finished(int result,
57 void *pp_cb,
58 void *pp_task_cb)
59 {
53 - strbuf_addstr(err, "asking for a quick stop\n");
60 + if (err)
61 + strbuf_addstr(err, "asking for a quick stop\n");
62 + else
63 + fprintf(stderr, "asking for a quick stop\n");
64 return 1;
65 }
66
@@ -407,6 +417,12 @@ int cmd__run_command(int argc, const char **argv)
417 if (!strcmp(argv[1], "run-command"))
418 exit(run_command(&proc));
419
420 + if (!strcmp(argv[1], "--ungroup")) {
421 + argv += 1;
422 + argc -= 1;
423 + run_processes_parallel_ungroup = 1;
424 + }
425 +
426 jobs = atoi(argv[2]);
427 strvec_clear(&proc.args);
428 strvec_pushv(&proc.args, (const char **)argv + 3);
t/t0061-run-command.sh
+30
@@ -134,16 +134,34 @@ test_expect_success 'run_command runs in parallel with more jobs available than
134 test_cmp expect actual
135 '
136
137 +test_expect_success 'run_command runs ungrouped in parallel with more jobs available than tasks' '
138 + test-tool run-command --ungroup run-command-parallel 5 sh -c "printf \"%s\n%s\n\" Hello World" >out 2>err &&
139 + test_line_count = 8 out &&
140 + test_line_count = 4 err
141 +'
142 +
143 test_expect_success 'run_command runs in parallel with as many jobs as tasks' '
144 test-tool run-command run-command-parallel 4 sh -c "printf \"%s\n%s\n\" Hello World" 2>actual &&
145 test_cmp expect actual
146 '
147
148 +test_expect_success 'run_command runs ungrouped in parallel with as many jobs as tasks' '
149 + test-tool run-command --ungroup run-command-parallel 4 sh -c "printf \"%s\n%s\n\" Hello World" >out 2>err &&
150 + test_line_count = 8 out &&
151 + test_line_count = 4 err
152 +'
153 +
154 test_expect_success 'run_command runs in parallel with more tasks than jobs available' '
155 test-tool run-command run-command-parallel 3 sh -c "printf \"%s\n%s\n\" Hello World" 2>actual &&
156 test_cmp expect actual
157 '
158
159 +test_expect_success 'run_command runs ungrouped in parallel with more tasks than jobs available' '
160 + test-tool run-command --ungroup run-command-parallel 3 sh -c "printf \"%s\n%s\n\" Hello World" >out 2>err &&
161 + test_line_count = 8 out &&
162 + test_line_count = 4 err
163 +'
164 +
165 cat >expect <<-EOF
166 preloaded output of a child
167 asking for a quick stop
@@ -158,6 +176,12 @@ test_expect_success 'run_command is asked to abort gracefully' '
176 test_cmp expect actual
177 '
178
179 +test_expect_success 'run_command is asked to abort gracefully (ungroup)' '
180 + test-tool run-command --ungroup run-command-abort 3 false >out 2>err &&
181 + test_must_be_empty out &&
182 + test_line_count = 6 err
183 +'
184 +
185 cat >expect <<-EOF
186 no further jobs available
187 EOF
@@ -167,6 +191,12 @@ test_expect_success 'run_command outputs ' '
191 test_cmp expect actual
192 '
193
194 +test_expect_success 'run_command outputs (ungroup) ' '
195 + test-tool run-command --ungroup run-command-no-jobs 3 sh -c "printf \"%s\n%s\n\" Hello World" >out 2>err &&
196 + test_must_be_empty out &&
197 + test_cmp expect err
198 +'
199 +
200 test_trace () {
201 expect="$1"
202 shift