git/list[1] front-page[2] threads[3] people[4] search[5] about
 

[PATCH v7 14/17] run-command: add stdin callback for parallelization

From
Emily Shaffer <emilyshaffer@google.com>
Date
Dec 22, 2020, 00:02 UTC
Message-ID
<20201222000220.1491091-15-emilyshaffer@google.com>
In-Reply-To
<20201222000220.1491091-1-emilyshaffer@google.com>

If a user of the run_processes_parallel() API wants to pipe a large amount of information to stdin of each parallel command, that information could exceed the buffer of the pipe allocated for that process's stdin. Generally this is solved by repeatedly writing to child_process.in between calls to start_command() and finish_command(); run_processes_parallel() did not provide users an opportunity to access child_process at that time.

Because the data might be extremely large (for example, a list of all refs received during a push from a client) simply taking a string_list or strbuf is not as scalable as using a callback; the rest of the run_processes_parallel() API also uses callbacks, so making this feature match the rest of the API reduces mental load on the user.

Signed-off-by: Emily Shaffer <emilyshaffer@google.com>
---
 builtin/fetch.c             |  1 +
 builtin/submodule--helper.c |  2 +-
 hook.c                      |  1 +
 run-command.c               | 54 +++++++++++++++++++++++++++++++++++--
 run-command.h               | 17 +++++++++++-
 submodule.c                 |  1 +
 t/helper/test-run-command.c | 31 ++++++++++++++++++---
 t/t0061-run-command.sh      | 30 +++++++++++++++++++++
 8 files changed, 129 insertions(+), 8 deletions(-)
diff --git a/builtin/fetch.c b/builtin/fetch.c
index ecf8537605..5e153b5193 100644
--- a/builtin/fetch.c
+++ b/builtin/fetch.c
@@ -1647,6 +1647,7 @@ static int fetch_multiple(struct string_list *list, int max_children)
 		result = run_processes_parallel_tr2(max_children,
 						    &fetch_next_remote,
 						    &fetch_failed_to_start,
+						    NULL,
 						    &fetch_finished,
 						    &state,
 						    "fetch", "parallel/fetch");
diff --git a/builtin/submodule--helper.c b/builtin/submodule--helper.c
index c30896c897..bb623c1852 100644
--- a/builtin/submodule--helper.c
+++ b/builtin/submodule--helper.c
@@ -2294,7 +2294,7 @@ static int update_submodules(struct submodule_update_clone *suc)
 	int i;
 
 	run_processes_parallel_tr2(suc->max_jobs, update_clone_get_next_task,
-				   update_clone_start_failure,
+				   update_clone_start_failure, NULL,
 				   update_clone_task_finished, suc, "submodule",
 				   "parallel/update");
 
