{"thread":{"id":"40187","subject":"[RFC PATCH 0/5] Progressing with `git submodule foreach_parallel`","startedAt":"2015-08-27T00:52:41Z","lastAt":"2015-08-28T15:34:49Z","messageCount":11,"participants":["Stefan Beller","Johannes Schindelin","Junio C Hamano"],"isPatch":true,"patchVersion":1,"patchTotal":5},"messages":[{"id":"268768","messageId":"1440636766-12738-1-git-send-email-sbeller@google.com","threadId":"40187","inReplyTo":null,"subject":"[RFC PATCH 0/5] Progressing with `git submodule foreach_parallel`","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-27T00:52:41Z","receivedAt":"2015-08-27T00:52:41Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"This series applies on top of 10d0bef9981c8045b8c9aac1ca7b8761896bed0c in\nthe branch origin/sb/submodule-helper\n\nThe first patch is a resend of module_clone, which now contains all\nmemory leak fixes. \n\nThe second patch adding the new task queue is rewritten to not\nuse semaphores any more, but only mutexes and condition variables\nas that's what we use at other places in the code base. So it is\nbetter to align to these places, it makes the threading pool also\neasier to understand internally.\nOriginally there were 2 semaphores to control the queue length\n(0 < queue length < number of threads), such that we don't overallocate\nmemory, but rather slow down the thread adding new tasks to the queue.\nThis has been removed and the queue may grow infinitely large.\n\nThe third patch adding `git submodule foreach_parallel` learned a\ntotally different behavior how to interact with the output channels\n(way better for human consumption as you have early feedback and\nfeedback all the time, ok-ish for machine consumption, but far from\nperfect).\n\nThe fourth patch rewriting index-pack was made smaller and doesn't \nrewrite the workflow any more, but only plugs in the new threading\nAPI instead of constructing the pthreads themselves.\nI started benchmarking that patch, but I seem to get only bogus results,\nso I cannot tell if it still makes it 2 % slower as Jeff claimed.\n\nThe fifth patch refactors pack-objects to be *way* more understandable IMHO!\nIt also switches to the new threading pool, but the refactor may be\nsuitable also if we decide to not use the threading API there.\nThe reason for it being more readable is the movement of the rebalancing\ninto the threads themselves, which then turns out to require a lot less\nof locking and communication variables.\n\nAny feedback is welcome!\nThanks,\nStefan\n\nStefan Beller (5):\n  submodule: implement `module_clone` as a builtin helper\n  thread-utils: add a threaded task queue\n  submodule: helper to run foreach in parallel\n  index-pack: Use the new worker pool\n  pack-objects: Use new worker pool\n\n builtin/index-pack.c        |  23 ++--\n builtin/pack-objects.c      | 175 ++++++++----------------\n builtin/submodule--helper.c | 318 +++++++++++++++++++++++++++++++++++++++++++-\n git-submodule.sh            |  91 ++-----------\n run-command.c               |  29 ++--\n thread-utils.c              | 227 +++++++++++++++++++++++++++++++\n thread-utils.h              |  35 +++++\n 7 files changed, 677 insertions(+), 221 deletions(-)\n\n-- \n2.5.0.264.g784836d\n"},{"id":"268770","messageId":"1440636766-12738-2-git-send-email-sbeller@google.com","threadId":"40187","inReplyTo":"1440636766-12738-1-git-send-email-sbeller@google.com","subject":"[PATCH 1/5] submodule: implement `module_clone` as a builtin helper","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-27T00:52:42Z","receivedAt":"2015-08-27T00:52:42Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"`module_clone` is part of the update command, which I want to convert\nto C next.\n\nSigned-off-by: Stefan Beller <sbeller@google.com>\nSigned-off-by: Junio C Hamano <gitster@pobox.com>\n---\n builtin/submodule--helper.c | 160 +++++++++++++++++++++++++++++++++++++++++++-\n git-submodule.sh            |  80 +---------------------\n 2 files changed, 162 insertions(+), 78 deletions(-)\n\ndiff --git a/builtin/submodule--helper.c b/builtin/submodule--helper.c\nindex 4b32a3c..f11fb9c 100644\n--- a/builtin/submodule--helper.c\n+++ b/builtin/submodule--helper.c\n@@ -8,6 +8,7 @@\n #include \"submodule.h\"\n #include \"submodule-config.h\"\n #include \"string-list.h\"\n+#include \"run-command.h\"\n \n static const struct cache_entry **ce_entries;\n static int ce_alloc, ce_used;\n@@ -124,6 +125,160 @@ static int module_name(int argc, const char **argv, const char *prefix)\n \treturn 0;\n }\n \n+static int clone_submodule(const char *path, const char *gitdir, const char *url,\n+\t\t\t   const char *depth, const char *reference, int quiet)\n+{\n+\tstruct child_process cp;\n+\tchild_process_init(&cp);\n+\n+\targv_array_push(&cp.args, \"clone\");\n+\targv_array_push(&cp.args, \"--no-checkout\");\n+\tif (quiet)\n+\t\targv_array_push(&cp.args, \"--quiet\");\n+\tif (depth && strcmp(depth, \"\")) {\n+\t\targv_array_push(&cp.args, \"--depth\");\n+\t\targv_array_push(&cp.args, depth);\n+\t}\n+\tif (reference && strcmp(reference, \"\")) {\n+\t\targv_array_push(&cp.args, \"--reference\");\n+\t\targv_array_push(&cp.args, reference);\n+\t}\n+\tif (gitdir) {\n+\t\targv_array_push(&cp.args, \"--separate-git-dir\");\n+\t\targv_array_push(&cp.args, gitdir);\n+\t}\n+\targv_array_push(&cp.args, url);\n+\targv_array_push(&cp.args, path);\n+\n+\tcp.git_cmd = 1;\n+\tcp.env = local_repo_env;\n+\n+\tcp.no_stdin = 1;\n+\tcp.no_stdout = 1;\n+\tcp.no_stderr = 1;\n+\n+\treturn run_command(&cp);\n+}\n+\n+/*\n+ * Clone a submodule\n+ *\n+ * $1 = submodule path\n+ * $2 = submodule name\n+ * $3 = URL to clone\n+ * $4 = reference repository to reuse (empty for independent)\n+ * $5 = depth argument for shallow clones (empty for deep)\n+ *\n+ * Prior to calling, cmd_update checks that a possibly existing\n+ * path is not a git repository.\n+ * Likewise, cmd_add checks that path does not exist at all,\n+ * since it is the location of a new submodule.\n+ */\n+static int module_clone(int argc, const char **argv, const char *prefix)\n+{\n+\tconst char *path = NULL, *name = NULL, *url = NULL, *reference = NULL, *depth = NULL;\n+\tint quiet = 0;\n+\tFILE *submodule_dot_git;\n+\tconst char *sm_gitdir, *p;\n+\tstruct strbuf rel_path = STRBUF_INIT;\n+\tstruct strbuf sb = STRBUF_INIT;\n+\n+\tstruct option module_update_options[] = {\n+\t\tOPT_STRING(0, \"prefix\", &alternative_path,\n+\t\t\t   N_(\"path\"),\n+\t\t\t   N_(\"alternative anchor for relative paths\")),\n+\t\tOPT_STRING(0, \"path\", &path,\n+\t\t\t   N_(\"path\"),\n+\t\t\t   N_(\"where the new submodule will be cloned to\")),\n+\t\tOPT_STRING(0, \"name\", &name,\n+\t\t\t   N_(\"string\"),\n+\t\t\t   N_(\"name of the new submodule\")),\n+\t\tOPT_STRING(0, \"url\", &url,\n+\t\t\t   N_(\"string\"),\n+\t\t\t   N_(\"url where to clone the submodule from\")),\n+\t\tOPT_STRING(0, \"reference\", &reference,\n+\t\t\t   N_(\"string\"),\n+\t\t\t   N_(\"reference repository\")),\n+\t\tOPT_STRING(0, \"depth\", &depth,\n+\t\t\t   N_(\"string\"),\n+\t\t\t   N_(\"depth for shallow clones\")),\n+\t\tOPT_END()\n+\t};\n+\n+\tstatic const char * const git_submodule_helper_usage[] = {\n+\t\tN_(\"git submodule--helper update [--prefix=<path>] [--quiet] [--remote] [-N|--no-fetch]\"\n+\t\t   \"[-f|--force] [--rebase|--merge] [--reference <repository>]\"\n+\t\t   \"[--depth <depth>] [--recursive] [--] [<path>...]\"),\n+\t\tNULL\n+\t};\n+\n+\targc = parse_options(argc, argv, prefix, module_update_options,\n+\t\t\t     git_submodule_helper_usage, 0);\n+\n+\tif (getenv(\"GIT_QUIET\"))\n+\t\tquiet = 1;\n+\n+\tstrbuf_addf(&sb, \"%s/modules/%s\", get_git_dir(), name);\n+\tsm_gitdir = strbuf_detach(&sb, NULL);\n+\n+\tif (!file_exists(sm_gitdir)) {\n+\t\tsafe_create_leading_directories_const(sm_gitdir);\n+\t\tif (clone_submodule(path, sm_gitdir, url, depth, reference, quiet))\n+\t\t\tdie(N_(\"Clone of '%s' into submodule path '%s' failed\"),\n+\t\t\t    url, path);\n+\t} else {\n+\t\tsafe_create_leading_directories_const(path);\n+\t\tunlink(sm_gitdir);\n+\t}\n+\n+\t/* Write a .git file in the submodule to redirect to the superproject. */\n+\tif (alternative_path && !strcmp(alternative_path, \"\")) {\n+\t\tp = relative_path(path, alternative_path, &sb);\n+\t\tstrbuf_reset(&sb);\n+\t} else\n+\t\tp = path;\n+\n+\tif (safe_create_leading_directories_const(p) < 0)\n+\t\tdie(\"Could not create directory '%s'\", p);\n+\n+\tstrbuf_addf(&sb, \"%s/.git\", p);\n+\n+\tif (safe_create_leading_directories_const(sb.buf) < 0)\n+\t\tdie(_(\"could not create leading directories of '%s'\"), sb.buf);\n+\tsubmodule_dot_git = fopen(sb.buf, \"w\");\n+\tif (!submodule_dot_git)\n+\t\tdie (\"Cannot open file '%s': %s\", sb.buf, strerror(errno));\n+\n+\tfprintf(submodule_dot_git, \"gitdir: %s\\n\",\n+\t\trelative_path(sm_gitdir, path, &rel_path));\n+\tif (fclose(submodule_dot_git))\n+\t\tdie(\"Could not close file %s\", sb.buf);\n+\tstrbuf_reset(&sb);\n+\n+\t/* Redirect the worktree of the submodule in the superprojects config */\n+\tif (!is_absolute_path(sm_gitdir)) {\n+\t\tchar *s = (char*)sm_gitdir;\n+\t\tif (strbuf_getcwd(&sb))\n+\t\t\tdie_errno(\"unable to get current working directory\");\n+\t\tstrbuf_addf(&sb, \"/%s\", sm_gitdir);\n+\t\tsm_gitdir = strbuf_detach(&sb, NULL);\n+\t\tfree(s);\n+\t}\n+\n+\tif (strbuf_getcwd(&sb))\n+\t\tdie_errno(\"unable to get current working directory\");\n+\tstrbuf_addf(&sb, \"/%s\", path);\n+\n+\tp = git_pathdup_submodule(path, \"config\");\n+\tif (!p)\n+\t\tdie(\"Could not get submodule directory for '%s'\", path);\n+\tgit_config_set_in_file(p, \"core.worktree\",\n+\t\t\t       relative_path(sb.buf, sm_gitdir, &rel_path));\n+\tstrbuf_release(&sb);\n+\tfree((char*)sm_gitdir);\n+\treturn 0;\n+}\n+\n int cmd_submodule__helper(int argc, const char **argv, const char *prefix)\n {\n \tif (argc < 2)\n@@ -135,6 +290,9 @@ int cmd_submodule__helper(int argc, const char **argv, const char *prefix)\n \tif (!strcmp(argv[1], \"module_name\"))\n \t\treturn module_name(argc - 2, argv + 2, prefix);\n \n+\tif (!strcmp(argv[1], \"module_clone\"))\n+\t\treturn module_clone(argc - 1, argv + 1, prefix);\n+\n usage:\n-\tusage(\"git submodule--helper [module_list module_name]\\n\");\n+\tusage(\"git submodule--helper [module_list module_name module_clone]\\n\");\n }\ndiff --git a/git-submodule.sh b/git-submodule.sh\nindex e6ff38d..fb5155e 100755\n--- a/git-submodule.sh\n+++ b/git-submodule.sh\n@@ -178,80 +178,6 @@ get_submodule_config () {\n \tprintf '%s' \"${value:-$default}\"\n }\n \n-#\n-# Clone a submodule\n-#\n-# $1 = submodule path\n-# $2 = submodule name\n-# $3 = URL to clone\n-# $4 = reference repository to reuse (empty for independent)\n-# $5 = depth argument for shallow clones (empty for deep)\n-#\n-# Prior to calling, cmd_update checks that a possibly existing\n-# path is not a git repository.\n-# Likewise, cmd_add checks that path does not exist at all,\n-# since it is the location of a new submodule.\n-#\n-module_clone()\n-{\n-\tsm_path=$1\n-\tname=$2\n-\turl=$3\n-\treference=\"$4\"\n-\tdepth=\"$5\"\n-\tquiet=\n-\tif test -n \"$GIT_QUIET\"\n-\tthen\n-\t\tquiet=-q\n-\tfi\n-\n-\tgitdir=\n-\tgitdir_base=\n-\tbase_name=$(dirname \"$name\")\n-\n-\tgitdir=$(git rev-parse --git-dir)\n-\tgitdir_base=\"$gitdir/modules/$base_name\"\n-\tgitdir=\"$gitdir/modules/$name\"\n-\n-\tif test -d \"$gitdir\"\n-\tthen\n-\t\tmkdir -p \"$sm_path\"\n-\t\trm -f \"$gitdir/index\"\n-\telse\n-\t\tmkdir -p \"$gitdir_base\"\n-\t\t(\n-\t\t\tclear_local_git_env\n-\t\t\tgit clone $quiet ${depth:+\"$depth\"} -n ${reference:+\"$reference\"} \\\n-\t\t\t\t--separate-git-dir \"$gitdir\" \"$url\" \"$sm_path\"\n-\t\t) ||\n-\t\tdie \"$(eval_gettext \"Clone of '\\$url' into submodule path '\\$sm_path' failed\")\"\n-\tfi\n-\n-\t# We already are at the root of the work tree but cd_to_toplevel will\n-\t# resolve any symlinks that might be present in $PWD\n-\ta=$(cd_to_toplevel && cd \"$gitdir\" && pwd)/\n-\tb=$(cd_to_toplevel && cd \"$sm_path\" && pwd)/\n-\t# Remove all common leading directories after a sanity check\n-\tif test \"${a#$b}\" != \"$a\" || test \"${b#$a}\" != \"$b\"; then\n-\t\tdie \"$(eval_gettext \"Gitdir '\\$a' is part of the submodule path '\\$b' or vice versa\")\"\n-\tfi\n-\twhile test \"${a%%/*}\" = \"${b%%/*}\"\n-\tdo\n-\t\ta=${a#*/}\n-\t\tb=${b#*/}\n-\tdone\n-\t# Now chop off the trailing '/'s that were added in the beginning\n-\ta=${a%/}\n-\tb=${b%/}\n-\n-\t# Turn each leading \"*/\" component into \"../\"\n-\trel=$(printf '%s\\n' \"$b\" | sed -e 's|[^/][^/]*|..|g')\n-\tprintf '%s\\n' \"gitdir: $rel/$a\" >\"$sm_path/.git\"\n-\n-\trel=$(printf '%s\\n' \"$a\" | sed -e 's|[^/][^/]*|..|g')\n-\t(clear_local_git_env; cd \"$sm_path\" && GIT_WORK_TREE=. git config core.worktree \"$rel/$b\")\n-}\n-\n isnumber()\n {\n \tn=$(($1 + 0)) 2>/dev/null && test \"$n\" = \"$1\"\n@@ -301,7 +227,7 @@ cmd_add()\n \t\t\tshift\n \t\t\t;;\n \t\t--depth=*)\n-\t\t\tdepth=$1\n+\t\t\tdepth=\"$1\"\n \t\t\t;;\n \t\t--)\n \t\t\tshift\n@@ -412,7 +338,7 @@ Use -f if you really want to add it.\" >&2\n \t\t\t\techo \"$(eval_gettext \"Reactivating local git directory for submodule '\\$sm_name'.\")\"\n \t\t\tfi\n \t\tfi\n-\t\tmodule_clone \"$sm_path\" \"$sm_name\" \"$realrepo\" \"$reference\" \"$depth\" || exit\n+\t\tgit submodule--helper module_clone --prefix \"$wt_prefix\" --path \"$sm_path\" --name \"$sm_name\" --url \"$realrepo\" \"$reference\" \"$depth\" || exit\n \t\t(\n \t\t\tclear_local_git_env\n \t\t\tcd \"$sm_path\" &&\n@@ -774,7 +700,7 @@ Maybe you want to use 'update --init'?\")\"\n \n \t\tif ! test -d \"$sm_path\"/.git && ! test -f \"$sm_path\"/.git\n \t\tthen\n-\t\t\tmodule_clone \"$sm_path\" \"$name\" \"$url\" \"$reference\" \"$depth\" || exit\n+\t\t\tgit submodule--helper module_clone --prefix \"$prefix\" --path \"$sm_path\" --name \"$name\" --url \"$url\" \"$reference\" \"$depth\" || exit\n \t\t\tcloned_modules=\"$cloned_modules;$name\"\n \t\t\tsubsha1=\n \t\telse\n-- \n2.5.0.264.g784836d\n"},{"id":"268771","messageId":"1440636766-12738-3-git-send-email-sbeller@google.com","threadId":"40187","inReplyTo":"1440636766-12738-1-git-send-email-sbeller@google.com","subject":"[PATCH 2/5] thread-utils: add a threaded task queue","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-27T00:52:43Z","receivedAt":"2015-08-27T00:52:43Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"This adds functionality to do work in a parallel threaded\nfashion while the boiler plate code for setting up threads\nand tearing them down as well as queuing up tasks is hidden\nbehind the new API.\n\nSigned-off-by: Stefan Beller <sbeller@google.com>\n---\n run-command.c  |  29 +++++---\n thread-utils.c | 227 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++\n thread-utils.h |  35 +++++++++\n 3 files changed, 279 insertions(+), 12 deletions(-)\n\ndiff --git a/run-command.c b/run-command.c\nindex 28e1d55..cb15cd9 100644\n--- a/run-command.c\n+++ b/run-command.c\n@@ -668,6 +668,22 @@ int git_atexit(void (*handler)(void))\n \n #endif\n \n+void setup_main_thread(void)\n+{\n+\tif (!main_thread_set) {\n+\t\t/*\n+\t\t * We assume that the first time that start_async is called\n+\t\t * it is from the main thread.\n+\t\t */\n+\t\tmain_thread_set = 1;\n+\t\tmain_thread = pthread_self();\n+\t\tpthread_key_create(&async_key, NULL);\n+\t\tpthread_key_create(&async_die_counter, NULL);\n+\t\tset_die_routine(die_async);\n+\t\tset_die_is_recursing_routine(async_die_is_recursing);\n+\t}\n+}\n+\n int start_async(struct async *async)\n {\n \tint need_in, need_out;\n@@ -740,18 +756,7 @@ int start_async(struct async *async)\n \telse if (async->out)\n \t\tclose(async->out);\n #else\n-\tif (!main_thread_set) {\n-\t\t/*\n-\t\t * We assume that the first time that start_async is called\n-\t\t * it is from the main thread.\n-\t\t */\n-\t\tmain_thread_set = 1;\n-\t\tmain_thread = pthread_self();\n-\t\tpthread_key_create(&async_key, NULL);\n-\t\tpthread_key_create(&async_die_counter, NULL);\n-\t\tset_die_routine(die_async);\n-\t\tset_die_is_recursing_routine(async_die_is_recursing);\n-\t}\n+\tsetup_main_thread();\n \n \tif (proc_in >= 0)\n \t\tset_cloexec(proc_in);\ndiff --git a/thread-utils.c b/thread-utils.c\nindex a2135e0..b45ab92 100644\n--- a/thread-utils.c\n+++ b/thread-utils.c\n@@ -1,5 +1,7 @@\n #include \"cache.h\"\n #include \"thread-utils.h\"\n+#include \"run-command.h\"\n+#include \"git-compat-util.h\"\n \n #if defined(hpux) || defined(__hpux) || defined(_hpux)\n #  include <sys/pstat.h>\n@@ -75,3 +77,228 @@ int init_recursive_mutex(pthread_mutex_t *m)\n \t}\n \treturn ret;\n }\n+\n+#ifndef NO_PTHREADS\n+struct job_list {\n+\tint (*fct)(struct task_queue *tq, void *task);\n+\tvoid *task;\n+\tstruct job_list *next;\n+};\n+\n+static pthread_t main_thread;\n+static int main_thread_set;\n+static pthread_key_t async_key;\n+static pthread_key_t async_die_counter;\n+\n+static NORETURN void die_async(const char *err, va_list params)\n+{\n+\tvreportf(\"fatal: \", err, params);\n+\n+\tif (!pthread_equal(main_thread, pthread_self()))\n+\t\tpthread_exit((void *)128);\n+\n+\texit(128);\n+}\n+\n+static int async_die_is_recursing(void)\n+{\n+\tvoid *ret = pthread_getspecific(async_die_counter);\n+\tpthread_setspecific(async_die_counter, (void *)1);\n+\treturn ret != NULL;\n+}\n+\n+/* FIXME: deduplicate this code with run-command.c */\n+static void setup_main_thread(void)\n+{\n+\tif (!main_thread_set) {\n+\t\tmain_thread_set = 1;\n+\t\tmain_thread = pthread_self();\n+\t\tpthread_key_create(&async_key, NULL);\n+\t\tpthread_key_create(&async_die_counter, NULL);\n+\t\tset_die_routine(die_async);\n+\t\tset_die_is_recursing_routine(async_die_is_recursing);\n+\t}\n+}\n+\n+struct task_queue {\n+\tpthread_mutex_t mutex;\n+\tpthread_cond_t cond_non_empty;\n+\n+\tint queued_tasks;\n+\tstruct job_list *first;\n+\tstruct job_list *last;\n+\n+\tpthread_t *threads;\n+\tunsigned max_threads;\n+\tunsigned max_tasks;\n+\n+\tvoid (*finish_function)(struct task_queue *tq);\n+\tint early_return;\n+};\n+\n+static void next_task(struct task_queue *tq,\n+\t\t      int (**fct)(struct task_queue *tq, void *task),\n+\t\t      void **task,\n+\t\t      int *early_return)\n+{\n+\tstruct job_list *job = NULL;\n+\n+\tpthread_mutex_lock(&tq->mutex);\n+\twhile (tq->queued_tasks == 0)\n+\t\tpthread_cond_wait(&tq->cond_non_empty, &tq->mutex);\n+\n+\ttq->early_return |= *early_return;\n+\n+\tif (!tq->early_return) {\n+\t\tjob = tq->first;\n+\t\ttq->first = job->next;\n+\t\tif (!tq->first)\n+\t\t\ttq->last = NULL;\n+\t\ttq->queued_tasks--;\n+\t}\n+\n+\tpthread_mutex_unlock(&tq->mutex);\n+\n+\tif (job) {\n+\t\t*fct = job->fct;\n+\t\t*task = job->task;\n+\t} else {\n+\t\t*fct = NULL;\n+\t\t*task = NULL;\n+\t}\n+\n+\tfree(job);\n+}\n+\n+static void *dispatcher(void *args)\n+{\n+\tvoid *task;\n+\tint (*fct)(struct task_queue *tq, void *task);\n+\tint early_return = 0;\n+\tstruct task_queue *tq = args;\n+\n+\tnext_task(tq, &fct, &task, &early_return);\n+\twhile (fct && !early_return) {\n+\t\tearly_return = fct(tq, task);\n+\t\tnext_task(tq, &fct, &task, &early_return);\n+\t}\n+\n+\tif (tq->finish_function)\n+\t\ttq->finish_function(tq);\n+\n+\tpthread_exit(0);\n+}\n+\n+struct task_queue *create_task_queue(unsigned max_threads)\n+{\n+\tstruct task_queue *tq = xmalloc(sizeof(*tq));\n+\n+\tint i, ret;\n+\tif (!max_threads)\n+\t\ttq->max_threads = online_cpus();\n+\telse\n+\t\ttq->max_threads = max_threads;\n+\n+\tpthread_mutex_init(&tq->mutex, NULL);\n+\tpthread_cond_init(&tq->cond_non_empty, NULL);\n+\n+\ttq->threads = xmalloc(tq->max_threads * sizeof(pthread_t));\n+\n+\ttq->queued_tasks = 0;\n+\ttq->first = NULL;\n+\ttq->last = NULL;\n+\n+\tsetup_main_thread();\n+\n+\tfor (i = 0; i < tq->max_threads; i++) {\n+\t\tret = pthread_create(&tq->threads[i], 0, &dispatcher, tq);\n+\t\tif (ret)\n+\t\t\tdie(\"unable to create thread: %s\", strerror(ret));\n+\t}\n+\n+\ttq->early_return = 0;\n+\n+\treturn tq;\n+}\n+\n+void add_task(struct task_queue *tq,\n+\t      int (*fct)(struct task_queue *tq, void *task),\n+\t      void *task)\n+{\n+\tstruct job_list *job_list;\n+\n+\tjob_list = xmalloc(sizeof(*job_list));\n+\tjob_list->task = task;\n+\tjob_list->fct = fct;\n+\tjob_list->next = NULL;\n+\n+\tpthread_mutex_lock(&tq->mutex);\n+\n+\tif (!tq->last) {\n+\t\ttq->last = job_list;\n+\t\ttq->first = tq->last;\n+\t} else {\n+\t\ttq->last->next = job_list;\n+\t\ttq->last = tq->last->next;\n+\t}\n+\ttq->queued_tasks++;\n+\n+\tpthread_mutex_unlock(&tq->mutex);\n+\tpthread_cond_signal(&tq->cond_non_empty);\n+}\n+\n+int finish_task_queue(struct task_queue *tq, void (*fct)(struct task_queue *tq))\n+{\n+\tint ret;\n+\tint i;\n+\n+\ttq->finish_function = fct;\n+\n+\tfor (i = 0; i < tq->max_threads; i++)\n+\t\tadd_task(tq, NULL, NULL);\n+\n+\tfor (i = 0; i < tq->max_threads; i++)\n+\t\tpthread_join(tq->threads[i], 0);\n+\n+\tpthread_mutex_destroy(&tq->mutex);\n+\tpthread_cond_destroy(&tq->cond_non_empty);\n+\n+\tif (tq->first)\n+\t\tdie(\"BUG: internal error with queuing jobs for threads\");\n+\n+\tfree(tq->threads);\n+\tret = tq->early_return;\n+\n+\tfree(tq);\n+\treturn ret;\n+}\n+#else /* NO_PTHREADS */\n+\n+struct task_queue {\n+\tint early_return;\n+};\n+\n+struct task_queue *create_task_queue(unsigned max_threads)\n+{\n+\tstruct task_queue *tq = xmalloc(sizeof(*tq));\n+\n+\ttq->early_return = 0;\n+}\n+\n+void add_task(struct task_queue *tq,\n+\t      int (*fct)(struct task_queue *tq, void *task),\n+\t      void *task)\n+{\n+\tif (tq->early_return)\n+\t\treturn;\n+\n+\ttq->early_return |= fct(tq, task);\n+}\n+\n+int finish_task_queue(struct task_queue *tq, void (*fct)(struct task_queue *tq))\n+{\n+\tint ret = tq->early_return;\n+\tfree(tq);\n+\treturn ret;\n+}\n+#endif\ndiff --git a/thread-utils.h b/thread-utils.h\nindex d9a769d..f41cfb1 100644\n--- a/thread-utils.h\n+++ b/thread-utils.h\n@@ -12,4 +12,39 @@ extern int init_recursive_mutex(pthread_mutex_t*);\n #define online_cpus() 1\n \n #endif\n+\n+/*\n+ * Creates a struct `task_queue`, which holds a list of tasks. Up to\n+ * `max_threads` threads are active to process the enqueued tasks\n+ * processing the tasks in a first in first out order.\n+ *\n+ * If `max_threads` is zero the number of cores available will be used.\n+ *\n+ * Currently this only works in environments with pthreads, in other\n+ * environments, the task will be processed sequentially in `add_task`.\n+ */\n+struct task_queue *create_task_queue(unsigned max_threads);\n+\n+/*\n+ * The function and data are put into the task queue.\n+ *\n+ * The function `fct` must not be NULL, as that's used internally\n+ * in `finish_task_queue` to signal shutdown. If the return code\n+ * of `fct` is unequal to 0, the tasks will stop eventually,\n+ * the current parallel tasks will be flushed out.\n+ */\n+void add_task(struct task_queue *tq,\n+\t      int (*fct)(struct task_queue *tq, void *task),\n+\t      void *task);\n+\n+/*\n+ * Waits for all tasks to be done and frees the object. The return code\n+ * is zero if all enqueued tasks were processed.\n+ *\n+ * The function `fct` is called once in each thread after the last task\n+ * for that thread was processed. If no thread local cleanup needs to be\n+ * performed, pass NULL.\n+ */\n+int finish_task_queue(struct task_queue *tq, void (*fct)(struct task_queue *tq));\n+\n #endif /* THREAD_COMPAT_H */\n-- \n2.5.0.264.g784836d\n"},{"id":"268772","messageId":"1440636766-12738-4-git-send-email-sbeller@google.com","threadId":"40187","inReplyTo":"1440636766-12738-1-git-send-email-sbeller@google.com","subject":"[PATCH 3/5] submodule: helper to run foreach in parallel","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-27T00:52:44Z","receivedAt":"2015-08-27T00:52:44Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"Similar to `git submodule foreach` the new command\n`git submodule foreach_parallel` will run a command\non each submodule.\n\nThe commands are run in parallel up to the number of\ncores by default, or you can specify '-j 4' tun just\nrun with 4 threads for example.\n\nOne major difference to `git submodule foreach` is the\nhandling of input and output to the commands. Because\nof the parallel nature of the execution it is not trivial\nhow to schedule the std{in,out,err} channel for submodule\nthe command is run in. So in this patch there is no\nsupport for stdin.\n\nThe goal of the output for std{out, err} is to look like\nthe single threaded version as much as possible, so\nstdout and stderr from one submodule operation are\nbuffered together in one single channel and output\ntogether when the output is allowed.\n\nTo do that, we'll have a mutex for the output, which\neach thread will try to acquire and directly pipe their\noutput to the standard output if they are lucky to\nget the mutex.\n\nIf they do not have the mutex each thread will buffer\ntheir output.\n\nExample:\nLet's assume we have 5 submodules A,B,C,D,E and the\noperation on each submodule takes a different amount\nof time (say `git fetch`), then the output of\n`git submodule foreach` might look like this:\n\n time -->\n output: |---A---|   |-B-|   |----C-----------|   |-D-|   |-E-|\n\nWhen we schedule these threads into two threads, a schedule\nand sample output over time may look like this:\n\nthread 1: |---A---|   |-D-|   |-E-|\nthread 2: |-B-|   |----C-----------|\n\noutput:   |---A---| B |----C-------| E D\n\nSo A will be perceived as it would run normally in\nthe single threaded version of foreach. As B has finished\nby the time the mutex becomes available, the whole buffer\nwill just be dumped into the standard output. This will be\nperceived by the user as just a 'very fast' operation of B.\nOnce that is done, C takes the mutex, and flushes the piled\nup buffer to standard output. In case the subcommand is a\ngit command, we have a progress display, which will just\nlook like the first half of C happend really quickly.\n\nNotice how E and D are put out in a different order than the\noriginal as the new parallel foreach doesn't care about the\norder.\n\nSo this way of output is really good for human consumption\nand not for machine consumption as you always see the progress,\nbut it is not easy to tell which output comes from which\ncommand as there is no indication other than displaying\n\"Entering <submodule path>\" for each beginning section of\noutput.\n\nMaybe we want to integrate the unthreaded foreach eventually\ninto the new code base in C and have special cases for that,\nsuch as accepting stdin again.\n\nSigned-off-by: Stefan Beller <sbeller@google.com>\n---\n builtin/submodule--helper.c | 160 +++++++++++++++++++++++++++++++++++++++++++-\n git-submodule.sh            |  11 ++-\n 2 files changed, 168 insertions(+), 3 deletions(-)\n\ndiff --git a/builtin/submodule--helper.c b/builtin/submodule--helper.c\nindex f11fb9c..2c06f28 100644\n--- a/builtin/submodule--helper.c\n+++ b/builtin/submodule--helper.c\n@@ -8,8 +8,11 @@\n #include \"submodule.h\"\n #include \"submodule-config.h\"\n #include \"string-list.h\"\n+#include \"thread-utils.h\"\n #include \"run-command.h\"\n-\n+#ifndef NO_PTHREADS\n+#include <semaphore.h>\n+#endif\n static const struct cache_entry **ce_entries;\n static int ce_alloc, ce_used;\n static const char *alternative_path;\n@@ -279,6 +282,155 @@ static int module_clone(int argc, const char **argv, const char *prefix)\n \treturn 0;\n }\n \n+struct submodule_args {\n+\tconst char *name;\n+\tconst char *path;\n+\tconst char *sha1;\n+\tconst char *toplevel;\n+\tconst char *prefix;\n+\tconst char **cmd;\n+\tpthread_mutex_t *mutex;\n+};\n+\n+int run_cmd_submodule(struct task_queue *aq, void *task)\n+{\n+\tint i, lock_acquired = 0;\n+\tstruct submodule_args *args = task;\n+\tstruct strbuf out = STRBUF_INIT;\n+\tstruct strbuf sb = STRBUF_INIT;\n+\tstruct child_process *cp = xmalloc(sizeof(*cp));\n+\tchar buf[1024];\n+\n+\tstrbuf_addf(&out, N_(\"Entering %s\\n\"), relative_path(args->path, args->prefix, &sb));\n+\n+\tchild_process_init(cp);\n+\targv_array_pushv(&cp->args, args->cmd);\n+\n+\targv_array_pushf(&cp->env_array, \"name=%s\", args->name);\n+\targv_array_pushf(&cp->env_array, \"path=%s\", args->path);\n+\targv_array_pushf(&cp->env_array, \"sha1=%s\", args->sha1);\n+\targv_array_pushf(&cp->env_array, \"toplevel=%s\", args->toplevel);\n+\n+\tfor (i = 0; local_repo_env[i]; i++)\n+\t\targv_array_push(&cp->env_array, local_repo_env[i]);\n+\n+\tcp->no_stdin = 1;\n+\tcp->out = 0;\n+\tcp->err = -1;\n+\tcp->dir = args->path;\n+\tcp->stdout_to_stderr = 1;\n+\tcp->use_shell = 1;\n+\n+\tif (start_command(cp)) {\n+\t\tdie(\"Could not start command\");\n+\t\tfor (i = 0; cp->args.argv; i++)\n+\t\t\tfprintf(stderr, \"%s\\n\", cp->args.argv[i]);\n+\t}\n+\n+\twhile (1) {\n+\t\tssize_t len = xread(cp->err, buf, sizeof(buf));\n+\t\tif (len < 0)\n+\t\t\tdie(\"Read from child failed\");\n+\t\telse if (len == 0)\n+\t\t\tbreak;\n+\t\telse {\n+\t\t\tstrbuf_add(&out, buf, len);\n+\t\t}\n+\t\tif (!pthread_mutex_trylock(args->mutex))\n+\t\t\tlock_acquired = 1;\n+\t\tif (lock_acquired) {\n+\t\t\tfputs(out.buf, stderr);\n+\t\t\tstrbuf_reset(&out);\n+\t\t}\n+\t}\n+\tif (finish_command(cp))\n+\t\tdie(\"command died with error\");\n+\n+\tif (!lock_acquired)\n+\t\tpthread_mutex_lock(args->mutex);\n+\n+\tfputs(out.buf, stderr);\n+\tpthread_mutex_unlock(args->mutex);\n+\n+\treturn 0;\n+}\n+\n+int module_foreach_parallel(int argc, const char **argv, const char *prefix)\n+{\n+\tint i, recursive = 0, number_threads = 0, quiet = 0;\n+\tstatic struct pathspec pathspec;\n+\tstruct strbuf sb = STRBUF_INIT;\n+\tstruct task_queue *aq;\n+\tchar **cmd;\n+\tconst char **nullargv = {NULL};\n+\tpthread_mutex_t mutex;\n+\n+\tstruct option module_update_options[] = {\n+\t\tOPT_STRING(0, \"prefix\", &alternative_path,\n+\t\t\t   N_(\"path\"),\n+\t\t\t   N_(\"alternative anchor for relative paths\")),\n+\t\tOPT_STRING(0, \"cmd\", &cmd,\n+\t\t\t   N_(\"string\"),\n+\t\t\t   N_(\"command to run\")),\n+\t\tOPT_BOOL('r', \"--recursive\", &recursive,\n+\t\t\t N_(\"Recurse into nexted submodules\")),\n+\t\tOPT_INTEGER('j', \"jobs\", &number_threads,\n+\t\t\t    N_(\"Recurse into nexted submodules\")),\n+\t\tOPT__QUIET(&quiet, N_(\"Suppress output\")),\n+\t\tOPT_END()\n+\t};\n+\n+\tstatic const char * const git_submodule_helper_usage[] = {\n+\t\tN_(\"git submodule--helper foreach [--prefix=<path>] [<path>...]\"),\n+\t\tNULL\n+\t};\n+\n+\targc = parse_options(argc, argv, prefix, module_update_options,\n+\t\t\t     git_submodule_helper_usage, 0);\n+\n+\tif (module_list_compute(0, nullargv, NULL, &pathspec) < 0)\n+\t\treturn 1;\n+\n+\tgitmodules_config();\n+\n+\tpthread_mutex_init(&mutex, NULL);\n+\taq = create_task_queue(number_threads);\n+\n+\tfor (i = 0; i < ce_used; i++) {\n+\t\tconst struct submodule *sub;\n+\t\tconst struct cache_entry *ce = ce_entries[i];\n+\t\tstruct submodule_args *args = malloc(sizeof(*args));\n+\n+\t\tif (ce_stage(ce))\n+\t\t\targs->sha1 = xstrdup(sha1_to_hex(null_sha1));\n+\t\telse\n+\t\t\targs->sha1 = xstrdup(sha1_to_hex(ce->sha1));\n+\n+\t\tstrbuf_reset(&sb);\n+\t\tstrbuf_addf(&sb, \"%s/.git\", ce->name);\n+\t\tif (!file_exists(sb.buf)) {\n+\t\t\tfree(args);\n+\t\t\tcontinue;\n+\t\t}\n+\n+\t\targs->path = ce->name;\n+\t\tsub = submodule_from_path(null_sha1, args->path);\n+\t\tif (!sub)\n+\t\t\tdie(\"No submodule mapping found in .gitmodules for path '%s'\", args->path);\n+\n+\t\targs->name = sub->name;\n+\t\targs->toplevel = xgetcwd();\n+\t\targs->cmd = argv;\n+\t\targs->mutex = &mutex;\n+\t\targs->prefix = alternative_path;\n+\t\tadd_task(aq, run_cmd_submodule, args);\n+\t}\n+\n+\tfinish_task_queue(aq, NULL);\n+\tpthread_mutex_destroy(&mutex);\n+\treturn 0;\n+}\n+\n int cmd_submodule__helper(int argc, const char **argv, const char *prefix)\n {\n \tif (argc < 2)\n@@ -293,6 +445,10 @@ int cmd_submodule__helper(int argc, const char **argv, const char *prefix)\n \tif (!strcmp(argv[1], \"module_clone\"))\n \t\treturn module_clone(argc - 1, argv + 1, prefix);\n \n+\tif (!strcmp(argv[1], \"foreach_parallel\"))\n+\t\treturn module_foreach_parallel(argc - 1, argv + 1, prefix);\n+\n usage:\n-\tusage(\"git submodule--helper [module_list module_name module_clone]\\n\");\n+\tfprintf(stderr, \"%s\", argv[1]);\n+\tusage(\"git submodule--helper [module_list module_name module_clone foreach_parallel]\\n\");\n }\ndiff --git a/git-submodule.sh b/git-submodule.sh\nindex fb5155e..f06488a 100755\n--- a/git-submodule.sh\n+++ b/git-submodule.sh\n@@ -431,6 +431,15 @@ cmd_foreach()\n }\n \n #\n+# Execute an arbitrary command sequence in each checked out\n+# submodule in parallel.\n+#\n+cmd_foreach_parallel()\n+{\n+\tgit submodule--helper foreach_parallel --prefix \"$wt_prefix\" $@\n+}\n+\n+#\n # Register submodules in .git/config\n #\n # $@ = requested paths (default to all)\n@@ -1225,7 +1234,7 @@ cmd_sync()\n while test $# != 0 && test -z \"$command\"\n do\n \tcase \"$1\" in\n-\tadd | foreach | init | deinit | update | status | summary | sync)\n+\tadd | foreach | foreach_parallel | init | deinit | update | status | summary | sync)\n \t\tcommand=$1\n \t\t;;\n \t-q|--quiet)\n-- \n2.5.0.264.g784836d\n"},{"id":"268773","messageId":"1440636766-12738-5-git-send-email-sbeller@google.com","threadId":"40187","inReplyTo":"1440636766-12738-1-git-send-email-sbeller@google.com","subject":"[PATCH 4/5] index-pack: Use the new worker pool","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-27T00:52:45Z","receivedAt":"2015-08-27T00:52:45Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"This demonstrates how the new threading API may be used.\nThere is no change in the workflow, just using the new\nthreading API instead of keeping track of the pthreads\nourselves.\n\nSigned-off-by: Stefan Beller <sbeller@google.com>\n---\n builtin/index-pack.c | 23 ++++++++++++-----------\n 1 file changed, 12 insertions(+), 11 deletions(-)\n\ndiff --git a/builtin/index-pack.c b/builtin/index-pack.c\nindex 3f10840..187b281 100644\n--- a/builtin/index-pack.c\n+++ b/builtin/index-pack.c\n@@ -1075,7 +1075,7 @@ static void resolve_base(struct object_entry *obj)\n }\n \n #ifndef NO_PTHREADS\n-static void *threaded_second_pass(void *data)\n+static int threaded_second_pass(struct task_queue *tq, void *data)\n {\n \tset_thread_data(data);\n \tfor (;;) {\n@@ -1096,7 +1096,7 @@ static void *threaded_second_pass(void *data)\n \n \t\tresolve_base(&objects[i]);\n \t}\n-\treturn NULL;\n+\treturn 0;\n }\n #endif\n \n@@ -1195,18 +1195,19 @@ static void resolve_deltas(void)\n \t\t\t\t\t  nr_ref_deltas + nr_ofs_deltas);\n \n #ifndef NO_PTHREADS\n-\tnr_dispatched = 0;\n+\n \tif (nr_threads > 1 || getenv(\"GIT_FORCE_THREADS\")) {\n+\t\tstruct task_queue *tq;\n+\t\tnr_dispatched = 0;\n \t\tinit_thread();\n-\t\tfor (i = 0; i < nr_threads; i++) {\n-\t\t\tint ret = pthread_create(&thread_data[i].thread, NULL,\n-\t\t\t\t\t\t threaded_second_pass, thread_data + i);\n-\t\t\tif (ret)\n-\t\t\t\tdie(_(\"unable to create thread: %s\"),\n-\t\t\t\t    strerror(ret));\n-\t\t}\n+\n+\t\ttq = create_task_queue(nr_threads);\n \t\tfor (i = 0; i < nr_threads; i++)\n-\t\t\tpthread_join(thread_data[i].thread, NULL);\n+\t\t\tadd_task(tq, threaded_second_pass, thread_data + i);\n+\n+\t\tif (finish_task_queue(tq, NULL))\n+\t\t\tdie(\"Not all threads have finished\");\n+\n \t\tcleanup_thread();\n \t\treturn;\n \t}\n-- \n2.5.0.264.g784836d\n"},{"id":"268769","messageId":"1440636766-12738-6-git-send-email-sbeller@google.com","threadId":"40187","inReplyTo":"1440636766-12738-1-git-send-email-sbeller@google.com","subject":"[PATCH 5/5] pack-objects: Use new worker pool","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-27T00:52:46Z","receivedAt":"2015-08-27T00:52:46Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"Before we had <n> threads doing the delta finding work, and the main thread\nwas load balancing the threads, i.e. moving work from a thread with a large\namount left to an idle thread whenever such a situation arose.\n\nThis moves the load balancing to the threads themselves. As soon as one\nthread is done working it will look at its peer threads and will pickup\nhalf the work load from the thread with the largest pending load.\n\nBy having the load balancing as part of the threads, the locking and\ncommunication model becomes easier, such that we don't need so many\nmutexes any more.\n\nIt also demonstrates the usage of the new threading pool being easily\napplied in different situations.\n\nSigned-off-by: Stefan Beller <sbeller@google.com>\n---\n builtin/pack-objects.c | 175 ++++++++++++++++---------------------------------\n 1 file changed, 57 insertions(+), 118 deletions(-)\n\ndiff --git a/builtin/pack-objects.c b/builtin/pack-objects.c\nindex 62cc16d..f46d2df 100644\n--- a/builtin/pack-objects.c\n+++ b/builtin/pack-objects.c\n@@ -17,6 +17,7 @@\n #include \"pack-objects.h\"\n #include \"progress.h\"\n #include \"refs.h\"\n+#include \"run-command.h\"\n #include \"streaming.h\"\n #include \"thread-utils.h\"\n #include \"pack-bitmap.h\"\n@@ -1887,26 +1888,12 @@ static void try_to_free_from_threads(size_t size)\n \n static try_to_free_t old_try_to_free_routine;\n \n-/*\n- * The main thread waits on the condition that (at least) one of the workers\n- * has stopped working (which is indicated in the .working member of\n- * struct thread_params).\n- * When a work thread has completed its work, it sets .working to 0 and\n- * signals the main thread and waits on the condition that .data_ready\n- * becomes 1.\n- */\n-\n struct thread_params {\n-\tpthread_t thread;\n \tstruct object_entry **list;\n \tunsigned list_size;\n \tunsigned remaining;\n \tint window;\n \tint depth;\n-\tint working;\n-\tint data_ready;\n-\tpthread_mutex_t mutex;\n-\tpthread_cond_t cond;\n \tunsigned *processed;\n };\n \n@@ -1933,7 +1920,52 @@ static void cleanup_threaded_search(void)\n \tpthread_mutex_destroy(&progress_mutex);\n }\n \n-static void *threaded_find_deltas(void *arg)\n+static struct thread_params *p;\n+\n+static void threaded_split_largest_workload(struct thread_params *target)\n+{\n+\tint i;\n+\n+\tstruct object_entry **list;\n+\tstruct thread_params *victim = NULL;\n+\tunsigned sub_size = 0;\n+\n+\t/* Find a victim */\n+\tprogress_lock();\n+\tfor (i = 0; i < delta_search_threads; i++)\n+\t\tif (p[i].remaining > 2*window &&\n+\t\t    (!victim || victim->remaining < p[i].remaining))\n+\t\t\tvictim = &p[i];\n+\n+\tif (victim) {\n+\t\tsub_size = victim->remaining / 2;\n+\t\tlist = victim->list + victim->list_size - sub_size;\n+\t\twhile (sub_size && list[0]->hash &&\n+\t\t       list[0]->hash == list[-1]->hash) {\n+\t\t\tlist++;\n+\t\t\tsub_size--;\n+\t\t}\n+\t\tif (!sub_size) {\n+\t\t\t/*\n+\t\t\t * It is possible for some \"paths\" to have\n+\t\t\t * so many objects that no hash boundary\n+\t\t\t * might be found.  Let's just steal the\n+\t\t\t * exact half in that case.\n+\t\t\t */\n+\t\t\tsub_size = victim->remaining / 2;\n+\t\t\tlist -= sub_size;\n+\t\t}\n+\t\tvictim->list_size -= sub_size;\n+\t\tvictim->remaining -= sub_size;\n+\n+\t\ttarget->list = list;\n+\t\ttarget->list_size = sub_size;\n+\t\ttarget->remaining = sub_size;\n+\t}\n+\tprogress_unlock();\n+}\n+\n+static int threaded_find_deltas(struct task_queue *tq, void *arg)\n {\n \tstruct thread_params *me = arg;\n \n@@ -1941,34 +1973,17 @@ static void *threaded_find_deltas(void *arg)\n \t\tfind_deltas(me->list, &me->remaining,\n \t\t\t    me->window, me->depth, me->processed);\n \n-\t\tprogress_lock();\n-\t\tme->working = 0;\n-\t\tpthread_cond_signal(&progress_cond);\n-\t\tprogress_unlock();\n-\n-\t\t/*\n-\t\t * We must not set ->data_ready before we wait on the\n-\t\t * condition because the main thread may have set it to 1\n-\t\t * before we get here. In order to be sure that new\n-\t\t * work is available if we see 1 in ->data_ready, it\n-\t\t * was initialized to 0 before this thread was spawned\n-\t\t * and we reset it to 0 right away.\n-\t\t */\n-\t\tpthread_mutex_lock(&me->mutex);\n-\t\twhile (!me->data_ready)\n-\t\t\tpthread_cond_wait(&me->cond, &me->mutex);\n-\t\tme->data_ready = 0;\n-\t\tpthread_mutex_unlock(&me->mutex);\n+\t\tthreaded_split_largest_workload(me);\n \t}\n-\t/* leave ->working 1 so that this doesn't get more work assigned */\n-\treturn NULL;\n+\n+\treturn 0;\n }\n \n static void ll_find_deltas(struct object_entry **list, unsigned list_size,\n \t\t\t   int window, int depth, unsigned *processed)\n {\n-\tstruct thread_params *p;\n-\tint i, ret, active_threads = 0;\n+\tstruct task_queue *tq;\n+\tint i;\n \n \tinit_threaded_search();\n \n@@ -1980,8 +1995,11 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,\n \tif (progress > pack_to_stdout)\n \t\tfprintf(stderr, \"Delta compression using up to %d threads.\\n\",\n \t\t\t\tdelta_search_threads);\n+\n \tp = xcalloc(delta_search_threads, sizeof(*p));\n \n+\ttq = create_task_queue(delta_search_threads);\n+\n \t/* Partition the work amongst work threads. */\n \tfor (i = 0; i < delta_search_threads; i++) {\n \t\tunsigned sub_size = list_size / (delta_search_threads - i);\n@@ -1993,8 +2011,6 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,\n \t\tp[i].window = window;\n \t\tp[i].depth = depth;\n \t\tp[i].processed = processed;\n-\t\tp[i].working = 1;\n-\t\tp[i].data_ready = 0;\n \n \t\t/* try to split chunks on \"path\" boundaries */\n \t\twhile (sub_size && sub_size < list_size &&\n@@ -2008,87 +2024,10 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,\n \n \t\tlist += sub_size;\n \t\tlist_size -= sub_size;\n+\t\tadd_task(tq, threaded_find_deltas, &p[i]);\n \t}\n \n-\t/* Start work threads. */\n-\tfor (i = 0; i < delta_search_threads; i++) {\n-\t\tif (!p[i].list_size)\n-\t\t\tcontinue;\n-\t\tpthread_mutex_init(&p[i].mutex, NULL);\n-\t\tpthread_cond_init(&p[i].cond, NULL);\n-\t\tret = pthread_create(&p[i].thread, NULL,\n-\t\t\t\t     threaded_find_deltas, &p[i]);\n-\t\tif (ret)\n-\t\t\tdie(\"unable to create thread: %s\", strerror(ret));\n-\t\tactive_threads++;\n-\t}\n-\n-\t/*\n-\t * Now let's wait for work completion.  Each time a thread is done\n-\t * with its work, we steal half of the remaining work from the\n-\t * thread with the largest number of unprocessed objects and give\n-\t * it to that newly idle thread.  This ensure good load balancing\n-\t * until the remaining object list segments are simply too short\n-\t * to be worth splitting anymore.\n-\t */\n-\twhile (active_threads) {\n-\t\tstruct thread_params *target = NULL;\n-\t\tstruct thread_params *victim = NULL;\n-\t\tunsigned sub_size = 0;\n-\n-\t\tprogress_lock();\n-\t\tfor (;;) {\n-\t\t\tfor (i = 0; !target && i < delta_search_threads; i++)\n-\t\t\t\tif (!p[i].working)\n-\t\t\t\t\ttarget = &p[i];\n-\t\t\tif (target)\n-\t\t\t\tbreak;\n-\t\t\tpthread_cond_wait(&progress_cond, &progress_mutex);\n-\t\t}\n-\n-\t\tfor (i = 0; i < delta_search_threads; i++)\n-\t\t\tif (p[i].remaining > 2*window &&\n-\t\t\t    (!victim || victim->remaining < p[i].remaining))\n-\t\t\t\tvictim = &p[i];\n-\t\tif (victim) {\n-\t\t\tsub_size = victim->remaining / 2;\n-\t\t\tlist = victim->list + victim->list_size - sub_size;\n-\t\t\twhile (sub_size && list[0]->hash &&\n-\t\t\t       list[0]->hash == list[-1]->hash) {\n-\t\t\t\tlist++;\n-\t\t\t\tsub_size--;\n-\t\t\t}\n-\t\t\tif (!sub_size) {\n-\t\t\t\t/*\n-\t\t\t\t * It is possible for some \"paths\" to have\n-\t\t\t\t * so many objects that no hash boundary\n-\t\t\t\t * might be found.  Let's just steal the\n-\t\t\t\t * exact half in that case.\n-\t\t\t\t */\n-\t\t\t\tsub_size = victim->remaining / 2;\n-\t\t\t\tlist -= sub_size;\n-\t\t\t}\n-\t\t\ttarget->list = list;\n-\t\t\tvictim->list_size -= sub_size;\n-\t\t\tvictim->remaining -= sub_size;\n-\t\t}\n-\t\ttarget->list_size = sub_size;\n-\t\ttarget->remaining = sub_size;\n-\t\ttarget->working = 1;\n-\t\tprogress_unlock();\n-\n-\t\tpthread_mutex_lock(&target->mutex);\n-\t\ttarget->data_ready = 1;\n-\t\tpthread_cond_signal(&target->cond);\n-\t\tpthread_mutex_unlock(&target->mutex);\n-\n-\t\tif (!sub_size) {\n-\t\t\tpthread_join(target->thread, NULL);\n-\t\t\tpthread_cond_destroy(&target->cond);\n-\t\t\tpthread_mutex_destroy(&target->mutex);\n-\t\t\tactive_threads--;\n-\t\t}\n-\t}\n+\tfinish_task_queue(tq, NULL);\n \tcleanup_threaded_search();\n \tfree(p);\n }\n-- \n2.5.0.264.g784836d\n"},{"id":"268784","messageId":"92ffde8a07de2a4d50f79ecb7350dbb3@www.dscho.org","threadId":"40187","inReplyTo":"1440636766-12738-2-git-send-email-sbeller@google.com","subject":"Re: [PATCH 1/5] submodule: implement `module_clone` as a builtin helper","fromName":"Johannes Schindelin","fromEmail":"johannes.schindelin@gmx.de","sentAt":"2015-08-27T12:36:08Z","receivedAt":"2015-08-27T12:36:08Z","isPatch":true,"sender":{"key":"johannes.schindelin@gmx.de","avatar":"https://avatars.githubusercontent.com/u/127790?v=4"},"body":"Hi Stefan,\n\nthank you so much for doing this. `git submodule` is really, really slow on Windows...\n\nOn 2015-08-27 02:52, Stefan Beller wrote:\n\n> diff --git a/builtin/submodule--helper.c b/builtin/submodule--helper.c\n> index 4b32a3c..f11fb9c 100644\n> --- a/builtin/submodule--helper.c\n> +++ b/builtin/submodule--helper.c\n> [...]\n> +\n> +/*\n> + * Clone a submodule\n> + *\n> + * $1 = submodule path\n> + * $2 = submodule name\n> + * $3 = URL to clone\n> + * $4 = reference repository to reuse (empty for independent)\n> + * $5 = depth argument for shallow clones (empty for deep)\n\nI think this description is now safely obsolete and can be deleted: you introduced explicit options like --depth to avoid the \"magic\" of positional parameters.\n\n> +static int module_clone(int argc, const char **argv, const char *prefix)\n> +{\n> +\tconst char *path = NULL, *name = NULL, *url = NULL, *reference =\n> NULL, *depth = NULL;\n\nThis line is a little long ;-)\n\n> @@ -135,6 +290,9 @@ int cmd_submodule__helper(int argc, const char\n> **argv, const char *prefix)\n>  \tif (!strcmp(argv[1], \"module_name\"))\n>  \t\treturn module_name(argc - 2, argv + 2, prefix);\n>  \n> +\tif (!strcmp(argv[1], \"module_clone\"))\n> +\t\treturn module_clone(argc - 1, argv + 1, prefix);\n> +\n>  usage:\n> -\tusage(\"git submodule--helper [module_list module_name]\\n\");\n> +\tusage(\"git submodule--helper [module_list module_name module_clone]\\n\");\n\nWas the convention not to use ( ... | ... | ... )?\n\nThanks,\nDscho\n"},{"id":"268791","messageId":"a35977776a7ea97b0ddd4d1921e3e95f@www.dscho.org","threadId":"40187","inReplyTo":"1440636766-12738-3-git-send-email-sbeller@google.com","subject":"Re: [PATCH 2/5] thread-utils: add a threaded task queue","fromName":"Johannes Schindelin","fromEmail":"johannes.schindelin@gmx.de","sentAt":"2015-08-27T12:59:33Z","receivedAt":"2015-08-27T12:59:33Z","isPatch":true,"sender":{"key":"johannes.schindelin@gmx.de","avatar":"https://avatars.githubusercontent.com/u/127790?v=4"},"body":"Hi Stefan,\n\nOn 2015-08-27 02:52, Stefan Beller wrote:\n\n> diff --git a/run-command.c b/run-command.c\n> index 28e1d55..cb15cd9 100644\n> --- a/run-command.c\n> +++ b/run-command.c\n> @@ -668,6 +668,22 @@ int git_atexit(void (*handler)(void))\n>  \n>  #endif\n>  \n> +void setup_main_thread(void)\n> [...]\n>\n> diff --git a/thread-utils.c b/thread-utils.c\n> index a2135e0..b45ab92 100644\n> --- a/thread-utils.c\n> +++ b/thread-utils.c\n> [...]\n> +/* FIXME: deduplicate this code with run-command.c */\n> +static void setup_main_thread(void)\n\nDo you remember off-hand why the code could not be moved to thread-utils.c wholesale? Just curious.\n\n> +#else /* NO_PTHREADS */\n> +\n> +struct task_queue {\n> +\tint early_return;\n> +};\n> +\n> +struct task_queue *create_task_queue(unsigned max_threads)\n> +{\n> +\tstruct task_queue *tq = xmalloc(sizeof(*tq));\n> +\n> +\ttq->early_return = 0;\n> +}\n> +\n> +void add_task(struct task_queue *tq,\n> +\t      int (*fct)(struct task_queue *tq, void *task),\n\nMight make sense to typedef this... Maybe task_t?\n\n> +\t      void *task)\n> +{\n> +\tif (tq->early_return)\n> +\t\treturn;\n\nAh, so \"early_return\" actually means \"interrupted\" or \"canceled\"?\n\nI guess I will have to set aside some time to wrap my head around the way tasks are handled here, in particular how the two `early_return` variables (`dispatcher()`'s local variable and the field in the `task_queue`) interact.\n\nThanks!\nDscho\n"},{"id":"268801","messageId":"CAGZ79kZQ6z8L6Cp-Z56EBq1e-700NsxW9XmFq7fXuMwtUyRSvg@mail.gmail.com","threadId":"40187","inReplyTo":"a35977776a7ea97b0ddd4d1921e3e95f@www.dscho.org","subject":"Re: [PATCH 2/5] thread-utils: add a threaded task queue","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-27T17:02:42Z","receivedAt":"2015-08-27T17:02:42Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Thu, Aug 27, 2015 at 5:59 AM, Johannes Schindelin\n<johannes.schindelin@gmx.de> wrote:\n> Hi Stefan,\n>\n> On 2015-08-27 02:52, Stefan Beller wrote:\n>\n>> diff --git a/run-command.c b/run-command.c\n>> index 28e1d55..cb15cd9 100644\n>> --- a/run-command.c\n>> +++ b/run-command.c\n>> @@ -668,6 +668,22 @@ int git_atexit(void (*handler)(void))\n>>\n>>  #endif\n>>\n>> +void setup_main_thread(void)\n>> [...]\n>>\n>> diff --git a/thread-utils.c b/thread-utils.c\n>> index a2135e0..b45ab92 100644\n>> --- a/thread-utils.c\n>> +++ b/thread-utils.c\n>> [...]\n>> +/* FIXME: deduplicate this code with run-command.c */\n>> +static void setup_main_thread(void)\n>\n> Do you remember off-hand why the code could not be moved to thread-utils.c wholesale? Just curious.\n\nThe code in run-command has a few things regarding the struct async\nhandling in there,\nwhich we don't  need/want. I just realized there is some duplicate\ncode, but I couldn't cut\nit clearly out.\n\n>\n>> +#else /* NO_PTHREADS */\n>> +\n>> +struct task_queue {\n>> +     int early_return;\n>> +};\n>> +\n>> +struct task_queue *create_task_queue(unsigned max_threads)\n>> +{\n>> +     struct task_queue *tq = xmalloc(sizeof(*tq));\n>> +\n>> +     tq->early_return = 0;\n>> +}\n>> +\n>> +void add_task(struct task_queue *tq,\n>> +           int (*fct)(struct task_queue *tq, void *task),\n>\n> Might make sense to typedef this... Maybe task_t?\n>\n>> +           void *task)\n>> +{\n>> +     if (tq->early_return)\n>> +             return;\n>\n> Ah, so \"early_return\" actually means \"interrupted\" or \"canceled\"?\n\nThe early_return is meant to return early in case of an error in some thread.\nIn the threaded version, the `dispatcher` is executed in each thread.\nIt gets its\nnew tasks via `next_task`, which takes the early_return value from the thread,\nORs it into the early_return of the task queue (which all threads have\naccess to).\nSo by the ORing into the task queues early_return the signal to abort\nearly is propagated to a place all threads have access to. And in case\nthat value is\nset, the `next_task` will return NULL as an indication to cleanup and\npthread_exit.\n\n>\n> I guess I will have to set aside some time to wrap my head around the way tasks are handled here, in particular how the two `early_return` variables (`dispatcher()`'s local variable and the field in the `task_queue`) interact.\n>\n> Thanks!\n> Dscho\n"},{"id":"268806","messageId":"CAGZ79kaDT5P_LgELhCA+TWc+caCFpZDcQYsDcjW0vVquXfj3HQ@mail.gmail.com","threadId":"40187","inReplyTo":"92ffde8a07de2a4d50f79ecb7350dbb3@www.dscho.org","subject":"Re: [PATCH 1/5] submodule: implement `module_clone` as a builtin helper","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-27T21:57:45Z","receivedAt":"2015-08-27T21:57:45Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Thu, Aug 27, 2015 at 5:36 AM, Johannes Schindelin\n<johannes.schindelin@gmx.de> wrote:\n> Hi Stefan,\n>\n> thank you so much for doing this. `git submodule` is really, really slow on Windows...\n>\n> On 2015-08-27 02:52, Stefan Beller wrote:\n>\n>> diff --git a/builtin/submodule--helper.c b/builtin/submodule--helper.c\n>> index 4b32a3c..f11fb9c 100644\n>> --- a/builtin/submodule--helper.c\n>> +++ b/builtin/submodule--helper.c\n>> [...]\n>> +\n>> +/*\n>> + * Clone a submodule\n>> + *\n>> + * $1 = submodule path\n>> + * $2 = submodule name\n>> + * $3 = URL to clone\n>> + * $4 = reference repository to reuse (empty for independent)\n>> + * $5 = depth argument for shallow clones (empty for deep)\n>\n> I think this description is now safely obsolete and can be deleted: you introduced explicit options like --depth to avoid the \"magic\" of positional parameters.\n\nok I'll remove that in a reroll.\n\n>\n>> +static int module_clone(int argc, const char **argv, const char *prefix)\n>> +{\n>> +     const char *path = NULL, *name = NULL, *url = NULL, *reference =\n>> NULL, *depth = NULL;\n>\n> This line is a little long ;-)\n\nfixed in reroll.\n\n>\n>> @@ -135,6 +290,9 @@ int cmd_submodule__helper(int argc, const char\n>> **argv, const char *prefix)\n>>       if (!strcmp(argv[1], \"module_name\"))\n>>               return module_name(argc - 2, argv + 2, prefix);\n>>\n>> +     if (!strcmp(argv[1], \"module_clone\"))\n>> +             return module_clone(argc - 1, argv + 1, prefix);\n>> +\n>>  usage:\n>> -     usage(\"git submodule--helper [module_list module_name]\\n\");\n>> +     usage(\"git submodule--helper [module_list module_name module_clone]\\n\");\n>\n> Was the convention not to use ( ... | ... | ... )?\n\ngit tag puts it as\n    usage: git tag [-a | -s | -u <key-id>]\n\nI looked at other --helper (git-difftool--helper which has no help to\nbe invoced on the command line\nand git-bisect--helper which doesn't have the problem as the help is clear).\n\nSo I'll go with [brackets | separated | by | pipes | inside ] then,\njust as git-tag.\n\n>\n> Thanks,\n> Dscho\n"},{"id":"268840","messageId":"xmqq1ten4eau.fsf@gitster.mtv.corp.google.com","threadId":"40187","inReplyTo":"a35977776a7ea97b0ddd4d1921e3e95f@www.dscho.org","subject":"Re: [PATCH 2/5] thread-utils: add a threaded task queue","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-08-28T15:34:49Z","receivedAt":"2015-08-28T15:34:49Z","isPatch":true,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"Johannes Schindelin <johannes.schindelin@gmx.de> writes:\n\n>> +void add_task(struct task_queue *tq,\n>> +\t      int (*fct)(struct task_queue *tq, void *task),\n>\n> Might make sense to typedef this... Maybe task_t?\n\nLet's not introduce user defined type that ends with _t that is seen\nglobally.\n\n>> +\t      void *task)\n>> +{\n>> +\tif (tq->early_return)\n>> +\t\treturn;\n>\n> Ah, so \"early_return\" actually means \"interrupted\" or \"canceled\"?\n>\n> I guess I will have to set aside some time to wrap my head around the\n> way tasks are handled here, in particular how the two `early_return`\n> variables (`dispatcher()`'s local variable and the field in the\n> task_queue`) interact.\n\nWe had a very similar conversation in $gmane/276324 as the\nearly-return and get_task interaction was not quite intuitive.\n\nI thought Stefan said something about this part of the logic being\nunreadable and needs rework.  Perhaps that will come in the next\nreroll, or something?\n\nI tend to agree with you that interrupted or cancelled would be a\ngood name for this thing; at least it would help understanding what\nis going on than \"early-return\".\n\nThanks.\n"}]}