{"thread":{"id":"40145","subject":"[PATCH 1/3] submodule: implement `module_clone` as a builtin helper","startedAt":"2015-08-21T01:40:35Z","lastAt":"2015-08-24T21:22:33Z","messageCount":16,"participants":["Stefan Beller","Junio C Hamano","Jeff King"],"isPatch":true,"patchVersion":1,"patchTotal":3},"messages":[{"id":"268396","messageId":"1440121237-24576-1-git-send-email-sbeller@google.com","threadId":"40145","inReplyTo":null,"subject":"[PATCH 1/3] submodule: implement `module_clone` as a builtin helper","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-21T01:40:35Z","receivedAt":"2015-08-21T01:40:35Z","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>\n---\n\nReplacing the latest patch in sb/submodule-helper as that contains\ndebug code in `cmd_submodule__helper`.\n\n builtin/submodule--helper.c | 156 +++++++++++++++++++++++++++++++++++++++++++-\n git-submodule.sh            |  80 +----------------------\n 2 files changed, 158 insertions(+), 78 deletions(-)\n\ndiff --git a/builtin/submodule--helper.c b/builtin/submodule--helper.c\nindex 4b32a3c..ae74b80 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,156 @@ 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+\tstrbuf_reset(&sb);\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\tstrbuf_addf(&sb, \"%s/%s\", xgetcwd(), sm_gitdir);\n+\t\tsm_gitdir = strbuf_detach(&sb, NULL);\n+\t\tstrbuf_reset(&sb);\n+\t\tfree(s);\n+\t}\n+\tstrbuf_addf(&sb, \"%s/%s\", xgetcwd(), 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+\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 +286,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.g01b5c38.dirty\n"},{"id":"268397","messageId":"1440121237-24576-2-git-send-email-sbeller@google.com","threadId":"40145","inReplyTo":"1440121237-24576-1-git-send-email-sbeller@google.com","subject":"[RFC PATCH 2/3] run-commands: add an async queue processor","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-21T01:40:36Z","receivedAt":"2015-08-21T01:40:36Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"This adds functionality to do work in parallel.\n\nThe whole life cycle of such a thread pool would look like\n\n    struct task_queue * tq = create_task_queue(32); // no of threads\n    for (...)\n        add_task(tq, process_one_item_function, item); // non blocking\n    ...\n    int ret = finish_task_queue(tq); // blocks until all tasks are done\n    if (!tq)\n        die (\"Not all items were be processed\");\n\nThe caller must take care of handling the output.\n\nSigned-off-by: Stefan Beller <sbeller@google.com>\n---\n\nI sent this a while ago to the list, no comments on it :(\nThe core functionality stayed the same, but I hope to improved naming and\nlocation of the code.\n\nThe WIP is only for the NO_PTHREADS case.\n\n\n run-command.c | 212 ++++++++++++++++++++++++++++++++++++++++++++++++++++++----\n run-command.h |  30 +++++++++\n 2 files changed, 230 insertions(+), 12 deletions(-)\n\ndiff --git a/run-command.c b/run-command.c\nindex 28e1d55..4029011 100644\n--- a/run-command.c\n+++ b/run-command.c\n@@ -4,6 +4,21 @@\n #include \"sigchain.h\"\n #include \"argv-array.h\"\n \n+#ifdef NO_PTHREADS\n+\n+#else\n+\n+#include \"thread-utils.h\"\n+\n+#include <pthread.h>\n+#include <semaphore.h>\n+#include <stdio.h>\n+#include <unistd.h>\n+\n+#endif\n+\n+#include \"git-compat-util.h\"\n+\n void child_process_init(struct child_process *child)\n {\n \tmemset(child, 0, sizeof(*child));\n@@ -668,6 +683,22 @@ int git_atexit(void (*handler)(void))\n \n #endif\n \n+void setup_main_thread()\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 +771,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);\n@@ -852,3 +872,171 @@ int capture_command(struct child_process *cmd, struct strbuf *buf, size_t hint)\n \tclose(cmd->out);\n \treturn finish_command(cmd);\n }\n+\n+#ifndef NO_PTHREADS\n+struct job_list {\n+\tint (*fct)(struct task_queue *aq, void *task);\n+\tvoid *task;\n+\tstruct job_list *next;\n+};\n+#endif\n+\n+struct task_queue {\n+#ifndef NO_PTHREADS\n+\t/*\n+\t * To avoid deadlocks always aquire the semaphores with lowest priority\n+\t * first, priorites are in descending order as listed.\n+\t *\n+\t * The `mutex` is a general purpose lock for modifying data in the async\n+\t * queue, such as adding a new task or adding a return value from\n+\t * an already run task.\n+\t *\n+\t * `workingcount` and `freecount` are opposing semaphores, the sum of\n+\t * their values should equal `max_threads` at any time while the `mutex`\n+\t * is available.\n+\t */\n+\tsem_t mutex;\n+\tsem_t workingcount;\n+\tsem_t freecount;\n+\n+\tpthread_t *threads;\n+\tunsigned max_threads;\n+\n+\tstruct job_list *first;\n+\tstruct job_list *last;\n+#endif\n+\tint early_return;\n+};\n+\n+#ifndef NO_PTHREADS\n+\n+static void get_task(struct task_queue *aq,\n+\t\t     int (**fct)(struct task_queue *aq, void *task),\n+\t\t     void **task,\n+\t\t     int *early_return)\n+{\n+\tstruct job_list *job;\n+\n+\tsem_wait(&aq->workingcount);\n+\tsem_wait(&aq->mutex);\n+\n+\tif (!aq->first)\n+\t\tdie(\"BUG: internal error with dequeuing jobs for threads\");\n+\tjob = aq->first;\n+\t*fct = job->fct;\n+\t*task = job->task;\n+\taq->early_return |= *early_return;\n+\t*early_return = aq->early_return;\n+\taq->first = job->next;\n+\tif (!aq->first)\n+\t\taq->last = NULL;\n+\n+\tsem_post(&aq->freecount);\n+\tsem_post(&aq->mutex);\n+\n+\tfree(job);\n+}\n+\n+static void* dispatcher(void *args)\n+{\n+\tvoid *task;\n+\tint (*fct)(struct task_queue *aq, void *data);\n+\tint early_return = 0;\n+\tstruct task_queue *aq = args;\n+\n+\tget_task(aq, &fct, &task, &early_return);\n+\twhile (fct || early_return != 0) {\n+\t\tearly_return = fct(aq, task);\n+\t\tget_task(aq, &fct, &task, &early_return);\n+\t}\n+\n+\tpthread_exit(0);\n+}\n+#endif\n+\n+struct task_queue *create_task_queue(unsigned max_threads)\n+{\n+\tstruct task_queue *aq = xmalloc(sizeof(*aq));\n+\n+#ifndef NO_PTHREADS\n+\tint i;\n+\tif (!max_threads)\n+\t\taq->max_threads = online_cpus();\n+\telse\n+\t\taq->max_threads = max_threads;\n+\n+\tsem_init(&aq->mutex, 0, 1);\n+\tsem_init(&aq->workingcount, 0, 0);\n+\tsem_init(&aq->freecount, 0, aq->max_threads);\n+\taq->threads = xmalloc(aq->max_threads * sizeof(pthread_t));\n+\n+\tfor (i = 0; i < aq->max_threads; i++)\n+\t\tpthread_create(&aq->threads[i], 0, &dispatcher, aq);\n+\n+\taq->first = NULL;\n+\taq->last = NULL;\n+\n+\tsetup_main_thread();\n+#endif\n+\taq->early_return = 0;\n+\n+\treturn aq;\n+}\n+\n+void add_task(struct task_queue *aq,\n+\t      int (*fct)(struct task_queue *aq, void *task),\n+\t      void *task)\n+{\n+#ifndef NO_PTHREADS\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+\tsem_wait(&aq->freecount);\n+\tsem_wait(&aq->mutex);\n+\n+\tif (!aq->last) {\n+\t\taq->last = job_list;\n+\t\taq->first = aq->last;\n+\t} else {\n+\t\taq->last->next = job_list;\n+\t\taq->last = aq->last->next;\n+\t}\n+\n+\tsem_post(&aq->workingcount);\n+\tsem_post(&aq->mutex);\n+#else\n+\tALLOC_GROW(aq->ret->ret, aq->ret->count + 1, aq->ret->alloc);\n+\taq->ret->ret[aq->ret->count++] = aq->function(job);\n+#endif\n+}\n+\n+int finish_task_queue(struct task_queue *aq)\n+{\n+\tint ret;\n+#ifndef NO_PTHREADS\n+\tint i;\n+\tfor (i = 0; i < aq->max_threads; i++)\n+\t\tadd_task(aq, NULL, NULL);\n+\n+\tfor (i = 0; i < aq->max_threads; i++)\n+\t\tpthread_join(aq->threads[i], 0);\n+\n+\tsem_destroy(&aq->mutex);\n+\tsem_destroy(&aq->workingcount);\n+\tsem_destroy(&aq->freecount);\n+\n+\tif (aq->first)\n+\t\tdie(\"BUG: internal error with queuing jobs for threads\");\n+\n+\tfree(aq->threads);\n+#endif\n+\tret = aq->early_return;\n+\n+\tfree(aq);\n+\treturn ret;\n+}\n+\ndiff --git a/run-command.h b/run-command.h\nindex 5b4425a..c2cfd49 100644\n--- a/run-command.h\n+++ b/run-command.h\n@@ -119,4 +119,34 @@ struct async {\n int start_async(struct async *async);\n int finish_async(struct async *async);\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 *aq,\n+\t      int (*fct)(struct task_queue *aq, 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+int finish_task_queue(struct task_queue *aq);\n+\n #endif\n-- \n2.5.0.264.g01b5c38.dirty\n"},{"id":"268398","messageId":"1440121237-24576-3-git-send-email-sbeller@google.com","threadId":"40145","inReplyTo":"1440121237-24576-1-git-send-email-sbeller@google.com","subject":"[WIP/PATCH 3/3] submodule: helper to run foreach in parallel","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-21T01:40:37Z","receivedAt":"2015-08-21T01:40:37Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"This runs a command on each submodule in parallel and should eventually\nreplace `git submodule foreach`.\n\nThere is a new option -j/--jobs (inspired by make) to specify the number\nof parallel threads.\n\nThe jobs=1 case needs to be special cases to exactly replicate the current\ndefault behavior of `git submodule foreach` such as working stdin input.\nFor more than one job there is no input possible and the output of both\nstdout/stderr of the command are put into the stderr in an ordered fashion,\ni.e. the tasks to not intermingle their output in races.\n\nwhat currently works:\n git submodule--helper foreach \"echo \\$toplevel-\\$name-\\$path-\\$sha1\"\nwhich I took for testing during development from t7407.\n\nSigned-off-by: Stefan Beller <sbeller@google.com>\n---\n\nThis is still WIP, but comments are welcome.\n\n builtin/submodule--helper.c | 147 +++++++++++++++++++++++++++++++++++++++++++-\n git-submodule.sh            |   9 +++\n 2 files changed, 154 insertions(+), 2 deletions(-)\n\ndiff --git a/builtin/submodule--helper.c b/builtin/submodule--helper.c\nindex ae74b80..9823302 100644\n--- a/builtin/submodule--helper.c\n+++ b/builtin/submodule--helper.c\n@@ -9,7 +9,9 @@\n #include \"submodule-config.h\"\n #include \"string-list.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@@ -275,6 +277,144 @@ 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+\tstruct submodule_output *out;\n+\tsem_t *mutex;\n+};\n+\n+int run_cmd_submodule(struct task_queue *aq, void *task)\n+{\n+\tint i;\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}\n+\tif (finish_command(cp))\n+\t\tdie(\"command died with error\");\n+\n+\tsem_wait(args->mutex);\n+\tfputs(out.buf, stderr);\n+\tsem_post(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 = 1, 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+\tsem_t *mutex = xmalloc(sizeof(*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+\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\tcontinue;\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 = xstrdup(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);\n+\treturn 0;\n+}\n+\n int cmd_submodule__helper(int argc, const char **argv, const char *prefix)\n {\n \tif (argc < 2)\n@@ -289,6 +429,9 @@ 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\"))\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+\tusage(\"git submodule--helper [module_list module_name module_clone foreach]\\n\");\n }\ndiff --git a/git-submodule.sh b/git-submodule.sh\nindex fb5155e..fa18434 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 module_foreach_parallel --prefix \"$wt_prefix\" $@\n+}\n+\n+#\n # Register submodules in .git/config\n #\n # $@ = requested paths (default to all)\n-- \n2.5.0.264.g01b5c38.dirty\n"},{"id":"268442","messageId":"xmqqegiw5uom.fsf@gitster.dls.corp.google.com","threadId":"40145","inReplyTo":"1440121237-24576-2-git-send-email-sbeller@google.com","subject":"Re: [RFC PATCH 2/3] run-commands: add an async queue processor","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-08-21T19:05:13Z","receivedAt":"2015-08-21T19:05:13Z","isPatch":true,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"Stefan Beller <sbeller@google.com> writes:\n\n> This adds functionality to do work in parallel.\n>\n> The whole life cycle of such a thread pool would look like\n>\n>     struct task_queue * tq = create_task_queue(32); // no of threads\n>     for (...)\n>         add_task(tq, process_one_item_function, item); // non blocking\n>     ...\n>     int ret = finish_task_queue(tq); // blocks until all tasks are done\n>     if (!tq)\n>         die (\"Not all items were be processed\");\n>\n> The caller must take care of handling the output.\n>\n> Signed-off-by: Stefan Beller <sbeller@google.com>\n> ---\n>\n> I sent this a while ago to the list, no comments on it :(\n\nThe primary reason I suspect is because you sent to a wrong set of\npeople.  Submodule folks have largely been working in the scripted\nones, and may not necessarily be the ones who are most familiar with\nthe run-command infrastructure.\n\n\"shortlog --no-merges\" tells me that the obvious suspects are j6t\nand peff.\n\n> The core functionality stayed the same, but I hope to improved naming and\n> location of the code.\n>\n> The WIP is only for the NO_PTHREADS case.\n\n>  run-command.c | 212 ++++++++++++++++++++++++++++++++++++++++++++++++++++++----\n>  run-command.h |  30 +++++++++\n>  2 files changed, 230 insertions(+), 12 deletions(-)\n>\n> diff --git a/run-command.c b/run-command.c\n> index 28e1d55..4029011 100644\n> --- a/run-command.c\n> +++ b/run-command.c\n> @@ -4,6 +4,21 @@\n>  #include \"sigchain.h\"\n>  #include \"argv-array.h\"\n>  \n> +#ifdef NO_PTHREADS\n> +\n> +#else\n> +\n> +#include \"thread-utils.h\"\n> +\n> +#include <pthread.h>\n> +#include <semaphore.h>\n> +#include <stdio.h>\n> +#include <unistd.h>\n> +\n> +#endif\n> +\n> +#include \"git-compat-util.h\"\n> +\n\nThis goes against the way we have been organizing the header files.\n\nhttp://thread.gmane.org/gmane.comp.version-control.git/276241/focus=276265\n\n> @@ -668,6 +683,22 @@ int git_atexit(void (*handler)(void))\n>  \n>  #endif\n>  \n> +void setup_main_thread()\n\nvoid setup_main_thread(void)\n\n> @@ -852,3 +872,171 @@ int capture_command(struct child_process *cmd, struct strbuf *buf, size_t hint)\n>  \tclose(cmd->out);\n>  \treturn finish_command(cmd);\n>  }\n> +\n> +#ifndef NO_PTHREADS\n> +struct job_list {\n> +\tint (*fct)(struct task_queue *aq, void *task);\n> +\tvoid *task;\n> +\tstruct job_list *next;\n> +};\n> +#endif\n> +\n> +struct task_queue {\n> +#ifndef NO_PTHREADS\n> +\t/*\n> +\t * To avoid deadlocks always aquire the semaphores with lowest priority\n\nacquire.\n\n> +\t * first, priorites are in descending order as listed.\n> +\t *\n> +\t * The `mutex` is a general purpose lock for modifying data in the async\n> +\t * queue, such as adding a new task or adding a return value from\n> +\t * an already run task.\n> +\t *\n> +\t * `workingcount` and `freecount` are opposing semaphores, the sum of\n> +\t * their values should equal `max_threads` at any time while the `mutex`\n> +\t * is available.\n> +\t */\n> +\tsem_t mutex;\n> +\tsem_t workingcount;\n> +\tsem_t freecount;\n> +\n> +\tpthread_t *threads;\n> +\tunsigned max_threads;\n> +\n> +\tstruct job_list *first;\n> +\tstruct job_list *last;\n> +#endif\n> +\tint early_return;\n> +};\n> +\n> +#ifndef NO_PTHREADS\n> +\n> +static void get_task(struct task_queue *aq,\n> +\t\t     int (**fct)(struct task_queue *aq, void *task),\n> +\t\t     void **task,\n> +\t\t     int *early_return)\n> +{\n> +\tstruct job_list *job;\n> +\n> +\tsem_wait(&aq->workingcount);\n> +\tsem_wait(&aq->mutex);\n> +\n> +\tif (!aq->first)\n> +\t\tdie(\"BUG: internal error with dequeuing jobs for threads\");\n> +\tjob = aq->first;\n> +\t*fct = job->fct;\n> +\t*task = job->task;\n> +\taq->early_return |= *early_return;\n> +\t*early_return = aq->early_return;\n> +\taq->first = job->next;\n> +\tif (!aq->first)\n> +\t\taq->last = NULL;\n> +\n> +\tsem_post(&aq->freecount);\n> +\tsem_post(&aq->mutex);\n> +\n> +\tfree(job);\n> +}\n> +\n> +static void* dispatcher(void *args)\n\nstatic void *dispatcher(....)\n\n> +{\n> +\tvoid *task;\n> +\tint (*fct)(struct task_queue *aq, void *data);\n\ns/data/task/?\n\n> +\tint early_return = 0;\n> +\tstruct task_queue *aq = args;\n> +\n> +\tget_task(aq, &fct, &task, &early_return);\n> +\twhile (fct || early_return != 0) {\n> +\t\tearly_return = fct(aq, task);\n> +\t\tget_task(aq, &fct, &task, &early_return);\n> +\t}\n\nIf the func said \"we are done, you may stop dispatching now\", do you\nstill want to do another get_task()?\n\n> +\tpthread_exit(0);\n> +}\n> +#endif\n> +\n> +struct task_queue *create_task_queue(unsigned max_threads)\n> +{\n> +\tstruct task_queue *aq = xmalloc(sizeof(*aq));\n> +\n> +#ifndef NO_PTHREADS\n> +\tint i;\n> +\tif (!max_threads)\n> +\t\taq->max_threads = online_cpus();\n> +\telse\n> +\t\taq->max_threads = max_threads;\n> +\n> +\tsem_init(&aq->mutex, 0, 1);\n> +\tsem_init(&aq->workingcount, 0, 0);\n> +\tsem_init(&aq->freecount, 0, aq->max_threads);\n> +\taq->threads = xmalloc(aq->max_threads * sizeof(pthread_t));\n> +\n> +\tfor (i = 0; i < aq->max_threads; i++)\n> +\t\tpthread_create(&aq->threads[i], 0, &dispatcher, aq);\n> +\n> +\taq->first = NULL;\n> +\taq->last = NULL;\n\n\nShouldn't these be initialized before letting threads call into\ndispatcher?  The workingcount semaphore that is initialized to 0 may\nprevent them from peeking into these pointers and barfing, but still...\n\n> +\n> +\tsetup_main_thread();\n> +#endif\n> +\taq->early_return = 0;\n> +\n> +\treturn aq;\n> +}\n"},{"id":"268443","messageId":"xmqq7foo5tty.fsf@gitster.dls.corp.google.com","threadId":"40145","inReplyTo":"1440121237-24576-3-git-send-email-sbeller@google.com","subject":"Re: [WIP/PATCH 3/3] submodule: helper to run foreach in parallel","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-08-21T19:23:37Z","receivedAt":"2015-08-21T19:23:37Z","isPatch":true,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"Stefan Beller <sbeller@google.com> writes:\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> +\tstruct submodule_output *out;\n> +\tsem_t *mutex;\n> +};\n\nI do not see what submodule_output looks like in the patch, but I\nguess you are capturing _all_ output from a task and emitting\neverything at the end, after the task is done?\n\nI have to wonder if that is what people would expect and more\nimportantly if that is the most useful.  I am sympathetic to the\ndesire to avoid the output totally mixed up from different processes\nemitting things in an uncoordinated manner, and \"slurp everything\nand then show everything in one go\" is certainly _one_ way to do so,\nbut early feedback also counts.  Besides, because the order in which\ntasks are dispatched and completed is unpredictable, you cannot\nexpect a machine parseable output _without_ assuming some help from\nthe command invoked by each task (e.g. by prefixing the task's output\nwith some string that identifies which submodule the output is about).\n\nOnce you assume that the command is _aware_ that it needs to help\nthe foreach-parallel infrastructure so that the user can sift their\ncollective outputs to make sense of them, wouldn't a line-buffered\nintermixing also acceptable, and wouldn't it be a much lower impact\napproach to solve the same problem?\n"},{"id":"268445","messageId":"20150821194454.GB26466@sigill.intra.peff.net","threadId":"40145","inReplyTo":"xmqqegiw5uom.fsf@gitster.dls.corp.google.com","subject":"Re: [RFC PATCH 2/3] run-commands: add an async queue processor","fromName":"Jeff King","fromEmail":"peff@peff.net","sentAt":"2015-08-21T19:44:54Z","receivedAt":"2015-08-21T19:44:54Z","isPatch":true,"sender":{"key":"peff@peff.net","avatar":"https://avatars.githubusercontent.com/u/45925?v=4"},"body":"On Fri, Aug 21, 2015 at 12:05:13PM -0700, Junio C Hamano wrote:\n\n> The primary reason I suspect is because you sent to a wrong set of\n> people.  Submodule folks have largely been working in the scripted\n> ones, and may not necessarily be the ones who are most familiar with\n> the run-command infrastructure.\n> \n> \"shortlog --no-merges\" tells me that the obvious suspects are j6t\n> and peff.\n\nNo good deed goes unpunished. ;)\n\nBefore even looking at the implementation, my first question would be\nwhether this pattern is applicable in several places in git (i.e., is it\nworth the extra complexity of abstracting out in the first place). I\nthink there are a few task-queue patterns already in git; for example\nthe delta search in pack-objects. Is the interface given here sufficient\nto convert pack-objects? Is the result nicer to read? Is it as\nefficient?\n\nWe do not need to convert all possible call-sites to the new abstracted\ncode at once. But I find that converting at least _one_ is a good litmus\ntest to confirm that a new interface is generally useful.\n\n-Peff\n"},{"id":"268446","messageId":"CAGZ79kZrcPAAt+miHDGQp=052S-q=JaKvvLgKHaPG+G6cDjBtg@mail.gmail.com","threadId":"40145","inReplyTo":"xmqqegiw5uom.fsf@gitster.dls.corp.google.com","subject":"Re: [RFC PATCH 2/3] run-commands: add an async queue processor","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-21T19:45:28Z","receivedAt":"2015-08-21T19:45:28Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Fri, Aug 21, 2015 at 12:05 PM, Junio C Hamano <gitster@pobox.com> wrote:\n> Stefan Beller <sbeller@google.com> writes:\n>\n>> This adds functionality to do work in parallel.\n>>\n>> The whole life cycle of such a thread pool would look like\n>>\n>>     struct task_queue * tq = create_task_queue(32); // no of threads\n>>     for (...)\n>>         add_task(tq, process_one_item_function, item); // non blocking\n>>     ...\n>>     int ret = finish_task_queue(tq); // blocks until all tasks are done\n>>     if (!tq)\n>>         die (\"Not all items were be processed\");\n>>\n>> The caller must take care of handling the output.\n>>\n>> Signed-off-by: Stefan Beller <sbeller@google.com>\n>> ---\n>>\n>> I sent this a while ago to the list, no comments on it :(\n>\n> The primary reason I suspect is because you sent to a wrong set of\n> people.  Submodule folks have largely been working in the scripted\n> ones, and may not necessarily be the ones who are most familiar with\n> the run-command infrastructure.\n>\n> \"shortlog --no-merges\" tells me that the obvious suspects are j6t\n> and peff.\n\nnoted.\n\n>\n>> The core functionality stayed the same, but I hope to improved naming and\n>> location of the code.\n>>\n>> The WIP is only for the NO_PTHREADS case.\n>\n>>  run-command.c | 212 ++++++++++++++++++++++++++++++++++++++++++++++++++++++----\n>>  run-command.h |  30 +++++++++\n>>  2 files changed, 230 insertions(+), 12 deletions(-)\n>>\n>> diff --git a/run-command.c b/run-command.c\n>> index 28e1d55..4029011 100644\n>> --- a/run-command.c\n>> +++ b/run-command.c\n>> @@ -4,6 +4,21 @@\n>>  #include \"sigchain.h\"\n>>  #include \"argv-array.h\"\n>>\n>> +#ifdef NO_PTHREADS\n>> +\n>> +#else\n>> +\n>> +#include \"thread-utils.h\"\n>> +\n>> +#include <pthread.h>\n>> +#include <semaphore.h>\n>> +#include <stdio.h>\n>> +#include <unistd.h>\n>> +\n>> +#endif\n>> +\n>> +#include \"git-compat-util.h\"\n>> +\n>\n> This goes against the way we have been organizing the header files.\n>\n> http://thread.gmane.org/gmane.comp.version-control.git/276241/focus=276265\n\nok\n>\n>> @@ -668,6 +683,22 @@ int git_atexit(void (*handler)(void))\n>>\n>>  #endif\n>>\n>> +void setup_main_thread()\n>\n> void setup_main_thread(void)\n>\n>> @@ -852,3 +872,171 @@ int capture_command(struct child_process *cmd, struct strbuf *buf, size_t hint)\n>>       close(cmd->out);\n>>       return finish_command(cmd);\n>>  }\n>> +\n>> +#ifndef NO_PTHREADS\n>> +struct job_list {\n>> +     int (*fct)(struct task_queue *aq, void *task);\n>> +     void *task;\n>> +     struct job_list *next;\n>> +};\n>> +#endif\n>> +\n>> +struct task_queue {\n>> +#ifndef NO_PTHREADS\n>> +     /*\n>> +      * To avoid deadlocks always aquire the semaphores with lowest priority\n>\n> acquire.\n>\n>> +      * first, priorites are in descending order as listed.\n>> +      *\n>> +      * The `mutex` is a general purpose lock for modifying data in the async\n>> +      * queue, such as adding a new task or adding a return value from\n>> +      * an already run task.\n>> +      *\n>> +      * `workingcount` and `freecount` are opposing semaphores, the sum of\n>> +      * their values should equal `max_threads` at any time while the `mutex`\n>> +      * is available.\n>> +      */\n>> +     sem_t mutex;\n>> +     sem_t workingcount;\n>> +     sem_t freecount;\n>> +\n>> +     pthread_t *threads;\n>> +     unsigned max_threads;\n>> +\n>> +     struct job_list *first;\n>> +     struct job_list *last;\n>> +#endif\n>> +     int early_return;\n>> +};\n>> +\n>> +#ifndef NO_PTHREADS\n>> +\n>> +static void get_task(struct task_queue *aq,\n>> +                  int (**fct)(struct task_queue *aq, void *task),\n>> +                  void **task,\n>> +                  int *early_return)\n>> +{\n>> +     struct job_list *job;\n>> +\n>> +     sem_wait(&aq->workingcount);\n>> +     sem_wait(&aq->mutex);\n>> +\n>> +     if (!aq->first)\n>> +             die(\"BUG: internal error with dequeuing jobs for threads\");\n>> +     job = aq->first;\n>> +     *fct = job->fct;\n>> +     *task = job->task;\n>> +     aq->early_return |= *early_return;\n>> +     *early_return = aq->early_return;\n>> +     aq->first = job->next;\n>> +     if (!aq->first)\n>> +             aq->last = NULL;\n>> +\n>> +     sem_post(&aq->freecount);\n>> +     sem_post(&aq->mutex);\n>> +\n>> +     free(job);\n>> +}\n>> +\n>> +static void* dispatcher(void *args)\n>\n> static void *dispatcher(....)\n>\n>> +{\n>> +     void *task;\n>> +     int (*fct)(struct task_queue *aq, void *data);\n>\n> s/data/task/?\n>\n>> +     int early_return = 0;\n>> +     struct task_queue *aq = args;\n>> +\n>> +     get_task(aq, &fct, &task, &early_return);\n>> +     while (fct || early_return != 0) {\n>> +             early_return = fct(aq, task);\n>> +             get_task(aq, &fct, &task, &early_return);\n>> +     }\n>\n> If the func said \"we are done, you may stop dispatching now\", do you\n> still want to do another get_task()?\n\nThis question shows me I messed up readability of this. `get_task` both gets\na new task as well as taking the value from early_return writing it to\naq->early_return,\nsuch that other threads are notified.\n\nMaybe I should do that explicitely here and not get the new task.\n\n>\n>> +     pthread_exit(0);\n>> +}\n>> +#endif\n>> +\n>> +struct task_queue *create_task_queue(unsigned max_threads)\n>> +{\n>> +     struct task_queue *aq = xmalloc(sizeof(*aq));\n>> +\n>> +#ifndef NO_PTHREADS\n>> +     int i;\n>> +     if (!max_threads)\n>> +             aq->max_threads = online_cpus();\n>> +     else\n>> +             aq->max_threads = max_threads;\n>> +\n>> +     sem_init(&aq->mutex, 0, 1);\n>> +     sem_init(&aq->workingcount, 0, 0);\n>> +     sem_init(&aq->freecount, 0, aq->max_threads);\n>> +     aq->threads = xmalloc(aq->max_threads * sizeof(pthread_t));\n>> +\n>> +     for (i = 0; i < aq->max_threads; i++)\n>> +             pthread_create(&aq->threads[i], 0, &dispatcher, aq);\n>> +\n>> +     aq->first = NULL;\n>> +     aq->last = NULL;\n>\n>\n> Shouldn't these be initialized before letting threads call into\n> dispatcher?  The workingcount semaphore that is initialized to 0 may\n> prevent them from peeking into these pointers and barfing, but still...\n\nThey are initialized to NULL as the empty queue doesn't need a\ncontainer element.\nDo we do queues in another way usually?\n\n>\n>> +\n>> +     setup_main_thread();\n>> +#endif\n>> +     aq->early_return = 0;\n>> +\n>> +     return aq;\n>> +}\n"},{"id":"268447","messageId":"CAGZ79kYM6m-Me=okbxMAY_NKDm40JwfgRutwTzYfT5=ez42=jg@mail.gmail.com","threadId":"40145","inReplyTo":"20150821194454.GB26466@sigill.intra.peff.net","subject":"Re: [RFC PATCH 2/3] run-commands: add an async queue processor","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-21T19:48:23Z","receivedAt":"2015-08-21T19:48:23Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Fri, Aug 21, 2015 at 12:44 PM, Jeff King <peff@peff.net> wrote:\n> On Fri, Aug 21, 2015 at 12:05:13PM -0700, Junio C Hamano wrote:\n>\n>> The primary reason I suspect is because you sent to a wrong set of\n>> people.  Submodule folks have largely been working in the scripted\n>> ones, and may not necessarily be the ones who are most familiar with\n>> the run-command infrastructure.\n>>\n>> \"shortlog --no-merges\" tells me that the obvious suspects are j6t\n>> and peff.\n>\n> No good deed goes unpunished. ;)\n>\n> Before even looking at the implementation, my first question would be\n> whether this pattern is applicable in several places in git (i.e., is it\n> worth the extra complexity of abstracting out in the first place). I\n> think there are a few task-queue patterns already in git; for example\n> the delta search in pack-objects. Is the interface given here sufficient\n> to convert pack-objects? Is the result nicer to read? Is it as\n> efficient?\n>\n> We do not need to convert all possible call-sites to the new abstracted\n> code at once. But I find that converting at least _one_ is a good litmus\n> test to confirm that a new interface is generally useful.\n\nOk, I'll head off to convert one place.\n\n>\n> -Peff\n"},{"id":"268448","messageId":"20150821195135.GA26720@sigill.intra.peff.net","threadId":"40145","inReplyTo":"CAGZ79kYM6m-Me=okbxMAY_NKDm40JwfgRutwTzYfT5=ez42=jg@mail.gmail.com","subject":"Re: [RFC PATCH 2/3] run-commands: add an async queue processor","fromName":"Jeff King","fromEmail":"peff@peff.net","sentAt":"2015-08-21T19:51:35Z","receivedAt":"2015-08-21T19:51:35Z","isPatch":true,"sender":{"key":"peff@peff.net","avatar":"https://avatars.githubusercontent.com/u/45925?v=4"},"body":"On Fri, Aug 21, 2015 at 12:48:23PM -0700, Stefan Beller wrote:\n\n> > Before even looking at the implementation, my first question would be\n> > whether this pattern is applicable in several places in git (i.e., is it\n> > worth the extra complexity of abstracting out in the first place). I\n> > think there are a few task-queue patterns already in git; for example\n> > the delta search in pack-objects. Is the interface given here sufficient\n> > to convert pack-objects? Is the result nicer to read? Is it as\n> > efficient?\n> >\n> > We do not need to convert all possible call-sites to the new abstracted\n> > code at once. But I find that converting at least _one_ is a good litmus\n> > test to confirm that a new interface is generally useful.\n> \n> Ok, I'll head off to convert one place.\n\nThanks. By the way, reading over what I wrote, it sounds a little\nharsher than I meant. It is not \"if you do not convert an existing site,\nwe cannot accept a new interface, period\". But I would like at least\nsome explanation as an alternative, like \"I'm pretty sure we can convert\nsite X to this, but it is not a good time to do so now because of Y\".\nWhere \"Y\" might be \"we need to do this other refactoring work first\", or\n\"it would be disruptive to other work in the area\".\n\n-Peff\n"},{"id":"268449","messageId":"CAGZ79katsSYno3H1+yxoKk4K-1VxH99p-F8uwaDiakeWTKs-fw@mail.gmail.com","threadId":"40145","inReplyTo":"20150821195135.GA26720@sigill.intra.peff.net","subject":"Re: [RFC PATCH 2/3] run-commands: add an async queue processor","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-21T20:12:15Z","receivedAt":"2015-08-21T20:12:15Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Fri, Aug 21, 2015 at 12:51 PM, Jeff King <peff@peff.net> wrote:\n> On Fri, Aug 21, 2015 at 12:48:23PM -0700, Stefan Beller wrote:\n>\n>> > Before even looking at the implementation, my first question would be\n>> > whether this pattern is applicable in several places in git (i.e., is it\n>> > worth the extra complexity of abstracting out in the first place). I\n>> > think there are a few task-queue patterns already in git; for example\n>> > the delta search in pack-objects. Is the interface given here sufficient\n>> > to convert pack-objects? Is the result nicer to read? Is it as\n>> > efficient?\n>> >\n>> > We do not need to convert all possible call-sites to the new abstracted\n>> > code at once. But I find that converting at least _one_ is a good litmus\n>> > test to confirm that a new interface is generally useful.\n>>\n>> Ok, I'll head off to convert one place.\n>\n> Thanks. By the way, reading over what I wrote, it sounds a little\n> harsher than I meant.\n\nIt did not sound harsh at all. I was just reading an internal mailing list,\nwhich cites over generalizing as a very bad practice worse than\npremature optimization, so I totally understand your concern and\nagree.\n\n> It is not \"if you do not convert an existing site,\n> we cannot accept a new interface, period\". But I would like at least\n> some explanation as an alternative, like \"I'm pretty sure we can convert\n> site X to this, but it is not a good time to do so now because of Y\".\n> Where \"Y\" might be \"we need to do this other refactoring work first\", or\n> \"it would be disruptive to other work in the area\".\n>\n> -Peff\n"},{"id":"268450","messageId":"CAGZ79kYxYawAtATAoTmgG42w7E+2fEPfHek5bpbMgZ32wya5Zw@mail.gmail.com","threadId":"40145","inReplyTo":"xmqq7foo5tty.fsf@gitster.dls.corp.google.com","subject":"Re: [WIP/PATCH 3/3] submodule: helper to run foreach in parallel","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-21T20:21:32Z","receivedAt":"2015-08-21T20:21:32Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Fri, Aug 21, 2015 at 12:23 PM, Junio C Hamano <gitster@pobox.com> wrote:\n> Stefan Beller <sbeller@google.com> writes:\n>\n>> +struct submodule_args {\n>> +     const char *name;\n>> +     const char *path;\n>> +     const char *sha1;\n>> +     const char *toplevel;\n>> +     const char *prefix;\n>> +     const char **cmd;\n>> +     struct submodule_output *out;\n>> +     sem_t *mutex;\n>> +};\n>\n> I do not see what submodule_output looks like in the patch, but I\n> guess you are capturing _all_ output from a task and emitting\n> everything at the end, after the task is done?\n\nIt was a leftover and will be removed. Sorry for wasting your time.\nSo jrnieder and me had a discussion on how to do get the output right.\n\nAnd we had 2 designs:\n\n1) The worker thread acquires a mutex to be allowed to write to\n  stdout. This makes early output easy for each thread in case of\n  error messages. It also requires less resources than 2).\n  Also the expected code complexity is lower as the decisions\n  are made locally.\n\n2) Have a single handler which deals with all output of all tasks.\n  This handler would `select` on a huge number of pipes from\n  all the tasks, (so we would require a lot of pipes). And this central\n  handler (which would be submodule_output in this case) would\n  take care of having no intermingled racy output from tasks, with\n  prefixes and all bells and whistles. This handler could also\n  give a progress meter (300 out of 500 submodules updated already)\n  for all threads, or when the last task is running switch to piping\n  the output of that task to stdout in real time, so you'd get progress\n  of the last task as you're already used to.\n\nThe outcome of the discussion was to split the worker pool/load\ndistribution from the output handling in any case as there is no need\nto couple them. In my very first designs I had the output handling\nas part of the asynchronous worker pool.\n\nThis RFC implements 1), that's why there is only the mutex.\nI plan on enhancing this to let the last task output in real time (no buffering)\nas well as another counter \"task n/m completed\" after each task\nis done.\n\n>\n> I have to wonder if that is what people would expect and more\n> importantly if that is the most useful.  I am sympathetic to the\n> desire to avoid the output totally mixed up from different processes\n> emitting things in an uncoordinated manner, and \"slurp everything\n> and then show everything in one go\" is certainly _one_ way to do so,\n> but early feedback also counts.  Besides, because the order in which\n> tasks are dispatched and completed is unpredictable, you cannot\n> expect a machine parseable output _without_ assuming some help from\n> the command invoked by each task (e.g. by prefixing the task's output\n> with some string that identifies which submodule the output is about).\n\nI was very focused on the \"submodule foreach\" output, which (in case of\nno -q given), displays a\n\n    \"Entering %s\"\n\nas the first line for each finished task. I suspect that is not enough to\nmake it a good machine parseable output. So I will prefix each\nline with a running number of the task. Maybe like this:\n\n  [001/500] Entering 'foo/bar'\n  [001/500] Here goes the text for task foo/bar\n  [001/500] Here goes another text for task foo/bar\n  [002/500] Entering 'foo/baz'\n  [002/500] Here goes the text for task foo/baz\n  [002/500] Here goes another text for task foo/bar\n  [003/500] Entering 'foo/bla'\n  ...\n\nThis will make the output for each task distinguishable from\nother tasks as well offering some sort of progress meter.\n(The fewer submodules you have the less fine grained\nthe progress bar becomes)\n\nJonathan suggested to add a capability to the git protocol for a\nmachine readable progress meter in another channel. so we do not\nneed to parse the current output Counting/Compressing/Transfer/etc\n\n>\n> Once you assume that the command is _aware_ that it needs to help\n> the foreach-parallel infrastructure so that the user can sift their\n> collective outputs to make sense of them, wouldn't a line-buffered\n> intermixing also acceptable, and wouldn't it be a much lower impact\n> approach to solve the same problem?\n"},{"id":"268452","messageId":"xmqq37zc5q8l.fsf@gitster.dls.corp.google.com","threadId":"40145","inReplyTo":"20150821194454.GB26466@sigill.intra.peff.net","subject":"Re: [RFC PATCH 2/3] run-commands: add an async queue processor","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-08-21T20:41:14Z","receivedAt":"2015-08-21T20:41:14Z","isPatch":true,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"Jeff King <peff@peff.net> writes:\n\n> On Fri, Aug 21, 2015 at 12:05:13PM -0700, Junio C Hamano wrote:\n>\n>> The primary reason I suspect is because you sent to a wrong set of\n>> people.  Submodule folks have largely been working in the scripted\n>> ones, and may not necessarily be the ones who are most familiar with\n>> the run-command infrastructure.\n>> \n>> \"shortlog --no-merges\" tells me that the obvious suspects are j6t\n>> and peff.\n>\n> No good deed goes unpunished. ;)\n>\n> Before even looking at the implementation, my first question would be\n> whether this pattern is applicable in several places in git (i.e., is it\n> worth the extra complexity of abstracting out in the first place). I\n> think there are a few task-queue patterns already in git; for example\n> the delta search in pack-objects. Is the interface given here sufficient\n> to convert pack-objects? Is the result nicer to read? Is it as\n> efficient?\n>\n> We do not need to convert all possible call-sites to the new abstracted\n> code at once. But I find that converting at least _one_ is a good litmus\n> test to confirm that a new interface is generally useful.\n\nAh, thanks for saying this.  I recall saying something similar\nearlier and totally agree with you.\n"},{"id":"268453","messageId":"xmqqy4h44bdp.fsf@gitster.dls.corp.google.com","threadId":"40145","inReplyTo":"CAGZ79kZrcPAAt+miHDGQp=052S-q=JaKvvLgKHaPG+G6cDjBtg@mail.gmail.com","subject":"Re: [RFC PATCH 2/3] run-commands: add an async queue processor","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-08-21T20:47:30Z","receivedAt":"2015-08-21T20:47:30Z","isPatch":true,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"Stefan Beller <sbeller@google.com> writes:\n\n>>> +struct task_queue *create_task_queue(unsigned max_threads)\n>>> +{\n>>> +     struct task_queue *aq = xmalloc(sizeof(*aq));\n>>> +\n>>> +#ifndef NO_PTHREADS\n>>> +     int i;\n>>> +     if (!max_threads)\n>>> +             aq->max_threads = online_cpus();\n>>> +     else\n>>> +             aq->max_threads = max_threads;\n>>> +\n>>> +     sem_init(&aq->mutex, 0, 1);\n>>> +     sem_init(&aq->workingcount, 0, 0);\n>>> +     sem_init(&aq->freecount, 0, aq->max_threads);\n>>> +     aq->threads = xmalloc(aq->max_threads * sizeof(pthread_t));\n>>> +\n>>> +     for (i = 0; i < aq->max_threads; i++)\n>>> +             pthread_create(&aq->threads[i], 0, &dispatcher, aq);\n>>> +\n>>> +     aq->first = NULL;\n>>> +     aq->last = NULL;\n>>\n>>\n>> Shouldn't these be initialized before letting threads call into\n>> dispatcher?  The workingcount semaphore that is initialized to 0 may\n>> prevent them from peeking into these pointers and barfing, but still...\n>\n> They are initialized to NULL as the empty queue doesn't need a\n> container element.\n> Do we do queues in another way usually?\n\nI do not think we are on the same wavelength.  What I meant was to\ndo this:\n\n\taq = xmalloc(...);\n        set up _everything_ in aq and make it a consistent state;\n        /* aq->first and aq->last are part of _everything_ in aq */\n        for (many times)\n        \tpthread_create(...);\n\n\t/* No aq->first = aq->last = NULL assignment here */\n\ninstead of\n\n\taq = xmalloc(...);\n        set up part of aq;\n        for (many times)\n        \tpthread_create(...);\n\tbelatedly initialize aq->first and aq->last and finally\n        aq becomes a consistent state.\n\nwhich is what we see above.  The latter works _only_ because the\nthreads created are blocked waiting on aq->workingcount which is\ninitialized to block before threads are created to run dispatch,\nand one of the early things dispatch does is to try acquiring that\nsemaphore to block before accessing aq->first and aq->last.\n"},{"id":"268456","messageId":"CAGZ79kaj0=2yw+KBLKQCkMzZpmcdFk93yDd7m+3x6Jn7gGap8w@mail.gmail.com","threadId":"40145","inReplyTo":"xmqqy4h44bdp.fsf@gitster.dls.corp.google.com","subject":"Re: [RFC PATCH 2/3] run-commands: add an async queue processor","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-21T20:56:50Z","receivedAt":"2015-08-21T20:56:50Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Fri, Aug 21, 2015 at 1:47 PM, Junio C Hamano <gitster@pobox.com> wrote:\n>\n> I do not think we are on the same wavelength.  What I meant was to\n> do this:\n>\n>         aq = xmalloc(...);\n>         set up _everything_ in aq and make it a consistent state;\n>         /* aq->first and aq->last are part of _everything_ in aq */\n>         for (many times)\n>                 pthread_create(...);\n>\n>         /* No aq->first = aq->last = NULL assignment here */\n>\n> instead of\n>\n>         aq = xmalloc(...);\n>         set up part of aq;\n>         for (many times)\n>                 pthread_create(...);\n>         belatedly initialize aq->first and aq->last and finally\n>         aq becomes a consistent state.\n>\n> which is what we see above.  The latter works _only_ because the\n> threads created are blocked waiting on aq->workingcount which is\n> initialized to block before threads are created to run dispatch,\n> and one of the early things dispatch does is to try acquiring that\n> semaphore to block before accessing aq->first and aq->last.\n\nI see your point and it makes sense to me as it makes the\nmental memory model cleaner\n\nI viewed the methods as atomic units (you would not call add_task\nbefore the constructor has finished, so no harm there. And a dispatcher\nin the pthread would just block on the `workingcount` semaphore as the\nsemaphores are the main inter thread communication. I viewed the queue\nas just a secondary thing to distribute the work load.\n"},{"id":"268464","messageId":"CAGZ79kbhRQ0SmSu6xiij6Gxdrtfh02knexz6+CNjihY4xbC83w@mail.gmail.com","threadId":"40145","inReplyTo":"20150821194454.GB26466@sigill.intra.peff.net","subject":"Re: [RFC PATCH 2/3] run-commands: add an async queue processor","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-21T23:40:09Z","receivedAt":"2015-08-21T23:40:09Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Fri, Aug 21, 2015 at 12:44 PM, Jeff King <peff@peff.net> wrote:\n> On Fri, Aug 21, 2015 at 12:05:13PM -0700, Junio C Hamano wrote:\n>\n>> The primary reason I suspect is because you sent to a wrong set of\n>> people.  Submodule folks have largely been working in the scripted\n>> ones, and may not necessarily be the ones who are most familiar with\n>> the run-command infrastructure.\n>>\n>> \"shortlog --no-merges\" tells me that the obvious suspects are j6t\n>> and peff.\n>\n> No good deed goes unpunished. ;)\n>\n> Before even looking at the implementation, my first question would be\n> whether this pattern is applicable in several places in git (i.e., is it\n> worth the extra complexity of abstracting out in the first place). I\n> think there are a few task-queue patterns already in git; for example\n> the delta search in pack-objects. Is the interface given here sufficient\n> to convert pack-objects? Is the result nicer to read? Is it as\n> efficient?\n\nI have converted index-pack threading now, and it looks quite easy:\n\ndiff --git a/builtin/index-pack.c b/builtin/index-pack.c\nindex 3f10840..159ee36 100644\n--- a/builtin/index-pack.c\n+++ b/builtin/index-pack.c\n@@ -11,6 +11,7 @@\n #include \"exec_cmd.h\"\n #include \"streaming.h\"\n #include \"thread-utils.h\"\n+#include \"run-command.h\"\n\n static const char index_pack_usage[] =\n \"git index-pack [-v] [-o <index-file>] [--keep | --keep=<msg>]\n[--verify] [--strict] (<pack-file> | --stdin [--fix-thin]\n[<pack-file>])\";\n@@ -1075,7 +1076,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        set_thread_data(data);\n        for (;;) {\n@@ -1096,7 +1097,7 @@ static void *threaded_second_pass(void *data)\n\n                resolve_base(&objects[i]);\n        }\n-       return NULL;\n+       return 0;\n }\n #endif\n\n@@ -1195,18 +1196,18 @@ static void resolve_deltas(void)\n                                          nr_ref_deltas + nr_ofs_deltas);\n\n #ifndef NO_PTHREADS\n-       nr_dispatched = 0;\n        if (nr_threads > 1 || getenv(\"GIT_FORCE_THREADS\")) {\n+               nr_dispatched = 0;\n                init_thread();\n-               for (i = 0; i < nr_threads; i++) {\n-                       int ret = pthread_create(&thread_data[i].thread, NULL,\n-                                                threaded_second_pass,\nthread_data + i);\n-                       if (ret)\n-                               die(_(\"unable to create thread: %s\"),\n-                                   strerror(ret));\n-               }\n+\n+               tq = create_task_queue(nr_threads);\n+\n                for (i = 0; i < nr_threads; i++)\n-                       pthread_join(thread_data[i].thread, NULL);\n+                       add_task(tq, threaded_second_pass, thread_data + i);\n+\n+               if (finish_task_queue(tq))\n+                       die(\"Not all threads have finished\");\n+\n                cleanup_thread();\n                return;\n        }\n---\n(tests pass)\nThis was cheating as I picked to convert index-pack as opposed to upload-pack\n(index-pack is very similar to a queued workload. This was just moving\nthe thread\ncreation into the new proposed queue processor.)\n\nI realize now this can be adapted a bit more, to show off the queue features\nbut would require a larger rewrite. So instead of just creating the threads and\nthen locking, we get rid of the worker lock like this:\n\ndiff --git a/builtin/index-pack.c b/builtin/index-pack.c\nindex 3f10840..797efea 100644\n--- a/builtin/index-pack.c\n+++ b/builtin/index-pack.c\n@@ -11,6 +11,7 @@\n #include \"exec_cmd.h\"\n #include \"streaming.h\"\n #include \"thread-utils.h\"\n+#include \"run-command.h\"\n\n static const char index_pack_usage[] =\n \"git index-pack [-v] [-o <index-file>] [--keep | --keep=<msg>]\n[--verify] [--strict] (<pack-file> | --stdin [--fix-thin]\n[<pack-file>])\";\n@@ -106,10 +107,6 @@ static pthread_mutex_t counter_mutex;\n #define counter_lock() lock_mutex(&counter_mutex)\n #define counter_unlock() unlock_mutex(&counter_mutex)\n\n-static pthread_mutex_t work_mutex;\n-#define work_lock() lock_mutex(&work_mutex)\n-#define work_unlock() unlock_mutex(&work_mutex)\n-\n static pthread_mutex_t deepest_delta_mutex;\n #define deepest_delta_lock() lock_mutex(&deepest_delta_mutex)\n #define deepest_delta_unlock() unlock_mutex(&deepest_delta_mutex)\n@@ -140,7 +137,6 @@ static void init_thread(void)\n  int i;\n  init_recursive_mutex(&read_mutex);\n  pthread_mutex_init(&counter_mutex, NULL);\n- pthread_mutex_init(&work_mutex, NULL);\n  pthread_mutex_init(&type_cas_mutex, NULL);\n  if (show_stat)\n  pthread_mutex_init(&deepest_delta_mutex, NULL);\n@@ -163,7 +159,6 @@ static void cleanup_thread(void)\n  threads_active = 0;\n  pthread_mutex_destroy(&read_mutex);\n  pthread_mutex_destroy(&counter_mutex);\n- pthread_mutex_destroy(&work_mutex);\n  pthread_mutex_destroy(&type_cas_mutex);\n  if (show_stat)\n  pthread_mutex_destroy(&deepest_delta_mutex);\n@@ -181,9 +176,6 @@ static void cleanup_thread(void)\n #define counter_lock()\n #define counter_unlock()\n\n-#define work_lock()\n-#define work_unlock()\n-\n #define deepest_delta_lock()\n #define deepest_delta_unlock()\n\n@@ -1075,28 +1067,24 @@ 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- set_thread_data(data);\n- for (;;) {\n- int i;\n- counter_lock();\n- display_progress(progress, nr_resolved_deltas);\n- counter_unlock();\n- work_lock();\n- while (nr_dispatched < nr_objects &&\n-       is_delta_type(objects[nr_dispatched].type))\n- nr_dispatched++;\n- if (nr_dispatched >= nr_objects) {\n- work_unlock();\n- break;\n- }\n- i = nr_dispatched++;\n- work_unlock();\n+ if (!get_thread_data()) {\n+ struct thread_local *t = xmalloc(sizeof(*t));\n+ t->pack_fd = open(curr_pack, O_RDONLY);\n+ if (t->pack_fd == -1)\n+ die_errno(_(\"unable to open %s\"), curr_pack);\n\n- resolve_base(&objects[i]);\n+ set_thread_data(t);\n+ /* TODO: I haven't figured out where to free this memory */\n  }\n- return NULL;\n+\n+ resolve_base(data);\n+\n+ counter_lock();\n+ display_progress(progress, nr_resolved_deltas);\n+ counter_unlock();\n+ return 0;\n }\n #endif\n\n@@ -1195,18 +1183,21 @@ static void resolve_deltas(void)\n   nr_ref_deltas + nr_ofs_deltas);\n\n #ifndef NO_PTHREADS\n- nr_dispatched = 0;\n  if (nr_threads > 1 || getenv(\"GIT_FORCE_THREADS\")) {\n+ struct task_queue *tq;\n+\n  init_thread();\n- for (i = 0; i < nr_threads; i++) {\n- int ret = pthread_create(&thread_data[i].thread, NULL,\n- threaded_second_pass, thread_data + i);\n- if (ret)\n- die(_(\"unable to create thread: %s\"),\n-    strerror(ret));\n- }\n- for (i = 0; i < nr_threads; i++)\n- pthread_join(thread_data[i].thread, NULL);\n+ tq = create_task_queue(nr_threads);\n+\n+ for (nr_dispatched = 0; nr_dispatched < nr_objects; nr_dispatched++)\n+ if (!is_delta_type(objects[nr_dispatched].type))\n+ add_task(tq, threaded_second_pass, &objects[nr_dispatched]);\n+\n+ if (finish_task_queue(tq))\n+ die(\"Not all threads have finished\");\n+\n+ /* Here might be a good place to free the thread local storage caches */\n+\n  cleanup_thread();\n  return;\n  }\n----\n\nThis looks very pleasant to me as we have no lock contention all the time,\nwith the worker lock, but the main thread can load the work queue easily\nwhile we directly start processing in the worker threads.\nThis also reduces the lines of code which usually is a good sign.\n\nHowever there is a memory leak in this second solution.\n\nHow do you think these two approaches look like?\n\n(I cannot really say objectively if it's easier to read, as I wrote the code,\nso of course it is easy for me)\n\nWhy did I not pick upload-pack?\n========================\n\nI could not spot easily how to make it a typical queuing problem.\nWe start in the threads, and once in a while we're like: \"Uhg, this\nthread has more load than the other, let's shove a bit over there\"\n\nSo what we would need there is splitting the work in smallest chunks\nfrom the beginning and just load it into the queue via add_task\n\n\n\n>\n> We do not need to convert all possible call-sites to the new abstracted\n> code at once. But I find that converting at least _one_ is a good litmus\n> test to confirm that a new interface is generally useful.\n>\n> -Peff\n"},{"id":"268607","messageId":"xmqqvbc42xgm.fsf@gitster.dls.corp.google.com","threadId":"40145","inReplyTo":"CAGZ79kbhRQ0SmSu6xiij6Gxdrtfh02knexz6+CNjihY4xbC83w@mail.gmail.com","subject":"Re: [RFC PATCH 2/3] run-commands: add an async queue processor","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-08-24T21:22:33Z","receivedAt":"2015-08-24T21:22:33Z","isPatch":true,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"Stefan Beller <sbeller@google.com> writes:\n\n> diff --git a/builtin/index-pack.c b/builtin/index-pack.c\n> index 3f10840..159ee36 100644\n> --- a/builtin/index-pack.c\n> +++ b/builtin/index-pack.c\n> @@ -11,6 +11,7 @@\n>  #include \"exec_cmd.h\"\n>  #include \"streaming.h\"\n>  #include \"thread-utils.h\"\n> +#include \"run-command.h\"\n>\n>  static const char index_pack_usage[] =\n>  \"git index-pack [-v] [-o <index-file>] [--keep | --keep=<msg>]\n> [--verify] [--strict] (<pack-file> | --stdin [--fix-thin]\n> [<pack-file>])\";\n> @@ -1075,7 +1076,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>         set_thread_data(data);\n>         for (;;) {\n> @@ -1096,7 +1097,7 @@ static void *threaded_second_pass(void *data)\n>\n>                 resolve_base(&objects[i]);\n>         }\n> -       return NULL;\n> +       return 0;\n>  }\n>  #endif\n>\n> @@ -1195,18 +1196,18 @@ static void resolve_deltas(void)\n>                                           nr_ref_deltas + nr_ofs_deltas);\n>\n>  #ifndef NO_PTHREADS\n> -       nr_dispatched = 0;\n>         if (nr_threads > 1 || getenv(\"GIT_FORCE_THREADS\")) {\n> +               nr_dispatched = 0;\n>                 init_thread();\n> -               for (i = 0; i < nr_threads; i++) {\n> -                       int ret = pthread_create(&thread_data[i].thread, NULL,\n> -                                                threaded_second_pass,\n> thread_data + i);\n> -                       if (ret)\n> -                               die(_(\"unable to create thread: %s\"),\n> -                                   strerror(ret));\n> -               }\n> +\n> +               tq = create_task_queue(nr_threads);\n> +\n>                 for (i = 0; i < nr_threads; i++)\n> -                       pthread_join(thread_data[i].thread, NULL);\n> +                       add_task(tq, threaded_second_pass, thread_data + i);\n> +\n> +               if (finish_task_queue(tq))\n> +                       die(\"Not all threads have finished\");\n> +\n>                 cleanup_thread();\n>                 return;\n>         }\n\nThis looks quite straight-forward, but that is not too surprising,\nas the \"dispatcher\" side naturally should have a similar logic to\nmanage threads by creating and joining them ;-)\n\n> @@ -1075,28 +1067,24 @@ 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> - set_thread_data(data);\n> - for (;;) {\n> - int i;\n> - counter_lock();\n> - display_progress(progress, nr_resolved_deltas);\n> - counter_unlock();\n> - work_lock();\n> - while (nr_dispatched < nr_objects &&\n> -       is_delta_type(objects[nr_dispatched].type))\n> - nr_dispatched++;\n> - if (nr_dispatched >= nr_objects) {\n> - work_unlock();\n> - break;\n> - }\n> - i = nr_dispatched++;\n> - work_unlock();\n> + if (!get_thread_data()) {\n> + struct thread_local *t = xmalloc(sizeof(*t));\n> + t->pack_fd = open(curr_pack, O_RDONLY);\n> + if (t->pack_fd == -1)\n> + die_errno(_(\"unable to open %s\"), curr_pack);\n>\n> - resolve_base(&objects[i]);\n> + set_thread_data(t);\n> + /* TODO: I haven't figured out where to free this memory */\n\nSorry but it is hard to grok what is going on in the code with\nsquished indentation.\n\n> Why did I not pick upload-pack?\n> ========================\n>\n> I could not spot easily how to make it a typical queuing problem.\n> We start in the threads, and once in a while we're like: \"Uhg, this\n> thread has more load than the other, let's shove a bit over there\"\n>\n> So what we would need there is splitting the work in smallest chunks\n> from the beginning and just load it into the queue via add_task\n\n... or a way for the overload and tasks to communicate with each\nother and rebalance.  If I am not mistaken, it has a big negative\nconsequence for pack-objects to split the work to too small a chunk,\nas the chunk boundary also becomes boundary of find delta bases.\n"}]}