{"thread":{"id":"40335","subject":"[PATCH 1/2] Sending \"Fetching submodule \" output to stderr","startedAt":"2015-09-11T23:09:52Z","lastAt":"2015-09-14T22:06:22Z","messageCount":13,"participants":["Stefan Beller","Junio C Hamano","Jeff King","Jonathan Nieder"],"isPatch":true,"patchVersion":1,"patchTotal":2},"messages":[{"id":"269805","messageId":"1442012994-20374-1-git-send-email-sbeller@google.com","threadId":"40335","inReplyTo":null,"subject":"[RFC PATCHv1 0/2] Parallel git submodule fetching","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-09-11T23:09:52Z","receivedAt":"2015-09-11T23:09:52Z","isPatch":false,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"This is a new approach to making git submodules faster without using threads.\nWe'll directly run commands in parallel without an intermediate thread pool.\n\npatch 1 is fixing an error, which was hard to come by in a parallel world,\nbut is not much of a problem in a single-childed world.\n\nThe second patch works fine, (I'd love to hear feedback on the percieved\nchange of output) though it has still two edges I want to tackle:\n\n* The fcntl call to make the pipes nonblocking is not checked for errors.\n* We also need to communicate back if one process cannot spawn or such,\n i.e. the reporting for all subprocesses combined needs to added. I think\n about adding another callback which is called to finish up a child process.\n\nEarlier attempts used a threading pool, which made the conversion very\neasy as you don't need to touch lots of existing code as you would just\nreplace the run_command(..) call by an add_task(..). The threading pool\nwould abstract away the whole parallelism and its problems. It would also\nhave the backpressure to the main program, add_task would just not\nreturn if there was no place left in the queue.\n\n[1] http://www.spinics.net/lists/git/msg258435.html\n\nAny feedback welcome!\nThanks,\nStefan\n\nJonathan Nieder (1):\n Sending \"Fetching submodule \" output to stderr\n\nStefan Beller (1):\n fetch: fetch submodules in parallel\n\n Documentation/fetch-options.txt | 7 ++\n builtin/fetch.c | 6 +-\n builtin/pull.c | 6 ++\n run-command.c | 144 ++++++++++++++++++++++++++++++++++++++++\n run-command.h | 29 ++++++++\n strbuf.c | 31 +++++++++\n strbuf.h | 1 +\n submodule.c | 99 +++++++++++++++++----------\n submodule.h | 2 +-\n t/t0061-run-command.sh | 16 +++++\n t/t5526-fetch-submodules.sh | 70 ++++++++++++-------\n test-run-command.c | 23 +++++++\n 12 files changed, 373 insertions(+), 61 deletions(-)\n\n-- \n2.6.0.rc0.131.gf624c3d\n"},{"id":"269804","messageId":"1442012994-20374-2-git-send-email-sbeller@google.com","threadId":"40335","inReplyTo":"1442012994-20374-1-git-send-email-sbeller@google.com","subject":"[PATCH 1/2] Sending \"Fetching submodule \" output to stderr","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-09-11T23:09:53Z","receivedAt":"2015-09-11T23:09:53Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"From: Jonathan Nieder \n\nThe \"Pushing submodule \" progress output correctly goes to\nstderr, but \"Fetching submodule \" is going to stdout by mistake.\nFix it to write to stderr.\n\nNoticed while trying to implement a parallel submodule fetch. When\nthis particular output line went to a different file descriptor, it\nwas buffered separately, resulting in wrongly interleaved output if\nwe copied it to the terminal naively.\n\nSigned-off-by: Jonathan Nieder \nSigned-off-by: Stefan Beller \n---\n submodule.c | 2 +-\n t/t5526-fetch-submodules.sh | 51 +++++++++++++++++++++++----------------------\n 2 files changed, 27 insertions(+), 26 deletions(-)\n\ndiff --git a/submodule.c b/submodule.c\nindex 9fcc86f..1d64e57 100644\n--- a/submodule.c\n+++ b/submodule.c\n@@ -694,7 +694,7 @@ int fetch_populated_submodules(const struct argv_array *options,\n \t\t\tgit_dir = submodule_git_dir.buf;\n \t\tif (is_directory(git_dir)) {\n \t\t\tif (!quiet)\n-\t\t\t\tprintf(\"Fetching submodule %s%s\\n\", prefix, ce->name);\n+\t\t\t\tfprintf(stderr, \"Fetching submodule %s%s\\n\", prefix, ce->name);\n \t\t\tcp.dir = submodule_path.buf;\n \t\t\targv_array_push(&argv, default_argv);\n \t\t\targv_array_push(&argv, \"--submodule-prefix\");\ndiff --git a/t/t5526-fetch-submodules.sh b/t/t5526-fetch-submodules.sh\nindex a4532b0..17759b1 100755\n--- a/t/t5526-fetch-submodules.sh\n+++ b/t/t5526-fetch-submodules.sh\n@@ -16,7 +16,8 @@ add_upstream_commit() {\n \t\tgit add subfile &&\n \t\tgit commit -m new subfile &&\n \t\thead2=$(git rev-parse --short HEAD) &&\n-\t\techo \"From $pwd/submodule\" > ../expect.err &&\n+\t\techo \"Fetching submodule submodule\" > ../expect.err &&\n+\t\techo \"From $pwd/submodule\" >> ../expect.err &&\n \t\techo \" $head1..$head2 master -> origin/master\" >> ../expect.err\n \t) &&\n \t(\n@@ -27,6 +28,7 @@ add_upstream_commit() {\n \t\tgit add deepsubfile &&\n \t\tgit commit -m new deepsubfile &&\n \t\thead2=$(git rev-parse --short HEAD) &&\n+\t\techo \"Fetching submodule submodule/subdir/deepsubmodule\" >> ../expect.err\n \t\techo \"From $pwd/deepsubmodule\" >> ../expect.err &&\n \t\techo \" $head1..$head2 master -> origin/master\" >> ../expect.err\n \t)\n@@ -56,9 +58,7 @@ test_expect_success setup '\n \t(\n \t\tcd downstream &&\n \t\tgit submodule update --init --recursive\n-\t) &&\n-\techo \"Fetching submodule submodule\" > expect.out &&\n-\techo \"Fetching submodule submodule/subdir/deepsubmodule\" >> expect.out\n+\t)\n '\n \n test_expect_success \"fetch --recurse-submodules recurses into submodules\" '\n@@ -67,7 +67,7 @@ test_expect_success \"fetch --recurse-submodules recurses into submodules\" '\n \t\tcd downstream &&\n \t\tgit fetch --recurse-submodules >../actual.out 2>../actual.err\n \t) &&\n-\ttest_i18ncmp expect.out actual.out &&\n+\ttest_must_be_empty actual.out &&\n \ttest_i18ncmp expect.err actual.err\n '\n \n@@ -96,7 +96,7 @@ test_expect_success \"using fetchRecurseSubmodules=true in .gitmodules recurses i\n \t\tgit config -f .gitmodules submodule.submodule.fetchRecurseSubmodules true &&\n \t\tgit fetch >../actual.out 2>../actual.err\n \t) &&\n-\ttest_i18ncmp expect.out actual.out &&\n+\ttest_must_be_empty actual.out &&\n \ttest_i18ncmp expect.err actual.err\n '\n \n@@ -127,7 +127,7 @@ test_expect_success \"--recurse-submodules overrides fetchRecurseSubmodules setti\n \t\tgit config --unset -f .gitmodules submodule.submodule.fetchRecurseSubmodules &&\n \t\tgit config --unset submodule.submodule.fetchRecurseSubmodules\n \t) &&\n-\ttest_i18ncmp expect.out actual.out &&\n+\ttest_must_be_empty actual.out &&\n \ttest_i18ncmp expect.err actual.err\n '\n \n@@ -146,7 +146,7 @@ test_expect_success \"--dry-run propagates to submodules\" '\n \t\tcd downstream &&\n \t\tgit fetch --recurse-submodules --dry-run >../actual.out 2>../actual.err\n \t) &&\n-\ttest_i18ncmp expect.out actual.out &&\n+\ttest_must_be_empty actual.out &&\n \ttest_i18ncmp expect.err actual.err\n '\n \n@@ -155,7 +155,7 @@ test_expect_success \"Without --dry-run propagates to submodules\" '\n \t\tcd downstream &&\n \t\tgit fetch --recurse-submodules >../actual.out 2>../actual.err\n \t) &&\n-\ttest_i18ncmp expect.out actual.out &&\n+\ttest_must_be_empty actual.out &&\n \ttest_i18ncmp expect.err actual.err\n '\n \n@@ -166,7 +166,7 @@ test_expect_success \"recurseSubmodules=true propagates into submodules\" '\n \t\tgit config fetch.recurseSubmodules true\n \t\tgit fetch >../actual.out 2>../actual.err\n \t) &&\n-\ttest_i18ncmp expect.out actual.out &&\n+\ttest_must_be_empty actual.out &&\n \ttest_i18ncmp expect.err actual.err\n '\n \n@@ -180,7 +180,7 @@ test_expect_success \"--recurse-submodules overrides config in submodule\" '\n \t\t) &&\n \t\tgit fetch --recurse-submodules >../actual.out 2>../actual.err\n \t) &&\n-\ttest_i18ncmp expect.out actual.out &&\n+\ttest_must_be_empty actual.out &&\n \ttest_i18ncmp expect.err actual.err\n '\n \n@@ -214,16 +214,15 @@ test_expect_success \"Recursion stops when no new submodule commits are fetched\"\n \tgit add submodule &&\n \tgit commit -m \"new submodule\" &&\n \thead2=$(git rev-parse --short HEAD) &&\n-\techo \"Fetching submodule submodule\" > expect.out.sub &&\n \techo \"From $pwd/.\" > expect.err.sub &&\n \techo \" $head1..$head2 master -> origin/master\" >>expect.err.sub &&\n-\thead -2 expect.err >> expect.err.sub &&\n+\thead -3 expect.err >> expect.err.sub &&\n \t(\n \t\tcd downstream &&\n \t\tgit fetch >../actual.out 2>../actual.err\n \t) &&\n \ttest_i18ncmp expect.err.sub actual.err &&\n-\ttest_i18ncmp expect.out.sub actual.out\n+\ttest_must_be_empty actual.out\n '\n \n test_expect_success \"Recursion doesn't happen when new superproject commits don't change any submodules\" '\n@@ -269,7 +268,7 @@ test_expect_success \"Recursion picks up config in submodule\" '\n \t\t)\n \t) &&\n \ttest_i18ncmp expect.err.sub actual.err &&\n-\ttest_i18ncmp expect.out actual.out\n+\ttest_must_be_empty actual.out\n '\n \n test_expect_success \"Recursion picks up all submodules when necessary\" '\n@@ -285,7 +284,8 @@ test_expect_success \"Recursion picks up all submodules when necessary\" '\n \t\tgit add subdir/deepsubmodule &&\n \t\tgit commit -m \"new deepsubmodule\"\n \t\thead2=$(git rev-parse --short HEAD) &&\n-\t\techo \"From $pwd/submodule\" > ../expect.err.sub &&\n+\t\techo \"Fetching submodule submodule\" > ../expect.err.sub &&\n+\t\techo \"From $pwd/submodule\" >> ../expect.err.sub &&\n \t\techo \" $head1..$head2 master -> origin/master\" >> ../expect.err.sub\n \t) &&\n \thead1=$(git rev-parse --short HEAD) &&\n@@ -295,13 +295,13 @@ test_expect_success \"Recursion picks up all submodules when necessary\" '\n \techo \"From $pwd/.\" > expect.err.2 &&\n \techo \" $head1..$head2 master -> origin/master\" >> expect.err.2 &&\n \tcat expect.err.sub >> expect.err.2 &&\n-\ttail -2 expect.err >> expect.err.2 &&\n+\ttail -3 expect.err >> expect.err.2 &&\n \t(\n \t\tcd downstream &&\n \t\tgit fetch >../actual.out 2>../actual.err\n \t) &&\n \ttest_i18ncmp expect.err.2 actual.err &&\n-\ttest_i18ncmp expect.out actual.out\n+\ttest_must_be_empty actual.out\n '\n \n test_expect_success \"'--recurse-submodules=on-demand' doesn't recurse when no new commits are fetched in the superproject (and ignores config)\" '\n@@ -317,7 +317,8 @@ test_expect_success \"'--recurse-submodules=on-demand' doesn't recurse when no ne\n \t\tgit add subdir/deepsubmodule &&\n \t\tgit commit -m \"new deepsubmodule\" &&\n \t\thead2=$(git rev-parse --short HEAD) &&\n-\t\techo \"From $pwd/submodule\" > ../expect.err.sub &&\n+\t\techo Fetching submodule submodule > ../expect.err.sub &&\n+\t\techo \"From $pwd/submodule\" >> ../expect.err.sub &&\n \t\techo \" $head1..$head2 master -> origin/master\" >> ../expect.err.sub\n \t) &&\n \t(\n@@ -335,7 +336,7 @@ test_expect_success \"'--recurse-submodules=on-demand' recurses as deep as necess\n \tgit add submodule &&\n \tgit commit -m \"new submodule\" &&\n \thead2=$(git rev-parse --short HEAD) &&\n-\ttail -2 expect.err > expect.err.deepsub &&\n+\ttail -3 expect.err > expect.err.deepsub &&\n \techo \"From $pwd/.\" > expect.err &&\n \techo \" $head1..$head2 master -> origin/master\" >>expect.err &&\n \tcat expect.err.sub >> expect.err &&\n@@ -354,7 +355,7 @@ test_expect_success \"'--recurse-submodules=on-demand' recurses as deep as necess\n \t\t\tgit config --unset -f .gitmodules submodule.subdir/deepsubmodule.fetchRecursive\n \t\t)\n \t) &&\n-\ttest_i18ncmp expect.out actual.out &&\n+\ttest_must_be_empty actual.out &&\n \ttest_i18ncmp expect.err actual.err\n '\n \n@@ -388,7 +389,7 @@ test_expect_success \"'fetch.recurseSubmodules=on-demand' overrides global config\n \thead2=$(git rev-parse --short HEAD) &&\n \techo \"From $pwd/.\" > expect.err.2 &&\n \techo \" $head1..$head2 master -> origin/master\" >>expect.err.2 &&\n-\thead -2 expect.err >> expect.err.2 &&\n+\thead -3 expect.err >> expect.err.2 &&\n \t(\n \t\tcd downstream &&\n \t\tgit config fetch.recurseSubmodules on-demand &&\n@@ -399,7 +400,7 @@ test_expect_success \"'fetch.recurseSubmodules=on-demand' overrides global config\n \t\tcd downstream &&\n \t\tgit config --unset fetch.recurseSubmodules\n \t) &&\n-\ttest_i18ncmp expect.out.sub actual.out &&\n+\ttest_must_be_empty actual.out &&\n \ttest_i18ncmp expect.err.2 actual.err\n '\n \n@@ -416,7 +417,7 @@ test_expect_success \"'submodule..fetchRecurseSubmodules=on-demand' override\n \thead2=$(git rev-parse --short HEAD) &&\n \techo \"From $pwd/.\" > expect.err.2 &&\n \techo \" $head1..$head2 master -> origin/master\" >>expect.err.2 &&\n-\thead -2 expect.err >> expect.err.2 &&\n+\thead -3 expect.err >> expect.err.2 &&\n \t(\n \t\tcd downstream &&\n \t\tgit config submodule.submodule.fetchRecurseSubmodules on-demand &&\n@@ -427,7 +428,7 @@ test_expect_success \"'submodule..fetchRecurseSubmodules=on-demand' override\n \t\tcd downstream &&\n \t\tgit config --unset submodule.submodule.fetchRecurseSubmodules\n \t) &&\n-\ttest_i18ncmp expect.out.sub actual.out &&\n+\ttest_must_be_empty actual.out &&\n \ttest_i18ncmp expect.err.2 actual.err\n '\n \n-- \n2.6.0.rc0.131.gf624c3d\n"},{"id":"269806","messageId":"1442012994-20374-3-git-send-email-sbeller@google.com","threadId":"40335","inReplyTo":"1442012994-20374-1-git-send-email-sbeller@google.com","subject":"[PATCH 2/2] fetch: fetch submodules in parallel","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-09-11T23:09:54Z","receivedAt":"2015-09-11T23:09:54Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"If we run external commands in parallel we cannot pipe the output directly\nto the our stdout/err as it would mix up. So each process's output will\nflow through a pipe, which we buffer. One subprocess can be directly\npiped to out stdout/err for a low latency feedback to the user.\n\nExample:\nLet's assume we have 5 submodules A,B,C,D,E and each fetch takes a\ndifferent amount of time as the different submodules vary in size, then\nthe output of fetches in sequential order might look like this:\n\n time -->\n output: |---A---| |-B-| |----C-----------| |-D-| |-E-|\n\nWhen we schedule these submodules into maximal two parallel processes,\na schedule and sample output over time may look like this:\n\nthread 1: |---A---| |-D-| |-E-|\n\nthread 2: |-B-| |----C-----------|\n\noutput: |---A---|B|------C-------|DE\n\nSo A will be perceived as it would run normally in the single child\nversion. As B has finished by the time A is done, we can dump its whole\nprogress buffer on stderr, such that it looks like it finished in no time.\nOnce that is done, C is determined to be the visible child and its progress\nwill be reported in real time.\n\nSo this way of output is really good for human consumption,\nas it only changes the timing, not the actual output.\n\nFor machine consumption the output needs to be prepared in\nthe tasks, by either having a prefix per line or per block\nto indicate whose tasks output is displayed.\n\nSigned-off-by: Stefan Beller \n---\n Documentation/fetch-options.txt | 7 ++\n builtin/fetch.c | 6 +-\n builtin/pull.c | 6 ++\n run-command.c | 144 ++++++++++++++++++++++++++++++++++++++++\n run-command.h | 29 ++++++++\n strbuf.c | 31 +++++++++\n strbuf.h | 1 +\n submodule.c | 99 +++++++++++++++++----------\n submodule.h | 2 +-\n t/t0061-run-command.sh | 16 +++++\n t/t5526-fetch-submodules.sh | 19 ++++++\n test-run-command.c | 23 +++++++\n 12 files changed, 347 insertions(+), 36 deletions(-)\n\ndiff --git a/Documentation/fetch-options.txt b/Documentation/fetch-options.txt\nindex 45583d8..e2a59c3 100644\n--- a/Documentation/fetch-options.txt\n+++ b/Documentation/fetch-options.txt\n@@ -100,6 +100,13 @@ ifndef::git-pull[]\n \treference to a commit that isn't already in the local submodule\n \tclone.\n \n+-j::\n+--jobs=::\n+\tNumber of parallel children to be used for fetching submodules.\n+\tEach will fetch from different submodules, such that fetching many\n+\tsubmodules will be faster. By default submodules will be fetched\n+\tone at a time\n+\n --no-recurse-submodules::\n \tDisable recursive fetching of submodules (this has the same effect as\n \tusing the '--recurse-submodules=no' option).\ndiff --git a/builtin/fetch.c b/builtin/fetch.c\nindex ee1f1a9..09ff837 100644\n--- a/builtin/fetch.c\n+++ b/builtin/fetch.c\n@@ -37,6 +37,7 @@ static int prune = -1; /* unspecified */\n static int all, append, dry_run, force, keep, multiple, update_head_ok, verbosity;\n static int progress = -1, recurse_submodules = RECURSE_SUBMODULES_DEFAULT;\n static int tags = TAGS_DEFAULT, unshallow, update_shallow;\n+static int max_children = 1;\n static const char *depth;\n static const char *upload_pack;\n static struct strbuf default_rla = STRBUF_INIT;\n@@ -99,6 +100,8 @@ static struct option builtin_fetch_options[] = {\n \t\t N_(\"fetch all tags and associated objects\"), TAGS_SET),\n \tOPT_SET_INT('n', NULL, &tags,\n \t\t N_(\"do not fetch all tags (--no-tags)\"), TAGS_UNSET),\n+\tOPT_INTEGER('j', \"jobs\", &max_children,\n+\t\t N_(\"number of threads used for fetching\")),\n \tOPT_BOOL('p', \"prune\", &prune,\n \t\t N_(\"prune remote-tracking branches no longer on remote\")),\n \t{ OPTION_CALLBACK, 0, \"recurse-submodules\", NULL, N_(\"on-demand\"),\n@@ -1217,7 +1220,8 @@ int cmd_fetch(int argc, const char **argv, const char *prefix)\n \t\tresult = fetch_populated_submodules(&options,\n \t\t\t\t\t\t submodule_prefix,\n \t\t\t\t\t\t recurse_submodules,\n-\t\t\t\t\t\t verbosity < 0);\n+\t\t\t\t\t\t verbosity < 0,\n+\t\t\t\t\t\t max_children);\n \t\targv_array_clear(&options);\n \t}\n \ndiff --git a/builtin/pull.c b/builtin/pull.c\nindex 722a83c..fbbda67 100644\n--- a/builtin/pull.c\n+++ b/builtin/pull.c\n@@ -94,6 +94,7 @@ static int opt_force;\n static char *opt_tags;\n static char *opt_prune;\n static char *opt_recurse_submodules;\n+static char *max_children;\n static int opt_dry_run;\n static char *opt_keep;\n static char *opt_depth;\n@@ -177,6 +178,9 @@ static struct option pull_options[] = {\n \t\tN_(\"on-demand\"),\n \t\tN_(\"control recursive fetching of submodules\"),\n \t\tPARSE_OPT_OPTARG),\n+\tOPT_PASSTHRU('j', \"jobs\", &max_children, N_(\"n\"),\n+\t\tN_(\"number of threads used for fetching submodules\"),\n+\t\tPARSE_OPT_OPTARG),\n \tOPT_BOOL(0, \"dry-run\", &opt_dry_run,\n \t\tN_(\"dry run\")),\n \tOPT_PASSTHRU('k', \"keep\", &opt_keep, NULL,\n@@ -524,6 +528,8 @@ static int run_fetch(const char *repo, const char **refspecs)\n \t\targv_array_push(&args, opt_prune);\n \tif (opt_recurse_submodules)\n \t\targv_array_push(&args, opt_recurse_submodules);\n+\tif (max_children)\n+\t\targv_array_push(&args, max_children);\n \tif (opt_dry_run)\n \t\targv_array_push(&args, \"--dry-run\");\n \tif (opt_keep)\ndiff --git a/run-command.c b/run-command.c\nindex 28e1d55..b8ff67b 100644\n--- a/run-command.c\n+++ b/run-command.c\n@@ -852,3 +852,147 @@ int capture_command(struct child_process *cmd, struct strbuf *buf, size_t hint)\n \tclose(cmd->out);\n \treturn finish_command(cmd);\n }\n+\n+int run_processes_async(int n, get_next_task fn, void *data)\n+{\n+\tint i, wait_status;\n+\tpid_t pid;\n+\n+\t/* no more tasks. Also set when aborting early. */\n+\tint all_tasks_started = 0;\n+\tint nr_processes = 0;\n+\tint child_in_foreground = 0;\n+\tstruct timeval timeout;\n+\tstruct child_process *children = xcalloc(n, sizeof(*children));\n+\tchar *slots = xcalloc(n, sizeof(*slots));\n+\tstruct strbuf *err = xcalloc(n, sizeof(*err));\n+\tfd_set fdset;\n+\tint maxfd;\n+\tstruct strbuf finished_children = STRBUF_INIT;\n+\tint flags;\n+\tfor (i = 0; i < n; i++)\n+\t\tstrbuf_init(&err[i], 0);\n+\n+\twhile (!all_tasks_started || nr_processes > 0) {\n+\t\t/* Start new processes. */\n+\t\twhile (!all_tasks_started && nr_processes < n) {\n+\t\t\tfor (i = 0; i < n; i++)\n+\t\t\t\tif (!slots[i])\n+\t\t\t\t\tbreak; /* found an empty slot */\n+\t\t\tif (i == n)\n+\t\t\t\tdie(\"BUG: bookkeeping is hard\");\n+\n+\t\t\tif (fn(data, &children[i], &err[i])) {\n+\t\t\t\tall_tasks_started = 1;\n+\t\t\t\tbreak;\n+\t\t\t}\n+\t\t\tif (start_command(&children[i]))\n+\t\t\t\tdie(_(\"Could not start child process\"));\n+\t\t\tflags = fcntl(children[i].err, F_GETFL);\n+\t\t\tfcntl(children[i].err, F_SETFL, flags | O_NONBLOCK);\n+\t\t\tnr_processes++;\n+\t\t\tslots[i] = 1;\n+\t\t}\n+\n+\t\t/* prepare data for select call */\n+\t\tFD_ZERO(&fdset);\n+\t\tmaxfd = 0;\n+\t\tfor (i = 0; i < n; i++) {\n+\t\t\tif (!slots[i])\n+\t\t\t\tcontinue;\n+\t\t\tFD_SET(children[i].err, &fdset);\n+\t\t\tif (children[i].err > maxfd)\n+\t\t\t\tmaxfd = children[i].err;\n+\t\t}\n+\t\ttimeout.tv_sec = 0;\n+\t\ttimeout.tv_usec = 500000;\n+\n+\t\ti = select(maxfd + 1, &fdset, NULL, NULL, &timeout);\n+\t\tif (i < 0) {\n+\t\t\tif (errno == EINTR)\n+\t\t\t\t/* A signal was caught; try again */\n+\t\t\t\tcontinue;\n+\t\t\telse if (errno == ENOMEM)\n+\t\t\t\tdie_errno(\"BUG: keeping track of fds is hard\");\n+\t\t\telse if (errno == EINVAL)\n+\t\t\t\tdie_errno(\"BUG: invalid arguments to select\");\n+\t\t\telse if (errno == EBADF)\n+\t\t\t\tdie_errno(\"BUG: keeping track of fds is hard\");\n+\t\t\telse\n+\t\t\t\tdie_errno(\"Unknown error with select\");\n+\t\t}\n+\n+\t\t/* Buffer output from all pipes. */\n+\t\tfor (i = 0; i < n; i++) {\n+\t\t\tif (!slots[i])\n+\t\t\t\tcontinue;\n+\t\t\tif (FD_ISSET(children[i].err, &fdset))\n+\t\t\t\tstrbuf_read_noblock(&err[i], children[i].err, 0);\n+\t\t\tif (child_in_foreground == i) {\n+\t\t\t\tfputs(err[i].buf, stderr);\n+\t\t\t\tstrbuf_reset(&err[i]);\n+\t\t\t\tfflush(stderr);\n+\t\t\t}\n+\t\t}\n+\n+\t\t/* Collect finished child processes. */\n+\t\twhile (nr_processes > 0) {\n+\t\t\tpid = waitpid(-1, &wait_status, WNOHANG);\n+\t\t\tif (pid == 0)\n+\t\t\t\t/* no child finished */\n+\t\t\t\tbreak;\n+\n+\t\t\tif (pid < 0) {\n+\t\t\t\tif (errno == EINTR)\n+\t\t\t\t\tbreak; /* just try again next time */\n+\t\t\t\tif (errno == EINVAL || errno == ECHILD)\n+\t\t\t\t\tdie_errno(\"wait\");\n+\t\t\t} else {\n+\t\t\t\t/* Find the finished child. */\n+\t\t\t\tfor (i = 0; i < n; i++)\n+\t\t\t\t\tif (slots[i] && pid == children[i].pid)\n+\t\t\t\t\t\tbreak;\n+\t\t\t\tif (i == n)\n+\t\t\t\t\t/* waitpid returned another process id which\n+\t\t\t\t\t * we are not waiting on, so ignore it*/\n+\t\t\t\t\tbreak;\n+\t\t\t}\n+\n+\t\t\tstrbuf_read_noblock(&err[i], children[i].err, 0);\n+\t\t\targv_array_clear(&children[i].args);\n+\t\t\targv_array_clear(&children[i].env_array);\n+\n+\t\t\tslots[i] = 0;\n+\t\t\tnr_processes--;\n+\n+\t\t\tif (i != child_in_foreground) {\n+\t\t\t\tstrbuf_addbuf(&finished_children, &err[i]);\n+\t\t\t\tstrbuf_reset(&err[i]);\n+\t\t\t} else {\n+\t\t\t\tfputs(err[i].buf, stderr);\n+\t\t\t\tstrbuf_reset(&err[i]);\n+\n+\t\t\t\t/* Output all other finished child processes */\n+\t\t\t\tfputs(finished_children.buf, stderr);\n+\t\t\t\tstrbuf_reset(&finished_children);\n+\n+\t\t\t\t/*\n+\t\t\t\t * Pick next process to output live.\n+\t\t\t\t * There can be no active process if n==1\n+\t\t\t\t * NEEDSWORK:\n+\t\t\t\t * For now we pick it randomly by doing a round\n+\t\t\t\t * robin. Later we may want to pick the one with\n+\t\t\t\t * the most output or the longest or shortest\n+\t\t\t\t * running process time.\n+\t\t\t\t */\n+\t\t\t\tfor (i = 0; i < n; i++)\n+\t\t\t\t\tif (slots[(child_in_foreground + i) % n])\n+\t\t\t\t\t\tbreak;\n+\t\t\t\tchild_in_foreground = (child_in_foreground + i) % n;\n+\t\t\t\tfputs(err[child_in_foreground].buf, stderr);\n+\t\t\t\tstrbuf_reset(&err[child_in_foreground]);\n+\t\t\t}\n+\t\t}\n+\t}\n+\treturn 0;\n+}\ndiff --git a/run-command.h b/run-command.h\nindex 5b4425a..8f53ad6 100644\n--- a/run-command.h\n+++ b/run-command.h\n@@ -119,4 +119,33 @@ struct async {\n int start_async(struct async *async);\n int finish_async(struct async *async);\n \n+/**\n+ * Return 0 if the next child is ready to run.\n+ * This callback takes care to initialize the child process and preload the\n+ * out and error channel. The preloading of these outpout channels is useful\n+ * if you want to have a message printed directly before the output of the\n+ * child process.\n+ *\n+ * Return != 0 if there are no more tasks to be processed.\n+ */\n+typedef int (*get_next_task)(void *data,\n+\t\t\t struct child_process *cp,\n+\t\t\t struct strbuf *err);\n+\n+/**\n+ * Runs up to n processes at the same time. Whenever a process can\n+ * be started, the callback `get_next_task` is called to obtain the\n+ * data fed to the child process.\n+ *\n+ * The children started via this function run in parallel and their output\n+ * to both stdout and stderr is buffered, while one of the children will\n+ * directly output to stdout/stderr.\n+ *\n+ * This leads to a problem with output from processes which put out to\n+ * stdout/err alternatingly as the buffering will not be able to replay\n+ * the\n+ */\n+\n+int run_processes_async(int n, get_next_task fn, void *data);\n+\n #endif\ndiff --git a/strbuf.c b/strbuf.c\nindex cce5eed..7f866c3 100644\n--- a/strbuf.c\n+++ b/strbuf.c\n@@ -384,6 +384,37 @@ ssize_t strbuf_read(struct strbuf *sb, int fd, size_t hint)\n \treturn sb->len - oldlen;\n }\n \n+ssize_t strbuf_read_noblock(struct strbuf *sb, int fd, size_t hint)\n+{\n+\tsize_t oldlen = sb->len;\n+\tsize_t oldalloc = sb->alloc;\n+\n+\tstrbuf_grow(sb, hint ? hint : 8192);\n+\tfor (;;) {\n+\t\tssize_t cnt;\n+\n+\t\tcnt = read(fd, sb->buf + sb->len, sb->alloc - sb->len - 1);\n+\t\tif (cnt < 0) {\n+\t\t\tif (errno == EINTR)\n+\t\t\t\tcontinue;\n+\t\t\tif (errno == EAGAIN)\n+\t\t\t\tbreak;\n+\t\t\tif (oldalloc == 0)\n+\t\t\t\tstrbuf_release(sb);\n+\t\t\telse\n+\t\t\t\tstrbuf_setlen(sb, oldlen);\n+\t\t\treturn -1;\n+\t\t}\n+\t\tif (!cnt)\n+\t\t\tbreak;\n+\t\tsb->len += cnt;\n+\t\tstrbuf_grow(sb, 8192);\n+\t}\n+\n+\tsb->buf[sb->len] = '\\0';\n+\treturn sb->len - oldlen;\n+}\n+\n #define STRBUF_MAXLINK (2*PATH_MAX)\n \n int strbuf_readlink(struct strbuf *sb, const char *path, size_t hint)\ndiff --git a/strbuf.h b/strbuf.h\nindex aef2794..7ea462b 100644\n--- a/strbuf.h\n+++ b/strbuf.h\n@@ -365,6 +365,7 @@ extern size_t strbuf_fread(struct strbuf *, size_t, FILE *);\n * any partial read is undone.\n */\n extern ssize_t strbuf_read(struct strbuf *, int fd, size_t hint);\n+extern ssize_t strbuf_read_noblock(struct strbuf *, int fd, size_t hint);\n \n /**\n * Read the contents of a file, specified by its path. The third argument\ndiff --git a/submodule.c b/submodule.c\nindex 1d64e57..6d757c6 100644\n--- a/submodule.c\n+++ b/submodule.c\n@@ -12,6 +12,7 @@\n #include \"sha1-array.h\"\n #include \"argv-array.h\"\n #include \"blob.h\"\n+#include \"thread-utils.h\"\n \n static int config_fetch_recurse_submodules = RECURSE_SUBMODULES_ON_DEMAND;\n static struct string_list changed_submodule_paths;\n@@ -615,37 +616,61 @@ static void calculate_changed_submodule_paths(void)\n \tinitialized_fetch_ref_tips = 0;\n }\n \n+struct submodule_parallel_fetch {\n+\tint count;\n+\tstruct argv_array args;\n+\tconst char *work_tree;\n+\tconst char *prefix;\n+\tint command_line_option;\n+\tint quiet;\n+};\n+#define SPF_INIT {0, ARGV_ARRAY_INIT, NULL}\n+\n+int get_next_submodule(void *data, struct child_process *cp,\n+\t\t struct strbuf *err);\n+\n int fetch_populated_submodules(const struct argv_array *options,\n \t\t\t const char *prefix, int command_line_option,\n-\t\t\t int quiet)\n+\t\t\t int quiet, int max_parallel_jobs)\n {\n \tint i, result = 0;\n-\tstruct child_process cp = CHILD_PROCESS_INIT;\n-\tstruct argv_array argv = ARGV_ARRAY_INIT;\n-\tconst char *work_tree = get_git_work_tree();\n-\tif (!work_tree)\n+\tstruct submodule_parallel_fetch spf = SPF_INIT;\n+\tspf.work_tree = get_git_work_tree();\n+\tspf.command_line_option = command_line_option;\n+\tspf.quiet = quiet;\n+\tspf.prefix = prefix;\n+\tif (!spf.work_tree)\n \t\tgoto out;\n \n \tif (read_cache() < 0)\n \t\tdie(\"index file corrupt\");\n \n-\targv_array_push(&argv, \"fetch\");\n+\targv_array_push(&spf.args, \"fetch\");\n \tfor (i = 0; i < options->argc; i++)\n-\t\targv_array_push(&argv, options->argv[i]);\n-\targv_array_push(&argv, \"--recurse-submodules-default\");\n+\t\targv_array_push(&spf.args, options->argv[i]);\n+\targv_array_push(&spf.args, \"--recurse-submodules-default\");\n \t/* default value, \"--submodule-prefix\" and its value are added later */\n \n-\tcp.env = local_repo_env;\n-\tcp.git_cmd = 1;\n-\tcp.no_stdin = 1;\n-\n \tcalculate_changed_submodule_paths();\n+\trun_processes_async(max_parallel_jobs, get_next_submodule, &spf);\n+\n+\targv_array_clear(&spf.args);\n+out:\n+\tstring_list_clear(&changed_submodule_paths, 1);\n+\treturn result;\n+}\n+\n+int get_next_submodule(void *data, struct child_process *cp,\n+\t\t struct strbuf *err)\n+{\n+\tint ret = 0;\n+\tstruct submodule_parallel_fetch *spf = data;\n \n-\tfor (i = 0; i < active_nr; i++) {\n+\tfor ( ; spf->count < active_nr; spf->count++) {\n \t\tstruct strbuf submodule_path = STRBUF_INIT;\n \t\tstruct strbuf submodule_git_dir = STRBUF_INIT;\n \t\tstruct strbuf submodule_prefix = STRBUF_INIT;\n-\t\tconst struct cache_entry *ce = active_cache[i];\n+\t\tconst struct cache_entry *ce = active_cache[spf->count];\n \t\tconst char *git_dir, *default_argv;\n \t\tconst struct submodule *submodule;\n \n@@ -657,7 +682,7 @@ int fetch_populated_submodules(const struct argv_array *options,\n \t\t\tsubmodule = submodule_from_name(null_sha1, ce->name);\n \n \t\tdefault_argv = \"yes\";\n-\t\tif (command_line_option == RECURSE_SUBMODULES_DEFAULT) {\n+\t\tif (spf->command_line_option == RECURSE_SUBMODULES_DEFAULT) {\n \t\t\tif (submodule &&\n \t\t\t submodule->fetch_recurse !=\n \t\t\t\t\t\tRECURSE_SUBMODULES_NONE) {\n@@ -680,40 +705,46 @@ int fetch_populated_submodules(const struct argv_array *options,\n \t\t\t\t\tdefault_argv = \"on-demand\";\n \t\t\t\t}\n \t\t\t}\n-\t\t} else if (command_line_option == RECURSE_SUBMODULES_ON_DEMAND) {\n+\t\t} else if (spf->command_line_option == RECURSE_SUBMODULES_ON_DEMAND) {\n \t\t\tif (!unsorted_string_list_lookup(&changed_submodule_paths, ce->name))\n \t\t\t\tcontinue;\n \t\t\tdefault_argv = \"on-demand\";\n \t\t}\n \n-\t\tstrbuf_addf(&submodule_path, \"%s/%s\", work_tree, ce->name);\n+\t\tstrbuf_addf(&submodule_path, \"%s/%s\", spf->work_tree, ce->name);\n \t\tstrbuf_addf(&submodule_git_dir, \"%s/.git\", submodule_path.buf);\n-\t\tstrbuf_addf(&submodule_prefix, \"%s%s/\", prefix, ce->name);\n+\t\tstrbuf_addf(&submodule_prefix, \"%s%s/\", spf->prefix, ce->name);\n \t\tgit_dir = read_gitfile(submodule_git_dir.buf);\n \t\tif (!git_dir)\n \t\t\tgit_dir = submodule_git_dir.buf;\n \t\tif (is_directory(git_dir)) {\n-\t\t\tif (!quiet)\n-\t\t\t\tfprintf(stderr, \"Fetching submodule %s%s\\n\", prefix, ce->name);\n-\t\t\tcp.dir = submodule_path.buf;\n-\t\t\targv_array_push(&argv, default_argv);\n-\t\t\targv_array_push(&argv, \"--submodule-prefix\");\n-\t\t\targv_array_push(&argv, submodule_prefix.buf);\n-\t\t\tcp.argv = argv.argv;\n-\t\t\tif (run_command(&cp))\n-\t\t\t\tresult = 1;\n-\t\t\targv_array_pop(&argv);\n-\t\t\targv_array_pop(&argv);\n-\t\t\targv_array_pop(&argv);\n+\t\t\tchild_process_init(cp);\n+\t\t\tcp->dir = strbuf_detach(&submodule_path, NULL);\n+\t\t\tcp->git_cmd = 1;\n+\t\t\tcp->no_stdout = 1;\n+\t\t\tcp->no_stdin = 1;\n+\t\t\tcp->stdout_to_stderr = 1;\n+\t\t\tcp->err = -1;\n+\t\t\tcp->env = local_repo_env;\n+\t\t\tif (!spf->quiet)\n+\t\t\t\tstrbuf_addf(err, \"Fetching submodule %s%s\\n\",\n+\t\t\t\t\t spf->prefix, ce->name);\n+\t\t\targv_array_init(&cp->args);\n+\t\t\targv_array_pushv(&cp->args, spf->args.argv);\n+\t\t\targv_array_push(&cp->args, default_argv);\n+\t\t\targv_array_push(&cp->args, \"--submodule-prefix\");\n+\t\t\targv_array_push(&cp->args, submodule_prefix.buf);\n+\t\t\tret = 1;\n \t\t}\n \t\tstrbuf_release(&submodule_path);\n \t\tstrbuf_release(&submodule_git_dir);\n \t\tstrbuf_release(&submodule_prefix);\n+\t\tif (ret) {\n+\t\t\tspf->count++;\n+\t\t\treturn 0;\n+\t\t}\n \t}\n-\targv_array_clear(&argv);\n-out:\n-\tstring_list_clear(&changed_submodule_paths, 1);\n-\treturn result;\n+\treturn 1;\n }\n \n unsigned is_submodule_modified(const char *path, int ignore_untracked)\ndiff --git a/submodule.h b/submodule.h\nindex 5507c3d..cbc0003 100644\n--- a/submodule.h\n+++ b/submodule.h\n@@ -31,7 +31,7 @@ void set_config_fetch_recurse_submodules(int value);\n void check_for_new_submodule_commits(unsigned char new_sha1[20]);\n int fetch_populated_submodules(const struct argv_array *options,\n \t\t\t const char *prefix, int command_line_option,\n-\t\t\t int quiet);\n+\t\t\t int quiet, int max_parallel_jobs);\n unsigned is_submodule_modified(const char *path, int ignore_untracked);\n int submodule_uses_gitfile(const char *path);\n int ok_to_remove_submodule(const char *path);\ndiff --git a/t/t0061-run-command.sh b/t/t0061-run-command.sh\nindex 9acf628..0970fb0 100755\n--- a/t/t0061-run-command.sh\n+++ b/t/t0061-run-command.sh\n@@ -47,4 +47,20 @@ test_expect_success POSIXPERM,SANITY 'unreadable directory in PATH' '\n \ttest_cmp expect actual\n '\n \n+cat >expect <<-EOF\n+Now running instance 0\n+Hello World\n+Now running instance 1\n+Hello World\n+Now running instance 2\n+Hello World\n+Now running instance 3\n+Hello World\n+EOF\n+\n+test_expect_success 'run_command runs in parallel' '\n+\ttest-run-command run-command-async sh -c \"echo Hello World >&2;\" 2>actual &&\n+\ttest_cmp expect actual\n+'\n+\n test_done\ndiff --git a/t/t5526-fetch-submodules.sh b/t/t5526-fetch-submodules.sh\nindex 17759b1..1b4ce69 100755\n--- a/t/t5526-fetch-submodules.sh\n+++ b/t/t5526-fetch-submodules.sh\n@@ -71,6 +71,16 @@ test_expect_success \"fetch --recurse-submodules recurses into submodules\" '\n \ttest_i18ncmp expect.err actual.err\n '\n \n+test_expect_success \"fetch --recurse-submodules -j2 has the same output behaviour\" '\n+\tadd_upstream_commit &&\n+\t(\n+\t\tcd downstream &&\n+\t\tgit fetch --recurse-submodules -j2 2>../actual.err\n+\t) &&\n+\ttest_must_be_empty actual.out &&\n+\ttest_i18ncmp expect.err actual.err\n+'\n+\n test_expect_success \"fetch alone only fetches superproject\" '\n \tadd_upstream_commit &&\n \t(\n@@ -140,6 +150,15 @@ test_expect_success \"--quiet propagates to submodules\" '\n \t! test -s actual.err\n '\n \n+test_expect_success \"--quiet propagates to parallel submodules\" '\n+\t(\n+\t\tcd downstream &&\n+\t\tgit fetch --recurse-submodules -j 2 --quiet >../actual.out 2>../actual.err\n+\t) &&\n+\t! test -s actual.out &&\n+\t! test -s actual.err\n+'\n+\n test_expect_success \"--dry-run propagates to submodules\" '\n \tadd_upstream_commit &&\n \t(\ndiff --git a/test-run-command.c b/test-run-command.c\nindex 89c7de2..4817f6e 100644\n--- a/test-run-command.c\n+++ b/test-run-command.c\n@@ -10,9 +10,29 @@\n \n #include \"git-compat-util.h\"\n #include \"run-command.h\"\n+#include \"argv-array.h\"\n+#include \"strbuf.h\"\n #include \n #include \n \n+static int number_callbacks;\n+int run_processes_async_next(void *data,\n+\t\t\t struct child_process *cp,\n+\t\t\t struct strbuf *err)\n+{\n+\tstruct child_process *d = data;\n+\tif (number_callbacks >= 4)\n+\t\treturn 1;\n+\n+\targv_array_pushv(&cp->args, d->argv);\n+\tcp->stdout_to_stderr = 1;\n+\tcp->no_stdin = 1;\n+\tcp->err = -1;\n+\tstrbuf_addf(err, \"Now running instance %d\\n\", number_callbacks);\n+\tnumber_callbacks++;\n+\treturn 0;\n+}\n+\n int main(int argc, char **argv)\n {\n \tstruct child_process proc = CHILD_PROCESS_INIT;\n@@ -30,6 +50,9 @@ int main(int argc, char **argv)\n \tif (!strcmp(argv[1], \"run-command\"))\n \t\texit(run_command(&proc));\n \n+\tif (!strcmp(argv[1], \"run-command-async\"))\n+\t\texit(run_processes_async(4, run_processes_async_next, &proc));\n+\n \tfprintf(stderr, \"check usage\\n\");\n \treturn 1;\n }\n-- \n2.6.0.rc0.131.gf624c3d\n"},{"id":"269833","messageId":"xmqqpp1nxxji.fsf@gitster.mtv.corp.google.com","threadId":"40335","inReplyTo":"1442012994-20374-3-git-send-email-sbeller@google.com","subject":"Re: [PATCH 2/2] fetch: fetch submodules in parallel","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-09-12T19:11:45Z","receivedAt":"2015-09-12T19:11:45Z","isPatch":true,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"Stefan Beller writes:\n\n> diff --git a/run-command.c b/run-command.c\n> index 28e1d55..b8ff67b 100644\n> --- a/run-command.c\n> +++ b/run-command.c\n> @@ -852,3 +852,147 @@ int capture_command(struct child_process *cmd, struct strbuf *buf, size_t hint)\n> \tclose(cmd->out);\n> \treturn finish_command(cmd);\n> }\n> +\n> +int run_processes_async(int n, get_next_task fn, void *data)\n> +{\n> +\tint i, wait_status;\n> +\tpid_t pid;\n> +\n> +\t/* no more tasks. Also set when aborting early. */\n> +\tint all_tasks_started = 0;\n> +\tint nr_processes = 0;\n> +\tint child_in_foreground = 0;\n> +\tstruct timeval timeout;\n> +\tstruct child_process *children = xcalloc(n, sizeof(*children));\n> +\tchar *slots = xcalloc(n, sizeof(*slots));\n> +\tstruct strbuf *err = xcalloc(n, sizeof(*err));\n> +\tfd_set fdset;\n> +\tint maxfd;\n> +\tstruct strbuf finished_children = STRBUF_INIT;\n> +\tint flags;\n> +\tfor (i = 0; i < n; i++)\n> +\t\tstrbuf_init(&err[i], 0);\n> +\n> +\twhile (!all_tasks_started || nr_processes > 0) {\n> +\t\t/* Start new processes. */\n> +\t\twhile (!all_tasks_started && nr_processes < n) {\n> +\t\t\tfor (i = 0; i < n; i++)\n> +\t\t\t\tif (!slots[i])\n> +\t\t\t\t\tbreak; /* found an empty slot */\n> +\t\t\tif (i == n)\n> +\t\t\t\tdie(\"BUG: bookkeeping is hard\");\n> +\n> +\t\t\tif (fn(data, &children[i], &err[i])) {\n> +\t\t\t\tall_tasks_started = 1;\n> +\t\t\t\tbreak;\n> +\t\t\t}\n> +\t\t\tif (start_command(&children[i]))\n> +\t\t\t\tdie(_(\"Could not start child process\"));\n> +\t\t\tflags = fcntl(children[i].err, F_GETFL);\n> +\t\t\tfcntl(children[i].err, F_SETFL, flags | O_NONBLOCK);\n\nThis function in run-command.c looks as if it is a generic helper to\nbe called by anybody, but it seems to only care about the standard\nerror and not the standard output stream, which means potential\nusers that do not dup them together cannot use it. Is that a big\ndownside, or is it sufficient to document the API to say that\nchildren must do so? I offhand do not think the latter is\nunreasonable, but that may be only because I haven't thought things\nthrough.\n\n> +\t\t\tnr_processes++;\n> +\t\t\tslots[i] = 1;\n> +\t\t}\n> +\n> +\t\t/* prepare data for select call */\n> +\t\tFD_ZERO(&fdset);\n> +\t\tmaxfd = 0;\n> +\t\tfor (i = 0; i < n; i++) {\n> +\t\t\tif (!slots[i])\n> +\t\t\t\tcontinue;\n> +\t\t\tFD_SET(children[i].err, &fdset);\n> +\t\t\tif (children[i].err > maxfd)\n> +\t\t\t\tmaxfd = children[i].err;\n> +\t\t}\n> +\t\ttimeout.tv_sec = 0;\n> +\t\ttimeout.tv_usec = 500000;\n> +\n> +\t\ti = select(maxfd + 1, &fdset, NULL, NULL, &timeout);\n\nI thought we try to use poll() and on systems with only select we\nallow compat/ to emulate in our code.\n\n> +\t\tif (i < 0) {\n> +\t\t\tif (errno == EINTR)\n> +\t\t\t\t/* A signal was caught; try again */\n> +\t\t\t\tcontinue;\n> +\t\t\telse if (errno == ENOMEM)\n> +\t\t\t\tdie_errno(\"BUG: keeping track of fds is hard\");\n> +\t\t\telse if (errno == EINVAL)\n> +\t\t\t\tdie_errno(\"BUG: invalid arguments to select\");\n> +\t\t\telse if (errno == EBADF)\n> +\t\t\t\tdie_errno(\"BUG: keeping track of fds is hard\");\n> +\t\t\telse\n> +\t\t\t\tdie_errno(\"Unknown error with select\");\n\nI doubt that the later part of elseif cascade adds any value. You\nwill see errno printed anyway.\n"},{"id":"269930","messageId":"CAGZ79kaBvVWT1OPMxUAU9N2oaC5TT5wwWew5jS0k_o5J10sKfA@mail.gmail.com","threadId":"40335","inReplyTo":"xmqqpp1nxxji.fsf@gitster.mtv.corp.google.com","subject":"Re: [PATCH 2/2] fetch: fetch submodules in parallel","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-09-14T16:46:58Z","receivedAt":"2015-09-14T16:46:58Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Sat, Sep 12, 2015 at 12:11 PM, Junio C Hamano wrote:\n>> + if (start_command(&children[i]))\n>> + die(_(\"Could not start child process\"));\n>> + flags = fcntl(children[i].err, F_GETFL);\n>> + fcntl(children[i].err, F_SETFL, flags | O_NONBLOCK);\n>\n> This function in run-command.c looks as if it is a generic helper to\n> be called by anybody, but it seems to only care about the standard\n> error and not the standard output stream, which means potential\n> users that do not dup them together cannot use it. Is that a big\n> downside, or is it sufficient to document the API to say that\n> children must do so? I offhand do not think the latter is\n> unreasonable, but that may be only because I haven't thought things\n> through.\n\nYes it ought to become a generic helper eventually.\n\nI tried implementing a buffering solution for both stdout and stderr,\nbut that doesn't really workout well if you consider interleaved output\non the pipes as we cannot accurately replay that later on. To do that\nwe would need to store the timing information of the channels, at least\nthe relative order of it like:\n\n (stdout, First comes text to stdout),\n (stderr, interrupted by text in stderr)\n (stdout, but stdout doesn't bother, blasting more text)\n (stderr, continues to interrupt)\n\nobtaining the information is inherently racy, as all we can do is\npolling/reading from both stdout/err as fast as possible but without\nproper synchronization mechanisms we cannot be sure.\n\nI will add documentation explaining why the async output case\nwill only deal with one channel. I chose stderr as that's already\navailable and needed in this use case.\n\n\n>\n>> + nr_processes++;\n>> + slots[i] = 1;\n>> + }\n>> +\n>> + /* prepare data for select call */\n>> + FD_ZERO(&fdset);\n>> + maxfd = 0;\n>> + for (i = 0; i < n; i++) {\n>> + if (!slots[i])\n>> + continue;\n>> + FD_SET(children[i].err, &fdset);\n>> + if (children[i].err > maxfd)\n>> + maxfd = children[i].err;\n>> + }\n>> + timeout.tv_sec = 0;\n>> + timeout.tv_usec = 500000;\n>> +\n>> + i = select(maxfd + 1, &fdset, NULL, NULL, &timeout);\n>\n> I thought we try to use poll() and on systems with only select we\n> allow compat/ to emulate in our code.\n\nI did not know that. I'll rewrite the patch to use poll instead.\n\n>\n>> + if (i < 0) {\n>> + if (errno == EINTR)\n>> + /* A signal was caught; try again */\n>> + continue;\n>> + else if (errno == ENOMEM)\n>> + die_errno(\"BUG: keeping track of fds is hard\");\n>> + else if (errno == EINVAL)\n>> + die_errno(\"BUG: invalid arguments to select\");\n>> + else if (errno == EBADF)\n>> + die_errno(\"BUG: keeping track of fds is hard\");\n>> + else\n>> + die_errno(\"Unknown error with select\");\n>\n> I doubt that the later part of elseif cascade adds any value. You\n> will see errno printed anyway.\n\nok.\n"},{"id":"269941","messageId":"20150914171736.GA1548@sigill.intra.peff.net","threadId":"40335","inReplyTo":"CAGZ79kaBvVWT1OPMxUAU9N2oaC5TT5wwWew5jS0k_o5J10sKfA@mail.gmail.com","subject":"Re: [PATCH 2/2] fetch: fetch submodules in parallel","fromName":"Jeff King","fromEmail":"peff@peff.net","sentAt":"2015-09-14T17:17:36Z","receivedAt":"2015-09-14T17:17:36Z","isPatch":true,"sender":{"key":"peff@peff.net","avatar":"https://avatars.githubusercontent.com/u/45925?v=4"},"body":"On Mon, Sep 14, 2015 at 09:46:58AM -0700, Stefan Beller wrote:\n\n> I tried implementing a buffering solution for both stdout and stderr,\n> but that doesn't really workout well if you consider interleaved output\n> on the pipes as we cannot accurately replay that later on. To do that\n> we would need to store the timing information of the channels, at least\n> the relative order of it like:\n> \n> (stdout, First comes text to stdout),\n> (stderr, interrupted by text in stderr)\n> (stdout, but stdout doesn't bother, blasting more text)\n> (stderr, continues to interrupt)\n> \n> obtaining the information is inherently racy, as all we can do is\n> polling/reading from both stdout/err as fast as possible but without\n> proper synchronization mechanisms we cannot be sure.\n\nI don't think you need exact timing information. This is no different\nthan running the commands themselves, with stdout and stderr writing to\na pty that your terminal emulator will then read() from. If the program\nproduces intermingled stdout/stderr that clogs up the terminal, that is\nits problem.\n\nThe only difference is that we're going to save it and later replay it\nall very quickly. So I think it would be sufficient just to retain the\noriginal order.\n\n> I will add documentation explaining why the async output case\n> will only deal with one channel. I chose stderr as that's already\n> available and needed in this use case.\n\nI suspect you could just set child->stdout_to_stderr in this case, and\nthen you get your ordering for free. But probably in the general case\npeople would want to run inspection commands that produce a useful\nstdout.\n\nTo handle multiple channels, I think you could just do a linked list of\nbuffers rather than a single strbuf. Like:\n\n struct io_chunk {\n\tint channel;\n\tchar *buf;\n\tsize_t len;\n\tstruct io_chunk *next;\n };\n\nand just keep appending chunks to the list (and to dump them, just walk\nthe list, writing each to the appropriate channel descriptor).\n\n-Peff\n"},{"id":"269948","messageId":"CAGZ79kZUdyLJzTiFXSgSezrPdG-aGgYJm6juAtpq8A_Je5JAfw@mail.gmail.com","threadId":"40335","inReplyTo":"20150914171736.GA1548@sigill.intra.peff.net","subject":"Re: [PATCH 2/2] fetch: fetch submodules in parallel","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-09-14T17:47:31Z","receivedAt":"2015-09-14T17:47:31Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Mon, Sep 14, 2015 at 10:17 AM, Jeff King wrote:\n> On Mon, Sep 14, 2015 at 09:46:58AM -0700, Stefan Beller wrote:\n>\n>> I tried implementing a buffering solution for both stdout and stderr,\n>> but that doesn't really workout well if you consider interleaved output\n>> on the pipes as we cannot accurately replay that later on. To do that\n>> we would need to store the timing information of the channels, at least\n>> the relative order of it like:\n>>\n>> (stdout, First comes text to stdout),\n>> (stderr, interrupted by text in stderr)\n>> (stdout, but stdout doesn't bother, blasting more text)\n>> (stderr, continues to interrupt)\n>>\n>> obtaining the information is inherently racy, as all we can do is\n>> polling/reading from both stdout/err as fast as possible but without\n>> proper synchronization mechanisms we cannot be sure.\n>\n> I don't think you need exact timing information. This is no different\n> than running the commands themselves, with stdout and stderr writing to\n> a pty that your terminal emulator will then read() from. If the program\n> produces intermingled stdout/stderr that clogs up the terminal, that is\n> its problem.\n>\n> The only difference is that we're going to save it and later replay it\n> all very quickly. So I think it would be sufficient just to retain the\n> original order.\n>\n>> I will add documentation explaining why the async output case\n>> will only deal with one channel. I chose stderr as that's already\n>> available and needed in this use case.\n>\n> I suspect you could just set child->stdout_to_stderr in this case, and\n> then you get your ordering for free. But probably in the general case\n> people would want to run inspection commands that produce a useful\n> stdout.\n>\n> To handle multiple channels, I think you could just do a linked list of\n> buffers rather than a single strbuf. Like:\n\nI will have no problem coding such a thing in a user program,\nbut how do you obtain this non racily from the child using the posix\nAPI?\n\nThe poll/select command may return more than one fd ready, so\nthen you don't know the ordering in which you would need to replay\nit. This may introduce subtle bugs?\n\nSo I'd rather come up with a solution buffering 2 channels once we need it,\nkeeping the stdout_to_stderr as a requirement for now.\n\n>\n> struct io_chunk {\n> int channel;\n> char *buf;\n> size_t len;\n> struct io_chunk *next;\n> };\n>\n> and just keep appending chunks to the list (and to dump them, just walk\n> the list, writing each to the appropriate channel descriptor).\n>\n> -Peff\n"},{"id":"269949","messageId":"20150914175509.GJ8165@google.com","threadId":"40335","inReplyTo":"20150914171736.GA1548@sigill.intra.peff.net","subject":"Re: [PATCH 2/2] fetch: fetch submodules in parallel","fromName":"Jonathan Nieder","fromEmail":"jrnieder@gmail.com","sentAt":"2015-09-14T17:55:09Z","receivedAt":"2015-09-14T17:55:09Z","isPatch":true,"sender":{"key":"jrnieder@gmail.com","avatar":"https://avatars.githubusercontent.com/u/281595?v=4"},"body":"Jeff King wrote:\n> On Mon, Sep 14, 2015 at 09:46:58AM -0700, Stefan Beller wrote:\n\n>> I tried implementing a buffering solution for both stdout and stderr,\n>> but that doesn't really workout well if you consider interleaved output\n>> on the pipes as we cannot accurately replay that later on.\n[...]\n>> obtaining the information is inherently racy\n[...]\n> I don't think you need exact timing information. This is no different\n> than running the commands themselves, with stdout and stderr writing to\n> a pty that your terminal emulator will then read() from. If the program\n> produces intermingled stdout/stderr that clogs up the terminal, that is\n> its problem.\n\nThe difference is that when stdout and stderr write to a pty, they write\nto the same pty. That is, suppose a child process does\n\n\twrite(1, \"A\\n\", 2);\n\twrite(2, \"B\\n\", 1);\n\twrite(1, \"C\\n\", 2);\n\nThen the output that should be echoed to the terminal is\n\n\tA\n\tB\n\tC\n\nNow the parent might do\n\n\tfor (;;) {\n\t\tint n = select(...);\n\t\t... do stuff ...\n\t}\n\nIf all three writes happen during the \"do stuff\" step, then *if* the\nchild's stdout and stderr went to different pipes, all the parent sees\nis\n\n\tchild's stdout: A\\nC\\n\n\tchild's stderr: B\\n\n\nThere is not sufficient information to recover the original output\norder. (Linux provides a pipe2(..., O_DIRECT) that almost provides\nsufficient information --- it tells you\n\n\tchild's stdout: \"A\\n\", \"C\\n\"\n\tchild's stderr: \"B\\n\"\n\nbut still doesn't give information about ordering.)\n\nThat's probably okay: in most git commands, stderr shows a combination\nof diagnostic output and progress information and stdout shows the\nactual result, so interleaving between the two is not too common.\n\nOne can imagine a \"git grep --recurse-submodules\" that wants to run a\ngrep in each submodule and combine their output in some appropriate\nway. It's not clear what order is best for that use case: stderr\n(errors, plus progress in some imaginary future) at the beginning to\nshow the story of how output was generated before the output? stderr\nat the end so errors are not hidden way up on the screen? Some kind\nof interleaving that pays attention to the format of stdout?\n\nThat is more complicated than the \"fetch --recurse-submodules\" case\nthat Stefan is currently tackling, so it seems wise to me to punt for\nnow.\n\nThanks and hope that helps,\nJonathan\n"},{"id":"269950","messageId":"xmqqsi6gx4uh.fsf@gitster.mtv.corp.google.com","threadId":"40335","inReplyTo":"20150914171736.GA1548@sigill.intra.peff.net","subject":"Re: [PATCH 2/2] fetch: fetch submodules in parallel","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-09-14T17:56:06Z","receivedAt":"2015-09-14T17:56:06Z","isPatch":true,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"Jeff King writes:\n\n> I don't think you need exact timing information. This is no different\n> than running the commands themselves, with stdout and stderr writing to\n> a pty that your terminal emulator will then read() from. If the program\n> produces intermingled stdout/stderr that clogs up the terminal, that is\n> its problem.\n>\n> The only difference is that we're going to save it and later replay it\n> all very quickly. So I think it would be sufficient just to retain the\n> original order.\n>\n>> I will add documentation explaining why the async output case\n>> will only deal with one channel. I chose stderr as that's already\n>> available and needed in this use case.\n>\n> I suspect you could just set child->stdout_to_stderr in this case, and\n> then you get your ordering for free.\n\nI think we are in agreement; that is exactly what I wanted to say\nwhen I said \"I offhand do not think the latter [i.e. the callers\nhave to dup them together] is unreasonable\". Thanks for stating it\nmore clearly and explicitly.\n\n> To handle multiple channels, I think you could just do a linked list of\n> buffers rather than a single strbuf. Like:\n>\n> struct io_chunk {\n> \tint channel;\n> \tchar *buf;\n> \tsize_t len;\n> \tstruct io_chunk *next;\n> };\n>\n> and just keep appending chunks to the list (and to dump them, just walk\n> the list, writing each to the appropriate channel descriptor).\n\nPerhaps, but let's not overdesign things before we have a concrete\nexample codepath that benefits from such a thing. It is hard to\ndetect a misdesign without a real usage pattern.\n"},{"id":"269952","messageId":"20150914180759.GA11920@sigill.intra.peff.net","threadId":"40335","inReplyTo":"20150914175509.GJ8165@google.com","subject":"Re: [PATCH 2/2] fetch: fetch submodules in parallel","fromName":"Jeff King","fromEmail":"peff@peff.net","sentAt":"2015-09-14T18:07:59Z","receivedAt":"2015-09-14T18:07:59Z","isPatch":true,"sender":{"key":"peff@peff.net","avatar":"https://avatars.githubusercontent.com/u/45925?v=4"},"body":"On Mon, Sep 14, 2015 at 10:55:09AM -0700, Jonathan Nieder wrote:\n\n> > I don't think you need exact timing information. This is no different\n> > than running the commands themselves, with stdout and stderr writing to\n> > a pty that your terminal emulator will then read() from. If the program\n> > produces intermingled stdout/stderr that clogs up the terminal, that is\n> > its problem.\n> \n> The difference is that when stdout and stderr write to a pty, they write\n> to the same pty. That is, suppose a child process does\n> \n> \twrite(1, \"A\\n\", 2);\n> \twrite(2, \"B\\n\", 1);\n> \twrite(1, \"C\\n\", 2);\n\nAh, right. The pty is where things get mixed, not the read() from the\nterminal. So it depends on the write() order. Thanks for the\nexplanation.\n\n> One can imagine a \"git grep --recurse-submodules\" that wants to run a\n> grep in each submodule and combine their output in some appropriate\n> way. It's not clear what order is best for that use case: stderr\n> (errors, plus progress in some imaginary future) at the beginning to\n> show the story of how output was generated before the output? stderr\n> at the end so errors are not hidden way up on the screen? Some kind\n> of interleaving that pays attention to the format of stdout?\n\nI'd suggest a \"best effort\" interleaving, where we select and preserve\nthe read() order. That makes the easy cases work (you get things in the\noriginal order), and the hard cases at least do something reasonable\n(we may reorder two items which come in the same atomic \"tick\" of our\nselect, but at least they are nearby).\n\nThat's just my gut feeling, though.\n\n> That is more complicated than the \"fetch --recurse-submodules\" case\n> that Stefan is currently tackling, so it seems wise to me to punt for\n> now.\n\nI can live with that.\n\n-Peff\n"},{"id":"269963","messageId":"1442267451-2527-1-git-send-email-sbeller@google.com","threadId":"40335","inReplyTo":"20150914180759.GA11920@sigill.intra.peff.net","subject":"[PATCHv2] fetch submodules in parallel","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-09-14T21:50:50Z","receivedAt":"2015-09-14T21:50:50Z","isPatch":false,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"This replaces the last patch of the \"Parallel git submodule fetching\"\nseries. Changes:\n\n* have correct return code in submodule fetching when one submodule fails\n* use poll instead of select now\n* broke down into more smaller functions instead of one giant.\n (I think it is an improvement, but I wouldn't be surprised if someone objects)\n* closed memory leaks\n* document the need for stdout_to_stderr\n\nI don't deem it RFC-ish any more but good to go.\n\nAny feedback welcome!\nThanks,\nStefan\n\n\nStefan Beller (1):\n fetch: fetch submodules in parallel\n\n Documentation/fetch-options.txt | 7 +\n builtin/fetch.c | 6 +-\n builtin/pull.c | 6 +\n run-command.c | 278 ++++++++++++++++++++++++++++++++++++----\n run-command.h | 36 ++++++\n strbuf.c | 31 +++++\n strbuf.h | 1 +\n submodule.c | 119 ++++++++++++-----\n submodule.h | 2 +-\n t/t0061-run-command.sh | 20 +++\n t/t5526-fetch-submodules.sh | 19 +++\n test-run-command.c | 24 ++++\n 12 files changed, 490 insertions(+), 59 deletions(-)\n\nInterdiff to RFCv1:\n\ndiff --git a/Documentation/fetch-options.txt b/Documentation/fetch-options.txt\nindex d432f98..6b109f6 100644\n--- a/Documentation/fetch-options.txt\n+++ b/Documentation/fetch-options.txt\n@@ -105,7 +105,7 @@ ifndef::git-pull[]\n \tNumber of parallel children to be used for fetching submodules.\n \tEach will fetch from different submodules, such that fetching many\n \tsubmodules will be faster. By default submodules will be fetched\n-\tone at a time\n+\tone at a time.\n \n --no-recurse-submodules::\n \tDisable recursive fetching of submodules (this has the same effect as\ndiff --git a/builtin/fetch.c b/builtin/fetch.c\nindex a1520bb..f28eac6 100644\n--- a/builtin/fetch.c\n+++ b/builtin/fetch.c\n@@ -101,7 +101,7 @@ static struct option builtin_fetch_options[] = {\n \tOPT_SET_INT('n', NULL, &tags,\n \t\t N_(\"do not fetch all tags (--no-tags)\"), TAGS_UNSET),\n \tOPT_INTEGER('j', \"jobs\", &max_children,\n-\t\t N_(\"number of threads used for fetching\")),\n+\t\t N_(\"number of submodules fetched in parallel\")),\n \tOPT_BOOL('p', \"prune\", &prune,\n \t\t N_(\"prune remote-tracking branches no longer on remote\")),\n \t{ OPTION_CALLBACK, 0, \"recurse-submodules\", NULL, N_(\"on-demand\"),\ndiff --git a/builtin/pull.c b/builtin/pull.c\nindex bc117e9..f0af196 100644\n--- a/builtin/pull.c\n+++ b/builtin/pull.c\n@@ -179,7 +179,7 @@ static struct option pull_options[] = {\n \t\tN_(\"control recursive fetching of submodules\"),\n \t\tPARSE_OPT_OPTARG),\n \tOPT_PASSTHRU('j', \"jobs\", &max_children, N_(\"n\"),\n-\t\tN_(\"number of threads used for fetching submodules\"),\n+\t\tN_(\"number of submodules pulled in parallel\"),\n \t\tPARSE_OPT_OPTARG),\n \tOPT_BOOL(0, \"dry-run\", &opt_dry_run,\n \t\tN_(\"dry run\")),\ndiff --git a/run-command.c b/run-command.c\nindex b8ff67b..6f6f9fb 100644\n--- a/run-command.c\n+++ b/run-command.c\n@@ -232,6 +232,35 @@ static inline void set_cloexec(int fd)\n \t\tfcntl(fd, F_SETFD, flags | FD_CLOEXEC);\n }\n \n+static int determine_return_value(int wait_status,\n+\t\t\t\t int *result,\n+\t\t\t\t int *error_code,\n+\t\t\t\t const char *argv0)\n+{\n+\tif (WIFSIGNALED(wait_status)) {\n+\t\t*result = WTERMSIG(wait_status);\n+\t\tif (*result != SIGINT && *result != SIGQUIT)\n+\t\t\terror(\"%s died of signal %d\", argv0, *result);\n+\t\t/*\n+\t\t * This return value is chosen so that code & 0xff\n+\t\t * mimics the exit code that a POSIX shell would report for\n+\t\t * a program that died from this signal.\n+\t\t */\n+\t\t*result += 128;\n+\t} else if (WIFEXITED(wait_status)) {\n+\t\t*result = WEXITSTATUS(wait_status);\n+\t\t/*\n+\t\t * Convert special exit code when execvp failed.\n+\t\t */\n+\t\tif (*result == 127) {\n+\t\t\t*result = -1;\n+\t\t\t*error_code = ENOENT;\n+\t\t}\n+\t} else\n+\t\treturn 1;\n+\treturn 0;\n+}\n+\n static int wait_or_whine(pid_t pid, const char *argv0)\n {\n \tint status, code = -1;\n@@ -244,29 +273,10 @@ static int wait_or_whine(pid_t pid, const char *argv0)\n \tif (waiting < 0) {\n \t\tfailed_errno = errno;\n \t\terror(\"waitpid for %s failed: %s\", argv0, strerror(errno));\n-\t} else if (waiting != pid) {\n-\t\terror(\"waitpid is confused (%s)\", argv0);\n-\t} else if (WIFSIGNALED(status)) {\n-\t\tcode = WTERMSIG(status);\n-\t\tif (code != SIGINT && code != SIGQUIT)\n-\t\t\terror(\"%s died of signal %d\", argv0, code);\n-\t\t/*\n-\t\t * This return value is chosen so that code & 0xff\n-\t\t * mimics the exit code that a POSIX shell would report for\n-\t\t * a program that died from this signal.\n-\t\t */\n-\t\tcode += 128;\n-\t} else if (WIFEXITED(status)) {\n-\t\tcode = WEXITSTATUS(status);\n-\t\t/*\n-\t\t * Convert special exit code when execvp failed.\n-\t\t */\n-\t\tif (code == 127) {\n-\t\t\tcode = -1;\n-\t\t\tfailed_errno = ENOENT;\n-\t\t}\n \t} else {\n-\t\terror(\"waitpid is confused (%s)\", argv0);\n+\t\tif (waiting != pid\n+\t\t || (determine_return_value(status, &code, &failed_errno, argv0) < 0))\n+\t\t\terror(\"waitpid is confused (%s)\", argv0);\n \t}\n \n \tclear_child_for_cleanup(pid);\n@@ -853,146 +863,226 @@ int capture_command(struct child_process *cmd, struct strbuf *buf, size_t hint)\n \treturn finish_command(cmd);\n }\n \n-int run_processes_async(int n, get_next_task fn, void *data)\n+static void unblock_fd(int fd)\n {\n-\tint i, wait_status;\n-\tpid_t pid;\n+\tint flags = fcntl(fd, F_GETFL);\n+\tif (flags < 0) {\n+\t\twarning(\"Could not get file status flags, \"\n+\t\t\t\"output will be degraded\");\n+\t\treturn;\n+\t}\n+\tif (fcntl(fd, F_SETFL, flags | O_NONBLOCK)) {\n+\t\t\twarning(\"Could not set file status flags, \"\n+\t\t\t\"output will be degraded\");\n+\t\treturn;\n+\t}\n+}\n \n-\t/* no more tasks. Also set when aborting early. */\n-\tint all_tasks_started = 0;\n-\tint nr_processes = 0;\n-\tint child_in_foreground = 0;\n-\tstruct timeval timeout;\n-\tstruct child_process *children = xcalloc(n, sizeof(*children));\n-\tchar *slots = xcalloc(n, sizeof(*slots));\n-\tstruct strbuf *err = xcalloc(n, sizeof(*err));\n-\tfd_set fdset;\n-\tint maxfd;\n-\tstruct strbuf finished_children = STRBUF_INIT;\n-\tint flags;\n-\tfor (i = 0; i < n; i++)\n-\t\tstrbuf_init(&err[i], 0);\n-\n-\twhile (!all_tasks_started || nr_processes > 0) {\n-\t\t/* Start new processes. */\n-\t\twhile (!all_tasks_started && nr_processes < n) {\n-\t\t\tfor (i = 0; i < n; i++)\n-\t\t\t\tif (!slots[i])\n-\t\t\t\t\tbreak; /* found an empty slot */\n-\t\t\tif (i == n)\n-\t\t\t\tdie(\"BUG: bookkeeping is hard\");\n-\n-\t\t\tif (fn(data, &children[i], &err[i])) {\n-\t\t\t\tall_tasks_started = 1;\n-\t\t\t\tbreak;\n-\t\t\t}\n-\t\t\tif (start_command(&children[i]))\n-\t\t\t\tdie(_(\"Could not start child process\"));\n-\t\t\tflags = fcntl(children[i].err, F_GETFL);\n-\t\t\tfcntl(children[i].err, F_SETFL, flags | O_NONBLOCK);\n-\t\t\tnr_processes++;\n-\t\t\tslots[i] = 1;\n-\t\t}\n+struct parallel_processes {\n+\tint max_number_processes;\n+\tvoid *data;\n+\tget_next_task fn;\n+\thandle_child_starting_failure fn_err;\n+\thandle_child_return_value fn_exit;\n+\n+\tint nr_processes;\n+\tint all_tasks_started;\n+\tint foreground_child;\n+\tchar *slots;\n+\tstruct child_process *children;\n+\tstruct pollfd *pfd;\n+\tstruct strbuf *err;\n+\tstruct strbuf finished_children;\n+};\n+\n+static void run_processes_parallel_init(struct parallel_processes *pp,\n+\t\t\t\t\tint n, void *data,\n+\t\t\t\t\tget_next_task fn,\n+\t\t\t\t\thandle_child_starting_failure fn_err,\n+\t\t\t\t\thandle_child_return_value fn_exit)\n+{\n+\tint i;\n+\n+\tpp->max_number_processes = n;\n+\tpp->data = data;\n+\tpp->fn = fn;\n+\tpp->fn_err = fn_err;\n+\tpp->fn_exit = fn_exit;\n+\n+\tpp->nr_processes = 0;\n+\tpp->all_tasks_started = 0;\n+\tpp->foreground_child = 0;\n+\tpp->slots = xcalloc(n, sizeof(*pp->slots));\n+\tpp->children = xcalloc(n, sizeof(*pp->children));\n+\tpp->pfd = xcalloc(n, sizeof(*pp->pfd));\n+\tpp->err = xcalloc(n, sizeof(*pp->err));\n+\tstrbuf_init(&pp->finished_children, 0);\n+\n+\tfor (i = 0; i < n; i++) {\n+\t\tstrbuf_init(&pp->err[i], 0);\n+\t\tpp->pfd[i].events = POLLIN;\n+\t\tpp->pfd[i].fd = -1;\n+\t}\n+}\n+\n+static void run_processes_parallel_cleanup(struct parallel_processes *pp)\n+{\n+\tint i;\n+\tfor (i = 0; i < pp->max_number_processes; i++)\n+\t\tstrbuf_release(&pp->err[i]);\n+\n+\tfree(pp->children);\n+\tfree(pp->slots);\n+\tfree(pp->pfd);\n+\tfree(pp->err);\n+\tstrbuf_release(&pp->finished_children);\n+}\n \n-\t\t/* prepare data for select call */\n-\t\tFD_ZERO(&fdset);\n-\t\tmaxfd = 0;\n-\t\tfor (i = 0; i < n; i++) {\n-\t\t\tif (!slots[i])\n-\t\t\t\tcontinue;\n-\t\t\tFD_SET(children[i].err, &fdset);\n-\t\t\tif (children[i].err > maxfd)\n-\t\t\t\tmaxfd = children[i].err;\n+static void run_processes_parallel_start_new(struct parallel_processes *pp)\n+{\n+\tint i;\n+\t/* Start new processes. */\n+\twhile (!pp->all_tasks_started\n+\t && pp->nr_processes < pp->max_number_processes) {\n+\t\tfor (i = 0; i < pp->max_number_processes; i++)\n+\t\t\tif (!pp->slots[i])\n+\t\t\t\tbreak; /* found an empty slot */\n+\t\tif (i == pp->max_number_processes)\n+\t\t\tdie(\"BUG: bookkeeping is hard\");\n+\n+\t\tif (pp->fn(pp->data, &pp->children[i], &pp->err[i])) {\n+\t\t\tpp->all_tasks_started = 1;\n+\t\t\tbreak;\n \t\t}\n-\t\ttimeout.tv_sec = 0;\n-\t\ttimeout.tv_usec = 500000;\n+\t\tif (start_command(&pp->children[i]))\n+\t\t\tpp->fn_err(pp->data, &pp->children[i], &pp->err[i]);\n \n-\t\ti = select(maxfd + 1, &fdset, NULL, NULL, &timeout);\n-\t\tif (i < 0) {\n-\t\t\tif (errno == EINTR)\n-\t\t\t\t/* A signal was caught; try again */\n-\t\t\t\tcontinue;\n-\t\t\telse if (errno == ENOMEM)\n-\t\t\t\tdie_errno(\"BUG: keeping track of fds is hard\");\n-\t\t\telse if (errno == EINVAL)\n-\t\t\t\tdie_errno(\"BUG: invalid arguments to select\");\n-\t\t\telse if (errno == EBADF)\n-\t\t\t\tdie_errno(\"BUG: keeping track of fds is hard\");\n-\t\t\telse\n-\t\t\t\tdie_errno(\"Unknown error with select\");\n+\t\tunblock_fd(pp->children[i].err);\n+\n+\t\tpp->nr_processes++;\n+\t\tpp->slots[i] = 1;\n+\t\tpp->pfd[i].fd = pp->children[i].err;\n+\t}\n+}\n+\n+static int run_processes_parallel_buffer_stderr(struct parallel_processes *pp)\n+{\n+\tint i;\n+\ti = poll(pp->pfd, pp->max_number_processes, 100);\n+\tif (i < 0) {\n+\t\tif (errno == EINTR)\n+\t\t\t/* A signal was caught; try again */\n+\t\t\treturn -1;\n+\t\telse {\n+\t\t\trun_processes_parallel_cleanup(pp);\n+\t\t\tdie_errno(\"poll\");\n \t\t}\n+\t}\n \n-\t\t/* Buffer output from all pipes. */\n-\t\tfor (i = 0; i < n; i++) {\n-\t\t\tif (!slots[i])\n-\t\t\t\tcontinue;\n-\t\t\tif (FD_ISSET(children[i].err, &fdset))\n-\t\t\t\tstrbuf_read_noblock(&err[i], children[i].err, 0);\n-\t\t\tif (child_in_foreground == i) {\n-\t\t\t\tfputs(err[i].buf, stderr);\n-\t\t\t\tstrbuf_reset(&err[i]);\n-\t\t\t\tfflush(stderr);\n-\t\t\t}\n+\t/* Buffer output from all pipes. */\n+\tfor (i = 0; i < pp->max_number_processes; i++) {\n+\t\tif (!pp->slots[i])\n+\t\t\tcontinue;\n+\t\tif (pp->pfd[i].revents & POLLIN)\n+\t\t\tstrbuf_read_noblock(&pp->err[i], pp->children[i].err, 0);\n+\t\tif (pp->foreground_child == i) {\n+\t\t\tfputs(pp->err[i].buf, stderr);\n+\t\t\tstrbuf_reset(&pp->err[i]);\n \t\t}\n+\t}\n+\treturn 0;\n+}\n \n-\t\t/* Collect finished child processes. */\n-\t\twhile (nr_processes > 0) {\n-\t\t\tpid = waitpid(-1, &wait_status, WNOHANG);\n-\t\t\tif (pid == 0)\n-\t\t\t\t/* no child finished */\n-\t\t\t\tbreak;\n-\n-\t\t\tif (pid < 0) {\n-\t\t\t\tif (errno == EINTR)\n-\t\t\t\t\tbreak; /* just try again next time */\n-\t\t\t\tif (errno == EINVAL || errno == ECHILD)\n-\t\t\t\t\tdie_errno(\"wait\");\n-\t\t\t} else {\n-\t\t\t\t/* Find the finished child. */\n-\t\t\t\tfor (i = 0; i < n; i++)\n-\t\t\t\t\tif (slots[i] && pid == children[i].pid)\n-\t\t\t\t\t\tbreak;\n-\t\t\t\tif (i == n)\n-\t\t\t\t\t/* waitpid returned another process id which\n-\t\t\t\t\t * we are not waiting on, so ignore it*/\n+\n+static void run_processes_parallel_collect_finished(struct parallel_processes *pp)\n+{\n+\tint i = 0;\n+\tpid_t pid;\n+\tint wait_status, code;\n+\tint n = pp->max_number_processes;\n+\t/* Collect finished child processes. */\n+\twhile (pp->nr_processes > 0) {\n+\t\tpid = waitpid(-1, &wait_status, WNOHANG);\n+\t\tif (pid == 0)\n+\t\t\treturn; /* no child finished */\n+\n+\t\tif (pid < 0) {\n+\t\t\tif (errno == EINTR)\n+\t\t\t\treturn; /* just try again next time */\n+\t\t\tif (errno == EINVAL || errno == ECHILD)\n+\t\t\t\tdie_errno(\"wait\");\n+\t\t} else {\n+\t\t\t/* Find the finished child. */\n+\t\t\tfor (i = 0; i < pp->max_number_processes; i++)\n+\t\t\t\tif (pp->slots[i] && pid == pp->children[i].pid)\n \t\t\t\t\tbreak;\n-\t\t\t}\n+\t\t\tif (i == pp->max_number_processes)\n+\t\t\t\t/*\n+\t\t\t\t * waitpid returned another process id\n+\t\t\t\t * which we are not waiting for.\n+\t\t\t\t */\n+\t\t\t\treturn;\n+\t\t}\n+\t\tstrbuf_read_noblock(&pp->err[i], pp->children[i].err, 0);\n \n-\t\t\tstrbuf_read_noblock(&err[i], children[i].err, 0);\n-\t\t\targv_array_clear(&children[i].args);\n-\t\t\targv_array_clear(&children[i].env_array);\n+\t\tif (determine_return_value(wait_status, &code, &errno,\n+\t\t\t\t\t pp->children[i].argv[0]) < 0)\n+\t\t\terror(\"waitpid is confused (%s)\",\n+\t\t\t pp->children[i].argv[0]);\n \n-\t\t\tslots[i] = 0;\n-\t\t\tnr_processes--;\n+\t\tpp->fn_exit(pp->data, &pp->children[i], code);\n \n-\t\t\tif (i != child_in_foreground) {\n-\t\t\t\tstrbuf_addbuf(&finished_children, &err[i]);\n-\t\t\t\tstrbuf_reset(&err[i]);\n-\t\t\t} else {\n-\t\t\t\tfputs(err[i].buf, stderr);\n-\t\t\t\tstrbuf_reset(&err[i]);\n+\t\targv_array_clear(&pp->children[i].args);\n+\t\targv_array_clear(&pp->children[i].env_array);\n \n-\t\t\t\t/* Output all other finished child processes */\n-\t\t\t\tfputs(finished_children.buf, stderr);\n-\t\t\t\tstrbuf_reset(&finished_children);\n+\t\tpp->nr_processes--;\n+\t\tpp->slots[i] = 0;\n+\t\tpp->pfd[i].fd = -1;\n \n-\t\t\t\t/*\n-\t\t\t\t * Pick next process to output live.\n-\t\t\t\t * There can be no active process if n==1\n-\t\t\t\t * NEEDSWORK:\n-\t\t\t\t * For now we pick it randomly by doing a round\n-\t\t\t\t * robin. Later we may want to pick the one with\n-\t\t\t\t * the most output or the longest or shortest\n-\t\t\t\t * running process time.\n-\t\t\t\t */\n-\t\t\t\tfor (i = 0; i < n; i++)\n-\t\t\t\t\tif (slots[(child_in_foreground + i) % n])\n-\t\t\t\t\t\tbreak;\n-\t\t\t\tchild_in_foreground = (child_in_foreground + i) % n;\n-\t\t\t\tfputs(err[child_in_foreground].buf, stderr);\n-\t\t\t\tstrbuf_reset(&err[child_in_foreground]);\n-\t\t\t}\n+\t\tif (i != pp->foreground_child) {\n+\t\t\tstrbuf_addbuf(&pp->finished_children, &pp->err[i]);\n+\t\t\tstrbuf_reset(&pp->err[i]);\n+\t\t} else {\n+\t\t\tfputs(pp->err[i].buf, stderr);\n+\t\t\tstrbuf_reset(&pp->err[i]);\n+\n+\t\t\t/* Output all other finished child processes */\n+\t\t\tfputs(pp->finished_children.buf, stderr);\n+\t\t\tstrbuf_reset(&pp->finished_children);\n+\n+\t\t\t/*\n+\t\t\t * Pick next process to output live.\n+\t\t\t * NEEDSWORK:\n+\t\t\t * For now we pick it randomly by doing a round\n+\t\t\t * robin. Later we may want to pick the one with\n+\t\t\t * the most output or the longest or shortest\n+\t\t\t * running process time.\n+\t\t\t */\n+\t\t\tfor (i = 0; i < n; i++)\n+\t\t\t\tif (pp->slots[(pp->foreground_child + i) % n])\n+\t\t\t\t\tbreak;\n+\t\t\tpp->foreground_child = (pp->foreground_child + i) % n;\n+\t\t\tfputs(pp->err[pp->foreground_child].buf, stderr);\n+\t\t\tstrbuf_reset(&pp->err[pp->foreground_child]);\n \t\t}\n \t}\n+}\n+\n+int run_processes_parallel(int n, void *data,\n+\t\t\t get_next_task fn,\n+\t\t\t handle_child_starting_failure fn_err,\n+\t\t\t handle_child_return_value fn_exit)\n+{\n+\tstruct parallel_processes pp;\n+\trun_processes_parallel_init(&pp, n, data, fn, fn_err, fn_exit);\n+\n+\twhile (!pp.all_tasks_started || pp.nr_processes > 0) {\n+\t\trun_processes_parallel_start_new(&pp);\n+\t\tif (run_processes_parallel_buffer_stderr(&pp))\n+\t\t\tcontinue;\n+\t\trun_processes_parallel_collect_finished(&pp);\n+\t}\n+\trun_processes_parallel_cleanup(&pp);\n+\n \treturn 0;\n }\ndiff --git a/run-command.h b/run-command.h\nindex 8f53ad6..0487f71 100644\n--- a/run-command.h\n+++ b/run-command.h\n@@ -120,32 +120,39 @@ int start_async(struct async *async);\n int finish_async(struct async *async);\n \n /**\n- * Return 0 if the next child is ready to run.\n- * This callback takes care to initialize the child process and preload the\n- * out and error channel. The preloading of these outpout channels is useful\n- * if you want to have a message printed directly before the output of the\n- * child process.\n+ * This callback should initialize the child process and preload the\n+ * error channel. The preloading of is useful if you want to have a message\n+ * printed directly before the output of the child process.\n+ * You MUST set stdout_to_stderr.\n *\n+ * Return 0 if the next child is ready to run.\n * Return != 0 if there are no more tasks to be processed.\n */\n typedef int (*get_next_task)(void *data,\n \t\t\t struct child_process *cp,\n \t\t\t struct strbuf *err);\n \n+typedef void (*handle_child_starting_failure)(void *data,\n+\t\t\t\t\t struct child_process *cp,\n+\t\t\t\t\t struct strbuf *err);\n+\n+typedef void (*handle_child_return_value)(void *data,\n+\t\t\t\t\t struct child_process *cp,\n+\t\t\t\t\t int result);\n+\n /**\n- * Runs up to n processes at the same time. Whenever a process can\n- * be started, the callback `get_next_task` is called to obtain the\n- * data fed to the child process.\n+ * Runs up to n processes at the same time. Whenever a process can be\n+ * started, the callback `get_next_task` is called to obtain the data\n+ * fed to the child process.\n *\n * The children started via this function run in parallel and their output\n- * to both stdout and stderr is buffered, while one of the children will\n- * directly output to stdout/stderr.\n- *\n- * This leads to a problem with output from processes which put out to\n- * stdout/err alternatingly as the buffering will not be able to replay\n- * the\n+ * to stderr is buffered, while one of the children will directly output\n+ * to stderr.\n */\n \n-int run_processes_async(int n, get_next_task fn, void *data);\n+int run_processes_parallel(int n, void *data,\n+\t\t\t get_next_task fn,\n+\t\t\t handle_child_starting_failure,\n+\t\t\t handle_child_return_value);\n \n #endif\ndiff --git a/submodule.c b/submodule.c\nindex 6d757c6..a0e06e8 100644\n--- a/submodule.c\n+++ b/submodule.c\n@@ -623,17 +623,32 @@ struct submodule_parallel_fetch {\n \tconst char *prefix;\n \tint command_line_option;\n \tint quiet;\n+\tint result;\n };\n-#define SPF_INIT {0, ARGV_ARRAY_INIT, NULL}\n+#define SPF_INIT {0, ARGV_ARRAY_INIT, NULL, NULL, 0, 0, 0}\n \n int get_next_submodule(void *data, struct child_process *cp,\n \t\t struct strbuf *err);\n \n+void handle_submodule_fetch_start_err(void *data, struct child_process *cp, struct strbuf *err)\n+{\n+\tstruct submodule_parallel_fetch *spf = data;\n+\tspf->result = 1;\n+}\n+\n+void handle_submodule_fetch_finish( void *data, struct child_process *cp, int retvalue)\n+{\n+\tstruct submodule_parallel_fetch *spf = data;\n+\n+\tif (retvalue)\n+\t\tspf->result = 1;\n+}\n+\n int fetch_populated_submodules(const struct argv_array *options,\n \t\t\t const char *prefix, int command_line_option,\n \t\t\t int quiet, int max_parallel_jobs)\n {\n-\tint i, result = 0;\n+\tint i;\n \tstruct submodule_parallel_fetch spf = SPF_INIT;\n \tspf.work_tree = get_git_work_tree();\n \tspf.command_line_option = command_line_option;\n@@ -652,12 +667,15 @@ int fetch_populated_submodules(const struct argv_array *options,\n \t/* default value, \"--submodule-prefix\" and its value are added later */\n \n \tcalculate_changed_submodule_paths();\n-\trun_processes_async(max_parallel_jobs, get_next_submodule, &spf);\n+\trun_processes_parallel(max_parallel_jobs, &spf,\n+\t\t\t get_next_submodule,\n+\t\t\t handle_submodule_fetch_start_err,\n+\t\t\t handle_submodule_fetch_finish);\n \n \targv_array_clear(&spf.args);\n out:\n \tstring_list_clear(&changed_submodule_paths, 1);\n-\treturn result;\n+\treturn spf.result;\n }\n \n int get_next_submodule(void *data, struct child_process *cp,\ndiff --git a/t/t0061-run-command.sh b/t/t0061-run-command.sh\nindex 0970fb0..37c89b9 100755\n--- a/t/t0061-run-command.sh\n+++ b/t/t0061-run-command.sh\n@@ -48,18 +48,22 @@ test_expect_success POSIXPERM,SANITY 'unreadable directory in PATH' '\n '\n \n cat >expect <<-EOF\n-Now running instance 0\n-Hello World\n-Now running instance 1\n-Hello World\n-Now running instance 2\n-Hello World\n-Now running instance 3\n-Hello World\n+preloaded output of a child\n+Hello\n+World\n+preloaded output of a child\n+Hello\n+World\n+preloaded output of a child\n+Hello\n+World\n+preloaded output of a child\n+Hello\n+World\n EOF\n \n test_expect_success 'run_command runs in parallel' '\n-\ttest-run-command run-command-async sh -c \"echo Hello World >&2;\" 2>actual &&\n+\ttest-run-command run-command-async sh -c \"printf \\\"%s\\n%s\\n\\\" Hello World\" 2>actual &&\n \ttest_cmp expect actual\n '\n \ndiff --git a/test-run-command.c b/test-run-command.c\nindex 4817f6e..71fd3ca 100644\n--- a/test-run-command.c\n+++ b/test-run-command.c\n@@ -16,9 +16,9 @@\n #include \n \n static int number_callbacks;\n-int run_processes_async_next(void *data,\n-\t\t\t struct child_process *cp,\n-\t\t\t struct strbuf *err)\n+int parallel_next(void *data,\n+\t\t struct child_process *cp,\n+\t\t struct strbuf *err)\n {\n \tstruct child_process *d = data;\n \tif (number_callbacks >= 4)\n@@ -28,7 +28,7 @@ int run_processes_async_next(void *data,\n \tcp->stdout_to_stderr = 1;\n \tcp->no_stdin = 1;\n \tcp->err = -1;\n-\tstrbuf_addf(err, \"Now running instance %d\\n\", number_callbacks);\n+\tstrbuf_addf(err, \"preloaded output of a child\\n\");\n \tnumber_callbacks++;\n \treturn 0;\n }\n@@ -51,7 +51,8 @@ int main(int argc, char **argv)\n \t\texit(run_command(&proc));\n \n \tif (!strcmp(argv[1], \"run-command-async\"))\n-\t\texit(run_processes_async(4, run_processes_async_next, &proc));\n+\t\texit(run_processes_parallel(4, &proc, parallel_next,\n+\t\t\t\t\t NULL, NULL));\n \n \tfprintf(stderr, \"check usage\\n\");\n \treturn 1;\n\n-- \n2.6.0.rc0.131.gf624c3d\n"},{"id":"269962","messageId":"1442267451-2527-2-git-send-email-sbeller@google.com","threadId":"40335","inReplyTo":"1442267451-2527-1-git-send-email-sbeller@google.com","subject":"[PATCHv2] fetch: fetch submodules in parallel","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-09-14T21:50:51Z","receivedAt":"2015-09-14T21:50:51Z","isPatch":false,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"If we run external commands in parallel we cannot pipe the output directly\nto the our stdout/err as it would mix up. So each process's output will\nflow through a pipe, which we buffer. One subprocess can be directly\npiped to out stdout/err for a low latency feedback to the user.\n\nExample:\nLet's assume we have 5 submodules A,B,C,D,E and each fetch takes a\ndifferent amount of time as the different submodules vary in size, then\nthe output of fetches in sequential order might look like this:\n\n time -->\n output: |---A---| |-B-| |----C-----------| |-D-| |-E-|\n\nWhen we schedule these submodules into maximal two parallel processes,\na schedule and sample output over time may look like this:\n\nthread 1: |---A---| |-D-| |-E-|\n\nthread 2: |-B-| |----C-----------|\n\noutput: |---A---|B|------C-------|DE\n\nSo A will be perceived as it would run normally in the single child\nversion. As B has finished by the time A is done, we can dump its whole\nprogress buffer on stderr, such that it looks like it finished in no time.\nOnce that is done, C is determined to be the visible child and its progress\nwill be reported in real time.\n\nSo this way of output is really good for human consumption,\nas it only changes the timing, not the actual output.\n\nFor machine consumption the output needs to be prepared in\nthe tasks, by either having a prefix per line or per block\nto indicate whose tasks output is displayed.\n\nSigned-off-by: Stefan Beller \n---\n Documentation/fetch-options.txt | 7 +\n builtin/fetch.c | 6 +-\n builtin/pull.c | 6 +\n run-command.c | 278 ++++++++++++++++++++++++++++++++++++----\n run-command.h | 36 ++++++\n strbuf.c | 31 +++++\n strbuf.h | 1 +\n submodule.c | 119 ++++++++++++-----\n submodule.h | 2 +-\n t/t0061-run-command.sh | 20 +++\n t/t5526-fetch-submodules.sh | 19 +++\n test-run-command.c | 24 ++++\n 12 files changed, 490 insertions(+), 59 deletions(-)\n\ndiff --git a/Documentation/fetch-options.txt b/Documentation/fetch-options.txt\nindex 45583d8..6b109f6 100644\n--- a/Documentation/fetch-options.txt\n+++ b/Documentation/fetch-options.txt\n@@ -100,6 +100,13 @@ ifndef::git-pull[]\n \treference to a commit that isn't already in the local submodule\n \tclone.\n \n+-j::\n+--jobs=::\n+\tNumber of parallel children to be used for fetching submodules.\n+\tEach will fetch from different submodules, such that fetching many\n+\tsubmodules will be faster. By default submodules will be fetched\n+\tone at a time.\n+\n --no-recurse-submodules::\n \tDisable recursive fetching of submodules (this has the same effect as\n \tusing the '--recurse-submodules=no' option).\ndiff --git a/builtin/fetch.c b/builtin/fetch.c\nindex ee1f1a9..f28eac6 100644\n--- a/builtin/fetch.c\n+++ b/builtin/fetch.c\n@@ -37,6 +37,7 @@ static int prune = -1; /* unspecified */\n static int all, append, dry_run, force, keep, multiple, update_head_ok, verbosity;\n static int progress = -1, recurse_submodules = RECURSE_SUBMODULES_DEFAULT;\n static int tags = TAGS_DEFAULT, unshallow, update_shallow;\n+static int max_children = 1;\n static const char *depth;\n static const char *upload_pack;\n static struct strbuf default_rla = STRBUF_INIT;\n@@ -99,6 +100,8 @@ static struct option builtin_fetch_options[] = {\n \t\t N_(\"fetch all tags and associated objects\"), TAGS_SET),\n \tOPT_SET_INT('n', NULL, &tags,\n \t\t N_(\"do not fetch all tags (--no-tags)\"), TAGS_UNSET),\n+\tOPT_INTEGER('j', \"jobs\", &max_children,\n+\t\t N_(\"number of submodules fetched in parallel\")),\n \tOPT_BOOL('p', \"prune\", &prune,\n \t\t N_(\"prune remote-tracking branches no longer on remote\")),\n \t{ OPTION_CALLBACK, 0, \"recurse-submodules\", NULL, N_(\"on-demand\"),\n@@ -1217,7 +1220,8 @@ int cmd_fetch(int argc, const char **argv, const char *prefix)\n \t\tresult = fetch_populated_submodules(&options,\n \t\t\t\t\t\t submodule_prefix,\n \t\t\t\t\t\t recurse_submodules,\n-\t\t\t\t\t\t verbosity < 0);\n+\t\t\t\t\t\t verbosity < 0,\n+\t\t\t\t\t\t max_children);\n \t\targv_array_clear(&options);\n \t}\n \ndiff --git a/builtin/pull.c b/builtin/pull.c\nindex 722a83c..f0af196 100644\n--- a/builtin/pull.c\n+++ b/builtin/pull.c\n@@ -94,6 +94,7 @@ static int opt_force;\n static char *opt_tags;\n static char *opt_prune;\n static char *opt_recurse_submodules;\n+static char *max_children;\n static int opt_dry_run;\n static char *opt_keep;\n static char *opt_depth;\n@@ -177,6 +178,9 @@ static struct option pull_options[] = {\n \t\tN_(\"on-demand\"),\n \t\tN_(\"control recursive fetching of submodules\"),\n \t\tPARSE_OPT_OPTARG),\n+\tOPT_PASSTHRU('j', \"jobs\", &max_children, N_(\"n\"),\n+\t\tN_(\"number of submodules pulled in parallel\"),\n+\t\tPARSE_OPT_OPTARG),\n \tOPT_BOOL(0, \"dry-run\", &opt_dry_run,\n \t\tN_(\"dry run\")),\n \tOPT_PASSTHRU('k', \"keep\", &opt_keep, NULL,\n@@ -524,6 +528,8 @@ static int run_fetch(const char *repo, const char **refspecs)\n \t\targv_array_push(&args, opt_prune);\n \tif (opt_recurse_submodules)\n \t\targv_array_push(&args, opt_recurse_submodules);\n+\tif (max_children)\n+\t\targv_array_push(&args, max_children);\n \tif (opt_dry_run)\n \t\targv_array_push(&args, \"--dry-run\");\n \tif (opt_keep)\ndiff --git a/run-command.c b/run-command.c\nindex 28e1d55..6f6f9fb 100644\n--- a/run-command.c\n+++ b/run-command.c\n@@ -232,6 +232,35 @@ static inline void set_cloexec(int fd)\n \t\tfcntl(fd, F_SETFD, flags | FD_CLOEXEC);\n }\n \n+static int determine_return_value(int wait_status,\n+\t\t\t\t int *result,\n+\t\t\t\t int *error_code,\n+\t\t\t\t const char *argv0)\n+{\n+\tif (WIFSIGNALED(wait_status)) {\n+\t\t*result = WTERMSIG(wait_status);\n+\t\tif (*result != SIGINT && *result != SIGQUIT)\n+\t\t\terror(\"%s died of signal %d\", argv0, *result);\n+\t\t/*\n+\t\t * This return value is chosen so that code & 0xff\n+\t\t * mimics the exit code that a POSIX shell would report for\n+\t\t * a program that died from this signal.\n+\t\t */\n+\t\t*result += 128;\n+\t} else if (WIFEXITED(wait_status)) {\n+\t\t*result = WEXITSTATUS(wait_status);\n+\t\t/*\n+\t\t * Convert special exit code when execvp failed.\n+\t\t */\n+\t\tif (*result == 127) {\n+\t\t\t*result = -1;\n+\t\t\t*error_code = ENOENT;\n+\t\t}\n+\t} else\n+\t\treturn 1;\n+\treturn 0;\n+}\n+\n static int wait_or_whine(pid_t pid, const char *argv0)\n {\n \tint status, code = -1;\n@@ -244,29 +273,10 @@ static int wait_or_whine(pid_t pid, const char *argv0)\n \tif (waiting < 0) {\n \t\tfailed_errno = errno;\n \t\terror(\"waitpid for %s failed: %s\", argv0, strerror(errno));\n-\t} else if (waiting != pid) {\n-\t\terror(\"waitpid is confused (%s)\", argv0);\n-\t} else if (WIFSIGNALED(status)) {\n-\t\tcode = WTERMSIG(status);\n-\t\tif (code != SIGINT && code != SIGQUIT)\n-\t\t\terror(\"%s died of signal %d\", argv0, code);\n-\t\t/*\n-\t\t * This return value is chosen so that code & 0xff\n-\t\t * mimics the exit code that a POSIX shell would report for\n-\t\t * a program that died from this signal.\n-\t\t */\n-\t\tcode += 128;\n-\t} else if (WIFEXITED(status)) {\n-\t\tcode = WEXITSTATUS(status);\n-\t\t/*\n-\t\t * Convert special exit code when execvp failed.\n-\t\t */\n-\t\tif (code == 127) {\n-\t\t\tcode = -1;\n-\t\t\tfailed_errno = ENOENT;\n-\t\t}\n \t} else {\n-\t\terror(\"waitpid is confused (%s)\", argv0);\n+\t\tif (waiting != pid\n+\t\t || (determine_return_value(status, &code, &failed_errno, argv0) < 0))\n+\t\t\terror(\"waitpid is confused (%s)\", argv0);\n \t}\n \n \tclear_child_for_cleanup(pid);\n@@ -852,3 +862,227 @@ int capture_command(struct child_process *cmd, struct strbuf *buf, size_t hint)\n \tclose(cmd->out);\n \treturn finish_command(cmd);\n }\n+\n+static void unblock_fd(int fd)\n+{\n+\tint flags = fcntl(fd, F_GETFL);\n+\tif (flags < 0) {\n+\t\twarning(\"Could not get file status flags, \"\n+\t\t\t\"output will be degraded\");\n+\t\treturn;\n+\t}\n+\tif (fcntl(fd, F_SETFL, flags | O_NONBLOCK)) {\n+\t\t\twarning(\"Could not set file status flags, \"\n+\t\t\t\"output will be degraded\");\n+\t\treturn;\n+\t}\n+}\n+\n+struct parallel_processes {\n+\tint max_number_processes;\n+\tvoid *data;\n+\tget_next_task fn;\n+\thandle_child_starting_failure fn_err;\n+\thandle_child_return_value fn_exit;\n+\n+\tint nr_processes;\n+\tint all_tasks_started;\n+\tint foreground_child;\n+\tchar *slots;\n+\tstruct child_process *children;\n+\tstruct pollfd *pfd;\n+\tstruct strbuf *err;\n+\tstruct strbuf finished_children;\n+};\n+\n+static void run_processes_parallel_init(struct parallel_processes *pp,\n+\t\t\t\t\tint n, void *data,\n+\t\t\t\t\tget_next_task fn,\n+\t\t\t\t\thandle_child_starting_failure fn_err,\n+\t\t\t\t\thandle_child_return_value fn_exit)\n+{\n+\tint i;\n+\n+\tpp->max_number_processes = n;\n+\tpp->data = data;\n+\tpp->fn = fn;\n+\tpp->fn_err = fn_err;\n+\tpp->fn_exit = fn_exit;\n+\n+\tpp->nr_processes = 0;\n+\tpp->all_tasks_started = 0;\n+\tpp->foreground_child = 0;\n+\tpp->slots = xcalloc(n, sizeof(*pp->slots));\n+\tpp->children = xcalloc(n, sizeof(*pp->children));\n+\tpp->pfd = xcalloc(n, sizeof(*pp->pfd));\n+\tpp->err = xcalloc(n, sizeof(*pp->err));\n+\tstrbuf_init(&pp->finished_children, 0);\n+\n+\tfor (i = 0; i < n; i++) {\n+\t\tstrbuf_init(&pp->err[i], 0);\n+\t\tpp->pfd[i].events = POLLIN;\n+\t\tpp->pfd[i].fd = -1;\n+\t}\n+}\n+\n+static void run_processes_parallel_cleanup(struct parallel_processes *pp)\n+{\n+\tint i;\n+\tfor (i = 0; i < pp->max_number_processes; i++)\n+\t\tstrbuf_release(&pp->err[i]);\n+\n+\tfree(pp->children);\n+\tfree(pp->slots);\n+\tfree(pp->pfd);\n+\tfree(pp->err);\n+\tstrbuf_release(&pp->finished_children);\n+}\n+\n+static void run_processes_parallel_start_new(struct parallel_processes *pp)\n+{\n+\tint i;\n+\t/* Start new processes. */\n+\twhile (!pp->all_tasks_started\n+\t && pp->nr_processes < pp->max_number_processes) {\n+\t\tfor (i = 0; i < pp->max_number_processes; i++)\n+\t\t\tif (!pp->slots[i])\n+\t\t\t\tbreak; /* found an empty slot */\n+\t\tif (i == pp->max_number_processes)\n+\t\t\tdie(\"BUG: bookkeeping is hard\");\n+\n+\t\tif (pp->fn(pp->data, &pp->children[i], &pp->err[i])) {\n+\t\t\tpp->all_tasks_started = 1;\n+\t\t\tbreak;\n+\t\t}\n+\t\tif (start_command(&pp->children[i]))\n+\t\t\tpp->fn_err(pp->data, &pp->children[i], &pp->err[i]);\n+\n+\t\tunblock_fd(pp->children[i].err);\n+\n+\t\tpp->nr_processes++;\n+\t\tpp->slots[i] = 1;\n+\t\tpp->pfd[i].fd = pp->children[i].err;\n+\t}\n+}\n+\n+static int run_processes_parallel_buffer_stderr(struct parallel_processes *pp)\n+{\n+\tint i;\n+\ti = poll(pp->pfd, pp->max_number_processes, 100);\n+\tif (i < 0) {\n+\t\tif (errno == EINTR)\n+\t\t\t/* A signal was caught; try again */\n+\t\t\treturn -1;\n+\t\telse {\n+\t\t\trun_processes_parallel_cleanup(pp);\n+\t\t\tdie_errno(\"poll\");\n+\t\t}\n+\t}\n+\n+\t/* Buffer output from all pipes. */\n+\tfor (i = 0; i < pp->max_number_processes; i++) {\n+\t\tif (!pp->slots[i])\n+\t\t\tcontinue;\n+\t\tif (pp->pfd[i].revents & POLLIN)\n+\t\t\tstrbuf_read_noblock(&pp->err[i], pp->children[i].err, 0);\n+\t\tif (pp->foreground_child == i) {\n+\t\t\tfputs(pp->err[i].buf, stderr);\n+\t\t\tstrbuf_reset(&pp->err[i]);\n+\t\t}\n+\t}\n+\treturn 0;\n+}\n+\n+\n+static void run_processes_parallel_collect_finished(struct parallel_processes *pp)\n+{\n+\tint i = 0;\n+\tpid_t pid;\n+\tint wait_status, code;\n+\tint n = pp->max_number_processes;\n+\t/* Collect finished child processes. */\n+\twhile (pp->nr_processes > 0) {\n+\t\tpid = waitpid(-1, &wait_status, WNOHANG);\n+\t\tif (pid == 0)\n+\t\t\treturn; /* no child finished */\n+\n+\t\tif (pid < 0) {\n+\t\t\tif (errno == EINTR)\n+\t\t\t\treturn; /* just try again next time */\n+\t\t\tif (errno == EINVAL || errno == ECHILD)\n+\t\t\t\tdie_errno(\"wait\");\n+\t\t} else {\n+\t\t\t/* Find the finished child. */\n+\t\t\tfor (i = 0; i < pp->max_number_processes; i++)\n+\t\t\t\tif (pp->slots[i] && pid == pp->children[i].pid)\n+\t\t\t\t\tbreak;\n+\t\t\tif (i == pp->max_number_processes)\n+\t\t\t\t/*\n+\t\t\t\t * waitpid returned another process id\n+\t\t\t\t * which we are not waiting for.\n+\t\t\t\t */\n+\t\t\t\treturn;\n+\t\t}\n+\t\tstrbuf_read_noblock(&pp->err[i], pp->children[i].err, 0);\n+\n+\t\tif (determine_return_value(wait_status, &code, &errno,\n+\t\t\t\t\t pp->children[i].argv[0]) < 0)\n+\t\t\terror(\"waitpid is confused (%s)\",\n+\t\t\t pp->children[i].argv[0]);\n+\n+\t\tpp->fn_exit(pp->data, &pp->children[i], code);\n+\n+\t\targv_array_clear(&pp->children[i].args);\n+\t\targv_array_clear(&pp->children[i].env_array);\n+\n+\t\tpp->nr_processes--;\n+\t\tpp->slots[i] = 0;\n+\t\tpp->pfd[i].fd = -1;\n+\n+\t\tif (i != pp->foreground_child) {\n+\t\t\tstrbuf_addbuf(&pp->finished_children, &pp->err[i]);\n+\t\t\tstrbuf_reset(&pp->err[i]);\n+\t\t} else {\n+\t\t\tfputs(pp->err[i].buf, stderr);\n+\t\t\tstrbuf_reset(&pp->err[i]);\n+\n+\t\t\t/* Output all other finished child processes */\n+\t\t\tfputs(pp->finished_children.buf, stderr);\n+\t\t\tstrbuf_reset(&pp->finished_children);\n+\n+\t\t\t/*\n+\t\t\t * Pick next process to output live.\n+\t\t\t * NEEDSWORK:\n+\t\t\t * For now we pick it randomly by doing a round\n+\t\t\t * robin. Later we may want to pick the one with\n+\t\t\t * the most output or the longest or shortest\n+\t\t\t * running process time.\n+\t\t\t */\n+\t\t\tfor (i = 0; i < n; i++)\n+\t\t\t\tif (pp->slots[(pp->foreground_child + i) % n])\n+\t\t\t\t\tbreak;\n+\t\t\tpp->foreground_child = (pp->foreground_child + i) % n;\n+\t\t\tfputs(pp->err[pp->foreground_child].buf, stderr);\n+\t\t\tstrbuf_reset(&pp->err[pp->foreground_child]);\n+\t\t}\n+\t}\n+}\n+\n+int run_processes_parallel(int n, void *data,\n+\t\t\t get_next_task fn,\n+\t\t\t handle_child_starting_failure fn_err,\n+\t\t\t handle_child_return_value fn_exit)\n+{\n+\tstruct parallel_processes pp;\n+\trun_processes_parallel_init(&pp, n, data, fn, fn_err, fn_exit);\n+\n+\twhile (!pp.all_tasks_started || pp.nr_processes > 0) {\n+\t\trun_processes_parallel_start_new(&pp);\n+\t\tif (run_processes_parallel_buffer_stderr(&pp))\n+\t\t\tcontinue;\n+\t\trun_processes_parallel_collect_finished(&pp);\n+\t}\n+\trun_processes_parallel_cleanup(&pp);\n+\n+\treturn 0;\n+}\ndiff --git a/run-command.h b/run-command.h\nindex 5b4425a..0487f71 100644\n--- a/run-command.h\n+++ b/run-command.h\n@@ -119,4 +119,40 @@ struct async {\n int start_async(struct async *async);\n int finish_async(struct async *async);\n \n+/**\n+ * This callback should initialize the child process and preload the\n+ * error channel. The preloading of is useful if you want to have a message\n+ * printed directly before the output of the child process.\n+ * You MUST set stdout_to_stderr.\n+ *\n+ * Return 0 if the next child is ready to run.\n+ * Return != 0 if there are no more tasks to be processed.\n+ */\n+typedef int (*get_next_task)(void *data,\n+\t\t\t struct child_process *cp,\n+\t\t\t struct strbuf *err);\n+\n+typedef void (*handle_child_starting_failure)(void *data,\n+\t\t\t\t\t struct child_process *cp,\n+\t\t\t\t\t struct strbuf *err);\n+\n+typedef void (*handle_child_return_value)(void *data,\n+\t\t\t\t\t struct child_process *cp,\n+\t\t\t\t\t int result);\n+\n+/**\n+ * Runs up to n processes at the same time. Whenever a process can be\n+ * started, the callback `get_next_task` is called to obtain the data\n+ * fed to the child process.\n+ *\n+ * The children started via this function run in parallel and their output\n+ * to stderr is buffered, while one of the children will directly output\n+ * to stderr.\n+ */\n+\n+int run_processes_parallel(int n, void *data,\n+\t\t\t get_next_task fn,\n+\t\t\t handle_child_starting_failure,\n+\t\t\t handle_child_return_value);\n+\n #endif\ndiff --git a/strbuf.c b/strbuf.c\nindex cce5eed..7f866c3 100644\n--- a/strbuf.c\n+++ b/strbuf.c\n@@ -384,6 +384,37 @@ ssize_t strbuf_read(struct strbuf *sb, int fd, size_t hint)\n \treturn sb->len - oldlen;\n }\n \n+ssize_t strbuf_read_noblock(struct strbuf *sb, int fd, size_t hint)\n+{\n+\tsize_t oldlen = sb->len;\n+\tsize_t oldalloc = sb->alloc;\n+\n+\tstrbuf_grow(sb, hint ? hint : 8192);\n+\tfor (;;) {\n+\t\tssize_t cnt;\n+\n+\t\tcnt = read(fd, sb->buf + sb->len, sb->alloc - sb->len - 1);\n+\t\tif (cnt < 0) {\n+\t\t\tif (errno == EINTR)\n+\t\t\t\tcontinue;\n+\t\t\tif (errno == EAGAIN)\n+\t\t\t\tbreak;\n+\t\t\tif (oldalloc == 0)\n+\t\t\t\tstrbuf_release(sb);\n+\t\t\telse\n+\t\t\t\tstrbuf_setlen(sb, oldlen);\n+\t\t\treturn -1;\n+\t\t}\n+\t\tif (!cnt)\n+\t\t\tbreak;\n+\t\tsb->len += cnt;\n+\t\tstrbuf_grow(sb, 8192);\n+\t}\n+\n+\tsb->buf[sb->len] = '\\0';\n+\treturn sb->len - oldlen;\n+}\n+\n #define STRBUF_MAXLINK (2*PATH_MAX)\n \n int strbuf_readlink(struct strbuf *sb, const char *path, size_t hint)\ndiff --git a/strbuf.h b/strbuf.h\nindex aef2794..7ea462b 100644\n--- a/strbuf.h\n+++ b/strbuf.h\n@@ -365,6 +365,7 @@ extern size_t strbuf_fread(struct strbuf *, size_t, FILE *);\n * any partial read is undone.\n */\n extern ssize_t strbuf_read(struct strbuf *, int fd, size_t hint);\n+extern ssize_t strbuf_read_noblock(struct strbuf *, int fd, size_t hint);\n \n /**\n * Read the contents of a file, specified by its path. The third argument\ndiff --git a/submodule.c b/submodule.c\nindex 1d64e57..a0e06e8 100644\n--- a/submodule.c\n+++ b/submodule.c\n@@ -12,6 +12,7 @@\n #include \"sha1-array.h\"\n #include \"argv-array.h\"\n #include \"blob.h\"\n+#include \"thread-utils.h\"\n \n static int config_fetch_recurse_submodules = RECURSE_SUBMODULES_ON_DEMAND;\n static struct string_list changed_submodule_paths;\n@@ -615,37 +616,79 @@ static void calculate_changed_submodule_paths(void)\n \tinitialized_fetch_ref_tips = 0;\n }\n \n+struct submodule_parallel_fetch {\n+\tint count;\n+\tstruct argv_array args;\n+\tconst char *work_tree;\n+\tconst char *prefix;\n+\tint command_line_option;\n+\tint quiet;\n+\tint result;\n+};\n+#define SPF_INIT {0, ARGV_ARRAY_INIT, NULL, NULL, 0, 0, 0}\n+\n+int get_next_submodule(void *data, struct child_process *cp,\n+\t\t struct strbuf *err);\n+\n+void handle_submodule_fetch_start_err(void *data, struct child_process *cp, struct strbuf *err)\n+{\n+\tstruct submodule_parallel_fetch *spf = data;\n+\tspf->result = 1;\n+}\n+\n+void handle_submodule_fetch_finish( void *data, struct child_process *cp, int retvalue)\n+{\n+\tstruct submodule_parallel_fetch *spf = data;\n+\n+\tif (retvalue)\n+\t\tspf->result = 1;\n+}\n+\n int fetch_populated_submodules(const struct argv_array *options,\n \t\t\t const char *prefix, int command_line_option,\n-\t\t\t int quiet)\n+\t\t\t int quiet, int max_parallel_jobs)\n {\n-\tint i, result = 0;\n-\tstruct child_process cp = CHILD_PROCESS_INIT;\n-\tstruct argv_array argv = ARGV_ARRAY_INIT;\n-\tconst char *work_tree = get_git_work_tree();\n-\tif (!work_tree)\n+\tint i;\n+\tstruct submodule_parallel_fetch spf = SPF_INIT;\n+\tspf.work_tree = get_git_work_tree();\n+\tspf.command_line_option = command_line_option;\n+\tspf.quiet = quiet;\n+\tspf.prefix = prefix;\n+\tif (!spf.work_tree)\n \t\tgoto out;\n \n \tif (read_cache() < 0)\n \t\tdie(\"index file corrupt\");\n \n-\targv_array_push(&argv, \"fetch\");\n+\targv_array_push(&spf.args, \"fetch\");\n \tfor (i = 0; i < options->argc; i++)\n-\t\targv_array_push(&argv, options->argv[i]);\n-\targv_array_push(&argv, \"--recurse-submodules-default\");\n+\t\targv_array_push(&spf.args, options->argv[i]);\n+\targv_array_push(&spf.args, \"--recurse-submodules-default\");\n \t/* default value, \"--submodule-prefix\" and its value are added later */\n \n-\tcp.env = local_repo_env;\n-\tcp.git_cmd = 1;\n-\tcp.no_stdin = 1;\n-\n \tcalculate_changed_submodule_paths();\n+\trun_processes_parallel(max_parallel_jobs, &spf,\n+\t\t\t get_next_submodule,\n+\t\t\t handle_submodule_fetch_start_err,\n+\t\t\t handle_submodule_fetch_finish);\n+\n+\targv_array_clear(&spf.args);\n+out:\n+\tstring_list_clear(&changed_submodule_paths, 1);\n+\treturn spf.result;\n+}\n+\n+int get_next_submodule(void *data, struct child_process *cp,\n+\t\t struct strbuf *err)\n+{\n+\tint ret = 0;\n+\tstruct submodule_parallel_fetch *spf = data;\n \n-\tfor (i = 0; i < active_nr; i++) {\n+\tfor ( ; spf->count < active_nr; spf->count++) {\n \t\tstruct strbuf submodule_path = STRBUF_INIT;\n \t\tstruct strbuf submodule_git_dir = STRBUF_INIT;\n \t\tstruct strbuf submodule_prefix = STRBUF_INIT;\n-\t\tconst struct cache_entry *ce = active_cache[i];\n+\t\tconst struct cache_entry *ce = active_cache[spf->count];\n \t\tconst char *git_dir, *default_argv;\n \t\tconst struct submodule *submodule;\n \n@@ -657,7 +700,7 @@ int fetch_populated_submodules(const struct argv_array *options,\n \t\t\tsubmodule = submodule_from_name(null_sha1, ce->name);\n \n \t\tdefault_argv = \"yes\";\n-\t\tif (command_line_option == RECURSE_SUBMODULES_DEFAULT) {\n+\t\tif (spf->command_line_option == RECURSE_SUBMODULES_DEFAULT) {\n \t\t\tif (submodule &&\n \t\t\t submodule->fetch_recurse !=\n \t\t\t\t\t\tRECURSE_SUBMODULES_NONE) {\n@@ -680,40 +723,46 @@ int fetch_populated_submodules(const struct argv_array *options,\n \t\t\t\t\tdefault_argv = \"on-demand\";\n \t\t\t\t}\n \t\t\t}\n-\t\t} else if (command_line_option == RECURSE_SUBMODULES_ON_DEMAND) {\n+\t\t} else if (spf->command_line_option == RECURSE_SUBMODULES_ON_DEMAND) {\n \t\t\tif (!unsorted_string_list_lookup(&changed_submodule_paths, ce->name))\n \t\t\t\tcontinue;\n \t\t\tdefault_argv = \"on-demand\";\n \t\t}\n \n-\t\tstrbuf_addf(&submodule_path, \"%s/%s\", work_tree, ce->name);\n+\t\tstrbuf_addf(&submodule_path, \"%s/%s\", spf->work_tree, ce->name);\n \t\tstrbuf_addf(&submodule_git_dir, \"%s/.git\", submodule_path.buf);\n-\t\tstrbuf_addf(&submodule_prefix, \"%s%s/\", prefix, ce->name);\n+\t\tstrbuf_addf(&submodule_prefix, \"%s%s/\", spf->prefix, ce->name);\n \t\tgit_dir = read_gitfile(submodule_git_dir.buf);\n \t\tif (!git_dir)\n \t\t\tgit_dir = submodule_git_dir.buf;\n \t\tif (is_directory(git_dir)) {\n-\t\t\tif (!quiet)\n-\t\t\t\tfprintf(stderr, \"Fetching submodule %s%s\\n\", prefix, ce->name);\n-\t\t\tcp.dir = submodule_path.buf;\n-\t\t\targv_array_push(&argv, default_argv);\n-\t\t\targv_array_push(&argv, \"--submodule-prefix\");\n-\t\t\targv_array_push(&argv, submodule_prefix.buf);\n-\t\t\tcp.argv = argv.argv;\n-\t\t\tif (run_command(&cp))\n-\t\t\t\tresult = 1;\n-\t\t\targv_array_pop(&argv);\n-\t\t\targv_array_pop(&argv);\n-\t\t\targv_array_pop(&argv);\n+\t\t\tchild_process_init(cp);\n+\t\t\tcp->dir = strbuf_detach(&submodule_path, NULL);\n+\t\t\tcp->git_cmd = 1;\n+\t\t\tcp->no_stdout = 1;\n+\t\t\tcp->no_stdin = 1;\n+\t\t\tcp->stdout_to_stderr = 1;\n+\t\t\tcp->err = -1;\n+\t\t\tcp->env = local_repo_env;\n+\t\t\tif (!spf->quiet)\n+\t\t\t\tstrbuf_addf(err, \"Fetching submodule %s%s\\n\",\n+\t\t\t\t\t spf->prefix, ce->name);\n+\t\t\targv_array_init(&cp->args);\n+\t\t\targv_array_pushv(&cp->args, spf->args.argv);\n+\t\t\targv_array_push(&cp->args, default_argv);\n+\t\t\targv_array_push(&cp->args, \"--submodule-prefix\");\n+\t\t\targv_array_push(&cp->args, submodule_prefix.buf);\n+\t\t\tret = 1;\n \t\t}\n \t\tstrbuf_release(&submodule_path);\n \t\tstrbuf_release(&submodule_git_dir);\n \t\tstrbuf_release(&submodule_prefix);\n+\t\tif (ret) {\n+\t\t\tspf->count++;\n+\t\t\treturn 0;\n+\t\t}\n \t}\n-\targv_array_clear(&argv);\n-out:\n-\tstring_list_clear(&changed_submodule_paths, 1);\n-\treturn result;\n+\treturn 1;\n }\n \n unsigned is_submodule_modified(const char *path, int ignore_untracked)\ndiff --git a/submodule.h b/submodule.h\nindex 5507c3d..cbc0003 100644\n--- a/submodule.h\n+++ b/submodule.h\n@@ -31,7 +31,7 @@ void set_config_fetch_recurse_submodules(int value);\n void check_for_new_submodule_commits(unsigned char new_sha1[20]);\n int fetch_populated_submodules(const struct argv_array *options,\n \t\t\t const char *prefix, int command_line_option,\n-\t\t\t int quiet);\n+\t\t\t int quiet, int max_parallel_jobs);\n unsigned is_submodule_modified(const char *path, int ignore_untracked);\n int submodule_uses_gitfile(const char *path);\n int ok_to_remove_submodule(const char *path);\ndiff --git a/t/t0061-run-command.sh b/t/t0061-run-command.sh\nindex 9acf628..37c89b9 100755\n--- a/t/t0061-run-command.sh\n+++ b/t/t0061-run-command.sh\n@@ -47,4 +47,24 @@ test_expect_success POSIXPERM,SANITY 'unreadable directory in PATH' '\n \ttest_cmp expect actual\n '\n \n+cat >expect <<-EOF\n+preloaded output of a child\n+Hello\n+World\n+preloaded output of a child\n+Hello\n+World\n+preloaded output of a child\n+Hello\n+World\n+preloaded output of a child\n+Hello\n+World\n+EOF\n+\n+test_expect_success 'run_command runs in parallel' '\n+\ttest-run-command run-command-async sh -c \"printf \\\"%s\\n%s\\n\\\" Hello World\" 2>actual &&\n+\ttest_cmp expect actual\n+'\n+\n test_done\ndiff --git a/t/t5526-fetch-submodules.sh b/t/t5526-fetch-submodules.sh\nindex 17759b1..1b4ce69 100755\n--- a/t/t5526-fetch-submodules.sh\n+++ b/t/t5526-fetch-submodules.sh\n@@ -71,6 +71,16 @@ test_expect_success \"fetch --recurse-submodules recurses into submodules\" '\n \ttest_i18ncmp expect.err actual.err\n '\n \n+test_expect_success \"fetch --recurse-submodules -j2 has the same output behaviour\" '\n+\tadd_upstream_commit &&\n+\t(\n+\t\tcd downstream &&\n+\t\tgit fetch --recurse-submodules -j2 2>../actual.err\n+\t) &&\n+\ttest_must_be_empty actual.out &&\n+\ttest_i18ncmp expect.err actual.err\n+'\n+\n test_expect_success \"fetch alone only fetches superproject\" '\n \tadd_upstream_commit &&\n \t(\n@@ -140,6 +150,15 @@ test_expect_success \"--quiet propagates to submodules\" '\n \t! test -s actual.err\n '\n \n+test_expect_success \"--quiet propagates to parallel submodules\" '\n+\t(\n+\t\tcd downstream &&\n+\t\tgit fetch --recurse-submodules -j 2 --quiet >../actual.out 2>../actual.err\n+\t) &&\n+\t! test -s actual.out &&\n+\t! test -s actual.err\n+'\n+\n test_expect_success \"--dry-run propagates to submodules\" '\n \tadd_upstream_commit &&\n \t(\ndiff --git a/test-run-command.c b/test-run-command.c\nindex 89c7de2..71fd3ca 100644\n--- a/test-run-command.c\n+++ b/test-run-command.c\n@@ -10,9 +10,29 @@\n \n #include \"git-compat-util.h\"\n #include \"run-command.h\"\n+#include \"argv-array.h\"\n+#include \"strbuf.h\"\n #include \n #include \n \n+static int number_callbacks;\n+int parallel_next(void *data,\n+\t\t struct child_process *cp,\n+\t\t struct strbuf *err)\n+{\n+\tstruct child_process *d = data;\n+\tif (number_callbacks >= 4)\n+\t\treturn 1;\n+\n+\targv_array_pushv(&cp->args, d->argv);\n+\tcp->stdout_to_stderr = 1;\n+\tcp->no_stdin = 1;\n+\tcp->err = -1;\n+\tstrbuf_addf(err, \"preloaded output of a child\\n\");\n+\tnumber_callbacks++;\n+\treturn 0;\n+}\n+\n int main(int argc, char **argv)\n {\n \tstruct child_process proc = CHILD_PROCESS_INIT;\n@@ -30,6 +50,10 @@ int main(int argc, char **argv)\n \tif (!strcmp(argv[1], \"run-command\"))\n \t\texit(run_command(&proc));\n \n+\tif (!strcmp(argv[1], \"run-command-async\"))\n+\t\texit(run_processes_parallel(4, &proc, parallel_next,\n+\t\t\t\t\t NULL, NULL));\n+\n \tfprintf(stderr, \"check usage\\n\");\n \treturn 1;\n }\n-- \n2.6.0.rc0.131.gf624c3d\n"},{"id":"269964","messageId":"xmqqmvwoveox.fsf@gitster.mtv.corp.google.com","threadId":"40335","inReplyTo":"1442267451-2527-1-git-send-email-sbeller@google.com","subject":"Re: [PATCHv2] fetch submodules in parallel","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-09-14T22:06:22Z","receivedAt":"2015-09-14T22:06:22Z","isPatch":false,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"Stefan Beller writes:\n\n> This replaces the last patch of the \"Parallel git submodule fetching\"\n> series. Changes:\n>\n> * have correct return code in submodule fetching when one submodule fails\n> * use poll instead of select now\n> * broke down into more smaller functions instead of one giant.\n> (I think it is an improvement, but I wouldn't be surprised if someone objects)\n> * closed memory leaks\n> * document the need for stdout_to_stderr\n>\n> I don't deem it RFC-ish any more but good to go.\n\nI didn't say this in the previous round because it smelled like an\nRFC, but for a real submission, 2/2 may be doing too many things at\nonce. I suspect this is more or less \"taste\" thing, so I won't mind\ntoo much as long as the reviewers are OK with it.\n"}]}