{"thread":{"id":"11310","subject":"[PATCH] threaded pack-objects: Use condition variables for thread communication.","startedAt":"2007-12-15T23:18:53Z","lastAt":"2007-12-17T19:12:52Z","messageCount":11,"participants":["Johannes Sixt","Peter Baumann","David Brown","Junio C Hamano","Nicolas Pitre","Dmitry Potapov"],"isPatch":true,"patchVersion":1,"patchTotal":null},"messages":[{"id":"63283","messageId":"200712160018.54171.johannes.sixt@telecom.at","threadId":"11310","inReplyTo":null,"subject":"[PATCH] threaded pack-objects: Use condition variables for thread communication.","fromName":"Johannes Sixt","fromEmail":"johannes.sixt@telecom.at","sentAt":"2007-12-15T23:18:53Z","receivedAt":"2007-12-15T23:18:53Z","isPatch":true,"sender":{"key":"j6t@kdbg.org","avatar":"https://avatars.githubusercontent.com/u/14810926?v=4"},"body":"In the threaded pack-objects code the main thread and the worker threads\nmust mutually signal that they have assigned a new pack of work or have\ncompleted their work, respectively. Previously, the code used mutexes that\nwere locked in one thread and unlocked from a different thread, which is\nbogus (and happens to work on Linux).\n\nHere we rectify the implementation by using condition variables: There is\none condition variable on which the main thread waits until a thread\nrequests new work; and each worker thread has its own condition variable\non which it waits until it is assigned new work or signaled to terminate.\n\nAs a cleanup, the worker threads are spawned only after the initial work\npackages have been assigned.\n\nSigned-off-by: Johannes Sixt <johannes.sixt@telecom.at>\n---\n builtin-pack-objects.c |  129 +++++++++++++++++++++++++++++------------------\n 1 files changed, 79 insertions(+), 50 deletions(-)\n\ndiff --git a/builtin-pack-objects.c b/builtin-pack-objects.c\nindex 7dd0d7f..451e48e 100644\n--- a/builtin-pack-objects.c\n+++ b/builtin-pack-objects.c\n@@ -1594,6 +1594,15 @@ static void find_deltas(struct object_entry **list, unsigned *list_size,\n \n #ifdef THREADED_DELTA_SEARCH\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@@ -1601,37 +1610,50 @@ struct thread_params {\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-static pthread_mutex_t data_request  = PTHREAD_MUTEX_INITIALIZER;\n-static pthread_mutex_t data_ready    = PTHREAD_MUTEX_INITIALIZER;\n-static pthread_mutex_t data_provider = PTHREAD_MUTEX_INITIALIZER;\n-static struct thread_params *data_requester;\n+static pthread_cond_t progress_cond = PTHREAD_COND_INITIALIZER;\n \n static void *threaded_find_deltas(void *arg)\n {\n \tstruct thread_params *me = arg;\n \n-\tfor (;;) {\n-\t\tpthread_mutex_lock(&data_request);\n-\t\tdata_requester = me;\n-\t\tpthread_mutex_unlock(&data_provider);\n-\t\tpthread_mutex_lock(&data_ready);\n-\t\tpthread_mutex_unlock(&data_request);\n-\n-\t\tif (!me->remaining)\n-\t\t\treturn NULL;\n-\n+\twhile (me->remaining) {\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\tprogress_unlock();\n+\t\tpthread_cond_signal(&progress_cond);\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}\n+\t/* leave ->working 1 so that this doesn't get more work assigned */\n+\treturn NULL;\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 *target, p[delta_search_threads];\n+\tstruct thread_params p[delta_search_threads];\n \tint i, ret, active_threads = 0;\n \n \tif (delta_search_threads <= 1) {\n@@ -1639,49 +1661,42 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,\n \t\treturn;\n \t}\n \n-\tpthread_mutex_lock(&data_provider);\n-\tpthread_mutex_lock(&data_ready);\n-\n-\t/* Start work threads. */\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+\n \t\tp[i].window = window;\n \t\tp[i].depth = depth;\n \t\tp[i].processed = processed;\n-\t\tp[i].remaining = 0;\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/* Then partition the work amongst them. */\n-\tfor (i = 0; i < delta_search_threads; i++) {\n-\t\tunsigned sub_size = list_size / (delta_search_threads - i);\n-\n-\t\tpthread_mutex_lock(&data_provider);\n-\t\ttarget = data_requester;\n-\t\tif (!sub_size) {\n-\t\t\tpthread_mutex_unlock(&data_ready);\n-\t\t\tpthread_join(target->thread, NULL);\n-\t\t\tactive_threads--;\n-\t\t\tcontinue;\n-\t\t}\n+\t\tp[i].working = 1;\n+\t\tp[i].data_ready = 0;\n+\t\tpthread_mutex_init(&p[i].mutex, NULL);\n+\t\tpthread_cond_init(&p[i].cond, NULL);\n \n \t\t/* try to split chunks on \"path\" boundaries */\n \t\twhile (sub_size < list_size && list[sub_size]->hash &&\n \t\t       list[sub_size]->hash == list[sub_size-1]->hash)\n \t\t\tsub_size++;\n \n-\t\ttarget->list = list;\n-\t\ttarget->list_size = sub_size;\n-\t\ttarget->remaining = sub_size;\n-\t\tpthread_mutex_unlock(&data_ready);\n+\t\tp[i].list = list;\n+\t\tp[i].list_size = sub_size;\n+\t\tp[i].remaining = sub_size;\n \n \t\tlist += sub_size;\n \t\tlist_size -= sub_size;\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\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@@ -1690,13 +1705,21 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,\n \t * until the remaining object list segments are simply too short\n \t * to be worth splitting anymore.\n \t */\n-\tdo {\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-\t\tpthread_mutex_lock(&data_provider);\n-\t\ttarget = data_requester;\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@@ -1723,17 +1746,23 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,\n \t\t\tvictim->list_size -= sub_size;\n \t\t\tvictim->remaining -= sub_size;\n \t\t}\n-\t\tprogress_unlock();\n-\n \t\ttarget->list_size = sub_size;\n \t\ttarget->remaining = sub_size;\n-\t\tpthread_mutex_unlock(&data_ready);\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_mutex_unlock(&target->mutex);\n+\t\tpthread_cond_signal(&target->cond);\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} while (active_threads);\n+\t}\n }\n \n #else\n-- \n1.5.3.6.954.g0c63\n"},{"id":"63318","messageId":"20071216120558.GA4999@xp.machine.xx","threadId":"11310","inReplyTo":"200712160018.54171.johannes.sixt@telecom.at","subject":"Re: [PATCH] threaded pack-objects: Use condition variables for thread communication.","fromName":"Peter Baumann","fromEmail":"waste.manager@gmx.de","sentAt":"2007-12-16T12:05:58Z","receivedAt":"2007-12-16T12:05:58Z","isPatch":true,"sender":{"key":"waste.manager@gmx.de","avatar":null},"body":"On Sun, Dec 16, 2007 at 12:18:53AM +0100, Johannes Sixt wrote:\n> In the threaded pack-objects code the main thread and the worker threads\n> must mutually signal that they have assigned a new pack of work or have\n> completed their work, respectively. Previously, the code used mutexes that\n> were locked in one thread and unlocked from a different thread, which is\n> bogus (and happens to work on Linux).\n> \n> Here we rectify the implementation by using condition variables: There is\n> one condition variable on which the main thread waits until a thread\n> requests new work; and each worker thread has its own condition variable\n> on which it waits until it is assigned new work or signaled to terminate.\n> \n> As a cleanup, the worker threads are spawned only after the initial work\n> packages have been assigned.\n> \n> Signed-off-by: Johannes Sixt <johannes.sixt@telecom.at>\n> ---\n>  builtin-pack-objects.c |  129 +++++++++++++++++++++++++++++------------------\n>  1 files changed, 79 insertions(+), 50 deletions(-)\n> \n> diff --git a/builtin-pack-objects.c b/builtin-pack-objects.c\n> index 7dd0d7f..451e48e 100644\n> --- a/builtin-pack-objects.c\n> +++ b/builtin-pack-objects.c\n> @@ -1594,6 +1594,15 @@ static void find_deltas(struct object_entry **list, unsigned *list_size,\n>  \n>  #ifdef THREADED_DELTA_SEARCH\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> @@ -1601,37 +1610,50 @@ struct thread_params {\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> -static pthread_mutex_t data_request  = PTHREAD_MUTEX_INITIALIZER;\n> -static pthread_mutex_t data_ready    = PTHREAD_MUTEX_INITIALIZER;\n> -static pthread_mutex_t data_provider = PTHREAD_MUTEX_INITIALIZER;\n> -static struct thread_params *data_requester;\n> +static pthread_cond_t progress_cond = PTHREAD_COND_INITIALIZER;\n>  \n>  static void *threaded_find_deltas(void *arg)\n>  {\n>  \tstruct thread_params *me = arg;\n>  \n> -\tfor (;;) {\n> -\t\tpthread_mutex_lock(&data_request);\n> -\t\tdata_requester = me;\n> -\t\tpthread_mutex_unlock(&data_provider);\n> -\t\tpthread_mutex_lock(&data_ready);\n> -\t\tpthread_mutex_unlock(&data_request);\n> -\n> -\t\tif (!me->remaining)\n> -\t\t\treturn NULL;\n> -\n> +\twhile (me->remaining) {\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\tprogress_unlock();\n> +\t\tpthread_cond_signal(&progress_cond);\n\nShouldn't the pthread_cond_signal be inside the lock?\ne.g. swap progress_unlock() with pthread_cond_signal(&progress_cond)\n\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}\n> +\t/* leave ->working 1 so that this doesn't get more work assigned */\n> +\treturn NULL;\n>  }\n\n[...]\n\n\n-Peter\n"},{"id":"63336","messageId":"200712161941.38109.johannes.sixt@telecom.at","threadId":"11310","inReplyTo":"20071216120558.GA4999@xp.machine.xx","subject":"Re: [PATCH] threaded pack-objects: Use condition variables for thread communication.","fromName":"Johannes Sixt","fromEmail":"johannes.sixt@telecom.at","sentAt":"2007-12-16T18:41:37Z","receivedAt":"2007-12-16T18:41:37Z","isPatch":true,"sender":{"key":"j6t@kdbg.org","avatar":"https://avatars.githubusercontent.com/u/14810926?v=4"},"body":"On Sunday 16 December 2007 13:05, Peter Baumann wrote:\n> On Sun, Dec 16, 2007 at 12:18:53AM +0100, Johannes Sixt wrote:\n> > +\n> > +\t\tprogress_lock();\n> > +\t\tme->working = 0;\n> > +\t\tprogress_unlock();\n> > +\t\tpthread_cond_signal(&progress_cond);\n>\n> Shouldn't the pthread_cond_signal be inside the lock?\n> e.g. swap progress_unlock() with pthread_cond_signal(&progress_cond)\n\nNo, that's not necessary. Both ways are correct, but if it's outside the lock \nthere is less contention on the mutex (because the waiting thread must \nacquire the mutex lock before it can return from pthread_cond_wait).\n\n-- Hannes\n"},{"id":"63337","messageId":"20071216190016.GC4999@xp.machine.xx","threadId":"11310","inReplyTo":"200712161941.38109.johannes.sixt@telecom.at","subject":"Re: [PATCH] threaded pack-objects: Use condition variables for thread communication.","fromName":"Peter Baumann","fromEmail":"waste.manager@gmx.de","sentAt":"2007-12-16T19:00:16Z","receivedAt":"2007-12-16T19:00:16Z","isPatch":true,"sender":{"key":"waste.manager@gmx.de","avatar":null},"body":"On Sun, Dec 16, 2007 at 07:41:37PM +0100, Johannes Sixt wrote:\n> On Sunday 16 December 2007 13:05, Peter Baumann wrote:\n> > On Sun, Dec 16, 2007 at 12:18:53AM +0100, Johannes Sixt wrote:\n> > > +\n> > > +\t\tprogress_lock();\n> > > +\t\tme->working = 0;\n> > > +\t\tprogress_unlock();\n> > > +\t\tpthread_cond_signal(&progress_cond);\n> >\n> > Shouldn't the pthread_cond_signal be inside the lock?\n> > e.g. swap progress_unlock() with pthread_cond_signal(&progress_cond)\n> \n> No, that's not necessary. Both ways are correct, but if it's outside the lock \n> there is less contention on the mutex (because the waiting thread must \n> acquire the mutex lock before it can return from pthread_cond_wait).\n> \n\nAt least I was told otherwise and [1] backs my knowledge up. Are you\nreally sure?\n\n-Peter\n\nhttp://docs.sun.com/app/docs/doc/806-5257/6je9h032r?a=view#sync-53686\n"},{"id":"63338","messageId":"20071216191828.GA5446@old.davidb.org","threadId":"11310","inReplyTo":"200712161941.38109.johannes.sixt@telecom.at","subject":"Re: [PATCH] threaded pack-objects: Use condition variables for thread communication.","fromName":"David Brown","fromEmail":"git@davidb.org","sentAt":"2007-12-16T19:18:28Z","receivedAt":"2007-12-16T19:18:28Z","isPatch":true,"sender":{"key":"git@davidb.org","avatar":"https://gravatar.com/avatar/94c86a2938470a74c2eac5e2b69afc0871f79a660295c02219597aba8cb101c1?d=mp&s=160"},"body":"On Sun, Dec 16, 2007 at 07:41:37PM +0100, Johannes Sixt wrote:\n>On Sunday 16 December 2007 13:05, Peter Baumann wrote:\n>> On Sun, Dec 16, 2007 at 12:18:53AM +0100, Johannes Sixt wrote:\n>> > +\n>> > +\t\tprogress_lock();\n>> > +\t\tme->working = 0;\n>> > +\t\tprogress_unlock();\n>> > +\t\tpthread_cond_signal(&progress_cond);\n>>\n>> Shouldn't the pthread_cond_signal be inside the lock?\n>> e.g. swap progress_unlock() with pthread_cond_signal(&progress_cond)\n>\n>No, that's not necessary. Both ways are correct, but if it's outside the lock \n>there is less contention on the mutex (because the waiting thread must \n>acquire the mutex lock before it can return from pthread_cond_wait).\n\nThe signal needs to be before the unlock if is more than one signaller.\nSince normal usage has the signal inside of the lock, most implementations\nshould do this efficiently.\n\nDave\n"},{"id":"63342","messageId":"200712162045.34456.johannes.sixt@telecom.at","threadId":"11310","inReplyTo":"20071216190016.GC4999@xp.machine.xx","subject":"[PATCH v2] threaded pack-objects: Use condition variables for thread communication.","fromName":"Johannes Sixt","fromEmail":"johannes.sixt@telecom.at","sentAt":"2007-12-16T19:45:34Z","receivedAt":"2007-12-16T19:45:34Z","isPatch":true,"sender":{"key":"j6t@kdbg.org","avatar":"https://avatars.githubusercontent.com/u/14810926?v=4"},"body":"In the threaded pack-objects code the main thread and the worker threads\nmust mutually signal that they have assigned a new pack of work or have\ncompleted their work, respectively. Previously, the code used mutexes that\nwere locked in one thread and unlocked from a different thread, which is\nbogus (and happens to work on Linux).\n\nHere we rectify the implementation by using condition variables: There is\none condition variable on which the main thread waits until a thread\nrequests new work; and each worker thread has its own condition variable\non which it waits until it is assigned new work or signaled to terminate.\n\nAs a cleanup, the worker threads are spawned only after the initial work\npackages have been assigned.\n\nSigned-off-by: Johannes Sixt <johannes.sixt@telecom.at>\n---\nOn Sunday 16 December 2007 20:00, Peter Baumann wrote:\n> On Sun, Dec 16, 2007 at 07:41:37PM +0100, Johannes Sixt wrote:\n> > On Sunday 16 December 2007 13:05, Peter Baumann wrote:\n> > > On Sun, Dec 16, 2007 at 12:18:53AM +0100, Johannes Sixt wrote:\n> > > > +\n> > > > +\t\tprogress_lock();\n> > > > +\t\tme->working = 0;\n> > > > +\t\tprogress_unlock();\n> > > > +\t\tpthread_cond_signal(&progress_cond);\n> > >\n> > > Shouldn't the pthread_cond_signal be inside the lock?\n> > > e.g. swap progress_unlock() with pthread_cond_signal(&progress_cond)\n> >\n> > No, that's not necessary. Both ways are correct, but if it's outside the\n> > lock there is less contention on the mutex (because the waiting thread\n> > must acquire the mutex lock before it can return from pthread_cond_wait).\n>\n> At least I was told otherwise and [1] backs my knowledge up. Are you\n> really sure?\n>\n> -Peter\n>\n> http://docs.sun.com/app/docs/doc/806-5257/6je9h032r?a=view#sync-53686\n\nUh, yes, that's a very compelling reason. I'm wrong. Here's an updated patch.\n\n-- Hannes\n\n builtin-pack-objects.c |  129 +++++++++++++++++++++++++++++------------------\n 1 files changed, 79 insertions(+), 50 deletions(-)\n\ndiff --git a/builtin-pack-objects.c b/builtin-pack-objects.c\nindex 7dd0d7f..4715d5f 100644\n--- a/builtin-pack-objects.c\n+++ b/builtin-pack-objects.c\n@@ -1594,6 +1594,15 @@ static void find_deltas(struct object_entry **list, unsigned *list_size,\n \n #ifdef THREADED_DELTA_SEARCH\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@@ -1601,37 +1610,50 @@ struct thread_params {\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-static pthread_mutex_t data_request  = PTHREAD_MUTEX_INITIALIZER;\n-static pthread_mutex_t data_ready    = PTHREAD_MUTEX_INITIALIZER;\n-static pthread_mutex_t data_provider = PTHREAD_MUTEX_INITIALIZER;\n-static struct thread_params *data_requester;\n+static pthread_cond_t progress_cond = PTHREAD_COND_INITIALIZER;\n \n static void *threaded_find_deltas(void *arg)\n {\n \tstruct thread_params *me = arg;\n \n-\tfor (;;) {\n-\t\tpthread_mutex_lock(&data_request);\n-\t\tdata_requester = me;\n-\t\tpthread_mutex_unlock(&data_provider);\n-\t\tpthread_mutex_lock(&data_ready);\n-\t\tpthread_mutex_unlock(&data_request);\n-\n-\t\tif (!me->remaining)\n-\t\t\treturn NULL;\n-\n+\twhile (me->remaining) {\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}\n+\t/* leave ->working 1 so that this doesn't get more work assigned */\n+\treturn NULL;\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 *target, p[delta_search_threads];\n+\tstruct thread_params p[delta_search_threads];\n \tint i, ret, active_threads = 0;\n \n \tif (delta_search_threads <= 1) {\n@@ -1639,49 +1661,42 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,\n \t\treturn;\n \t}\n \n-\tpthread_mutex_lock(&data_provider);\n-\tpthread_mutex_lock(&data_ready);\n-\n-\t/* Start work threads. */\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+\n \t\tp[i].window = window;\n \t\tp[i].depth = depth;\n \t\tp[i].processed = processed;\n-\t\tp[i].remaining = 0;\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/* Then partition the work amongst them. */\n-\tfor (i = 0; i < delta_search_threads; i++) {\n-\t\tunsigned sub_size = list_size / (delta_search_threads - i);\n-\n-\t\tpthread_mutex_lock(&data_provider);\n-\t\ttarget = data_requester;\n-\t\tif (!sub_size) {\n-\t\t\tpthread_mutex_unlock(&data_ready);\n-\t\t\tpthread_join(target->thread, NULL);\n-\t\t\tactive_threads--;\n-\t\t\tcontinue;\n-\t\t}\n+\t\tp[i].working = 1;\n+\t\tp[i].data_ready = 0;\n+\t\tpthread_mutex_init(&p[i].mutex, NULL);\n+\t\tpthread_cond_init(&p[i].cond, NULL);\n \n \t\t/* try to split chunks on \"path\" boundaries */\n \t\twhile (sub_size < list_size && list[sub_size]->hash &&\n \t\t       list[sub_size]->hash == list[sub_size-1]->hash)\n \t\t\tsub_size++;\n \n-\t\ttarget->list = list;\n-\t\ttarget->list_size = sub_size;\n-\t\ttarget->remaining = sub_size;\n-\t\tpthread_mutex_unlock(&data_ready);\n+\t\tp[i].list = list;\n+\t\tp[i].list_size = sub_size;\n+\t\tp[i].remaining = sub_size;\n \n \t\tlist += sub_size;\n \t\tlist_size -= sub_size;\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\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@@ -1690,13 +1705,21 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,\n \t * until the remaining object list segments are simply too short\n \t * to be worth splitting anymore.\n \t */\n-\tdo {\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-\t\tpthread_mutex_lock(&data_provider);\n-\t\ttarget = data_requester;\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@@ -1723,17 +1746,23 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,\n \t\t\tvictim->list_size -= sub_size;\n \t\t\tvictim->remaining -= sub_size;\n \t\t}\n-\t\tprogress_unlock();\n-\n \t\ttarget->list_size = sub_size;\n \t\ttarget->remaining = sub_size;\n-\t\tpthread_mutex_unlock(&data_ready);\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} while (active_threads);\n+\t}\n }\n \n #else\n-- \n1.5.3.6.954.g0c63\n"},{"id":"63354","messageId":"7vy7bui90w.fsf@gitster.siamese.dyndns.org","threadId":"11310","inReplyTo":"200712162045.34456.johannes.sixt@telecom.at","subject":"Re: [PATCH v2] threaded pack-objects: Use condition variables for thread communication.","fromName":"Junio C Hamano","fromEmail":"gitster@pobox.com","sentAt":"2007-12-16T22:12:31Z","receivedAt":"2007-12-16T22:12:31Z","isPatch":true,"sender":{"key":"gitster@pobox.com","avatar":"https://avatars.githubusercontent.com/u/54884?v=4"},"body":"Looks nice and I think the clean-up makes the logic easier to follow as\nwell.\n"},{"id":"63365","messageId":"alpine.LFD.0.999999.0712162203470.8467@xanadu.home","threadId":"11310","inReplyTo":"200712162045.34456.johannes.sixt@telecom.at","subject":"Re: [PATCH v2] threaded pack-objects: Use condition variables for thread communication.","fromName":"Nicolas Pitre","fromEmail":"nico@cam.org","sentAt":"2007-12-17T03:13:38Z","receivedAt":"2007-12-17T03:13:38Z","isPatch":true,"sender":{"key":"nico@fluxnic.net","avatar":"https://avatars.githubusercontent.com/u/702790?v=4"},"body":"On Sun, 16 Dec 2007, Johannes Sixt wrote:\n\n> Signed-off-by: Johannes Sixt <johannes.sixt@telecom.at>\n\nOne micro nit:\n\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\nThere's a spurious ; here.\n\nOtherwise:\n\nAcked-by: Nicolas Pitre <nico@cam.org>\n\n\nNicolas\n"},{"id":"63367","messageId":"20071217042648.GI14173@dpotapov.dyndns.org","threadId":"11310","inReplyTo":"20071216190016.GC4999@xp.machine.xx","subject":"Re: [PATCH] threaded pack-objects: Use condition variables for thread communication.","fromName":"Dmitry Potapov","fromEmail":"dpotapov@gmail.com","sentAt":"2007-12-17T04:26:48Z","receivedAt":"2007-12-17T04:26:48Z","isPatch":true,"sender":{"key":"dpotapov@gmail.com","avatar":"https://avatars.githubusercontent.com/u/6568595?v=4"},"body":"On Sun, Dec 16, 2007 at 08:00:16PM +0100, Peter Baumann wrote:\n> On Sun, Dec 16, 2007 at 07:41:37PM +0100, Johannes Sixt wrote:\n> > On Sunday 16 December 2007 13:05, Peter Baumann wrote:\n> > > On Sun, Dec 16, 2007 at 12:18:53AM +0100, Johannes Sixt wrote:\n> > > > +\n> > > > +\t\tprogress_lock();\n> > > > +\t\tme->working = 0;\n> > > > +\t\tprogress_unlock();\n> > > > +\t\tpthread_cond_signal(&progress_cond);\n> > >\n> > > Shouldn't the pthread_cond_signal be inside the lock?\n> > > e.g. swap progress_unlock() with pthread_cond_signal(&progress_cond)\n> > \n> > No, that's not necessary. Both ways are correct, but if it's outside the lock \n> > there is less contention on the mutex (because the waiting thread must \n> > acquire the mutex lock before it can return from pthread_cond_wait).\n> > \n> \n> At least I was told otherwise and [1] backs my knowledge up. Are you\n> really sure?\n> \n> -Peter\n> \n> http://docs.sun.com/app/docs/doc/806-5257/6je9h032r?a=view#sync-53686\n\nThe POSIX standard clearly says that usage of pthread_cond_signal ouside\nof the mutex protection is allowed:\n\n===\nThe pthread_cond_signal() or pthread_cond_broadcast() functions may be\ncalled by a thread whether or not it currently owns the mutex that\nthreads calling pthread_cond_wait() or pthread_cond_timedwait() have\nassociated with the condition variable during their waits; however, if\npredictable scheduling behaviour is required, then that mutex is locked\nby the thread calling pthread_cond_signal() or pthread_cond_broadcast().\n===\nhttp://www.opengroup.org/onlinepubs/007908775/xsh/pthread_cond_signal.html\n\nAnd the argumentation provided by Sun's manual:\n> Otherwise, the condition variable could be signaled between the test\n> of the associated condition and blocking in pthread_cond_wait(), which\n> can cause an infinite wait.\n\nsounds strange to me. Indeed the condition variable could be signaled\nbetween the test and blocking, but the condition itself is false, so the\nthread should be blocked anyway. The only effect of signalling outside\nof the mutex protection is that the thread blocked on the condition\nvariable can be waken up while the associated condition is still false,\nbut it is not a problem.\n\nAs to what is more optimal, it is less clear. Because on one side,\nsignaling under the lock increases contention, but, on the other hand,\nit avoids spurious wakeup. So, I suppose it could be implementation\nspecific.\n\nDmitry\n"},{"id":"63372","messageId":"476628D1.4020300@viscovery.net","threadId":"11310","inReplyTo":"200712162045.34456.johannes.sixt@telecom.at","subject":"Re: [PATCH v2] threaded pack-objects: Use condition variables for thread communication.","fromName":"Johannes Sixt","fromEmail":"j.sixt@viscovery.net","sentAt":"2007-12-17T07:44:17Z","receivedAt":"2007-12-17T07:44:17Z","isPatch":true,"sender":{"key":"j6t@kdbg.org","avatar":"https://avatars.githubusercontent.com/u/14810926?v=4"},"body":"I just discovered a theoretical resource leakage:\n\n> +\t/* Partition the work amongst work threads. */\n>  \tfor (i = 0; i < delta_search_threads; i++) {\n...\n> +\t\tpthread_mutex_init(&p[i].mutex, NULL);\n> +\t\tpthread_cond_init(&p[i].cond, NULL);\n\nThese two initializations...\n\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\n... should go here because...\n\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> +\twhile (active_threads) {\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\n... we tear down only those for which we actually started a thread.\n\n>  \t\t\tactive_threads--;\n>  \t\t}\n> +\t}\n\nWill send a patch this evening.\n\n-- Hannes\n"},{"id":"63462","messageId":"200712172012.52374.johannes.sixt@telecom.at","threadId":"11310","inReplyTo":"476628D1.4020300@viscovery.net","subject":"[PATCH] Plug a resource leak in threaded pack-objects code.","fromName":"Johannes Sixt","fromEmail":"johannes.sixt@telecom.at","sentAt":"2007-12-17T19:12:52Z","receivedAt":"2007-12-17T19:12:52Z","isPatch":true,"sender":{"key":"j6t@kdbg.org","avatar":"https://avatars.githubusercontent.com/u/14810926?v=4"},"body":"A mutex and a condition variable is allocated for each thread and torn\ndown when the thread terminates. However, for certain workloads it can\nhappen that some threads are actually not started at all. In this case\nwe would leak the mutex and condition variable. Now we allocate them only\nfor those threads that are actually started.\n\nSigned-off-by: Johannes Sixt <johannes.sixt@telecom.at>\n---\n> I just discovered a theoretical resource leakage:\n>\n> Will send a patch this evening.\n\nHere it is.\n\n-- Hannes\n\n builtin-pack-objects.c |    4 ++--\n 1 files changed, 2 insertions(+), 2 deletions(-)\n\ndiff --git a/builtin-pack-objects.c b/builtin-pack-objects.c\nindex 5765d02..e0ce114 100644\n--- a/builtin-pack-objects.c\n+++ b/builtin-pack-objects.c\n@@ -1670,8 +1670,6 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,\n \t\tp[i].processed = processed;\n \t\tp[i].working = 1;\n \t\tp[i].data_ready = 0;\n-\t\tpthread_mutex_init(&p[i].mutex, NULL);\n-\t\tpthread_cond_init(&p[i].cond, NULL);\n \n \t\t/* try to split chunks on \"path\" boundaries */\n \t\twhile (sub_size < list_size && list[sub_size]->hash &&\n@@ -1690,6 +1688,8 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,\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-- \n1.5.4.rc0.37.g78e7\n"}]}