diff --git a/hook.c b/hook.c
index eea90ec1d0..312ede1251 100644
--- a/hook.c
+++ b/hook.c
@@ -368,6 +368,7 @@ int run_hooks(const char *hookname, struct run_hooks_opt *options)
 	run_processes_parallel_tr2(options->jobs,
 				   pick_next_hook,
 				   notify_start_failure,
+				   NULL,
 				   notify_hook_finished,
 				   &cb_data,
 				   "hook",
diff --git a/run-command.c b/run-command.c
index 80c8c97bc1..7b65c087f8 100644
--- a/run-command.c
+++ b/run-command.c
@@ -1548,6 +1548,7 @@ struct parallel_processes {
 
 	get_next_task_fn get_next_task;
 	start_failure_fn start_failure;
+	feed_pipe_fn feed_pipe;
 	task_finished_fn task_finished;
 
 	struct {
@@ -1575,6 +1576,13 @@ static int default_start_failure(struct strbuf *out,
 	return 0;
 }
 
+static int default_feed_pipe(struct strbuf *pipe,
+			     void *pp_cb,
+			     void *pp_task_cb)
+{
+	return 1;
+}
+
 static int default_task_finished(int result,
 				 struct strbuf *out,
 				 void *pp_cb,
@@ -1605,6 +1613,7 @@ static void pp_init(struct parallel_processes *pp,
 		    int n,
 		    get_next_task_fn get_next_task,
 		    start_failure_fn start_failure,
+		    feed_pipe_fn feed_pipe,
 		    task_finished_fn task_finished,
 		    void *data)
 {
@@ -1623,6 +1632,7 @@ static void pp_init(struct parallel_processes *pp,
 	pp->get_next_task = get_next_task;
 
 	pp->start_failure = start_failure ? start_failure : default_start_failure;
+	pp->feed_pipe = feed_pipe ? feed_pipe : default_feed_pipe;
 	pp->task_finished = task_finished ? task_finished : default_task_finished;
 
 	pp->nr_processes = 0;
@@ -1715,6 +1725,37 @@ static int pp_start_one(struct parallel_processes *pp)
 	return 0;
 }
 
+static void pp_buffer_stdin(struct parallel_processes *pp)
+{
+	int i;
+	struct strbuf sb = STRBUF_INIT;
+
+	/* Buffer stdin for each pipe. */
+	for (i = 0; i < pp->max_processes; i++) {
+		if (pp->children[i].state == GIT_CP_WORKING &&
+		    pp->children[i].process.in > 0) {
+			int done;
+			strbuf_reset(&sb);
+			done = pp->feed_pipe(&sb, pp->data,
+					      pp->children[i].data);
+			if (sb.len) {
+				if (write_in_full(pp->children[i].process.in,
+					      sb.buf, sb.len) < 0) {
+					if (errno != EPIPE)
+						die_errno("write");
+					done = 1;
+				}
+			}
+			if (done) {
+				close(pp->children[i].process.in);
+				pp->children[i].process.in = 0;
+			}
+		}
+	}
+
+	strbuf_release(&sb);
+}
+
 static void pp_buffer_stderr(struct parallel_processes *pp, int output_timeout)
 {
 	int i;
@@ -1779,6 +1820,7 @@ static int pp_collect_finished(struct parallel_processes *pp)
 		pp->nr_processes--;
 		pp->children[i].state = GIT_CP_FREE;
 		pp->pfd[i].fd = -1;
+		pp->children[i].process.in = 0;
 		child_process_init(&pp->children[i].process);
 
 		if (i != pp->output_owner) {
@@ -1812,6 +1854,7 @@ static int pp_collect_finished(struct parallel_processes *pp)
 int run_processes_parallel(int n,
 			   get_next_task_fn get_next_task,
 			   start_failure_fn start_failure,
+			   feed_pipe_fn feed_pipe,
 			   task_finished_fn task_finished,
 			   void *pp_cb)
 {
@@ -1820,7 +1863,9 @@ int run_processes_parallel(int n,
 	int spawn_cap = 4;
 	struct parallel_processes pp;
 
-	pp_init(&pp, n, get_next_task, start_failure, task_finished, pp_cb);
+	sigchain_push(SIGPIPE, SIG_IGN);
+
+	pp_init(&pp, n, get_next_task, start_failure, feed_pipe, task_finished, pp_cb);
 	while (1) {
 		for (i = 0;
 		    i < spawn_cap && !pp.shutdown &&
@@ -1837,6 +1882,7 @@ int run_processes_parallel(int n,
 		}
 		if (!pp.nr_processes)
 			break;
+		pp_buffer_stdin(&pp);
 		pp_buffer_stderr(&pp, output_timeout);
 		pp_output(&pp);
 		code = pp_collect_finished(&pp);
@@ -1848,11 +1894,15 @@ int run_processes_parallel(int n,
 	}
 
 	pp_cleanup(&pp);
+
+	sigchain_pop(SIGPIPE);
+
 	return 0;
 }
 
 int run_processes_parallel_tr2(int n, get_next_task_fn get_next_task,
 			       start_failure_fn start_failure,
+			       feed_pipe_fn feed_pipe,
 			       task_finished_fn task_finished, void *pp_cb,
 			       const char *tr2_category, const char *tr2_label)
 {
@@ -1862,7 +1912,7 @@ int run_processes_parallel_tr2(int n, get_next_task_fn get_next_task,
 				   ((n < 1) ? online_cpus() : n));
 
 	result = run_processes_parallel(n, get_next_task, start_failure,
-					task_finished, pp_cb);
+					feed_pipe, task_finished, pp_cb);
 
 	trace2_region_leave(tr2_category, tr2_label, NULL);
 
diff --git a/run-command.h b/run-command.h
index 6472b38bde..e058c0e2c8 100644
--- a/run-command.h
+++ b/run-command.h
@@ -436,6 +436,20 @@ typedef int (*start_failure_fn)(struct strbuf *out,
 				void *pp_cb,
 				void *pp_task_cb);
 
+/**
+ * This callback is called repeatedly on every child process who requests
+ * start_command() to create a pipe by setting child_process.in < 0.
+ *
+ * pp_cb is the callback cookie as passed into run_processes_parallel, and
+ * pp_task_cb is the callback cookie as passed into get_next_task_fn.
+ * The contents of 'send' will be read into the pipe and passed to the pipe.
+ *
+ * Return nonzero to close the pipe.
+ */
+typedef int (*feed_pipe_fn)(struct strbuf *pipe,
+			    void *pp_cb,
+			    void *pp_task_cb);
+
 /**
  * This callback is called on every child process that finished processing.
  *
@@ -470,10 +484,11 @@ typedef int (*task_finished_fn)(int result,
 int run_processes_parallel(int n,
 			   get_next_task_fn,
 			   start_failure_fn,
+			   feed_pipe_fn,
 			   task_finished_fn,
 			   void *pp_cb);
 int run_processes_parallel_tr2(int n, get_next_task_fn, start_failure_fn,
-			       task_finished_fn, void *pp_cb,
+			       feed_pipe_fn, task_finished_fn, void *pp_cb,
 			       const char *tr2_category, const char *tr2_label);
 
 #endif
diff --git a/submodule.c b/submodule.c
index b3bb59f066..953f41818c 100644
--- a/submodule.c
+++ b/submodule.c
@@ -1638,6 +1638,7 @@ int fetch_populated_submodules(struct repository *r,
 	run_processes_parallel_tr2(max_parallel_jobs,
 				   get_next_submodule,
 				   fetch_start_failure,
+				   NULL,
 				   fetch_finish,
 				   &spf,
 				   "submodule", "parallel/fetch");
diff --git a/t/helper/test-run-command.c b/t/helper/test-run-command.c
index 7ae03dc712..9348184d30 100644
--- a/t/helper/test-run-command.c
+++ b/t/helper/test-run-command.c
@@ -32,8 +32,13 @@ static int parallel_next(struct child_process *cp,
 		return 0;
 
 	strvec_pushv(&cp->args, d->argv);
+	cp->in = d->in;
+	cp->no_stdin = d->no_stdin;
 	strbuf_addstr(err, "preloaded output of a child\n");
 	number_callbacks++;
+
+	*task_cb = xmalloc(sizeof(int));
+	*(int*)(*task_cb) = 2;
 	return 1;
 }
 
@@ -55,6 +60,17 @@ static int task_finished(int result,
 	return 1;
 }
 
+static int test_stdin(struct strbuf *pipe, void *cb, void *task_cb)
+{
+	int *lines_remaining = task_cb;
+
+	if (*lines_remaining)
+		strbuf_addf(pipe, "sample stdin %d\n", --(*lines_remaining));
+
+	return !(*lines_remaining);
+}
+
+
 struct testsuite {
 	struct string_list tests, failed;
 	int next;
@@ -185,7 +201,7 @@ static int testsuite(int argc, const char **argv)
 		suite.tests.nr, max_jobs);
 
 	ret = run_processes_parallel(max_jobs, next_test, test_failed,
-				     test_finished, &suite);
+				     test_stdin, test_finished, &suite);
 
 	if (suite.failed.nr > 0) {
 		ret = 1;
@@ -413,15 +429,22 @@ int cmd__run_command(int argc, const char **argv)
 
 	if (!strcmp(argv[1], "run-command-parallel"))
 		exit(run_processes_parallel(jobs, parallel_next,
-					    NULL, NULL, &proc));
+					    NULL, NULL, NULL, &proc));
 
 	if (!strcmp(argv[1], "run-command-abort"))
 		exit(run_processes_parallel(jobs, parallel_next,
-					    NULL, task_finished, &proc));
+					    NULL, NULL, task_finished, &proc));
 
 	if (!strcmp(argv[1], "run-command-no-jobs"))
 		exit(run_processes_parallel(jobs, no_job,
-					    NULL, task_finished, &proc));
+					    NULL, NULL, task_finished, &proc));
+
+	if (!strcmp(argv[1], "run-command-stdin")) {
+		proc.in = -1;
+		proc.no_stdin = 0;
+		exit (run_processes_parallel(jobs, parallel_next, NULL,
+					     test_stdin, NULL, &proc));
+	}
 
 	fprintf(stderr, "check usage\n");
 	return 1;
diff --git a/t/t0061-run-command.sh b/t/t0061-run-command.sh
index 7d599675e3..87759482ad 100755
--- a/t/t0061-run-command.sh
+++ b/t/t0061-run-command.sh
@@ -143,6 +143,36 @@ test_expect_success 'run_command runs in parallel with more tasks than jobs avai
 	test_cmp expect actual
 '
 
+cat >expect <<-EOF
+preloaded output of a child
+listening for stdin:
+sample stdin 1
+sample stdin 0
+preloaded output of a child
+listening for stdin:
+sample stdin 1
+sample stdin 0
+preloaded output of a child
+listening for stdin:
+sample stdin 1
+sample stdin 0
+preloaded output of a child
+listening for stdin:
+sample stdin 1
+sample stdin 0
+EOF
+
+test_expect_success 'run_command listens to stdin' '
+	write_script stdin-script <<-\EOF &&
+	echo "listening for stdin:"
+	while read line; do
+		echo "$line"
+	done
+	EOF
+	test-tool run-command run-command-stdin 2 ./stdin-script 2>actual &&
+	test_cmp expect actual
+'
+
 cat >expect <<-EOF
 preloaded output of a child
 asking for a quick stop

-- 
2.28.0.rc0.142.g3c755180ce-goog
Previous: Emily ShafferNext: Jonathan Tan
Message 63 of 89 in “propose config-based hooks (part I)”
  1. 00/17 propose config-based hooks (part I)Emily Shaffer, Dec 5, 2020
  2. 01/17 doc: propose hooks managed by the configEmily Shaffer, Dec 5, 2020
  3. 02/17 hook: scaffolding for git-hook subcommandEmily Shaffer, Dec 5, 2020
  4. 03/17 hook: add list commandEmily Shaffer, Dec 5, 2020
  5. 04/17 hook: include hookdir hook in listEmily Shaffer, Dec 5, 2020
  6. 06/17 hook: implement hookcmd.<name>.skipEmily Shaffer, Dec 5, 2020
  7. 05/17 hook: respect hook.runHookDirEmily Shaffer, Dec 5, 2020
  8. 08/17 hook: add 'run' subcommandEmily Shaffer, Dec 5, 2020
  9. Phillip WoodDec 11, 2020
  10. Emily ShafferDec 15, 2020
  11. 09/17 hook: replace find_hook() with hook_exists()Emily Shaffer, Dec 5, 2020
  12. 07/17 parse-options: parse into strvecEmily Shaffer, Dec 5, 2020
  13. 11/17 run-command: allow stdin for run_processes_parallelEmily Shaffer, Dec 5, 2020
  14. 10/17 hook: support passing stdin to hooksEmily Shaffer, Dec 5, 2020
  15. 12/17 hook: allow parallel hook executionEmily Shaffer, Dec 5, 2020
  16. 14/17 run-command: add stdin callback for parallelizationEmily Shaffer, Dec 5, 2020
  17. 13/17 hook: allow specifying working directory for hooksEmily Shaffer, Dec 5, 2020
  18. 17/17 hooks: allow callers to capture outputEmily Shaffer, Dec 5, 2020
  19. 15/17 hook: provide stdin by string_list or callbackEmily Shaffer, Dec 5, 2020
  20. SZEDER GáborDec 8, 2020
  21. Emily ShafferDec 8, 2020
  22. 16/17 run-command: allow capturing of collated outputEmily Shaffer, Dec 5, 2020
  23. Josh SteadmonDec 16, 2020
  24. Junio C HamanoDec 16, 2020
  25. Emily ShafferDec 16, 2020
  26. Junio C HamanoDec 16, 2020
  27. Emily ShafferDec 18, 2020
  28. Junio C HamanoDec 18, 2020
  29. 00/17 propose config-based hooks (part I)Emily Shaffer, Dec 22, 2020
  30. 01/17 doc: propose hooks managed by the configEmily Shaffer, Dec 22, 2020
  31. Ævar Arnfjörð BjarmasonJan 23, 2021
  32. Emily ShafferJan 29, 2021
  33. Junio C HamanoFeb 1, 2021
  34. Emily ShafferMar 10, 2021
  35. 02/17 hook: scaffolding for git-hook subcommandEmily Shaffer, Dec 22, 2020
  36. 03/17 hook: add list commandEmily Shaffer, Dec 22, 2020
  37. Jonathan TanJan 31, 2021
  38. Emily ShafferFeb 9, 2021
  39. 04/17 hook: include hookdir hook in listEmily Shaffer, Dec 22, 2020
  40. Jonathan TanJan 31, 2021
  41. Emily ShafferFeb 9, 2021
  42. 05/17 hook: respect hook.runHookDirEmily Shaffer, Dec 22, 2020
  43. Jonathan TanJan 31, 2021
  44. Emily ShafferFeb 9, 2021
  45. 06/17 hook: implement hookcmd.<name>.skipEmily Shaffer, Dec 22, 2020
  46. Jonathan TanJan 31, 2021
  47. Emily ShafferFeb 9, 2021
  48. 08/17 hook: add 'run' subcommandEmily Shaffer, Dec 22, 2020
  49. Jonathan TanJan 31, 2021
  50. Emily ShafferFeb 11, 2021
  51. 07/17 parse-options: parse into strvecEmily Shaffer, Dec 22, 2020
  52. 09/17 hook: replace find_hook() with hook_exists()Emily Shaffer, Dec 22, 2020
  53. Jonathan TanJan 31, 2021
  54. Emily ShafferFeb 12, 2021
  55. Emily ShafferFeb 18, 2021
  56. 10/17 hook: support passing stdin to hooksEmily Shaffer, Dec 22, 2020
  57. 12/17 hook: allow parallel hook executionEmily Shaffer, Dec 22, 2020
  58. Jonathan TanFeb 1, 2021
  59. Emily ShafferFeb 22, 2021
  60. 11/17 run-command: allow stdin for run_processes_parallelEmily Shaffer, Dec 22, 2020
  61. Jonathan TanFeb 1, 2021
  62. Emily ShafferFeb 19, 2021
  63. 14/17 run-command: add stdin callback for parallelizationEmily Shaffer, Dec 22, 2020
  64. Jonathan TanFeb 1, 2021
  65. Emily ShafferFeb 22, 2021
  66. Jonathan TanFeb 23, 2021
  67. Emily ShafferMar 10, 2021
  68. 13/17 hook: allow specifying working directory for hooksEmily Shaffer, Dec 22, 2020
  69. 15/17 hook: provide stdin by string_list or callbackEmily Shaffer, Dec 22, 2020
  70. Jonathan TanFeb 1, 2021
  71. Emily ShafferFeb 23, 2021
  72. Jonathan TanFeb 25, 2021
  73. Emily ShafferMar 2, 2021
  74. Jonathan TanMar 2, 2021
  75. 16/17 run-command: allow capturing of collated outputEmily Shaffer, Dec 22, 2020
  76. 17/17 hooks: allow callers to capture outputEmily Shaffer, Dec 22, 2020
  77. Junio C HamanoDec 22, 2020
  78. Emily ShafferDec 28, 2020
  79. Junio C HamanoDec 28, 2020
  80. 18/17 doc: make git-hook.txt point of truthEmily Shaffer, Dec 28, 2020
  81. Emily ShafferDec 28, 2020
  82. Emily ShafferJan 29, 2021
  83. Josh SteadmonFeb 16, 2021
  84. Junio C HamanoFeb 16, 2021
  85. Josh SteadmonFeb 17, 2021
  86. Junio C HamanoFeb 17, 2021
  87. Junio C HamanoFeb 25, 2021
  88. Emily ShafferMar 1, 2021
  89. Junio C HamanoMar 1, 2021

Read the whole thread, see it on lore, or plain text.

$ cat FOOTERMessages come from the public archive at lore.kernel.org/git, fetched every hour. The front page is chosen and written each morning by an AI editor and can be wrong; the threads themselves are the record. About and API. For agents: an MCP server at https://gitlist.dev/mcp, and any thread, story or person page as Markdown by adding .md to its URL (or sending Accept: text/markdown). Details in /llms.txt.