threads / rfc / 40028

RFC patch, 4 partsAdd a workdispatcher to get work done in parallel

Subject: [RFC PATCH 2/4] Add a workdispatcher to get work done in parallel

## tl;dr

11 messages between Aug 6, 2015 and Aug 6, 2015. Diffs are folded; open one to read it.

replies: 10people: 4as markdown or json

Stefan Beller· Aug 6, 2015, 17:35 UTC · lore

[RFC/PATCH 0/4] parallel fetch for submodules

When I was looking at the branches of Jens for work done on submodules not yet upstream I found a commit "WIP threaded submodule fetching[1], and I was side tracked wanting to present a different approach to that.

The first patch is a bit unrelated as it relates to the rewrite of git-submodule.sh but also has code in submodule.c and the following patches modify code just around that, so I did not remove that patch from this series. It is the same I sent yesterday.

The next patch 2/4 presents a framework for parallel threaded work. It allows to setup a worker pool of <n> threads and then have a queue of tasks which are worked on by the threads. The patch is a the one which I'd request most comments on as I think that can be reused in a variety of situations (parallel checkout of files, parallel fetch of different remotes, or such).

I consider the third patch farely boring as it adds argv_array_copy, so I would not expect much discussion there.

The last patch 4/4 presents the new workdispatcher from 2/4 in use with just one unsolved problem of how to handle the output of the parallel commands to stdout and stderr. It may be useful to put handling of parallel outputs into the work dispatcher.

[1] https://github.com/jlehmann/git-submod-enhancements/commit/47597753206d40e234a47392e258065c9489e2b3

This series applies on top of origin/sb/submodule-helper (d2c6c09ac819, submodule: implement `module_list` as a builtin helper) and can also be found at https://github.com/stefanbeller/git/tree/parallel-submodule-fetch

Stefan Beller (4):
  submodule: implement `module_name` as a builtin helper
  Add a workdispatcher to get work done in parallel
  argv_array: add argv_array_clone to clone an existing argv array
  submodule: add infrastructure to fetch submodules in parallel
 Makefile                    |   1 +
 argv-array.c                |  13 ++++
 argv-array.h                |   1 +
 builtin/fetch.c             |   3 +-
 builtin/submodule--helper.c |  23 ++++++
 git-submodule.sh            |  32 ++------
 submodule.c                 |  92 ++++++++++++++++------
 submodule.h                 |   3 +-
 workdispatcher.c            | 184 ++++++++++++++++++++++++++++++++++++++++++++
 workdispatcher.h            |  29 +++++++
 10 files changed, 332 insertions(+), 49 deletions(-)
 create mode 100644 workdispatcher.c
 create mode 100644 workdispatcher.h
-- 
2.5.0.239.g9728e1d.dirty
Stefan Beller· Aug 6, 2015, 17:35 UTC · re: Stefan Beller · lore

[PATCH 1/4] submodule: implement `module_name` as a builtin helper

This implements the helper `module_name` in C instead of shell, yielding a nice performance boost.

Before this patch, I measured a time (best out of three):
  $ time ./t7400-submodule-basic.sh  >/dev/null
    real	0m11.066s
    user	0m3.348s
    sys	0m8.534s
With this patch applied I measured (also best out of three)
  $ time ./t7400-submodule-basic.sh  >/dev/null
    real	0m10.063s
    user	0m3.044s
    sys	0m7.487s
Signed-off-by: Stefan Beller <sbeller@google.com>
---
 builtin/submodule--helper.c | 23 +++++++++++++++++++++++
 git-submodule.sh            | 32 +++++++-------------------------
 submodule.c                 | 18 +++++++++++++-----
 submodule.h                 |  1 +
 4 files changed, 44 insertions(+), 30 deletions(-)
Show changes to 4 files +44 −30

builtin/submodule--helper.c, git-submodule.sh, submodule.c, submodule.h

diff --git a/builtin/submodule--helper.c b/builtin/submodule--helper.c
index cb18ddf..3713c4c 100644
--- a/builtin/submodule--helper.c
+++ b/builtin/submodule--helper.c
@@ -5,6 +5,8 @@
 #include "pathspec.h"
 #include "dir.h"
 #include "utf8.h"
+#include "submodule.h"
+#include "string-list.h"
 
 static char *ps_matched;
 static const struct cache_entry **ce_entries;
@@ -98,6 +100,24 @@ static int module_list(int argc, const char **argv, const char *prefix)
 	return 0;
 }
 
+static int module_name(int argc, const char **argv, const char *prefix)
+{
+	const char *name;
+
+	if (argc != 1)
+		usage("git submodule--helper module_name <path>\n");
+
+	gitmodules_config();
+	name = submodule_name_for_path(argv[0]);
+
+	if (name)
+		printf("%s\n", name);
+	else
+		die("No submodule mapping found in .gitmodules for path '%s'", argv[0]);
+
+	return 0;
+}
+
 int cmd_submodule__helper(int argc, const char **argv, const char *prefix)
 {
 	if (argc < 2)
@@ -106,6 +126,9 @@ int cmd_submodule__helper(int argc, const char **argv, const char *prefix)
 	if (!strcmp(argv[1], "module_list"))
 		return module_list(argc - 1, argv + 1, prefix);
 
+	if (!strcmp(argv[1], "module_name"))
+		return module_name(argc - 2, argv + 2, prefix);
+
 usage:
 	usage("git submodule--helper module_list\n");
 }
