{"thread":{"id":"29767","subject":"[PATCH 1/2] index-pack: split second pass obj handling into own function","startedAt":"2012-02-28T04:36:52Z","lastAt":"2012-03-02T18:53:00Z","messageCount":5,"participants":["Nguyễn Thái Ngọc Duy","Junio C Hamano"],"isPatch":true,"patchVersion":1,"patchTotal":2},"messages":[{"id":"185589","messageId":"1330403813-2770-1-git-send-email-pclouds@gmail.com","threadId":"29767","inReplyTo":null,"subject":"[PATCH 1/2] index-pack: split second pass obj handling into own function","fromName":"Nguyễn Thái Ngọc Duy","fromEmail":"pclouds@gmail.com","sentAt":"2012-02-28T04:36:52Z","receivedAt":"2012-02-28T04:36:52Z","isPatch":true,"sender":{"key":"pclouds@gmail.com","avatar":"https://avatars.githubusercontent.com/u/720?v=4"},"body":"\nSigned-off-by: Nguyễn Thái Ngọc Duy <pclouds@gmail.com>\n---\n builtin/index-pack.c |   31 ++++++++++++++++++-------------\n 1 files changed, 18 insertions(+), 13 deletions(-)\n\ndiff --git a/builtin/index-pack.c b/builtin/index-pack.c\nindex dd1c5c9..918684f 100644\n--- a/builtin/index-pack.c\n+++ b/builtin/index-pack.c\n@@ -682,6 +682,23 @@ static int compare_delta_entry(const void *a, const void *b)\n \t\t\t\t   objects[delta_b->obj_no].type);\n }\n \n+/*\n+ * Second pass:\n+ * - for all non-delta objects, look if it is used as a base for\n+ *   deltas;\n+ * - if used as a base, uncompress the object and apply all deltas,\n+ *   recursively checking if the resulting object is used as a base\n+ *   for some more deltas.\n+ */\n+static void second_pass(struct object_entry *obj)\n+{\n+\tstruct base_data *base_obj = alloc_base_data();\n+\tbase_obj->obj = obj;\n+\tbase_obj->data = NULL;\n+\tfind_unresolved_deltas(base_obj);\n+\tdisplay_progress(progress, nr_resolved_deltas);\n+}\n+\n /* Parse all objects and return the pack content SHA1 hash */\n static void parse_pack_objects(unsigned char *sha1)\n {\n@@ -736,26 +753,14 @@ static void parse_pack_objects(unsigned char *sha1)\n \tqsort(deltas, nr_deltas, sizeof(struct delta_entry),\n \t      compare_delta_entry);\n \n-\t/*\n-\t * Second pass:\n-\t * - for all non-delta objects, look if it is used as a base for\n-\t *   deltas;\n-\t * - if used as a base, uncompress the object and apply all deltas,\n-\t *   recursively checking if the resulting object is used as a base\n-\t *   for some more deltas.\n-\t */\n \tif (verbose)\n \t\tprogress = start_progress(\"Resolving deltas\", nr_deltas);\n \tfor (i = 0; i < nr_objects; i++) {\n \t\tstruct object_entry *obj = &objects[i];\n-\t\tstruct base_data *base_obj = alloc_base_data();\n \n \t\tif (is_delta_type(obj->type))\n \t\t\tcontinue;\n-\t\tbase_obj->obj = obj;\n-\t\tbase_obj->data = NULL;\n-\t\tfind_unresolved_deltas(base_obj);\n-\t\tdisplay_progress(progress, nr_resolved_deltas);\n+\t\tsecond_pass(obj);\n \t}\n }\n \n-- \n1.7.3.1.256.g2539c.dirty\n"},{"id":"185590","messageId":"1330403813-2770-2-git-send-email-pclouds@gmail.com","threadId":"29767","inReplyTo":"1330403813-2770-1-git-send-email-pclouds@gmail.com","subject":"[PATCH 2/2] index-pack: support multithreaded delta resolving","fromName":"Nguyễn Thái Ngọc Duy","fromEmail":"pclouds@gmail.com","sentAt":"2012-02-28T04:36:53Z","receivedAt":"2012-02-28T04:36:53Z","isPatch":true,"sender":{"key":"pclouds@gmail.com","avatar":"https://avatars.githubusercontent.com/u/720?v=4"},"body":"This puts delta resolving on each base on a separate thread, one base\ncache per thread.\n\nAn experiment on a 24 core machine with git.git shows that performance\ndoes not increase proportional to the number of cores. So by default,\nwe use maximum 3 cores.\n\n$ /usr/bin/time ~/t/git index-pack --threads=1 -v --stdin < XXX.pack\nReceiving objects: 100% (146564/146564), 53.99 MiB | 17.47 MiB/s, done.\nResolving deltas: 100% (109205/109205), done.\npack    d5471e8365717a5812cbc81ec7277cb697a80f08\n11.58user 0.37system 0:12.04elapsed 99%CPU (0avgtext+0avgdata 375088maxresident)k\n0inputs+118592outputs (0major+56894minor)pagefaults 0swaps\n\n$ ... --threads=2 ...\n14.58user 0.47system 0:09.99elapsed 150%CPU (0avgtext+0avgdata 411536maxresident)k\n0inputs+118592outputs (0major+79961minor)pagefaults 0swaps\n\n$ ... --threads=3 ...\n14.36user 0.64system 0:08.12elapsed 184%CPU (0avgtext+0avgdata 393312maxresident)k\n0inputs+118592outputs (0major+50998minor)pagefaults 0swaps\n\n$ ... --threads=4 ...\n15.81user 0.71system 0:08.17elapsed 202%CPU (0avgtext+0avgdata 419152maxresident)k\n0inputs+118592outputs (0major+54907minor)pagefaults 0swaps\n\n$ ... --threads=5 ...\n14.76user 0.72system 0:07.06elapsed 219%CPU (0avgtext+0avgdata 414112maxresident)k\n0inputs+118592outputs (0major+59547minor)pagefaults 0swaps\n\n$ ... --threads=8 ...\n15.98user 0.81system 0:07.71elapsed 217%CPU (0avgtext+0avgdata 429904maxresident)k\n0inputs+118592outputs (0major+66221minor)pagefaults 0swaps\n\n$ ... --threads=12 ...\n15.81user 0.74system 0:09.60elapsed 172%CPU (0avgtext+0avgdata 442336maxresident)k\n0inputs+118592outputs (0major+61353minor)pagefaults 0swaps\n\n$ ... --threads=16 ...\n15.41user 0.57system 0:11.62elapsed 137%CPU (0avgtext+0avgdata 451728maxresident)k\n0inputs+118592outputs (0major+63569minor)pagefaults 0swaps\n\n$ ... --threads=24 ...\n15.84user 0.63system 0:12.83elapsed 128%CPU (0avgtext+0avgdata 475728maxresident)k\n0inputs+118592outputs (0major+58013minor)pagefaults 0swaps\n\nSigned-off-by: Nguyễn Thái Ngọc Duy <pclouds@gmail.com>\n---\n On linux-2.6.git, 2 cores, real time from 3m18 went down to around 2m.\n\n Documentation/git-index-pack.txt |   10 ++\n Makefile                         |    2 +-\n builtin/index-pack.c             |  188 ++++++++++++++++++++++++++++++++++----\n 3 files changed, 180 insertions(+), 20 deletions(-)\n\ndiff --git a/Documentation/git-index-pack.txt b/Documentation/git-index-pack.txt\nindex 909687f..39e6d0d 100644\n--- a/Documentation/git-index-pack.txt\n+++ b/Documentation/git-index-pack.txt\n@@ -74,6 +74,16 @@ OPTIONS\n --strict::\n \tDie, if the pack contains broken objects or links.\n \n+--threads=<n>::\n+\tSpecifies the number of threads to spawn when resolving\n+\tdeltas. This requires that index-pack be compiled with\n+\tpthreads otherwise this option is ignored with a warning.\n+\tThis is meant to reduce packing time on multiprocessor\n+\tmachines. The required amount of memory for the delta search\n+\twindow is however multiplied by the number of threads.\n+\tSpecifying 0 will cause git to auto-detect the number of CPU's\n+\tand use maximum 3 threads.\n+\n \n Note\n ----\ndiff --git a/Makefile b/Makefile\nindex 1fb1705..5fae875 100644\n--- a/Makefile\n+++ b/Makefile\n@@ -2159,7 +2159,7 @@ builtin/branch.o builtin/checkout.o builtin/clone.o builtin/reset.o branch.o tra\n builtin/bundle.o bundle.o transport.o: bundle.h\n builtin/bisect--helper.o builtin/rev-list.o bisect.o: bisect.h\n builtin/clone.o builtin/fetch-pack.o transport.o: fetch-pack.h\n-builtin/grep.o builtin/pack-objects.o transport-helper.o thread-utils.o: thread-utils.h\n+builtin/index-pack.o builtin/grep.o builtin/pack-objects.o transport-helper.o thread-utils.o: thread-utils.h\n builtin/send-pack.o transport.o: send-pack.h\n builtin/log.o builtin/shortlog.o: shortlog.h\n builtin/prune.o builtin/reflog.o reachable.o: reachable.h\ndiff --git a/builtin/index-pack.c b/builtin/index-pack.c\nindex 918684f..e331f23 100644\n--- a/builtin/index-pack.c\n+++ b/builtin/index-pack.c\n@@ -9,6 +9,7 @@\n #include \"progress.h\"\n #include \"fsck.h\"\n #include \"exec_cmd.h\"\n+#include \"thread-utils.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@@ -38,6 +39,15 @@ struct base_data {\n \tint ofs_first, ofs_last;\n };\n \n+struct thread_local {\n+#ifndef NO_PTHREADS\n+\tpthread_t thread;\n+#endif\n+\tstruct base_data *base_cache;\n+\tsize_t base_cache_used;\n+\tint nr_resolved_deltas;\n+};\n+\n /*\n  * Even if sizeof(union delta_base) == 24 on 64-bit archs, we really want\n  * to memcmp() only the first 20 bytes.\n@@ -54,11 +64,12 @@ struct delta_entry {\n \n static struct object_entry *objects;\n static struct delta_entry *deltas;\n-static struct base_data *base_cache;\n-static size_t base_cache_used;\n+static struct thread_local *thread_data;\n static int nr_objects;\n+static int nr_processed;\n static int nr_deltas;\n static int nr_resolved_deltas;\n+static int nr_threads;\n \n static int from_stdin;\n static int strict;\n@@ -75,6 +86,42 @@ static git_SHA_CTX input_ctx;\n static uint32_t input_crc32;\n static int input_fd, output_fd, pack_fd;\n \n+#ifndef NO_PTHREADS\n+\n+static pthread_mutex_t read_mutex;\n+#define read_lock()\t\tpthread_mutex_lock(&read_mutex)\n+#define read_unlock()\t\tpthread_mutex_unlock(&read_mutex)\n+\n+static pthread_mutex_t work_mutex;\n+#define work_lock()\t\tpthread_mutex_lock(&work_mutex)\n+#define work_unlock()\t\tpthread_mutex_unlock(&work_mutex)\n+\n+/*\n+ * Mutex and conditional variable can't be statically-initialized on Windows.\n+ */\n+static void init_thread(void)\n+{\n+\tinit_recursive_mutex(&read_mutex);\n+\tpthread_mutex_init(&work_mutex, NULL);\n+}\n+\n+static void cleanup_thread(void)\n+{\n+\tpthread_mutex_destroy(&read_mutex);\n+\tpthread_mutex_destroy(&work_mutex);\n+}\n+\n+#else\n+\n+#define read_lock()\n+#define read_unlock()\n+\n+#define work_lock()\n+#define work_unlock()\n+\n+#endif\n+\n+\n static int mark_link(struct object *obj, int type, void *data)\n {\n \tif (!obj)\n@@ -223,6 +270,18 @@ static NORETURN void bad_object(unsigned long offset, const char *format, ...)\n \tdie(\"pack has bad object at offset %lu: %s\", offset, buf);\n }\n \n+static struct thread_local *get_thread_data()\n+{\n+#ifndef NO_PTHREADS\n+\tint i;\n+\tpthread_t self = pthread_self();\n+\tfor (i = 1; i < nr_threads; i++)\n+\t\tif (self == thread_data[i].thread)\n+\t\t\treturn &thread_data[i];\n+#endif\n+\treturn &thread_data[0];\n+}\n+\n static struct base_data *alloc_base_data(void)\n {\n \tstruct base_data *base = xmalloc(sizeof(struct base_data));\n@@ -237,15 +296,16 @@ static void free_base_data(struct base_data *c)\n \tif (c->data) {\n \t\tfree(c->data);\n \t\tc->data = NULL;\n-\t\tbase_cache_used -= c->size;\n+\t\tget_thread_data()->base_cache_used -= c->size;\n \t}\n }\n \n static void prune_base_data(struct base_data *retain)\n {\n \tstruct base_data *b;\n-\tfor (b = base_cache;\n-\t     base_cache_used > delta_base_cache_limit && b;\n+\tstruct thread_local *data = get_thread_data();\n+\tfor (b = data->base_cache;\n+\t     data->base_cache_used > delta_base_cache_limit && b;\n \t     b = b->child) {\n \t\tif (b->data && b != retain)\n \t\t\tfree_base_data(b);\n@@ -257,22 +317,23 @@ static void link_base_data(struct base_data *base, struct base_data *c)\n \tif (base)\n \t\tbase->child = c;\n \telse\n-\t\tbase_cache = c;\n+\t\tget_thread_data()->base_cache = c;\n \n \tc->base = base;\n \tc->child = NULL;\n \tif (c->data)\n-\t\tbase_cache_used += c->size;\n+\t\tget_thread_data()->base_cache_used += c->size;\n \tprune_base_data(c);\n }\n \n static void unlink_base_data(struct base_data *c)\n {\n-\tstruct base_data *base = c->base;\n+\tstruct base_data *base;\n+\tbase = c->base;\n \tif (base)\n \t\tbase->child = NULL;\n \telse\n-\t\tbase_cache = NULL;\n+\t\tget_thread_data()->base_cache = NULL;\n \tfree_base_data(c);\n }\n \n@@ -461,19 +522,24 @@ static void sha1_object(const void *data, unsigned long size,\n \t\t\tenum object_type type, unsigned char *sha1)\n {\n \thash_sha1_file(data, size, typename(type), sha1);\n+\tread_lock();\n \tif (has_sha1_file(sha1)) {\n \t\tvoid *has_data;\n \t\tenum object_type has_type;\n \t\tunsigned long has_size;\n \t\thas_data = read_sha1_file(sha1, &has_type, &has_size);\n+\t\tread_unlock();\n \t\tif (!has_data)\n \t\t\tdie(\"cannot read existing object %s\", sha1_to_hex(sha1));\n \t\tif (size != has_size || type != has_type ||\n \t\t    memcmp(data, has_data, size) != 0)\n \t\t\tdie(\"SHA1 COLLISION FOUND WITH %s !\", sha1_to_hex(sha1));\n \t\tfree(has_data);\n-\t}\n+\t} else\n+\t\tread_unlock();\n+\n \tif (strict) {\n+\t\tread_lock();\n \t\tif (type == OBJ_BLOB) {\n \t\t\tstruct blob *blob = lookup_blob(sha1);\n \t\t\tif (blob)\n@@ -507,6 +573,7 @@ static void sha1_object(const void *data, unsigned long size,\n \t\t\t}\n \t\t\tobj->flags |= FLAG_CHECKED;\n \t\t}\n+\t\tread_unlock();\n \t}\n }\n \n@@ -552,7 +619,7 @@ static void *get_base_data(struct base_data *c)\n \t\tif (!delta_nr) {\n \t\t\tc->data = get_data_from_pack(obj);\n \t\t\tc->size = obj->size;\n-\t\t\tbase_cache_used += c->size;\n+\t\t\tget_thread_data()->base_cache_used += c->size;\n \t\t\tprune_base_data(c);\n \t\t}\n \t\tfor (; delta_nr > 0; delta_nr--) {\n@@ -568,7 +635,7 @@ static void *get_base_data(struct base_data *c)\n \t\t\tfree(raw);\n \t\t\tif (!c->data)\n \t\t\t\tbad_object(obj->idx.offset, \"failed to apply delta\");\n-\t\t\tbase_cache_used += c->size;\n+\t\t\tget_thread_data()->base_cache_used += c->size;\n \t\t\tprune_base_data(c);\n \t\t}\n \t\tfree(delta);\n@@ -596,7 +663,7 @@ static void resolve_delta(struct object_entry *delta_obj,\n \t\tbad_object(delta_obj->idx.offset, \"failed to apply delta\");\n \tsha1_object(result->data, result->size, delta_obj->real_type,\n \t\t    delta_obj->idx.sha1);\n-\tnr_resolved_deltas++;\n+\tget_thread_data()->nr_resolved_deltas++;\n }\n \n static struct base_data *find_unresolved_deltas_1(struct base_data *base,\n@@ -696,7 +763,35 @@ static void second_pass(struct object_entry *obj)\n \tbase_obj->obj = obj;\n \tbase_obj->data = NULL;\n \tfind_unresolved_deltas(base_obj);\n-\tdisplay_progress(progress, nr_resolved_deltas);\n+}\n+\n+static void *threaded_second_pass(void *arg)\n+{\n+\tstruct thread_local *data = get_thread_data();\n+\tfor (;;) {\n+\t\tint i, nr = 16;\n+\t\twork_lock();\n+\t\tnr_resolved_deltas += data->nr_resolved_deltas;\n+\t\tdisplay_progress(progress, nr_resolved_deltas);\n+\t\tdata->nr_resolved_deltas = 0;\n+\t\twhile (nr_processed < nr_objects &&\n+\t\t       is_delta_type(objects[nr_processed].type))\n+\t\t\tnr_processed++;\n+\t\tif (nr_processed >= nr_objects) {\n+\t\t\twork_unlock();\n+\t\t\tbreak;\n+\t\t}\n+\t\ti = nr_processed;\n+\t\tnr_processed += nr;\n+\t\twork_unlock();\n+\n+\t\tfor (; nr && i < nr_objects; i++, nr--) {\n+\t\t\tif (is_delta_type(objects[i].type))\n+\t\t\t\tcontinue;\n+\t\t\tsecond_pass(&objects[i]);\n+\t\t}\n+\t}\n+\treturn NULL;\n }\n \n /* Parse all objects and return the pack content SHA1 hash */\n@@ -755,13 +850,30 @@ static void parse_pack_objects(unsigned char *sha1)\n \n \tif (verbose)\n \t\tprogress = start_progress(\"Resolving deltas\", nr_deltas);\n-\tfor (i = 0; i < nr_objects; i++) {\n-\t\tstruct object_entry *obj = &objects[i];\n \n-\t\tif (is_delta_type(obj->type))\n-\t\t\tcontinue;\n-\t\tsecond_pass(obj);\n+\tnr_processed = 0;\n+#ifndef NO_PTHREADS\n+\tif (nr_threads > 1) {\n+\t\tinit_thread();\n+\t\tfor (i = 1; 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, NULL);\n+\t\t\tif (ret)\n+\t\t\t\tdie(\"unable to create thread: %s\", strerror(ret));\n+\t\t}\n+\t\tfor (i = 1; i < nr_threads; i++) {\n+\t\t\tpthread_join(thread_data[i].thread, NULL);\n+\t\t\tthread_data[i].thread = 0;\n+\t\t}\n+\t\tcleanup_thread();\n+\n+\t\t/* stop get_thread_data() from looking up beyond the\n+\t\t   first item, when fix_unresolved_deltas() runs */\n+\t\tnr_threads = 1;\n+\t\treturn;\n \t}\n+#endif\n+\tthreaded_second_pass(thread_data);\n }\n \n static int write_compressed(struct sha1file *f, void *in, unsigned int size)\n@@ -967,6 +1079,17 @@ static int git_index_pack_config(const char *k, const char *v, void *cb)\n \t\t\tdie(\"bad pack.indexversion=%\"PRIu32, opts->version);\n \t\treturn 0;\n \t}\n+\tif (!strcmp(k, \"pack.threads\")) {\n+\t\tnr_threads = git_config_int(k, v);\n+\t\tif (nr_threads < 0)\n+\t\t\tdie(\"invalid number of threads specified (%d)\",\n+\t\t\t    nr_threads);\n+#ifdef NO_PTHREADS\n+\t\tif (nr_threads != 1)\n+\t\t\twarning(\"no threads support, ignoring %s\", k);\n+#endif\n+\t\treturn 0;\n+\t}\n \treturn git_default_config(k, v, cb);\n }\n \n@@ -1125,6 +1248,16 @@ int cmd_index_pack(int argc, const char **argv, const char *prefix)\n \t\t\t\tkeep_msg = \"\";\n \t\t\t} else if (!prefixcmp(arg, \"--keep=\")) {\n \t\t\t\tkeep_msg = arg + 7;\n+\t\t\t} else if (!prefixcmp(arg, \"--threads=\")) {\n+\t\t\t\tchar *end;\n+\t\t\t\tnr_threads = strtoul(arg+10, &end, 0);\n+\t\t\t\tif (!arg[10] || *end || nr_threads < 0)\n+\t\t\t\t\tusage(index_pack_usage);\n+#ifdef NO_PTHREADS\n+\t\t\t\tif (nr_threads != 1)\n+\t\t\t\t\twarning(\"no threads support, \"\n+\t\t\t\t\t\t\"ignoring %s\", arg);\n+#endif\n \t\t\t} else if (!prefixcmp(arg, \"--pack_header=\")) {\n \t\t\t\tstruct pack_header *hdr;\n \t\t\t\tchar *c;\n@@ -1196,6 +1329,23 @@ int cmd_index_pack(int argc, const char **argv, const char *prefix)\n \tif (strict)\n \t\topts.flags |= WRITE_IDX_STRICT;\n \n+#ifndef NO_PTHREADS\n+\tif (!nr_threads) {\n+\t\tnr_threads = online_cpus();\n+\t\t/* An experiment showed that more threads does not mean faster */\n+\t\tif (nr_threads > 3)\n+\t\t\tnr_threads = 3;\n+\t}\n+\t/* reserve thread_data[0] for the main thread */\n+\tif (nr_threads > 1)\n+\t\tnr_threads++;\n+#else\n+\tif (nr_threads != 1)\n+\t\twarning(\"no threads support, ignoring --threads\");\n+\tnr_threads = 1;\n+#endif\n+\tthread_data = xcalloc(nr_threads, sizeof(*thread_data));\n+\n \tcurr_pack = open_pack_file(pack_name);\n \tparse_pack_header();\n \tobjects = xcalloc(nr_objects + 1, sizeof(struct object_entry));\n-- \n1.7.3.1.256.g2539c.dirty\n"},{"id":"185884","messageId":"7v4nu7ilee.fsf@alter.siamese.dyndns.org","threadId":"29767","inReplyTo":"1330403813-2770-2-git-send-email-pclouds@gmail.com","subject":"Re: [PATCH 2/2] index-pack: support multithreaded delta resolving","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2012-03-02T06:09:45Z","receivedAt":"2012-03-02T06:09:45Z","isPatch":true,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"When applied to 25a7850 and then merged to 'pu', the result fails to\ncorrectly produce pack .idx file.\n\nI spent an hour or so this afternoon, scratching my head, staring at the\noutput from tests added to t5510 by today's tr/maint-bundle-boundary topic\nto see where it breaks.  Its last test creates a bundle that has three\nobjects, extracts a pack from it, and runs \"index-pack --fix-thin\" on it.\n\nThis topic makes it fail with \"fatal: pack has 1 unresolved deltas\".\n"},{"id":"185920","messageId":"1330695744-13854-1-git-send-email-pclouds@gmail.com","threadId":"29767","inReplyTo":"7v4nu7ilee.fsf@alter.siamese.dyndns.org","subject":"[PATCH v2 2/2] index-pack: support multithreaded delta resolving","fromName":"Nguyễn Thái Ngọc Duy","fromEmail":"pclouds@gmail.com","sentAt":"2012-03-02T13:42:24Z","receivedAt":"2012-03-02T13:42:24Z","isPatch":true,"sender":{"key":"pclouds@gmail.com","avatar":"https://avatars.githubusercontent.com/u/720?v=4"},"body":"This puts delta resolving on each base on a separate thread, one base\ncache per thread.\n\nAn experiment on a 24 core machine with git.git shows that performance\ndoes not increase proportional to the number of cores. So by default,\nwe use maximum 3 cores.\n\n$ /usr/bin/time ~/t/git index-pack --threads=1 -v --stdin < XXX.pack\nReceiving objects: 100% (146564/146564), 53.99 MiB | 17.47 MiB/s, done.\nResolving deltas: 100% (109205/109205), done.\npack    d5471e8365717a5812cbc81ec7277cb697a80f08\n11.58user 0.37system 0:12.04elapsed 99%CPU (0avgtext+0avgdata 375088maxresident)k\n0inputs+118592outputs (0major+56894minor)pagefaults 0swaps\n\n$ ... --threads=2 ...\n14.58user 0.47system 0:09.99elapsed 150%CPU (0avgtext+0avgdata 411536maxresident)k\n0inputs+118592outputs (0major+79961minor)pagefaults 0swaps\n\n$ ... --threads=3 ...\n14.36user 0.64system 0:08.12elapsed 184%CPU (0avgtext+0avgdata 393312maxresident)k\n0inputs+118592outputs (0major+50998minor)pagefaults 0swaps\n\n$ ... --threads=4 ...\n15.81user 0.71system 0:08.17elapsed 202%CPU (0avgtext+0avgdata 419152maxresident)k\n0inputs+118592outputs (0major+54907minor)pagefaults 0swaps\n\n$ ... --threads=5 ...\n14.76user 0.72system 0:07.06elapsed 219%CPU (0avgtext+0avgdata 414112maxresident)k\n0inputs+118592outputs (0major+59547minor)pagefaults 0swaps\n\n$ ... --threads=8 ...\n15.98user 0.81system 0:07.71elapsed 217%CPU (0avgtext+0avgdata 429904maxresident)k\n0inputs+118592outputs (0major+66221minor)pagefaults 0swaps\n\n$ ... --threads=12 ...\n15.81user 0.74system 0:09.60elapsed 172%CPU (0avgtext+0avgdata 442336maxresident)k\n0inputs+118592outputs (0major+61353minor)pagefaults 0swaps\n\n$ ... --threads=16 ...\n15.41user 0.57system 0:11.62elapsed 137%CPU (0avgtext+0avgdata 451728maxresident)k\n0inputs+118592outputs (0major+63569minor)pagefaults 0swaps\n\n$ ... --threads=24 ...\n15.84user 0.63system 0:12.83elapsed 128%CPU (0avgtext+0avgdata 475728maxresident)k\n0inputs+118592outputs (0major+58013minor)pagefaults 0swaps\n\nSigned-off-by: Nguyễn Thái Ngọc Duy <pclouds@gmail.com>\n---\n 2012/3/2 Junio C Hamano <gitster@pobox.com>:\n > When applied to 25a7850 and then merged to 'pu', the result fails to\n > correctly produce pack .idx file.\n >\n > I spent an hour or so this afternoon, scratching my head, staring at the\n > output from tests added to t5510 by today's tr/maint-bundle-boundary topic\n > to see where it breaks.  Its last test creates a bundle that has three\n > objects, extracts a pack from it, and runs \"index-pack --fix-thin\" on it.\n >\n > This topic makes it fail with \"fatal: pack has 1 unresolved deltas\".\n\n And it thought it was good enough to CC you. Apparently parallel\n programming is hard. I make two mistakes:\n\n 1. I make each thread save resolved delta counter in\n    thread_data[].nr_resolved_deltas, then accumulate all of them to\n    global nr_resolved_deltas later. This plan does not work with\n    fix_unresolved_deltas() because it runs in single thread mode. It\n    stores the counter in thread_data[0], but the counter is never\n    added back to global nr_resolved_deltas. This makes t5510 fail.\n\n 2. The reason I put nr_resolved_deltas to thread-local struct is to\n    avoid locking. But I'm wrong. I still have two places where\n    thread_data[].nr_resolved_deltas can be changed: increment in\n    resolve_delta() and and reset in threaded_second_pass().\n\n So locking is required for changing nr_resolved_deltas anyway, I have\n removed thread_data[].nr_resolved_deltas and do the locking on global\n nr_resolved_deltas properly. \"pu\" seems to be happy with the updated\n series.\n\n One other thing. I did not consider to run fix_unresolved_deltas() in\n parallel originally because I didn't think it could be done. It can.\n But I'm not sure it's worth the effort. Anyway we can do that later\n if it turns out worth it.\n\n Documentation/git-index-pack.txt |   10 ++\n Makefile                         |    2 +-\n builtin/index-pack.c             |  214 ++++++++++++++++++++++++++++++++++----\n 3 files changed, 206 insertions(+), 20 deletions(-)\n\ndiff --git a/Documentation/git-index-pack.txt b/Documentation/git-index-pack.txt\nindex 909687f..39e6d0d 100644\n--- a/Documentation/git-index-pack.txt\n+++ b/Documentation/git-index-pack.txt\n@@ -74,6 +74,16 @@ OPTIONS\n --strict::\n \tDie, if the pack contains broken objects or links.\n \n+--threads=<n>::\n+\tSpecifies the number of threads to spawn when resolving\n+\tdeltas. This requires that index-pack be compiled with\n+\tpthreads otherwise this option is ignored with a warning.\n+\tThis is meant to reduce packing time on multiprocessor\n+\tmachines. The required amount of memory for the delta search\n+\twindow is however multiplied by the number of threads.\n+\tSpecifying 0 will cause git to auto-detect the number of CPU's\n+\tand use maximum 3 threads.\n+\n \n Note\n ----\ndiff --git a/Makefile b/Makefile\nindex 1fb1705..5fae875 100644\n--- a/Makefile\n+++ b/Makefile\n@@ -2159,7 +2159,7 @@ builtin/branch.o builtin/checkout.o builtin/clone.o builtin/reset.o branch.o tra\n builtin/bundle.o bundle.o transport.o: bundle.h\n builtin/bisect--helper.o builtin/rev-list.o bisect.o: bisect.h\n builtin/clone.o builtin/fetch-pack.o transport.o: fetch-pack.h\n-builtin/grep.o builtin/pack-objects.o transport-helper.o thread-utils.o: thread-utils.h\n+builtin/index-pack.o builtin/grep.o builtin/pack-objects.o transport-helper.o thread-utils.o: thread-utils.h\n builtin/send-pack.o transport.o: send-pack.h\n builtin/log.o builtin/shortlog.o: shortlog.h\n builtin/prune.o builtin/reflog.o reachable.o: reachable.h\ndiff --git a/builtin/index-pack.c b/builtin/index-pack.c\nindex 918684f..edd7cbd 100644\n--- a/builtin/index-pack.c\n+++ b/builtin/index-pack.c\n@@ -9,6 +9,7 @@\n #include \"progress.h\"\n #include \"fsck.h\"\n #include \"exec_cmd.h\"\n+#include \"thread-utils.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@@ -39,6 +40,19 @@ struct base_data {\n };\n \n /*\n+ * Thread-local data for find_unresolved_deltas(). The main process\n+ * also takes thread_data[0] to run find_unresolved_deltas() as part\n+ * of fix_unresolved_deltas().\n+ */\n+struct thread_local {\n+#ifndef NO_PTHREADS\n+\tpthread_t thread;\n+#endif\n+\tstruct base_data *base_cache;\n+\tsize_t base_cache_used;\n+};\n+\n+/*\n  * Even if sizeof(union delta_base) == 24 on 64-bit archs, we really want\n  * to memcmp() only the first 20 bytes.\n  */\n@@ -54,11 +68,12 @@ struct delta_entry {\n \n static struct object_entry *objects;\n static struct delta_entry *deltas;\n-static struct base_data *base_cache;\n-static size_t base_cache_used;\n+static struct thread_local *thread_data;\n static int nr_objects;\n+static int nr_processed;\n static int nr_deltas;\n static int nr_resolved_deltas;\n+static int nr_threads;\n \n static int from_stdin;\n static int strict;\n@@ -75,6 +90,49 @@ static git_SHA_CTX input_ctx;\n static uint32_t input_crc32;\n static int input_fd, output_fd, pack_fd;\n \n+#ifndef NO_PTHREADS\n+\n+static pthread_mutex_t read_mutex;\n+#define read_lock()\t\tpthread_mutex_lock(&read_mutex)\n+#define read_unlock()\t\tpthread_mutex_unlock(&read_mutex)\n+\n+static pthread_mutex_t counter_mutex;\n+#define counter_lock()\t\tpthread_mutex_lock(&counter_mutex)\n+#define counter_unlock()\t\tpthread_mutex_unlock(&counter_mutex)\n+\n+static pthread_mutex_t work_mutex;\n+#define work_lock()\t\tpthread_mutex_lock(&work_mutex)\n+#define work_unlock()\t\tpthread_mutex_unlock(&work_mutex)\n+\n+/*\n+ * Mutex and conditional variable can't be statically-initialized on Windows.\n+ */\n+static void init_thread(void)\n+{\n+\tinit_recursive_mutex(&read_mutex);\n+\tpthread_mutex_init(&work_mutex, NULL);\n+}\n+\n+static void cleanup_thread(void)\n+{\n+\tpthread_mutex_destroy(&read_mutex);\n+\tpthread_mutex_destroy(&work_mutex);\n+}\n+\n+#else\n+\n+#define read_lock()\n+#define read_unlock()\n+\n+#define counter_lock()\n+#define counter_unlock()\n+\n+#define work_lock()\n+#define work_unlock()\n+\n+#endif\n+\n+\n static int mark_link(struct object *obj, int type, void *data)\n {\n \tif (!obj)\n@@ -223,6 +281,36 @@ static NORETURN void bad_object(unsigned long offset, const char *format, ...)\n \tdie(\"pack has bad object at offset %lu: %s\", offset, buf);\n }\n \n+static struct thread_local *get_thread_data(void)\n+{\n+#ifndef NO_PTHREADS\n+\tint i;\n+\tpthread_t self = pthread_self();\n+\tfor (i = 1; i < nr_threads; i++)\n+\t\tif (self == thread_data[i].thread)\n+\t\t\treturn &thread_data[i];\n+#endif\n+\treturn &thread_data[0];\n+}\n+\n+static void resolve_one_delta(void)\n+{\n+#ifndef NO_PTHREADS\n+\tint i;\n+\tpthread_t self = pthread_self();\n+\tfor (i = 1; i < nr_threads; i++)\n+\t\tif (self == thread_data[i].thread) {\n+\t\t\tcounter_lock();\n+\t\t\tnr_resolved_deltas++;\n+\t\t\tcounter_unlock();\n+\t\t\treturn;\n+\t\t}\n+#endif\n+\tassert(nr_threads == 1 &&\n+\t       \"This should only be reached when all threads are gone\");\n+\tnr_resolved_deltas++;\n+}\n+\n static struct base_data *alloc_base_data(void)\n {\n \tstruct base_data *base = xmalloc(sizeof(struct base_data));\n@@ -237,15 +325,16 @@ static void free_base_data(struct base_data *c)\n \tif (c->data) {\n \t\tfree(c->data);\n \t\tc->data = NULL;\n-\t\tbase_cache_used -= c->size;\n+\t\tget_thread_data()->base_cache_used -= c->size;\n \t}\n }\n \n static void prune_base_data(struct base_data *retain)\n {\n \tstruct base_data *b;\n-\tfor (b = base_cache;\n-\t     base_cache_used > delta_base_cache_limit && b;\n+\tstruct thread_local *data = get_thread_data();\n+\tfor (b = data->base_cache;\n+\t     data->base_cache_used > delta_base_cache_limit && b;\n \t     b = b->child) {\n \t\tif (b->data && b != retain)\n \t\t\tfree_base_data(b);\n@@ -257,22 +346,23 @@ static void link_base_data(struct base_data *base, struct base_data *c)\n \tif (base)\n \t\tbase->child = c;\n \telse\n-\t\tbase_cache = c;\n+\t\tget_thread_data()->base_cache = c;\n \n \tc->base = base;\n \tc->child = NULL;\n \tif (c->data)\n-\t\tbase_cache_used += c->size;\n+\t\tget_thread_data()->base_cache_used += c->size;\n \tprune_base_data(c);\n }\n \n static void unlink_base_data(struct base_data *c)\n {\n-\tstruct base_data *base = c->base;\n+\tstruct base_data *base;\n+\tbase = c->base;\n \tif (base)\n \t\tbase->child = NULL;\n \telse\n-\t\tbase_cache = NULL;\n+\t\tget_thread_data()->base_cache = NULL;\n \tfree_base_data(c);\n }\n \n@@ -461,19 +551,24 @@ static void sha1_object(const void *data, unsigned long size,\n \t\t\tenum object_type type, unsigned char *sha1)\n {\n \thash_sha1_file(data, size, typename(type), sha1);\n+\tread_lock();\n \tif (has_sha1_file(sha1)) {\n \t\tvoid *has_data;\n \t\tenum object_type has_type;\n \t\tunsigned long has_size;\n \t\thas_data = read_sha1_file(sha1, &has_type, &has_size);\n+\t\tread_unlock();\n \t\tif (!has_data)\n \t\t\tdie(\"cannot read existing object %s\", sha1_to_hex(sha1));\n \t\tif (size != has_size || type != has_type ||\n \t\t    memcmp(data, has_data, size) != 0)\n \t\t\tdie(\"SHA1 COLLISION FOUND WITH %s !\", sha1_to_hex(sha1));\n \t\tfree(has_data);\n-\t}\n+\t} else\n+\t\tread_unlock();\n+\n \tif (strict) {\n+\t\tread_lock();\n \t\tif (type == OBJ_BLOB) {\n \t\t\tstruct blob *blob = lookup_blob(sha1);\n \t\t\tif (blob)\n@@ -507,6 +602,7 @@ static void sha1_object(const void *data, unsigned long size,\n \t\t\t}\n \t\t\tobj->flags |= FLAG_CHECKED;\n \t\t}\n+\t\tread_unlock();\n \t}\n }\n \n@@ -552,7 +648,7 @@ static void *get_base_data(struct base_data *c)\n \t\tif (!delta_nr) {\n \t\t\tc->data = get_data_from_pack(obj);\n \t\t\tc->size = obj->size;\n-\t\t\tbase_cache_used += c->size;\n+\t\t\tget_thread_data()->base_cache_used += c->size;\n \t\t\tprune_base_data(c);\n \t\t}\n \t\tfor (; delta_nr > 0; delta_nr--) {\n@@ -568,7 +664,7 @@ static void *get_base_data(struct base_data *c)\n \t\t\tfree(raw);\n \t\t\tif (!c->data)\n \t\t\t\tbad_object(obj->idx.offset, \"failed to apply delta\");\n-\t\t\tbase_cache_used += c->size;\n+\t\t\tget_thread_data()->base_cache_used += c->size;\n \t\t\tprune_base_data(c);\n \t\t}\n \t\tfree(delta);\n@@ -596,7 +692,7 @@ static void resolve_delta(struct object_entry *delta_obj,\n \t\tbad_object(delta_obj->idx.offset, \"failed to apply delta\");\n \tsha1_object(result->data, result->size, delta_obj->real_type,\n \t\t    delta_obj->idx.sha1);\n-\tnr_resolved_deltas++;\n+\tresolve_one_delta();\n }\n \n static struct base_data *find_unresolved_deltas_1(struct base_data *base,\n@@ -696,7 +792,32 @@ static void second_pass(struct object_entry *obj)\n \tbase_obj->obj = obj;\n \tbase_obj->data = NULL;\n \tfind_unresolved_deltas(base_obj);\n-\tdisplay_progress(progress, nr_resolved_deltas);\n+}\n+\n+static void *threaded_second_pass(void *arg)\n+{\n+\tfor (;;) {\n+\t\tint i, nr = 16;\n+\t\twork_lock();\n+\t\tdisplay_progress(progress, nr_resolved_deltas);\n+\t\twhile (nr_processed < nr_objects &&\n+\t\t       is_delta_type(objects[nr_processed].type))\n+\t\t\tnr_processed++;\n+\t\tif (nr_processed >= nr_objects) {\n+\t\t\twork_unlock();\n+\t\t\tbreak;\n+\t\t}\n+\t\ti = nr_processed;\n+\t\tnr_processed += nr;\n+\t\twork_unlock();\n+\n+\t\tfor (; nr && i < nr_objects; i++, nr--) {\n+\t\t\tif (is_delta_type(objects[i].type))\n+\t\t\t\tcontinue;\n+\t\t\tsecond_pass(&objects[i]);\n+\t\t}\n+\t}\n+\treturn NULL;\n }\n \n /* Parse all objects and return the pack content SHA1 hash */\n@@ -755,13 +876,30 @@ static void parse_pack_objects(unsigned char *sha1)\n \n \tif (verbose)\n \t\tprogress = start_progress(\"Resolving deltas\", nr_deltas);\n-\tfor (i = 0; i < nr_objects; i++) {\n-\t\tstruct object_entry *obj = &objects[i];\n \n-\t\tif (is_delta_type(obj->type))\n-\t\t\tcontinue;\n-\t\tsecond_pass(obj);\n+\tnr_processed = 0;\n+#ifndef NO_PTHREADS\n+\tif (nr_threads > 1) {\n+\t\tinit_thread();\n+\t\tfor (i = 1; 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, NULL);\n+\t\t\tif (ret)\n+\t\t\t\tdie(\"unable to create thread: %s\", strerror(ret));\n+\t\t}\n+\t\tfor (i = 1; i < nr_threads; i++) {\n+\t\t\tpthread_join(thread_data[i].thread, NULL);\n+\t\t\tthread_data[i].thread = 0;\n+\t\t}\n+\t\tcleanup_thread();\n+\n+\t\t/* stop get_thread_data() from looking up beyond the\n+\t\t   first item, when fix_unresolved_deltas() runs */\n+\t\tnr_threads = 1;\n+\t\treturn;\n \t}\n+#endif\n+\tthreaded_second_pass(thread_data);\n }\n \n static int write_compressed(struct sha1file *f, void *in, unsigned int size)\n@@ -967,6 +1105,17 @@ static int git_index_pack_config(const char *k, const char *v, void *cb)\n \t\t\tdie(\"bad pack.indexversion=%\"PRIu32, opts->version);\n \t\treturn 0;\n \t}\n+\tif (!strcmp(k, \"pack.threads\")) {\n+\t\tnr_threads = git_config_int(k, v);\n+\t\tif (nr_threads < 0)\n+\t\t\tdie(\"invalid number of threads specified (%d)\",\n+\t\t\t    nr_threads);\n+#ifdef NO_PTHREADS\n+\t\tif (nr_threads != 1)\n+\t\t\twarning(\"no threads support, ignoring %s\", k);\n+#endif\n+\t\treturn 0;\n+\t}\n \treturn git_default_config(k, v, cb);\n }\n \n@@ -1125,6 +1274,16 @@ int cmd_index_pack(int argc, const char **argv, const char *prefix)\n \t\t\t\tkeep_msg = \"\";\n \t\t\t} else if (!prefixcmp(arg, \"--keep=\")) {\n \t\t\t\tkeep_msg = arg + 7;\n+\t\t\t} else if (!prefixcmp(arg, \"--threads=\")) {\n+\t\t\t\tchar *end;\n+\t\t\t\tnr_threads = strtoul(arg+10, &end, 0);\n+\t\t\t\tif (!arg[10] || *end || nr_threads < 0)\n+\t\t\t\t\tusage(index_pack_usage);\n+#ifdef NO_PTHREADS\n+\t\t\t\tif (nr_threads != 1)\n+\t\t\t\t\twarning(\"no threads support, \"\n+\t\t\t\t\t\t\"ignoring %s\", arg);\n+#endif\n \t\t\t} else if (!prefixcmp(arg, \"--pack_header=\")) {\n \t\t\t\tstruct pack_header *hdr;\n \t\t\t\tchar *c;\n@@ -1196,6 +1355,23 @@ int cmd_index_pack(int argc, const char **argv, const char *prefix)\n \tif (strict)\n \t\topts.flags |= WRITE_IDX_STRICT;\n \n+#ifndef NO_PTHREADS\n+\tif (!nr_threads) {\n+\t\tnr_threads = online_cpus();\n+\t\t/* An experiment showed that more threads does not mean faster */\n+\t\tif (nr_threads > 3)\n+\t\t\tnr_threads = 3;\n+\t}\n+\t/* reserve thread_data[0] for the main thread */\n+\tif (nr_threads > 1)\n+\t\tnr_threads++;\n+#else\n+\tif (nr_threads != 1)\n+\t\twarning(\"no threads support, ignoring --threads\");\n+\tnr_threads = 1;\n+#endif\n+\tthread_data = xcalloc(nr_threads, sizeof(*thread_data));\n+\n \tcurr_pack = open_pack_file(pack_name);\n \tparse_pack_header();\n \tobjects = xcalloc(nr_objects + 1, sizeof(struct object_entry));\n-- \n1.7.8.36.g69ee2\n"},{"id":"185940","messageId":"7vzkbydecz.fsf@alter.siamese.dyndns.org","threadId":"29767","inReplyTo":"1330695744-13854-1-git-send-email-pclouds@gmail.com","subject":"Re: [PATCH v2 2/2] index-pack: support multithreaded delta resolving","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2012-03-02T18:53:00Z","receivedAt":"2012-03-02T18:53:00Z","isPatch":true,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"Nguyễn Thái Ngọc Duy  <pclouds@gmail.com> writes:\n\n>  One other thing. I did not consider to run fix_unresolved_deltas() in\n>  parallel originally because I didn't think it could be done. It can.\n>  But I'm not sure it's worth the effort. Anyway we can do that later\n>  if it turns out worth it.\n\nMy hunch agrees with your \"not sure it's worth\", as these are objects at\nthe boundary of --thin transfer, which should only be proportional to the\nsize of a single snapshot, not the depth of history, and I think your \"we\ncan do that later\" is a sound judgement.\n\nThanks.\n"}]}