{"thread":{"id":"40174","subject":"[PATCH 2/5] thread-utils: add a threaded task queue","startedAt":"2015-08-25T17:28:21Z","lastAt":"2015-08-26T17:21:08Z","messageCount":19,"participants":["Stefan Beller","Jeff King","Junio C Hamano"],"isPatch":true,"patchVersion":1,"patchTotal":5},"messages":[{"id":"268661","messageId":"1440523706-23041-1-git-send-email-sbeller@google.com","threadId":"40174","inReplyTo":null,"subject":"[RFC PATCH 0/5] Demonstrate new parallel threading API","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-25T17:28:21Z","receivedAt":"2015-08-25T17:28:21Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"This series build on top of origin/sb/submodule-helper.\nThe first patch is a fixup to the last commit in the target branch\nto fix a memory leak. The patch is not really part of the series, but\nas I chose to build on top of that series I can fix it up as we go.\n\nThe patch 2 adds a new API to easily use a threaded work pool.\n\npatch 3 adds the command `submodule foreach_parallel` (completely untested, RFC!)\nwhich was the original goal of this series. But oh well, I got tricked by Peff\nto prove how awesome the new API for parallel threading is, so that's shown off\nin patches 4 and 5.\n\nNote: Both patch 4 and 5 delete more lines of code than they add, improving \nreadability a lot as you can focus on the actual task and not on the threading\nstuff.\n\nAny feedback welcome!\nThanks,\nStefan\n\nStefan Beller (5):\n  FIXUP 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        |  71 ++++++-------\n builtin/pack-objects.c      | 175 +++++++++++---------------------\n builtin/submodule--helper.c | 159 +++++++++++++++++++++++++++--\n git-submodule.sh            |   9 ++\n run-command.c               |  29 +++---\n thread-utils.c              | 237 ++++++++++++++++++++++++++++++++++++++++++++\n thread-utils.h              |  40 ++++++++\n 7 files changed, 545 insertions(+), 175 deletions(-)\n\n-- \n2.5.0.400.gff86faf\n"},{"id":"268665","messageId":"1440523706-23041-2-git-send-email-sbeller@google.com","threadId":"40174","inReplyTo":"1440523706-23041-1-git-send-email-sbeller@google.com","subject":"[PATCH 1/5] FIXUP submodule: implement `module_clone` as a builtin helper","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-25T17:28:22Z","receivedAt":"2015-08-25T17:28:22Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"This closes the memory leaks as pointed out by Jeff.\n\nSigned-off-by: Stefan Beller <sbeller@google.com>\n---\n builtin/submodule--helper.c | 11 +++++++----\n 1 file changed, 7 insertions(+), 4 deletions(-)\n\ndiff --git a/builtin/submodule--helper.c b/builtin/submodule--helper.c\nindex ae74b80..7e298b4 100644\n--- a/builtin/submodule--helper.c\n+++ b/builtin/submodule--helper.c\n@@ -220,7 +220,6 @@ static int module_clone(int argc, const char **argv, const char *prefix)\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@@ -259,12 +258,16 @@ static int module_clone(int argc, const char **argv, const char *prefix)\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\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\tstrbuf_reset(&sb);\n \t\tfree(s);\n \t}\n-\tstrbuf_addf(&sb, \"%s/%s\", xgetcwd(), path);\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-- \n2.5.0.400.gff86faf\n"},{"id":"268660","messageId":"1440523706-23041-3-git-send-email-sbeller@google.com","threadId":"40174","inReplyTo":"1440523706-23041-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-25T17:28:23Z","receivedAt":"2015-08-25T17:28:23Z","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 | 237 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++\n thread-utils.h |  40 ++++++++++\n 3 files changed, 294 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..936b3672 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,238 @@ 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+\t/*\n+\t * To avoid deadlocks always acquire 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+\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+\tsem_wait(&tq->workingcount);\n+\tsem_wait(&tq->mutex);\n+\n+\tif (*early_return) {\n+\t\ttq->early_return |= *early_return;\n+\t\t*fct = NULL;\n+\t\t*task = NULL;\n+\t} else {\n+\t\tif (!tq->first)\n+\t\t\tdie(\"BUG: internal error with dequeuing jobs for threads\");\n+\n+\t\tjob = tq->first;\n+\t\t*fct = job->fct;\n+\t\t*task = job->task;\n+\n+\t\ttq->first = job->next;\n+\t\tif (!tq->first)\n+\t\t\ttq->last = NULL;\n+\t}\n+\n+\tsem_post(&tq->freecount);\n+\tsem_post(&tq->mutex);\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 != 0) {\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+\tsem_init(&tq->mutex, 0, 1);\n+\tsem_init(&tq->workingcount, 0, 0);\n+\tsem_init(&tq->freecount, 0, tq->max_threads);\n+\ttq->threads = xmalloc(tq->max_threads * sizeof(pthread_t));\n+\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+\tsem_wait(&tq->freecount);\n+\tsem_wait(&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+\n+\tsem_post(&tq->workingcount);\n+\tsem_post(&tq->mutex);\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+\tsem_destroy(&tq->mutex);\n+\tsem_destroy(&tq->workingcount);\n+\tsem_destroy(&tq->freecount);\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..977d37b 100644\n--- a/thread-utils.h\n+++ b/thread-utils.h\n@@ -7,9 +7,49 @@\n extern int online_cpus(void);\n extern int init_recursive_mutex(pthread_mutex_t*);\n \n+#include <pthread.h>\n+#include <semaphore.h>\n+#include <stdio.h>\n+#include <unistd.h>\n+\n #else\n \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.400.gff86faf\n"},{"id":"268664","messageId":"1440523706-23041-4-git-send-email-sbeller@google.com","threadId":"40174","inReplyTo":"1440523706-23041-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-25T17:28:24Z","receivedAt":"2015-08-25T17:28:24Z","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 builtin/submodule--helper.c | 148 +++++++++++++++++++++++++++++++++++++++++++-\n git-submodule.sh            |   9 +++\n 2 files changed, 155 insertions(+), 2 deletions(-)\n\ndiff --git a/builtin/submodule--helper.c b/builtin/submodule--helper.c\nindex 7e298b4..adfa0e4 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@@ -278,6 +281,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, NULL);\n+\treturn 0;\n+}\n+\n int cmd_submodule__helper(int argc, const char **argv, const char *prefix)\n {\n \tif (argc < 2)\n@@ -292,6 +433,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.400.gff86faf\n"},{"id":"268662","messageId":"1440523706-23041-5-git-send-email-sbeller@google.com","threadId":"40174","inReplyTo":"1440523706-23041-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-25T17:28:25Z","receivedAt":"2015-08-25T17:28:25Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"By treating each object as its own task the workflow is easier to follow\nas the function used in the worker threads doesn't need any control logic\nany more.\n\nSigned-off-by: Stefan Beller <sbeller@google.com>\n---\n builtin/index-pack.c | 71 +++++++++++++++++++++++-----------------------------\n 1 file changed, 32 insertions(+), 39 deletions(-)\n\ndiff --git a/builtin/index-pack.c b/builtin/index-pack.c\nindex 3f10840..826bd22 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>] [--verify] [--strict] (<pack-file> | --stdin [--fix-thin] [<pack-file>])\";\n@@ -95,7 +96,6 @@ static const char *curr_pack;\n #ifndef NO_PTHREADS\n \n static struct thread_local *thread_data;\n-static int nr_dispatched;\n static int threads_active;\n \n static pthread_mutex_t read_mutex;\n@@ -106,10 +106,6 @@ static pthread_mutex_t counter_mutex;\n #define counter_lock()\t\tlock_mutex(&counter_mutex)\n #define counter_unlock()\tunlock_mutex(&counter_mutex)\n \n-static pthread_mutex_t work_mutex;\n-#define work_lock()\t\tlock_mutex(&work_mutex)\n-#define work_unlock()\t\tunlock_mutex(&work_mutex)\n-\n static pthread_mutex_t deepest_delta_mutex;\n #define deepest_delta_lock()\tlock_mutex(&deepest_delta_mutex)\n #define deepest_delta_unlock()\tunlock_mutex(&deepest_delta_mutex)\n@@ -140,7 +136,6 @@ static void init_thread(void)\n \tint i;\n \tinit_recursive_mutex(&read_mutex);\n \tpthread_mutex_init(&counter_mutex, NULL);\n-\tpthread_mutex_init(&work_mutex, NULL);\n \tpthread_mutex_init(&type_cas_mutex, NULL);\n \tif (show_stat)\n \t\tpthread_mutex_init(&deepest_delta_mutex, NULL);\n@@ -163,7 +158,6 @@ static void cleanup_thread(void)\n \tthreads_active = 0;\n \tpthread_mutex_destroy(&read_mutex);\n \tpthread_mutex_destroy(&counter_mutex);\n-\tpthread_mutex_destroy(&work_mutex);\n \tpthread_mutex_destroy(&type_cas_mutex);\n \tif (show_stat)\n \t\tpthread_mutex_destroy(&deepest_delta_mutex);\n@@ -181,9 +175,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 +1066,29 @@ 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-\t\tint i;\n-\t\tcounter_lock();\n-\t\tdisplay_progress(progress, nr_resolved_deltas);\n-\t\tcounter_unlock();\n-\t\twork_lock();\n-\t\twhile (nr_dispatched < nr_objects &&\n-\t\t       is_delta_type(objects[nr_dispatched].type))\n-\t\t\tnr_dispatched++;\n-\t\tif (nr_dispatched >= nr_objects) {\n-\t\t\twork_unlock();\n-\t\t\tbreak;\n-\t\t}\n-\t\ti = nr_dispatched++;\n-\t\twork_unlock();\n+\tif (!get_thread_data()) {\n+\t\tstruct thread_local *t = xmalloc(sizeof(*t));\n+\t\tt->pack_fd = open(curr_pack, O_RDONLY);\n+\t\tif (t->pack_fd == -1)\n+\t\t\tdie_errno(_(\"unable to open %s\"), curr_pack);\n \n-\t\tresolve_base(&objects[i]);\n+\t\tset_thread_data(t);\n \t}\n-\treturn NULL;\n+\n+\tresolve_base(data);\n+\n+\tcounter_lock();\n+\tdisplay_progress(progress, nr_resolved_deltas);\n+\tcounter_unlock();\n+\treturn 0;\n+}\n+\n+void cleanup_threaded_second_pass(struct task_queue *aq)\n+{\n+\tstruct thread_local *t = get_thread_data();\n+\tfree(t);\n }\n #endif\n \n@@ -1195,18 +1187,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 \tif (nr_threads > 1 || getenv(\"GIT_FORCE_THREADS\")) {\n+\t\tstruct task_queue *tq;\n+\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-\t\tfor (i = 0; i < nr_threads; i++)\n-\t\t\tpthread_join(thread_data[i].thread, NULL);\n+\t\ttq = create_task_queue(nr_threads);\n+\n+\t\tfor (i = 0; i < nr_objects; i++)\n+\t\t\tif (!is_delta_type(objects[i].type))\n+\t\t\t\tadd_task(tq, threaded_second_pass, &objects[i]);\n+\n+\t\tif (finish_task_queue(tq, cleanup_threaded_second_pass))\n+\t\t\tdie(\"Not all threads have finished\");\n+\n \t\tcleanup_thread();\n \t\treturn;\n \t}\n-- \n2.5.0.400.gff86faf\n"},{"id":"268663","messageId":"1440523706-23041-6-git-send-email-sbeller@google.com","threadId":"40174","inReplyTo":"1440523706-23041-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-25T17:28:26Z","receivedAt":"2015-08-25T17:28:26Z","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. It also demonstrates the usage of the new threading\npool being useful.\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.400.gff86faf\n"},{"id":"268680","messageId":"20150825190329.GA11271@sigill.intra.peff.net","threadId":"40174","inReplyTo":"1440523706-23041-5-git-send-email-sbeller@google.com","subject":"Re: [PATCH 4/5] index-pack: Use the new worker pool","fromName":"Jeff King","fromEmail":"peff@peff.net","sentAt":"2015-08-25T19:03:30Z","receivedAt":"2015-08-25T19:03:30Z","isPatch":true,"sender":{"key":"peff@peff.net","avatar":"https://avatars.githubusercontent.com/u/45925?v=4"},"body":"On Tue, Aug 25, 2015 at 10:28:25AM -0700, Stefan Beller wrote:\n\n> By treating each object as its own task the workflow is easier to follow\n> as the function used in the worker threads doesn't need any control logic\n> any more.\n\nHave you tried running t/perf/p5302 on this?\n\nI seem to get a pretty consistent 2%-ish slowdown, both against git.git\nand linux.git. That's not a lot, but I'm wondering if there is some\nlow-hanging fruit in the locking, or in the pattern of work being\ndispatched. Or it may just be noise, but it seems fairly consistent.\n\n-Peff\n"},{"id":"268682","messageId":"CAGZ79kaAT1C1XscOmkN8UzeJ7cmwOhxiTu1HsoTwN+HR5H2A5Q@mail.gmail.com","threadId":"40174","inReplyTo":"20150825190329.GA11271@sigill.intra.peff.net","subject":"Re: [PATCH 4/5] index-pack: Use the new worker pool","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-25T19:23:31Z","receivedAt":"2015-08-25T19:23:31Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Tue, Aug 25, 2015 at 12:03 PM, Jeff King <peff@peff.net> wrote:\n> On Tue, Aug 25, 2015 at 10:28:25AM -0700, Stefan Beller wrote:\n>\n>> By treating each object as its own task the workflow is easier to follow\n>> as the function used in the worker threads doesn't need any control logic\n>> any more.\n>\n> Have you tried running t/perf/p5302 on this?\n>\n> I seem to get a pretty consistent 2%-ish slowdown, both against git.git\n> and linux.git. That's not a lot, but I'm wondering if there is some\n> low-hanging fruit in the locking, or in the pattern of work being\n> dispatched. Or it may just be noise, but it seems fairly consistent.\n\nI did not run any perf tests, just the standard tests.\n\nMaybe the progress display can be moved out to another thread,\nsuch that it doesn't block the threads doing actual work.\nAlso the progress display can be done a bit more sloppy,\nwe are allowed to drop old events, if we have newer information.\n[Though this reasoning is not looking at the threading pool code,\nbecause ..um.. my code is perfect]\n\n>\n> -Peff\n"},{"id":"268690","messageId":"xmqqfv37xfrc.fsf@gitster.dls.corp.google.com","threadId":"40174","inReplyTo":"20150825190329.GA11271@sigill.intra.peff.net","subject":"Re: [PATCH 4/5] index-pack: Use the new worker pool","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-08-25T20:41:27Z","receivedAt":"2015-08-25T20:41:27Z","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 Tue, Aug 25, 2015 at 10:28:25AM -0700, Stefan Beller wrote:\n>\n>> By treating each object as its own task the workflow is easier to follow\n>> as the function used in the worker threads doesn't need any control logic\n>> any more.\n>\n> Have you tried running t/perf/p5302 on this?\n>\n> I seem to get a pretty consistent 2%-ish slowdown, both against git.git\n> and linux.git. That's not a lot, but I'm wondering if there is some\n> low-hanging fruit in the locking, or in the pattern of work being\n> dispatched. Or it may just be noise, but it seems fairly consistent.\n\nThe pattern of work dispatch hopefully is the same, no?  add_task()\ndoes the \"append at the end\" thing and next_task() picks from the\nfront of the queue.  The original is \"we have globally N things,\nso far M things have been handled, and we want a new one, so we pick\nthe M+1th one and do it\".\n\nThe amount of memory that is used to represent a single task may be\nmuch larger than the original, with overhead coming from job_list\nstructure and the doubly-linked list.  We may not be able to spin up\n30 threads and throw a million tasks at them using this, because of\nthe overhead.  It would be more suited to handle a pattern in which\nan overlord actively creates new tasks while worker threads chew\nthem, using the add_task/dispatch as the medium for communication\nbetween them.\n"},{"id":"268692","messageId":"CAGZ79kZ9V35KmJSExt=EqYFcyOSysaFsJA0E+dx46Qg+Pm+Ybg@mail.gmail.com","threadId":"40174","inReplyTo":"xmqqfv37xfrc.fsf@gitster.dls.corp.google.com","subject":"Re: [PATCH 4/5] index-pack: Use the new worker pool","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-25T20:59:04Z","receivedAt":"2015-08-25T20:59:04Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Tue, Aug 25, 2015 at 1:41 PM, Junio C Hamano <gitster@pobox.com> wrote:\n> Jeff King <peff@peff.net> writes:\n>\n>> On Tue, Aug 25, 2015 at 10:28:25AM -0700, Stefan Beller wrote:\n>>\n>>> By treating each object as its own task the workflow is easier to follow\n>>> as the function used in the worker threads doesn't need any control logic\n>>> any more.\n>>\n>> Have you tried running t/perf/p5302 on this?\n>>\n>> I seem to get a pretty consistent 2%-ish slowdown, both against git.git\n>> and linux.git. That's not a lot, but I'm wondering if there is some\n>> low-hanging fruit in the locking, or in the pattern of work being\n>> dispatched. Or it may just be noise, but it seems fairly consistent.\n>\n> The pattern of work dispatch hopefully is the same, no?  add_task()\n> does the \"append at the end\" thing and next_task() picks from the\n> front of the queue.  The original is \"we have globally N things,\n> so far M things have been handled, and we want a new one, so we pick\n> the M+1th one and do it\".\n>\n> The amount of memory that is used to represent a single task may be\n> much larger than the original, with overhead coming from job_list\n> structure and the doubly-linked list.  We may not be able to spin up\n> 30 threads and throw a million tasks at them using this, because of\n> the overhead.\n\nI thought about making the add_task block after a certain size of job_list\nsuch that there is always just enough tasks available to process.\nThen we would not need to used a huge amount of memory for having\na long line of waiting tasks; the shorter the line is the less we can tolerate\nburst behavior in the threads.\n\n> It would be more suited to handle a pattern in which\n> an overlord actively creates new tasks while worker threads chew\n> them, using the add_task/dispatch as the medium for communication\n> between them.\n\nNot sure I follow there.\n\nOriginal implementation:\nWe have M threads sitting around the table, all of them trying to obtain food\nfrom the one bowl on the table and then eating it.\nOnce the bowl is all eaten, we can stop.\n\nNew pattern:\nOne cook puts all the food items on the sushi-go-round-belt with a fancy plate\nand the threads grab them one by one still using locks (but they are internal to\nthe belt).\n\nAre you saying we're content with just a bowl and everyone helps themselves\nfor getting food?\n"},{"id":"268693","messageId":"xmqq7fojxeh3.fsf@gitster.dls.corp.google.com","threadId":"40174","inReplyTo":"1440523706-23041-4-git-send-email-sbeller@google.com","subject":"Re: [PATCH 3/5] submodule: helper to run foreach in parallel","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-08-25T21:09:12Z","receivedAt":"2015-08-25T21:09:12Z","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 runs a command on each submodule in parallel and should eventually\n> replace `git submodule foreach`.\n>\n> There is a new option -j/--jobs (inspired by make) to specify the number\n> of parallel threads.\n>\n> The jobs=1 case needs to be special cases to exactly replicate the current\n> default behavior of `git submodule foreach` such as working stdin input.\n\n\"git submodule foreach-parallel -j1\" feels like a roundabout way to\nsay \"git submodule foreach\"; \"I want you to go parallel.  Oh, not\nreally, I want you to do one thing at a time\".\n\nI do not think these two have to be equivalent---users who need the\ntraditional one-by-one \"foreach\" behaviour (including the standard\ninput and other aspects that are unreasonable to expect\n\"foreach-parallel -jN\" to replicate) can and should choose to use\n\"foreach\", not \"foreach-parallel -j1\".\n\nIn any case, I do not think I saw any special casing of -j1 case that\nattempts to leave the standard input operational.  Did I miss\nsomething, or is the above describing what is left to do?\n\n> For more than one job there is no input possible and the output of both\n> stdout/stderr of the command are put into the stderr in an ordered fashion,\n> i.e. the tasks to not intermingle their output in races.\n\nTo rephrase what I said earlier, \"for parallel version, the above\nthings happen, even with numthreads==1\", is perfectly fine.\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\nSo the standard error and the standard output are mixed to a single\npipe ...\n\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\n... and the whole thing is accumulated in core???\n\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... and emitted to standard error?\n\nI would have expected that the standard error would be left alone\n(i.e. letting warnings from multiple jobs to be mixed together\nsimply because everybody writes to the same file descriptor), while\nthe standard output would be line-buffered, perhaps captured by the\nabove loop and then emitted under mutex, or something.\n\nI think I said this earlier, but latency to the first output counts\nas an activity feedback mechanism.\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, NULL);\n\nThis is very nice.  Declare a task queue with N workers, throw bunch\nof task to it and then wait for all of them to complete.  Things\ncan't be simpler than that ;-).  One thing that other callers of the\nmechanism might want may be to plug and unplug the task queue, but\nthat can wait until the need arises.\n"},{"id":"268694","messageId":"xmqq37z7xebe.fsf@gitster.dls.corp.google.com","threadId":"40174","inReplyTo":"CAGZ79kZ9V35KmJSExt=EqYFcyOSysaFsJA0E+dx46Qg+Pm+Ybg@mail.gmail.com","subject":"Re: [PATCH 4/5] index-pack: Use the new worker pool","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-08-25T21:12:37Z","receivedAt":"2015-08-25T21:12: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> Not sure I follow there.\n>\n> Original implementation:\n> We have M threads sitting around the table, all of them trying to obtain food\n> from the one bowl on the table and then eating it.\n> Once the bowl is all eaten, we can stop.\n>\n> New pattern:\n> One cook puts all the food items on the sushi-go-round-belt with a fancy plate\n> and the threads grab them one by one still using locks (but they are internal to\n> the belt).\n>\n> Are you saying we're content with just a bowl and everyone helps themselves\n> for getting food?\n\nNo.  I am questioning how big overhead is for having the\ngo-round-belt that must hold all dishes to be eaten, which did not\nexist in the original arrangement.\n"},{"id":"268695","messageId":"CAGZ79kb2N_5_tJv-GURL9_ESFs=pHp=L-Mujn3Df_+-T74_9Dg@mail.gmail.com","threadId":"40174","inReplyTo":"xmqq7fojxeh3.fsf@gitster.dls.corp.google.com","subject":"Re: [PATCH 3/5] submodule: helper to run foreach in parallel","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-25T21:42:25Z","receivedAt":"2015-08-25T21:42:25Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Tue, Aug 25, 2015 at 2:09 PM, Junio C Hamano <gitster@pobox.com> wrote:\n> Stefan Beller <sbeller@google.com> writes:\n>\n>> This runs a command on each submodule in parallel and should eventually\n>> replace `git submodule foreach`.\n>>\n>> There is a new option -j/--jobs (inspired by make) to specify the number\n>> of parallel threads.\n>>\n>> The jobs=1 case needs to be special cases to exactly replicate the current\n>> default behavior of `git submodule foreach` such as working stdin input.\n>\n> \"git submodule foreach-parallel -j1\" feels like a roundabout way to\n> say \"git submodule foreach\"; \"I want you to go parallel.  Oh, not\n> really, I want you to do one thing at a time\".\n\nEventually I want to drop the -parallel, such that git foreach ... will\njust work as it always had, and if there is a --jobs argument we\nmay want to change the semantics a bit to make it work with the\npiping to stdout/err.\n\n>\n> I do not think these two have to be equivalent---users who need the\n> traditional one-by-one \"foreach\" behaviour (including the standard\n> input and other aspects that are unreasonable to expect\n> \"foreach-parallel -jN\" to replicate) can and should choose to use\n> \"foreach\", not \"foreach-parallel -j1\".\n>\n> In any case, I do not think I saw any special casing of -j1 case that\n> attempts to leave the standard input operational.  Did I miss\n> something, or is the above describing what is left to do?\n\nIt is the later, I am not quite confident yet about this very patch,\nbut I was rather waiting for feedback what people deem is important.\n\n>\n>> For more than one job there is no input possible and the output of both\n>> stdout/stderr of the command are put into the stderr in an ordered fashion,\n>> i.e. the tasks to not intermingle their output in races.\n>\n> To rephrase what I said earlier, \"for parallel version, the above\n> things happen, even with numthreads==1\", is perfectly fine.\n\n>\n>> +     cp->no_stdin = 1;\n>> +     cp->out = 0;\n>> +     cp->err = -1;\n>> +     cp->dir = args->path;\n>> +     cp->stdout_to_stderr = 1;\n>\n> So the standard error and the standard output are mixed to a single\n> pipe ...\n\nI was very focused on fetch, which would report progress and\ninformation to stderr only.\n\n>\n>> +     cp->use_shell = 1;\n>> +\n>> +     if (start_command(cp)) {\n>> +             die(\"Could not start command\");\n>> +             for (i = 0; cp->args.argv; i++)\n>> +                     fprintf(stderr, \"%s\\n\", cp->args.argv[i]);\n>> +     }\n>> +\n>> +     while (1) {\n>> +             ssize_t len = xread(cp->err, buf, sizeof(buf));\n>> +             if (len < 0)\n>> +                     die(\"Read from child failed\");\n>> +             else if (len == 0)\n>> +                     break;\n>> +             else {\n>> +                     strbuf_add(&out, buf, len);\n>> +             }\n>\n> ... and the whole thing is accumulated in core???\n\nThe pipes have a limit, so we need to empty them to prevent back-pressure?\nAnd because we want to have the output of one task at a time, we need to\nsave it up until we can put out the whole output, no?\n\n>\n>> +     }\n>> +     if (finish_command(cp))\n>> +             die(\"command died with error\");\n>> +\n>> +     sem_wait(args->mutex);\n>> +     fputs(out.buf, stderr);\n>> +     sem_post(args->mutex);\n>\n> ... and emitted to standard error?\n>\n> I would have expected that the standard error would be left alone\n\n`git fetch` which may be a good candidate for such an operation\nprovides progress on stderr, and we don't want to intermingle\n2 different submodule fetch progress displays\n(\"I need to work offline for a bit, so let me get all of the latest stuff,\nso I'll run `git submodule foreach -j 16 -- git fetch --all\" though ideally\nwe want to have `git fetch --recurse-submodules -j16` instead )\n\n> (i.e. letting warnings from multiple jobs to be mixed together\n> simply because everybody writes to the same file descriptor), while\n> the standard output would be line-buffered, perhaps captured by the\n> above loop and then emitted under mutex, or something.\n\n>\n> I think I said this earlier, but latency to the first output counts\n\n\"to the first stderr\"\nin this case?\n\nSo you would want one channel (stderr) for a fast reporting possibility\nand another channel (stdout) for a well ordered output mode to cover\nboth options. And the command to be run must adhere to the our\nselection of stderr for fast reporting and stdout for output to wait on.\n\n> as an activity feedback mechanism.\n>\n>> +     if (module_list_compute(0, nullargv, NULL, &pathspec) < 0)\n>> +             return 1;\n>> +\n>> +     gitmodules_config();\n>> +\n>> +     aq = create_task_queue(number_threads);\n>> +\n>> +     for (i = 0; i < ce_used; i++) {\n>> +             const struct submodule *sub;\n>> +             const struct cache_entry *ce = ce_entries[i];\n>> +             struct submodule_args *args = malloc(sizeof(*args));\n>> +\n>> +             if (ce_stage(ce))\n>> +                     args->sha1 = xstrdup(sha1_to_hex(null_sha1));\n>> +             else\n>> +                     args->sha1 = xstrdup(sha1_to_hex(ce->sha1));\n>> +\n>> +             strbuf_reset(&sb);\n>> +             strbuf_addf(&sb, \"%s/.git\", ce->name);\n>> +             if (!file_exists(sb.buf))\n>> +                     continue;\n>> +\n>> +             args->path = ce->name;\n>> +             sub = submodule_from_path(null_sha1, args->path);\n>> +             if (!sub)\n>> +                     die(\"No submodule mapping found in .gitmodules for path '%s'\", args->path);\n>> +\n>> +             args->name = sub->name;\n>> +             args->toplevel = xstrdup(xgetcwd());\n>> +             args->cmd = argv;\n>> +             args->mutex = mutex;\n>> +             args->prefix = alternative_path;\n>> +             add_task(aq, run_cmd_submodule, args);\n>> +     }\n>> +\n>> +     finish_task_queue(aq, NULL);\n>\n> This is very nice.  Declare a task queue with N workers, throw bunch\n> of task to it and then wait for all of them to complete.  Things\n> can't be simpler than that ;-).  One thing that other callers of the\n> mechanism might want may be to plug and unplug the task queue, but\n> that can wait until the need arises.\n\nI think a queue is the simplest thing at the moment. Eventually we\nmay want to have a way to specify a DAG of workers (e.g. 2 tasks reading from\nfiles to a buffer, 4 tasks to preprocess the output from the first 2 tasks, and\nthen 16 tasks to do heavy workload processing, and just one thread doing the\nstdout/err handling).\n\nIn the current patches the task queue is passed around and every worker thread\nhas access to it, in case it wants to start new tasks itself, so it\nmay be doable.\n\nWhy would we want to unplug the task queue from somewhere else?\n"},{"id":"268699","messageId":"xmqqy4gzvwh5.fsf@gitster.dls.corp.google.com","threadId":"40174","inReplyTo":"CAGZ79kb2N_5_tJv-GURL9_ESFs=pHp=L-Mujn3Df_+-T74_9Dg@mail.gmail.com","subject":"Re: [PATCH 3/5] submodule: helper to run foreach in parallel","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-08-25T22:23:18Z","receivedAt":"2015-08-25T22:23:18Z","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>>> +     while (1) {\n>>> +             ssize_t len = xread(cp->err, buf, sizeof(buf));\n>>> +             if (len < 0)\n>>> +                     die(\"Read from child failed\");\n>>> +             else if (len == 0)\n>>> +                     break;\n>>> +             else {\n>>> +                     strbuf_add(&out, buf, len);\n>>> +             }\n>>\n>> ... and the whole thing is accumulated in core???\n>\n> The pipes have a limit, so we need to empty them to prevent back-pressure?\n\nOf course.  But that does not lead to \"we hold everything in core\".\nThis side could choose to emit (under protection of args->mutex)\nearly, e.g. after reading a line, emit it to our standard output (or\nour standard error).\n\n> And because we want to have the output of one task at a time, we need to\n> save it up until we can put out the whole output, no?\n\nI do not necessarily agree, and I think I said that already:\n\n  http://thread.gmane.org/gmane.comp.version-control.git/276273/focus=276321\n\n>>> +     }\n>>> +     if (finish_command(cp))\n>>> +             die(\"command died with error\");\n>>> +\n>>> +     sem_wait(args->mutex);\n>>> +     fputs(out.buf, stderr);\n>>> +     sem_post(args->mutex);\n>>\n>> ... and emitted to standard error?\n>>\n>> I would have expected that the standard error would be left alone\n>\n> `git fetch` which may be a good candidate for such an operation\n> provides progress on stderr, and we don't want to intermingle\n> 2 different submodule fetch progress displays\n> (\"I need to work offline for a bit, so let me get all of the latest stuff,\n> so I'll run `git submodule foreach -j 16 -- git fetch --all\" though ideally\n> we want to have `git fetch --recurse-submodules -j16` instead )\n>\n>> (i.e. letting warnings from multiple jobs to be mixed together\n>> simply because everybody writes to the same file descriptor), while\n>> the standard output would be line-buffered, perhaps captured by the\n>> above loop and then emitted under mutex, or something.\n>\n>>\n>> I think I said this earlier, but latency to the first output counts\n>\n> \"to the first stderr\"\n> in this case?\n\nI didn't mean \"output==the standard output stream\".  As I said in\n$gmane/276321, an early output, as an indication that we are doing\nsomething, is important.\n\n> Why would we want to unplug the task queue from somewhere else?\n\nWhen you have a dispatcher more intelligent than a stupid FIFO, I\nwould imagine that you would want to be able to do this pattern,\nespecially when coming up with a task (not performing a task) takes\nnon-trivial amount of work:\n\n\tprepare task queue and have N threads waiting on it;\n\n\tplug the queue, i.e. tell threads that do not start picking\n\ttasks out of it yet;\n\n\tlarge enough loop to fill the queue to a reasonable size\n\twhile keeping the threads waiting;\n\n\tunplug the queue.  Now the threads can pick tasks from the\n\tqueue, but they have many to choose from, and a dispatcher\n\tcan do better than simple FIFO can take advantage of it;\n\n\tkeep filling the queue with more tasks, if necessary;\n\n        and finally, wait for everything to finish.\n\nWithout \"plug/unplug\" interface, you _could_ do the above by doing\nsomething stupid like\n\n\tprepare a task queue and have N threads waiting on it;\n\n\tloop to find enough number of tasks but do not put them to\n\ttask queue, as FIFO will eat them one-by-one; instead hold\n\tonto them in a custom data structure that is outside the\n\ttask queue system;\n\n\ttight and quick loop to move them to the task queue;\n\n\tkeep finding more tasks and feed them to the task queue;\n\n        and finally, wait for everything to finish.\n"},{"id":"268700","messageId":"CAGZ79kaMeaTiQgVxjf5aPAtLVw224Hf9gzDNS85+9C+rj_keWA@mail.gmail.com","threadId":"40174","inReplyTo":"xmqq37z7xebe.fsf@gitster.dls.corp.google.com","subject":"Re: [PATCH 4/5] index-pack: Use the new worker pool","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-25T22:39:02Z","receivedAt":"2015-08-25T22:39:02Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Tue, Aug 25, 2015 at 2:12 PM, Junio C Hamano <gitster@pobox.com> wrote:\n> Stefan Beller <sbeller@google.com> writes:\n>\n>> Not sure I follow there.\n>>\n>> Original implementation:\n>> We have M threads sitting around the table, all of them trying to obtain food\n>> from the one bowl on the table and then eating it.\n>> Once the bowl is all eaten, we can stop.\n>>\n>> New pattern:\n>> One cook puts all the food items on the sushi-go-round-belt with a fancy plate\n>> and the threads grab them one by one still using locks (but they are internal to\n>> the belt).\n>>\n>> Are you saying we're content with just a bowl and everyone helps themselves\n>> for getting food?\n>\n> No.  I am questioning how big overhead is for having the\n> go-round-belt that must hold all dishes to be eaten, which did not\n> exist in the original arrangement.\n>\n>\n\nThen please don't pick up this patch. This and patch 5 are there to convince\nJeff this is a good API, worth being introduced and not over engineered, just\nsolving a problem we're interested in with a minimal amount of code to side\ntrack from the actual goal we want to pursue.\n"},{"id":"268701","messageId":"xmqqtwrnvvie.fsf@gitster.dls.corp.google.com","threadId":"40174","inReplyTo":"xmqqy4gzvwh5.fsf@gitster.dls.corp.google.com","subject":"Re: [PATCH 3/5] submodule: helper to run foreach in parallel","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-08-25T22:44:09Z","receivedAt":"2015-08-25T22:44:09Z","isPatch":true,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"Junio C Hamano <gitster@pobox.com> writes:\n\n>> Why would we want to unplug the task queue from somewhere else?\n>\n> When you have a dispatcher more intelligent than a stupid FIFO, I\n> would imagine that you would want to be able to do this pattern,\n> especially when coming up with a task (not performing a task) takes\n> non-trivial amount of work:\n>\n> \tprepare task queue and have N threads waiting on it;\n>\n> \tplug the queue, i.e. tell threads that do not start picking\n> \ttasks out of it yet;\n\ns/that do not/not to/; sorry for grammo resulting from a lot of\nediting while doing other things X-<.\n\n> ...\n>       and finally, wait for everything to finish.\n>\n> Without \"plug/unplug\" interface, you _could_ do the above by doing\n> something stupid like\n> ...\n>       and finally, wait for everything to finish.\n\n... but having to use yet another queue outside only because the\ntask queue cannot be plugged feels stupid.\n\nBut because we do not yet have a way to plug custom dispatching\nlogic to the dispatcher, the above is in the \"but that can wait\nuntil the need arises.\" category, as I said.\n"},{"id":"268702","messageId":"xmqqpp2bvv87.fsf@gitster.dls.corp.google.com","threadId":"40174","inReplyTo":"CAGZ79kaMeaTiQgVxjf5aPAtLVw224Hf9gzDNS85+9C+rj_keWA@mail.gmail.com","subject":"Re: [PATCH 4/5] index-pack: Use the new worker pool","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2015-08-25T22:50:16Z","receivedAt":"2015-08-25T22:50:16Z","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> Then please don't pick up this patch. This and patch 5 are there to convince\n> Jeff this is a good API, worth being introduced and not over engineered, just\n> solving a problem we're interested in with a minimal amount of code to side\n> track from the actual goal we want to pursue.\n\nDon't worry.  I did not have intention to queue index-pack and\npack-objects rewrite unless the use of the new API makes them\ndemonstratably better.\n\nThe criteria for \"worth being introduced\" would include \"not\nexcessively heavyweight\", I would think.  And it was good that we\nhave these two patches to judge if the earlier one (2-3/5) is a good\nthing to add in a concrete way.  Perhaps the 2% slowdown might be\nshowing the performance characteristics of your worker pool model,\nand Peff's suggestion to at least see (if not solve) where the\noverhead is coming from was a very reasonable one, I would think.\n"},{"id":"268737","messageId":"20150826170647.GA1870@sigill.intra.peff.net","threadId":"40174","inReplyTo":"1440523706-23041-4-git-send-email-sbeller@google.com","subject":"Re: [PATCH 3/5] submodule: helper to run foreach in parallel","fromName":"Jeff King","fromEmail":"peff@peff.net","sentAt":"2015-08-26T17:06:48Z","receivedAt":"2015-08-26T17:06:48Z","isPatch":true,"sender":{"key":"peff@peff.net","avatar":"https://avatars.githubusercontent.com/u/45925?v=4"},"body":"On Tue, Aug 25, 2015 at 10:28:24AM -0700, Stefan Beller wrote:\n\n> +int module_foreach_parallel(int argc, const char **argv, const char *prefix)\n> +{\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\"args\" and \"args->sha1\" go out of scope and leak here.\n\n> +\t\targs->name = sub->name;\n> +\t\targs->toplevel = xstrdup(xgetcwd());\n\nAnother xgetcwd leak. :) I think this one can just drop the xstrdup.\n\n(Both of these were spotted by Coverity. I know you have played with it\na little, so if you are actually reading the emails it sends, I'll stop\nrelaying them).\n\n-Peff\n"},{"id":"268738","messageId":"CAGZ79kaWh5F_d9TGHFe-fAd5arFicaAVcmYnWnAS2Wn1PoGMzg@mail.gmail.com","threadId":"40174","inReplyTo":"20150826170647.GA1870@sigill.intra.peff.net","subject":"Re: [PATCH 3/5] submodule: helper to run foreach in parallel","fromName":"Stefan Beller","fromEmail":"sbeller@google.com","sentAt":"2015-08-26T17:21:08Z","receivedAt":"2015-08-26T17:21:08Z","isPatch":true,"sender":{"key":"stefanbeller@gmail.com","avatar":"https://avatars.githubusercontent.com/u/455868?v=4"},"body":"On Wed, Aug 26, 2015 at 10:06 AM, Jeff King <peff@peff.net> wrote:\n> On Tue, Aug 25, 2015 at 10:28:24AM -0700, Stefan Beller wrote:\n>\n>> +int module_foreach_parallel(int argc, const char **argv, const char *prefix)\n>> +{\n>> [...]\n>> +     for (i = 0; i < ce_used; i++) {\n>> +             const struct submodule *sub;\n>> +             const struct cache_entry *ce = ce_entries[i];\n>> +             struct submodule_args *args = malloc(sizeof(*args));\n>> +\n>> +             if (ce_stage(ce))\n>> +                     args->sha1 = xstrdup(sha1_to_hex(null_sha1));\n>> +             else\n>> +                     args->sha1 = xstrdup(sha1_to_hex(ce->sha1));\n>> +\n>> +             strbuf_reset(&sb);\n>> +             strbuf_addf(&sb, \"%s/.git\", ce->name);\n>> +             if (!file_exists(sb.buf))\n>> +                     continue;\n>\n> \"args\" and \"args->sha1\" go out of scope and leak here.\n>\n>> +             args->name = sub->name;\n>> +             args->toplevel = xstrdup(xgetcwd());\n>\n> Another xgetcwd leak. :) I think this one can just drop the xstrdup.\n>\n> (Both of these were spotted by Coverity. I know you have played with it\n> a little, so if you are actually reading the emails it sends, I'll stop\n> relaying them).\n\nFixing those memleaks was the first thing I did this morning. :)\n\nAnd yeah I do pay attention to these emails. (I maintain the automatic testing\nevery other day, so there is that.)\n\n>\n> -Peff\n"}]}