diff --git a/git-submodule.sh b/git-submodule.sh
index af9ecef..e6ff38d 100755
--- a/git-submodule.sh
+++ b/git-submodule.sh
@@ -178,24 +178,6 @@ get_submodule_config () {
 	printf '%s' "${value:-$default}"
 }
 
-
-#
-# Map submodule path to submodule name
-#
-# $1 = path
-#
-module_name()
-{
-	# Do we have "submodule.<something>.path = $1" defined in .gitmodules file?
-	sm_path="$1"
-	re=$(printf '%s\n' "$1" | sed -e 's/[].[^$\\*]/\\&/g')
-	name=$( git config -f .gitmodules --get-regexp '^submodule\..*\.path$' |
-		sed -n -e 's|^submodule\.\(.*\)\.path '"$re"'$|\1|p' )
-	test -z "$name" &&
-	die "$(eval_gettext "No submodule mapping found in .gitmodules for path '\$sm_path'")"
-	printf '%s\n' "$name"
-}
-
 #
 # Clone a submodule
 #
@@ -498,7 +480,7 @@ cmd_foreach()
 		then
 			displaypath=$(relative_path "$sm_path")
 			say "$(eval_gettext "Entering '\$prefix\$displaypath'")"
-			name=$(module_name "$sm_path")
+			name=$(git submodule--helper module_name "$sm_path")
 			(
 				prefix="$prefix$sm_path/"
 				clear_local_git_env
@@ -554,7 +536,7 @@ cmd_init()
 	while read mode sha1 stage sm_path
 	do
 		die_if_unmatched "$mode"
-		name=$(module_name "$sm_path") || exit
+		name=$(git submodule--helper module_name "$sm_path") || exit
 
 		displaypath=$(relative_path "$sm_path")
 
@@ -636,7 +618,7 @@ cmd_deinit()
 	while read mode sha1 stage sm_path
 	do
 		die_if_unmatched "$mode"
-		name=$(module_name "$sm_path") || exit
+		name=$(git submodule--helper module_name "$sm_path") || exit
 
 		displaypath=$(relative_path "$sm_path")
 
@@ -758,7 +740,7 @@ cmd_update()
 			echo >&2 "Skipping unmerged submodule $prefix$sm_path"
 			continue
 		fi
-		name=$(module_name "$sm_path") || exit
+		name=$(git submodule--helper module_name "$sm_path") || exit
 		url=$(git config submodule."$name".url)
 		branch=$(get_submodule_config "$name" branch master)
 		if ! test -z "$update"
@@ -1022,7 +1004,7 @@ cmd_summary() {
 			# Respect the ignore setting for --for-status.
 			if test -n "$for_status"
 			then
-				name=$(module_name "$sm_path")
+				name=$(git submodule--helper module_name "$sm_path")
 				ignore_config=$(get_submodule_config "$name" ignore none)
 				test $status != A && test $ignore_config = all && continue
 			fi
@@ -1184,7 +1166,7 @@ cmd_status()
 	while read mode sha1 stage sm_path
 	do
 		die_if_unmatched "$mode"
-		name=$(module_name "$sm_path") || exit
+		name=$(git submodule--helper module_name "$sm_path") || exit
 		url=$(git config submodule."$name".url)
 		displaypath=$(relative_path "$prefix$sm_path")
 		if test "$stage" = U
@@ -1261,7 +1243,7 @@ cmd_sync()
 	while read mode sha1 stage sm_path
 	do
 		die_if_unmatched "$mode"
-		name=$(module_name "$sm_path")
+		name=$(git submodule--helper module_name "$sm_path")
 		url=$(git config -f .gitmodules --get submodule."$name".url)
 
 		# Possibly a url relative to parent
diff --git a/submodule.c b/submodule.c
index 15e90d1..872967f 100644
--- a/submodule.c
+++ b/submodule.c
@@ -686,6 +686,16 @@ static void calculate_changed_submodule_paths(void)
 	initialized_fetch_ref_tips = 0;
 }
 
+const char* submodule_name_for_path(const char* path)
+{
+	struct string_list_item *item;
+	item = unsorted_string_list_lookup(&config_name_for_path, path);
+	if (item)
+		return item->util;
+	else
+		return NULL;
+}
+
 int fetch_populated_submodules(const struct argv_array *options,
 			       const char *prefix, int command_line_option,
 			       int quiet)
@@ -693,7 +703,7 @@ int fetch_populated_submodules(const struct argv_array *options,
 	int i, result = 0;
 	struct child_process cp = CHILD_PROCESS_INIT;
 	struct argv_array argv = ARGV_ARRAY_INIT;
-	struct string_list_item *name_for_path;
+	const char *name_for_path;
 	const char *work_tree = get_git_work_tree();
 	if (!work_tree)
 		goto out;
@@ -723,10 +733,8 @@ int fetch_populated_submodules(const struct argv_array *options,
 		if (!S_ISGITLINK(ce->ce_mode))
 			continue;
 
-		name = ce->name;
-		name_for_path = unsorted_string_list_lookup(&config_name_for_path, ce->name);
-		if (name_for_path)
-			name = name_for_path->util;
+		name_for_path = submodule_name_for_path(ce->name);
+		name =  name_for_path ? name_for_path : ce->name;
 
 		default_argv = "yes";
 		if (command_line_option == RECURSE_SUBMODULES_DEFAULT) {
diff --git a/submodule.h b/submodule.h
index 7beec48..e3dd854 100644
--- a/submodule.h
+++ b/submodule.h
@@ -41,5 +41,6 @@ int find_unpushed_submodules(unsigned char new_sha1[20], const char *remotes_nam
 		struct string_list *needs_pushing);
 int push_unpushed_submodules(unsigned char new_sha1[20], const char *remotes_name);
 void connect_work_tree_and_git_dir(const char *work_tree, const char *git_dir);
+const char* submodule_name_for_path(const char* path);
 
 #endif
-- 
2.5.0.239.g9728e1d.dirty
Jens Lehmann· Aug 6, 2015, 19:49 UTC · re: Stefan Beller · lore

Re: [PATCH 1/4] submodule: implement `module_name` as a builtin helper

Am 06.08.2015 um 19:35 schrieb Stefan Beller:
Show 19 quoted lines
> This implements the helper `module_name` in C instead of shell,
> yielding a nice performance boost.
>
> Before this patch, I measured a time (best out of three):
>
>    $ time ./t7400-submodule-basic.sh  >/dev/null
>      real	0m11.066s
>      user	0m3.348s
>      sys	0m8.534s
>
> With this patch applied I measured (also best out of three)
>
>    $ time ./t7400-submodule-basic.sh  >/dev/null
>      real	0m10.063s
>      user	0m3.044s
>      sys	0m7.487s
>
> Signed-off-by: Stefan Beller <sbeller@google.com>
> ---
Please see my comments in the other thread concerning this patch.

And wouldn't it make more sense to keep this patch together with the "submodule: implement `module_list` as a builtin helper" in its own "submodule-helper" series and have the following three patches in a separate "parallel fetch for submodules" series?

Show 201 quoted lines
>   builtin/submodule--helper.c | 23 +++++++++++++++++++++++
>   git-submodule.sh            | 32 +++++++-------------------------
>   submodule.c                 | 18 +++++++++++++-----
>   submodule.h                 |  1 +
>   4 files changed, 44 insertions(+), 30 deletions(-)
>
> diff --git a/builtin/submodule--helper.c b/builtin/submodule--helper.c
> index cb18ddf..3713c4c 100644
> --- a/builtin/submodule--helper.c
> +++ b/builtin/submodule--helper.c
> @@ -5,6 +5,8 @@
>   #include "pathspec.h"
>   #include "dir.h"
>   #include "utf8.h"
> +#include "submodule.h"
> +#include "string-list.h"
>
>   static char *ps_matched;
>   static const struct cache_entry **ce_entries;
> @@ -98,6 +100,24 @@ static int module_list(int argc, const char **argv, const char *prefix)
>   	return 0;
>   }
>
> +static int module_name(int argc, const char **argv, const char *prefix)
> +{
> +	const char *name;
> +
> +	if (argc != 1)
> +		usage("git submodule--helper module_name <path>\n");
> +
> +	gitmodules_config();
> +	name = submodule_name_for_path(argv[0]);
> +
> +	if (name)
> +		printf("%s\n", name);
> +	else
> +		die("No submodule mapping found in .gitmodules for path '%s'", argv[0]);
> +
> +	return 0;
> +}
> +
>   int cmd_submodule__helper(int argc, const char **argv, const char *prefix)
>   {
>   	if (argc < 2)
> @@ -106,6 +126,9 @@ int cmd_submodule__helper(int argc, const char **argv, const char *prefix)
>   	if (!strcmp(argv[1], "module_list"))
>   		return module_list(argc - 1, argv + 1, prefix);
>
> +	if (!strcmp(argv[1], "module_name"))
> +		return module_name(argc - 2, argv + 2, prefix);
> +
>   usage:
>   	usage("git submodule--helper module_list\n");
>   }
> diff --git a/git-submodule.sh b/git-submodule.sh
> index af9ecef..e6ff38d 100755
> --- a/git-submodule.sh
> +++ b/git-submodule.sh
> @@ -178,24 +178,6 @@ get_submodule_config () {
>   	printf '%s' "${value:-$default}"
>   }
>
> -
> -#
> -# Map submodule path to submodule name
> -#
> -# $1 = path
> -#
> -module_name()
> -{
> -	# Do we have "submodule.<something>.path = $1" defined in .gitmodules file?
> -	sm_path="$1"
> -	re=$(printf '%s\n' "$1" | sed -e 's/[].[^$\\*]/\\&/g')
> -	name=$( git config -f .gitmodules --get-regexp '^submodule\..*\.path$' |
> -		sed -n -e 's|^submodule\.\(.*\)\.path '"$re"'$|\1|p' )
> -	test -z "$name" &&
> -	die "$(eval_gettext "No submodule mapping found in .gitmodules for path '\$sm_path'")"
> -	printf '%s\n' "$name"
> -}
> -
>   #
>   # Clone a submodule
>   #
> @@ -498,7 +480,7 @@ cmd_foreach()
>   		then
>   			displaypath=$(relative_path "$sm_path")
>   			say "$(eval_gettext "Entering '\$prefix\$displaypath'")"
> -			name=$(module_name "$sm_path")
> +			name=$(git submodule--helper module_name "$sm_path")
>   			(
>   				prefix="$prefix$sm_path/"
>   				clear_local_git_env
> @@ -554,7 +536,7 @@ cmd_init()
>   	while read mode sha1 stage sm_path
>   	do
>   		die_if_unmatched "$mode"
> -		name=$(module_name "$sm_path") || exit
> +		name=$(git submodule--helper module_name "$sm_path") || exit
>
>   		displaypath=$(relative_path "$sm_path")
>
> @@ -636,7 +618,7 @@ cmd_deinit()
>   	while read mode sha1 stage sm_path
>   	do
>   		die_if_unmatched "$mode"
> -		name=$(module_name "$sm_path") || exit
> +		name=$(git submodule--helper module_name "$sm_path") || exit
>
>   		displaypath=$(relative_path "$sm_path")
>
> @@ -758,7 +740,7 @@ cmd_update()
>   			echo >&2 "Skipping unmerged submodule $prefix$sm_path"
>   			continue
>   		fi
> -		name=$(module_name "$sm_path") || exit
> +		name=$(git submodule--helper module_name "$sm_path") || exit
>   		url=$(git config submodule."$name".url)
>   		branch=$(get_submodule_config "$name" branch master)
>   		if ! test -z "$update"
> @@ -1022,7 +1004,7 @@ cmd_summary() {
>   			# Respect the ignore setting for --for-status.
>   			if test -n "$for_status"
>   			then
> -				name=$(module_name "$sm_path")
> +				name=$(git submodule--helper module_name "$sm_path")
>   				ignore_config=$(get_submodule_config "$name" ignore none)
>   				test $status != A && test $ignore_config = all && continue
>   			fi
> @@ -1184,7 +1166,7 @@ cmd_status()
>   	while read mode sha1 stage sm_path
>   	do
>   		die_if_unmatched "$mode"
> -		name=$(module_name "$sm_path") || exit
> +		name=$(git submodule--helper module_name "$sm_path") || exit
>   		url=$(git config submodule."$name".url)
>   		displaypath=$(relative_path "$prefix$sm_path")
>   		if test "$stage" = U
> @@ -1261,7 +1243,7 @@ cmd_sync()
>   	while read mode sha1 stage sm_path
>   	do
>   		die_if_unmatched "$mode"
> -		name=$(module_name "$sm_path")
> +		name=$(git submodule--helper module_name "$sm_path")
>   		url=$(git config -f .gitmodules --get submodule."$name".url)
>
>   		# Possibly a url relative to parent
> diff --git a/submodule.c b/submodule.c
> index 15e90d1..872967f 100644
> --- a/submodule.c
> +++ b/submodule.c
> @@ -686,6 +686,16 @@ static void calculate_changed_submodule_paths(void)
>   	initialized_fetch_ref_tips = 0;
>   }
>
> +const char* submodule_name_for_path(const char* path)
> +{
> +	struct string_list_item *item;
> +	item = unsorted_string_list_lookup(&config_name_for_path, path);
> +	if (item)
> +		return item->util;
> +	else
> +		return NULL;
> +}
> +
>   int fetch_populated_submodules(const struct argv_array *options,
>   			       const char *prefix, int command_line_option,
>   			       int quiet)
> @@ -693,7 +703,7 @@ int fetch_populated_submodules(const struct argv_array *options,
>   	int i, result = 0;
>   	struct child_process cp = CHILD_PROCESS_INIT;
>   	struct argv_array argv = ARGV_ARRAY_INIT;
> -	struct string_list_item *name_for_path;
> +	const char *name_for_path;
>   	const char *work_tree = get_git_work_tree();
>   	if (!work_tree)
>   		goto out;
> @@ -723,10 +733,8 @@ int fetch_populated_submodules(const struct argv_array *options,
>   		if (!S_ISGITLINK(ce->ce_mode))
>   			continue;
>
> -		name = ce->name;
> -		name_for_path = unsorted_string_list_lookup(&config_name_for_path, ce->name);
> -		if (name_for_path)
> -			name = name_for_path->util;
> +		name_for_path = submodule_name_for_path(ce->name);
> +		name =  name_for_path ? name_for_path : ce->name;
>
>   		default_argv = "yes";
>   		if (command_line_option == RECURSE_SUBMODULES_DEFAULT) {
> diff --git a/submodule.h b/submodule.h
> index 7beec48..e3dd854 100644
> --- a/submodule.h
> +++ b/submodule.h
> @@ -41,5 +41,6 @@ int find_unpushed_submodules(unsigned char new_sha1[20], const char *remotes_nam
>   		struct string_list *needs_pushing);
>   int push_unpushed_submodules(unsigned char new_sha1[20], const char *remotes_name);
>   void connect_work_tree_and_git_dir(const char *work_tree, const char *git_dir);
> +const char* submodule_name_for_path(const char* path);
>
>   #endif
>
Jens Lehmann· Aug 6, 2015, 19:54 UTC · re: Jens Lehmann · lore

Re: [PATCH 1/4] submodule: implement `module_name` as a builtin helper

Am 06.08.2015 um 21:49 schrieb Jens Lehmann:
> And wouldn't it make more sense to keep this patch together with
> the "submodule: implement `module_list` as a builtin helper" in
> its own "submodule-helper" series and have the following three
> patches in a separate "parallel fetch for submodules" series?

Please scratch that, I just now read your comment in the cover letter on that ...

Stefan Beller· Aug 6, 2015, 17:35 UTC · re: Stefan Beller · lore
This adds infrastructure code to work a set of tasks from a thread pool.
The whole life cycle of such a thread pool would look like
    struct workdispatcher *wd;
    struct return_values *rv;
    
    wd = create_workdispatcher(&command_for_task, max_parallel_jobs);
    for (...) {
        prepare(pointer_to_task);
        add_task(wd, pointer_to_task);
    }
    rv = wait_workdispatcher(wd);
Signed-off-by: Stefan Beller <sbeller@google.com>
---
 Makefile         |   1 +
 workdispatcher.c | 184 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
 workdispatcher.h |  29 +++++++++
 3 files changed, 214 insertions(+)
 create mode 100644 workdispatcher.c
 create mode 100644 workdispatcher.h
Show changes to 3 files +214 −0

Makefile, workdispatcher.c, workdispatcher.h

diff --git a/Makefile b/Makefile
index 6fb7484..2d8803c 100644
--- a/Makefile
+++ b/Makefile
@@ -805,6 +805,7 @@ LIB_OBJS += version.o
 LIB_OBJS += versioncmp.o
 LIB_OBJS += walker.o
 LIB_OBJS += wildmatch.o
+LIB_OBJS += workdispatcher.o
 LIB_OBJS += wrapper.o
 LIB_OBJS += write_or_die.o
 LIB_OBJS += ws.o
diff --git a/workdispatcher.c b/workdispatcher.c
new file mode 100644
index 0000000..adfedd9
--- /dev/null
+++ b/workdispatcher.c
@@ -0,0 +1,184 @@
+#include "cache.h"
+#include "workdispatcher.h"
+
+#ifndef NO_PTHREADS
+#include <pthread.h>
+#include <semaphore.h>
+#include <stdio.h>
+#include <unistd.h>
+
+#include "git-compat-util.h"
+struct job_list {
+	void *item;
+	struct job_list *next;
+};
+#endif
+
+struct workdispatcher {
+#ifndef NO_PTHREADS
+	/*
+	 * To avoid deadlocks always aquire the semaphores with lowest priority
+	 * first, priorites are in descending order as listed.
+	 *
+	 * The `mutex` is a general purpose lock for modifying data in the work
+	 * dispatcher, such as adding a new task or adding a return value from
+	 * an already run task.
+	 *
+	 * `workingcount` and `freecount` are opposing semaphores, the sum of
+	 * their values should equal `max_threads` at any time while the `mutex`
+	 * is available.
+	 */
+	sem_t mutex;
+	sem_t workingcount;
+	sem_t freecount;
+
+	pthread_t *threads;
+	unsigned max_threads;
+
+	struct job_list *first;
+	struct job_list *last;
+#endif
+	void *(*function)(void*);
+	struct return_values *ret;
+};
+
+#ifndef NO_PTHREADS
+static unsigned number_cores(void)
+{
+	int count = sysconf(_SC_NPROCESSORS_ONLN);
+	if (count < 1) {
+		fprintf(stderr, "Number of CPUs online reported %d. "
+			"Using one core.\n", count);
+		count = 1;
+	}
+	return count;
+}
+
+void *get_task(struct workdispatcher *wd)
+{
+	void *ret;
+	struct job_list *job;
+
+	sem_wait(&wd->workingcount);
+	sem_wait(&wd->mutex);
+
+	if (!wd->first)
+		die("BUG: internal error with dequeuing jobs for threads");
+	job = wd->first;
+	ret = job->item;
+	wd->first = job->next;
+	if (!wd->first)
+		wd->last = NULL;
+
+	sem_post(&wd->freecount);
+	sem_post(&wd->mutex);
+
+	free(job);
+	return ret;
+}
+
+void* dispatcher(void *args)
+{
+	struct workdispatcher *wd = args;
+	void *job = get_task(wd);
+	while (job) {
+		void *retvalue = wd->function(job);
+
+		sem_wait(&wd->mutex);
+		struct return_values *rv = wd->ret;
+		ALLOC_GROW(rv->ret, rv->count + 1, rv->alloc);
+		wd->ret->ret[rv->count++] = retvalue;
+		sem_post(&wd->mutex);
+
+		job = get_task(wd);
+	}
+
+	pthread_exit(0);
+}
+#endif
+
+struct workdispatcher *create_workdispatcher(void *function(void*),
+					     unsigned max_threads)
+{
+	struct workdispatcher *wd = xmalloc(sizeof(*wd));
+
+#ifndef NO_PTHREADS
+	int i;
+	if (!max_threads)
+		wd->max_threads = number_cores();
+	else
+		wd->max_threads = max_threads;
+
+	sem_init(&wd->mutex, 0, 1);
+	sem_init(&wd->workingcount, 0, 0);
+	sem_init(&wd->freecount, 0, wd->max_threads);
+	wd->threads = xmalloc(wd->max_threads * sizeof(pthread_t));
+
+	for (i = 0; i < wd->max_threads; i++)
+		pthread_create(&wd->threads[i], 0, &dispatcher, wd);
+
+	wd->first = NULL;
+	wd->last = NULL;
+#endif
+	wd->function = function;
+	wd->ret = xmalloc(sizeof(*wd->ret));
+	wd->ret->ret = NULL;
+	wd->ret->count = 0;
+	wd->ret->alloc = 0;
+
+	return wd;
+}
+
+void add_task(struct workdispatcher *wd, void *job)
+{
+#ifndef NO_PTHREADS
+	struct job_list *job_list;
+
+	job_list = xmalloc(sizeof(*job_list));
+	job_list->item = job;
+	job_list->next = NULL;
+
+	sem_wait(&wd->freecount);
+	sem_wait(&wd->mutex);
+
+	if (!wd->last) {
+		wd->last = job_list;
+		wd->first = wd->last;
+	} else {
+		wd->last->next = job_list;
+		wd->last = wd->last->next;
+	}
+
+	sem_post(&wd->workingcount);
+	sem_post(&wd->mutex);
+#else
+	ALLOC_GROW(wd->ret->ret, wd->ret->count + 1, wd->ret->alloc);
+	wd->ret->ret[wd->ret->count++] = wd->function(job);
+#endif
+}
+
+struct return_values *wait_workdispatcher(struct workdispatcher *wd)
+{
+	struct return_values *ret;
+#ifndef NO_PTHREADS
+	int i;
+	for (i = 0; i < wd->max_threads; i++)
+		add_task(wd, NULL);
+
+	for (i = 0; i < wd->max_threads; i++)
+		pthread_join(wd->threads[i], 0);
+
+	sem_destroy(&wd->mutex);
+	sem_destroy(&wd->workingcount);
+	sem_destroy(&wd->freecount);
+
+	if (wd->first)
+		die("BUG: internal error with queuing jobs for threads");
+
+	free(wd->threads);
+#endif
+	ret = wd->ret;
+
+	free(wd);
+	return ret;
+}
diff --git a/workdispatcher.h b/workdispatcher.h
new file mode 100644
index 0000000..9f78124
--- /dev/null
+++ b/workdispatcher.h
@@ -0,0 +1,29 @@
+#ifndef WORKDISPATCHER
+#define WORKDISPATCHER
+
+struct return_values {
+	void **ret;
+	int count, alloc;
+};
+
+/*
+ * Creates a struct workdispatcher, which holds a job list and assigns the
+ * jobs to be processed to a number of threads `maxthreads`.
+ * Within the threads the function `fct` is called with the pointer as
+ * given in add_task.
+ *
+ * If `maxthreads` is zero the number of cores available will be used.
+ *
+ * Currently this only works in environments with pthreads, in other
+ * environments, the task will be processed directly in `add_task`.
+ */
+struct workdispatcher *create_workdispatcher(void *fct(void*),
+					     unsigned maxthreads);
+
+/* Waits for all tasks to be done and frees the object. */
+struct return_values *wait_workdispatcher(struct workdispatcher *wd);
+
+/* `task` must not be NULL, as that's used internally to signal shutdown. */
+void add_task(struct workdispatcher *wd, void *task);
+
+#endif
-- 
2.5.0.239.g9728e1d.dirty
Stefan Beller· Aug 6, 2015, 17:35 UTC · re: Stefan Beller · lore

[PATCH 3/4] argv_array: add argv_array_copy

The copied argv array shall be an identical deep copy except for the internal allocation value.

CC: Jeff King <peff@peff.net>
Signed-off-by: Stefan Beller <sbeller@google.com>
---
 argv-array.c | 13 +++++++++++++
 argv-array.h |  1 +
 2 files changed, 14 insertions(+)
Show changes to 2 files +14 −0

argv-array.c, argv-array.h

diff --git a/argv-array.c b/argv-array.c
index 256741d..6d9c1dd 100644
--- a/argv-array.c
+++ b/argv-array.c
@@ -68,3 +68,16 @@ void argv_array_clear(struct argv_array *array)
 	}
 	argv_array_init(array);
 }
+
+void argv_array_copy(struct argv_array *src, struct argv_array *dst)
+{
+	int i;
+
+	dst->argv = xmalloc((src->argc + 1) * sizeof(*dst->argv));
+	dst->argc = src->argc;
+	dst->alloc = src->argc;
+	for (i = 0; i < dst->argc ; i++)
+		dst->argv[i] = xstrdup(src->argv[i]);
+	dst->argv[dst->argc] = NULL;
+}
+
diff --git a/argv-array.h b/argv-array.h
index c65e6e8..247627da 100644
--- a/argv-array.h
+++ b/argv-array.h
@@ -19,5 +19,6 @@ LAST_ARG_MUST_BE_NULL
 void argv_array_pushl(struct argv_array *, ...);
 void argv_array_pop(struct argv_array *);
 void argv_array_clear(struct argv_array *);
+void argv_array_copy(struct argv_array *src, struct argv_array *dst);
 
 #endif /* ARGV_ARRAY_H */
-- 
2.5.0.239.g9728e1d.dirty
Eric Sunshine· Aug 6, 2015, 18:18 UTC · re: Stefan Beller · lore

Re: [PATCH 3/4] argv_array: add argv_array_copy

On Thu, Aug 6, 2015 at 1:35 PM, Stefan Beller <sbeller@google.com> wrote:
Show 11 quoted lines
> The copied argv array shall be an identical deep copy except for
> the internal allocation value.
>
> Signed-off-by: Stefan Beller <sbeller@google.com>
> ---
> diff --git a/argv-array.c b/argv-array.c
> index 256741d..6d9c1dd 100644
> --- a/argv-array.c
> +++ b/argv-array.c
> @@ -68,3 +68,16 @@ void argv_array_clear(struct argv_array *array)
> +void argv_array_copy(struct argv_array *src, struct argv_array *dst)
'src' should be 'const'.

Typical Unix argument order has 'dst' first and 'src' second, i.e strcpy(dst, src). Is it worth deviating from that precedent?

> +{
> +       int i;
> +
> +       dst->argv = xmalloc((src->argc + 1) * sizeof(*dst->argv));

What happens if 'dst' already has content? Isn't that being leaked here? At the very least, don't you want to argv_array_clear(dst)?

> +       dst->argc = src->argc;
> +       dst->alloc = src->argc;
This is wrong, of course. The number allocated is actually argc+1, not argc.
> +       for (i = 0; i < dst->argc ; i++)

While it's not wrong per se to use dst->argc as the terminating condition, it is potentially misleading and confusing. Instead using src->argc as the terminating condition will better telegraph that the copy process is indeed predicated upon 'src'.

> +               dst->argv[i] = xstrdup(src->argv[i]);
> +       dst->argv[dst->argc] = NULL;

It's not clear why you want to hand-code the low-level functionality again (such as array allocation and string duplication), risking (and indeed making) errors in the process, when you could instead re-use existing argv_array code. I would have expected to see argv_array_copy() implemented as:

    argv_array_clear(dst);
    for (i = 0; i < src->argc; i++)
        argv_array_push(dst, src->argv[i]);
which provides far fewer opportunities for errors to creep in.

Moreover, this function might be too special-purpose. That is, why does it need to overwrite 'dst'? Can't you achieve the same functionality by merely appending to 'dst', and leave it up to the caller to decide whether 'dst' should be cleared beforehand or not? If so, then you can drop the argv_array_clear(dst) from the above.

However, that begs the question: Why do you need argv_array_copy() at all? Isn't the same functionality already provided by argv_array_pushv()? To wit, a caller which wants to copy from 'src' to 'dst' can already do:

    struct argv_array src = ...;
    struct argv_array dst = ARGV_ARRAY_INIT;
    argv_array_pushv(&dst, src->argv);
Show 15 quoted lines
> +}
> +
> diff --git a/argv-array.h b/argv-array.h
> index c65e6e8..247627da 100644
> --- a/argv-array.h
> +++ b/argv-array.h
> @@ -19,5 +19,6 @@ LAST_ARG_MUST_BE_NULL
>  void argv_array_pushl(struct argv_array *, ...);
>  void argv_array_pop(struct argv_array *);
>  void argv_array_clear(struct argv_array *);
> +void argv_array_copy(struct argv_array *src, struct argv_array *dst);
>
>  #endif /* ARGV_ARRAY_H */
> --
> 2.5.0.239.g9728e1d.dirty
Jeff King· Aug 6, 2015, 18:52 UTC · re: Eric Sunshine · lore

Re: [PATCH 3/4] argv_array: add argv_array_copy

On Thu, Aug 06, 2015 at 02:18:26PM -0400, Eric Sunshine wrote:
Show 8 quoted lines
> However, that begs the question: Why do you need argv_array_copy() at
> all? Isn't the same functionality already provided by
> argv_array_pushv()? To wit, a caller which wants to copy from 'src' to
> 'dst' can already do:
> 
>     struct argv_array src = ...;
>     struct argv_array dst = ARGV_ARRAY_INIT;
>     argv_array_pushv(&dst, src->argv);

Thanks for reviewing this, Eric. Everything you said is exactly what I was thinking, too, especially this last part.

-Peff
Stefan Beller· Aug 6, 2015, 17:35 UTC · re: Stefan Beller · lore

[RFC PATCH 4/4] submodule: add infrastructure to fetch submodules in parallel

This makes use of the new workdispatcher to fetch a number of submodules at the same time.

Still todo: sort the output of the fetch commands. I am unsure if this should be hooked into the workdispatcher as the problem of sorted output will appear likely again, so a general solution would not hurt.

Signed-off-by: Stefan Beller <sbeller@google.com>
---
 builtin/fetch.c |  3 ++-
 submodule.c     | 74 ++++++++++++++++++++++++++++++++++++++++++++-------------
 submodule.h     |  2 +-
 3 files changed, 60 insertions(+), 19 deletions(-)
Show changes to 3 files +60 −19

builtin/fetch.c, submodule.c, submodule.h

diff --git a/builtin/fetch.c b/builtin/fetch.c
index 8d5b2db..9053e8b 100644
--- a/builtin/fetch.c
+++ b/builtin/fetch.c
@@ -1207,7 +1207,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,
+						    1);
 		argv_array_clear(&options);
 	}
 
diff --git a/submodule.c b/submodule.c
index 872967f..0b2842b 100644
--- a/submodule.c
+++ b/submodule.c
@@ -11,6 +11,7 @@
 #include "sha1-array.h"
 #include "argv-array.h"
 #include "blob.h"
+#include "workdispatcher.h"
 
 static struct string_list config_name_for_path;
 static struct string_list config_fetch_recurse_submodules_for_name;
@@ -696,13 +697,49 @@ const char* submodule_name_for_path(const char* path)
 		return NULL;
 }
 
+struct submodule_parallel_fetch {
+	struct child_process cp;
+	struct argv_array argv;
+	struct strbuf sb;
+	int quiet;
+};
+
+void submodule_parallel_fetch_init(struct submodule_parallel_fetch *spf)
+{
+	child_process_init(&spf->cp);
+	argv_array_init(&spf->argv);
+	strbuf_init(&spf->sb, 0);
+	spf->quiet = 0;
+}
+
+void *run_command_and_cleanup(void *arg)
+{
+	struct submodule_parallel_fetch *spf = arg;
+	void *ret = NULL;
+
+	if (!spf->quiet)
+		puts(spf->sb.buf);
+
+	spf->cp.argv = spf->argv.argv;
+
+	if (run_command(&spf->cp))
+		ret = (void *)1;
+
+	strbuf_release(&spf->cp);
+	argv_array_clear(spf->argv);
+	free(spf);
+	return ret;
+}
+
 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 workdispatcher *wd;
+	struct return_values *rv;
 	struct argv_array argv = ARGV_ARRAY_INIT;
+	struct submodule_parallel_fetch *spf;
 	const char *name_for_path;
 	const char *work_tree = get_git_work_tree();
 	if (!work_tree)
@@ -717,12 +754,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();
 
+	wd = create_workdispatcher(&run_command_and_cleanup, max_parallel_jobs);
 	for (i = 0; i < active_nr; i++) {
 		struct strbuf submodule_path = STRBUF_INIT;
 		struct strbuf submodule_git_dir = STRBUF_INIT;
@@ -771,24 +805,30 @@ int fetch_populated_submodules(const struct argv_array *options,
 		if (!git_dir)
 			git_dir = submodule_git_dir.buf;
 		if (is_directory(git_dir)) {
+			spf = xmalloc(sizeof(*spf));
+			submodule_parallel_fetch_init(spf);
+			spf->cp.env = local_repo_env;
+			spf->cp.git_cmd = 1;
+			spf->cp.no_stdin = 1;
+			spf->cp.dir = strbuf_detach(&submodule_path, NULL);
+			spf->quiet = quiet;
 			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->sb, "Fetching submodule %s%s", prefix, ce->name);
+			argv_array_copy(&argv, &spf->argv);
+			argv_array_push(&spf->argv, default_argv);
+			argv_array_push(&spf->argv, "--submodule-prefix");
+			argv_array_push(&spf->argv, submodule_prefix.buf);
+			add_task(wd, spf);
 		}
 		strbuf_release(&submodule_path);
 		strbuf_release(&submodule_git_dir);
 		strbuf_release(&submodule_prefix);
 	}
 	argv_array_clear(&argv);
+	rv = wait_workdispatcher(wd);
+	for (i = 0; i < rv->count; i++)
+		if (rv->ret[i])
+			result = 1;
 out:
 	string_list_clear(&changed_submodule_paths, 1);
 	return result;
diff --git a/submodule.h b/submodule.h
index e3dd854..51195ea 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.239.g9728e1d.dirty
Jens Lehmann· Aug 6, 2015, 20:08 UTC · re: Stefan Beller · lore

Re: [RFC/PATCH 0/4] parallel fetch for submodules

Am 06.08.2015 um 19:35 schrieb Stefan Beller:
> When I was looking at the branches of Jens for work done on submodules
> not yet upstream I found a commit "WIP threaded submodule fetching[1],
> and I was side tracked wanting to present a different approach to that.

Cool. I didn't follow that route further than building a proof of concept because I ran into a nasty DNS-timeout on my router at home and at work we host all repos on a not-so-beefy server making parallel fetch rather pointless. But I suspect this approach will bring down fetch times for some users.

Maybe we could also re-use parallel fetch for multiple upstreams in the superproject when doing a "git fetch --all" without too much extra work?

Stefan Beller· Aug 6, 2015, 20:44 UTC · re: Jens Lehmann · lore

Re: [RFC/PATCH 0/4] parallel fetch for submodules

On Thu, Aug 6, 2015 at 1:08 PM, Jens Lehmann <Jens.Lehmann@web.de> wrote:
Show 12 quoted lines
> Am 06.08.2015 um 19:35 schrieb Stefan Beller:
>>
>> When I was looking at the branches of Jens for work done on submodules
>> not yet upstream I found a commit "WIP threaded submodule fetching[1],
>> and I was side tracked wanting to present a different approach to that.
>
>
> Cool. I didn't follow that route further than building a proof of
> concept because I ran into a nasty DNS-timeout on my router at home
> and at work we host all repos on a not-so-beefy server making parallel
> fetch rather pointless. But I suspect this approach will bring down
> fetch times for some users.
The difference between your proof of concept and mine is to have the number
of threads easier configurable. (Think of "git fetch --recurse-submodules=yes
 -j4", similar to "make -j4" splitting work up to 4 different programs
at the same
time. And that thing would be possible to add in this series)

If you fetch lots of submodules, both the client and server load should come in an ping-pong on-off pattern as the client waits for the server to prepare stuff and get it sent to it and then the client needs time to resolve deltas and write to disk. Depending on the duty cycle of each, a different number of parallel threads make sense (I would expected that they shift their phases against each other by pure randomness, i.e. one thread is currently resolving deltas while the other thread is telling the server to get some work done, so both the client and server get utilized at the same time).

>
> Maybe we could also re-use parallel fetch for multiple upstreams in
> the superproject when doing a "git fetch --all" without too much
> extra work?

That's why I tried advertising RFC patch 2/4 as I believe it could be easily reused for such things like "git fetch --all", but maybe other people see problems with it (over/under engineered, wrong things added, but other critical things missing) so I'd like to hear opinions on that. :)

← back to recent threads