Re: [PATCH 7/9] fetch: fetch submodules in parallel

[Stefan Beller <sbeller@google.com>]

On Thu, Aug 27, 2015 at 6:14 PM, Stefan Beller <sbeller@google.com> wrote:
> This makes use of the new task queue and the syncing feature of
> run-command to fetch a number of submodules at the same time.
>
> The output will look like it would have been run sequential,
> but faster.

And it breaks the tests t5526-fetch-submodules.sh as the output is done
on stderr only, instead of putting "Fetching submodule <submodule-path>
to stdout. :(

I guess combining stdout and stderr is not a good strategy after all now.
Advantages:
* The order is preserved if everything is in one stream.
Disadvantages:
* It's a change, which may not be expected.
* It's harder to make it machine parsable, as that probably
   relied on having 2 channels.

So now I need to come up with a way to either buffer 2 channels at the same
time, or we need to redefine parallel submodule fetch output semantics a bit.

>
> Signed-off-by: Stefan Beller <sbeller@google.com>
> ---
>  Documentation/fetch-options.txt |   7 +++
>  builtin/fetch.c                 |   6 ++-
>  builtin/pull.c                  |   6 +++
>  submodule.c                     | 100 +++++++++++++++++++++++++++++++++-------
>  submodule.h                     |   2 +-
>  5 files changed, 102 insertions(+), 19 deletions(-)
>
> diff --git a/Documentation/fetch-options.txt b/Documentation/fetch-options.txt
> index 45583d8..e2a59c3 100644
> --- a/Documentation/fetch-options.txt
> +++ b/Documentation/fetch-options.txt
> @@ -100,6 +100,13 @@ ifndef::git-pull[]
>         reference to a commit that isn't already in the local submodule
>         clone.
>
> +-j::
> +--jobs=<n>::
> +       Number of threads to be used for fetching submodules. Each thread
> +       will fetch from different submodules, such that fetching many
> +       submodules will be faster. By default the number of cpus will
> +       be used .
> +
>  --no-recurse-submodules::
>         Disable recursive fetching of submodules (this has the same effect as
>         using the '--recurse-submodules=no' option).
> diff --git a/builtin/fetch.c b/builtin/fetch.c
> index ee1f1a9..636707e 100644
> --- a/builtin/fetch.c
> +++ b/builtin/fetch.c
> @@ -37,6 +37,7 @@ static int prune = -1; /* unspecified */
>  static int all, append, dry_run, force, keep, multiple, update_head_ok, verbosity;
>  static int progress = -1, recurse_submodules = RECURSE_SUBMODULES_DEFAULT;
>  static int tags = TAGS_DEFAULT, unshallow, update_shallow;
> +static int max_threads;
>  static const char *depth;
>  static const char *upload_pack;
>  static struct strbuf default_rla = STRBUF_INIT;
> @@ -99,6 +100,8 @@ static struct option builtin_fetch_options[] = {
>                     N_("fetch all tags and associated objects"), TAGS_SET),
>         OPT_SET_INT('n', NULL, &tags,
>                     N_("do not fetch all tags (--no-tags)"), TAGS_UNSET),
> +       OPT_INTEGER('j', "jobs", &max_threads,
> +                   N_("number of threads used for fetching")),
>         OPT_BOOL('p', "prune", &prune,
>                  N_("prune remote-tracking branches no longer on remote")),
>         { OPTION_CALLBACK, 0, "recurse-submodules", NULL, N_("on-demand"),
> @@ -1217,7 +1220,8 @@ int cmd_fetch(int argc, const char **argv, const char *prefix)
>                 result = fetch_populated_submodules(&options,
>                                                     submodule_prefix,
>                                                     recurse_submodules,
> -                                                   verbosity < 0);
> +                                                   verbosity < 0,
> +                                                   max_threads);
>                 argv_array_clear(&options);
>         }
>
> diff --git a/builtin/pull.c b/builtin/pull.c
> index 722a83c..fbbda67 100644
> --- a/builtin/pull.c
> +++ b/builtin/pull.c
> @@ -94,6 +94,7 @@ static int opt_force;
>  static char *opt_tags;
>  static char *opt_prune;
>  static char *opt_recurse_submodules;
> +static char *max_threads;
>  static int opt_dry_run;
>  static char *opt_keep;
>  static char *opt_depth;
> @@ -177,6 +178,9 @@ static struct option pull_options[] = {
>                 N_("on-demand"),
>                 N_("control recursive fetching of submodules"),
>                 PARSE_OPT_OPTARG),
> +       OPT_PASSTHRU('j', "jobs", &max_threads, N_("n"),
> +               N_("number of threads used for fetching submodules"),
> +               PARSE_OPT_OPTARG),
>         OPT_BOOL(0, "dry-run", &opt_dry_run,
>                 N_("dry run")),
>         OPT_PASSTHRU('k', "keep", &opt_keep, NULL,
> @@ -524,6 +528,8 @@ static int run_fetch(const char *repo, const char **refspecs)
>                 argv_array_push(&args, opt_prune);
>         if (opt_recurse_submodules)
>                 argv_array_push(&args, opt_recurse_submodules);
> +       if (max_threads)
> +               argv_array_push(&args, max_threads);
>         if (opt_dry_run)
>                 argv_array_push(&args, "--dry-run");
>         if (opt_keep)
> diff --git a/submodule.c b/submodule.c
> index 9fcc86f..50266a8 100644
> --- a/submodule.c
> +++ b/submodule.c
> @@ -12,6 +12,7 @@
>  #include "sha1-array.h"
>  #include "argv-array.h"
>  #include "blob.h"
> +#include "thread-utils.h"
>
>  static int config_fetch_recurse_submodules = RECURSE_SUBMODULES_ON_DEMAND;
>  static struct string_list changed_submodule_paths;
> @@ -615,13 +616,79 @@ static void calculate_changed_submodule_paths(void)
>         initialized_fetch_ref_tips = 0;
>  }
>
> +struct submodule_parallel_fetch {
> +       struct child_process cp;
> +       struct argv_array argv;
> +       int *result;
> +};
> +
> +#ifndef NO_PTHREADS
> +static pthread_mutex_t output_mutex = PTHREAD_MUTEX_INITIALIZER;
> +
> +static void set_output_mutex(struct submodule_parallel_fetch *spf)
> +{
> +       spf->cp.sync_mutex = &output_mutex;
> +}
> +
> +#define lock_output_mutex() pthread_mutex_lock(&output_mutex)
> +
> +#define unlock_output_mutex() pthread_mutex_unlock(&output_mutex)
> +
> +static void destroy_output_mutex()
> +{
> +       pthread_mutex_destroy(&output_mutex);
> +}
> +
> +#else
> +#define set_output_mutex()
> +#define destroy_output_mutex()
> +#define lock_output_mutex()
> +#define unlock_output_mutex()
> +#endif
> +
> +static struct submodule_parallel_fetch *submodule_parallel_fetch_create()
> +{
> +       struct submodule_parallel_fetch *spf = xmalloc(sizeof(*spf));
> +       child_process_init(&spf->cp);
> +       spf->cp.env = local_repo_env;
> +       spf->cp.git_cmd = 1;
> +       spf->cp.no_stdin = 1;
> +       spf->cp.stdout_to_stderr = 1;
> +       spf->cp.sync_buf = xmalloc(sizeof(spf->cp.sync_buf));
> +       strbuf_init(spf->cp.sync_buf, 0);
> +
> +       argv_array_init(&spf->argv);
> +       return spf;
> +}
> +
> +static int run_command_and_cleanup(struct task_queue *tq, void *arg)
> +{
> +       int code;
> +       struct submodule_parallel_fetch *spf = arg;
> +
> +       spf->cp.argv = spf->argv.argv;
> +
> +       code = run_command(&spf->cp);
> +       if (code) {
> +               lock_output_mutex();
> +               *spf->result = code;
> +               unlock_output_mutex();
> +       }
> +
> +       argv_array_clear(&spf->argv);
> +       free((char*)spf->cp.dir);
> +       free(spf);
> +       return 0;
> +}
> +
>  int fetch_populated_submodules(const struct argv_array *options,
>                                const char *prefix, int command_line_option,
> -                              int quiet)
> +                              int quiet, int max_parallel_jobs)
>  {
>         int i, result = 0;
> -       struct child_process cp = CHILD_PROCESS_INIT;
> +       struct task_queue *tq;
>         struct argv_array argv = ARGV_ARRAY_INIT;
> +       struct submodule_parallel_fetch *spf;
>         const char *work_tree = get_git_work_tree();
>         if (!work_tree)
>                 goto out;
> @@ -635,12 +702,9 @@ int fetch_populated_submodules(const struct argv_array *options,
>         argv_array_push(&argv, "--recurse-submodules-default");
>         /* default value, "--submodule-prefix" and its value are added later */
>
> -       cp.env = local_repo_env;
> -       cp.git_cmd = 1;
> -       cp.no_stdin = 1;
> -
>         calculate_changed_submodule_paths();
>
> +       tq = create_task_queue(max_parallel_jobs);
>         for (i = 0; i < active_nr; i++) {
>                 struct strbuf submodule_path = STRBUF_INIT;
>                 struct strbuf submodule_git_dir = STRBUF_INIT;
> @@ -693,24 +757,26 @@ int fetch_populated_submodules(const struct argv_array *options,
>                 if (!git_dir)
>                         git_dir = submodule_git_dir.buf;
>                 if (is_directory(git_dir)) {
> +                       spf = submodule_parallel_fetch_create();
> +                       spf->result = &result;
> +                       spf->cp.dir = strbuf_detach(&submodule_path, NULL);
>                         if (!quiet)
> -                               printf("Fetching submodule %s%s\n", prefix, ce->name);
> -                       cp.dir = submodule_path.buf;
> -                       argv_array_push(&argv, default_argv);
> -                       argv_array_push(&argv, "--submodule-prefix");
> -                       argv_array_push(&argv, submodule_prefix.buf);
> -                       cp.argv = argv.argv;
> -                       if (run_command(&cp))
> -                               result = 1;
> -                       argv_array_pop(&argv);
> -                       argv_array_pop(&argv);
> -                       argv_array_pop(&argv);
> +                               strbuf_addf(spf->cp.sync_buf,
> +                                           "Fetching submodule %s%s",
> +                                           prefix, ce->name);
> +                       argv_array_pushv(&spf->argv, argv.argv);
> +                       argv_array_push(&spf->argv, default_argv);
> +                       argv_array_push(&spf->argv, "--submodule-prefix");
> +                       argv_array_push(&spf->argv, submodule_prefix.buf);
> +                       set_output_mutex(spf);
> +                       add_task(tq, run_command_and_cleanup, spf);
>                 }
>                 strbuf_release(&submodule_path);
>                 strbuf_release(&submodule_git_dir);
>                 strbuf_release(&submodule_prefix);
>         }
>         argv_array_clear(&argv);
> +       destroy_output_mutex();
>  out:
>         string_list_clear(&changed_submodule_paths, 1);
>         return result;
> diff --git a/submodule.h b/submodule.h
> index 5507c3d..cbc0003 100644
> --- a/submodule.h
> +++ b/submodule.h
> @@ -31,7 +31,7 @@ void set_config_fetch_recurse_submodules(int value);
>  void check_for_new_submodule_commits(unsigned char new_sha1[20]);
>  int fetch_populated_submodules(const struct argv_array *options,
>                                const char *prefix, int command_line_option,
> -                              int quiet);
> +                              int quiet, int max_parallel_jobs);
>  unsigned is_submodule_modified(const char *path, int ignore_untracked);
>  int submodule_uses_gitfile(const char *path);
>  int ok_to_remove_submodule(const char *path);
> --
> 2.5.0.264.g5e52b0d
>