[PATCH] threaded pack-objects: Use condition variables for thread communication.

[PATCH] threaded pack-objects: Use condition variables for thread communication.

[Johannes Sixt]

In the threaded pack-objects code the main thread and the worker threads
must mutually signal that they have assigned a new pack of work or have
completed their work, respectively. Previously, the code used mutexes that
were locked in one thread and unlocked from a different thread, which is
bogus (and happens to work on Linux).

Here we rectify the implementation by using condition variables: There is
one condition variable on which the main thread waits until a thread
requests new work; and each worker thread has its own condition variable
on which it waits until it is assigned new work or signaled to terminate.

As a cleanup, the worker threads are spawned only after the initial work
packages have been assigned.

Signed-off-by: Johannes Sixt <johannes.sixt@telecom.at>
---
 builtin-pack-objects.c |  129 +++++++++++++++++++++++++++++------------------
 1 files changed, 79 insertions(+), 50 deletions(-)

diff --git a/builtin-pack-objects.c b/builtin-pack-objects.c
index 7dd0d7f..451e48e 100644
--- a/builtin-pack-objects.c
+++ b/builtin-pack-objects.c
@@ -1594,6 +1594,15 @@ static void find_deltas(struct object_entry **list, unsigned *list_size,
 
 #ifdef THREADED_DELTA_SEARCH
 
+/*
+ * The main thread waits on the condition that (at least) one of the workers
+ * has stopped working (which is indicated in the .working member of
+ * struct thread_params).
+ * When a work thread has completed its work, it sets .working to 0 and
+ * signals the main thread and waits on the condition that .data_ready
+ * becomes 1.
+ */
+
 struct thread_params {
 	pthread_t thread;
 	struct object_entry **list;
@@ -1601,37 +1610,50 @@ struct thread_params {
 	unsigned remaining;
 	int window;
 	int depth;
+	int working;
+	int data_ready;
+	pthread_mutex_t mutex;
+	pthread_cond_t cond;
 	unsigned *processed;
 };
 
-static pthread_mutex_t data_request  = PTHREAD_MUTEX_INITIALIZER;
-static pthread_mutex_t data_ready    = PTHREAD_MUTEX_INITIALIZER;
-static pthread_mutex_t data_provider = PTHREAD_MUTEX_INITIALIZER;
-static struct thread_params *data_requester;
+static pthread_cond_t progress_cond = PTHREAD_COND_INITIALIZER;
 
 static void *threaded_find_deltas(void *arg)
 {
 	struct thread_params *me = arg;
 
-	for (;;) {
-		pthread_mutex_lock(&data_request);
-		data_requester = me;
-		pthread_mutex_unlock(&data_provider);
-		pthread_mutex_lock(&data_ready);
-		pthread_mutex_unlock(&data_request);
-
-		if (!me->remaining)
-			return NULL;
-
+	while (me->remaining) {
 		find_deltas(me->list, &me->remaining,
 			    me->window, me->depth, me->processed);
+
+		progress_lock();
+		me->working = 0;
+		progress_unlock();
+		pthread_cond_signal(&progress_cond);
+
+		/*
+		 * We must not set ->data_ready before we wait on the
+		 * condition because the main thread may have set it to 1
+		 * before we get here. In order to be sure that new
+		 * work is available if we see 1 in ->data_ready, it
+		 * was initialized to 0 before this thread was spawned
+		 * and we reset it to 0 right away.
+		 */
+		pthread_mutex_lock(&me->mutex);
+		while (!me->data_ready)
+			pthread_cond_wait(&me->cond, &me->mutex);
+		me->data_ready = 0;
+		pthread_mutex_unlock(&me->mutex);
 	}
+	/* leave ->working 1 so that this doesn't get more work assigned */
+	return NULL;
 }
 
 static void ll_find_deltas(struct object_entry **list, unsigned list_size,
 			   int window, int depth, unsigned *processed)
 {
-	struct thread_params *target, p[delta_search_threads];
+	struct thread_params p[delta_search_threads];
 	int i, ret, active_threads = 0;
 
 	if (delta_search_threads <= 1) {
@@ -1639,49 +1661,42 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,
 		return;
 	}
 
-	pthread_mutex_lock(&data_provider);
-	pthread_mutex_lock(&data_ready);
-
-	/* Start work threads. */
+	/* Partition the work amongst work threads. */
 	for (i = 0; i < delta_search_threads; i++) {
+		unsigned sub_size = list_size / (delta_search_threads - i);
+
 		p[i].window = window;
 		p[i].depth = depth;
 		p[i].processed = processed;
-		p[i].remaining = 0;
-		ret = pthread_create(&p[i].thread, NULL,
-				     threaded_find_deltas, &p[i]);
-		if (ret)
-			die("unable to create thread: %s", strerror(ret));
-		active_threads++;
-	}
-
-	/* Then partition the work amongst them. */
-	for (i = 0; i < delta_search_threads; i++) {
-		unsigned sub_size = list_size / (delta_search_threads - i);
-
-		pthread_mutex_lock(&data_provider);
-		target = data_requester;
-		if (!sub_size) {
-			pthread_mutex_unlock(&data_ready);
-			pthread_join(target->thread, NULL);
-			active_threads--;
-			continue;
-		}
+		p[i].working = 1;
+		p[i].data_ready = 0;
+		pthread_mutex_init(&p[i].mutex, NULL);
+		pthread_cond_init(&p[i].cond, NULL);
 
 		/* try to split chunks on "path" boundaries */
 		while (sub_size < list_size && list[sub_size]->hash &&
 		       list[sub_size]->hash == list[sub_size-1]->hash)
 			sub_size++;
 
-		target->list = list;
-		target->list_size = sub_size;
-		target->remaining = sub_size;
-		pthread_mutex_unlock(&data_ready);
+		p[i].list = list;
+		p[i].list_size = sub_size;
+		p[i].remaining = sub_size;
 
 		list += sub_size;
 		list_size -= sub_size;
 	}
 
+	/* Start work threads. */
+	for (i = 0; i < delta_search_threads; i++) {
+		if (!p[i].list_size)
+			continue;
+		ret = pthread_create(&p[i].thread, NULL,
+				     threaded_find_deltas, &p[i]);
+		if (ret)
+			die("unable to create thread: %s", strerror(ret));
+		active_threads++;
+	}
+
 	/*
 	 * Now let's wait for work completion.  Each time a thread is done
 	 * with its work, we steal half of the remaining work from the
@@ -1690,13 +1705,21 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,
 	 * until the remaining object list segments are simply too short
 	 * to be worth splitting anymore.
 	 */
-	do {
+	while (active_threads) {
+		struct thread_params *target = NULL;
 		struct thread_params *victim = NULL;
 		unsigned sub_size = 0;
-		pthread_mutex_lock(&data_provider);
-		target = data_requester;
 
 		progress_lock();
+		for (;;) {
+			for (i = 0; !target && i < delta_search_threads; i++)
+				if (!p[i].working)
+					target = &p[i];
+			if (target)
+				break;
+			pthread_cond_wait(&progress_cond, &progress_mutex);
+		};
+
 		for (i = 0; i < delta_search_threads; i++)
 			if (p[i].remaining > 2*window &&
 			    (!victim || victim->remaining < p[i].remaining))
@@ -1723,17 +1746,23 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,
 			victim->list_size -= sub_size;
 			victim->remaining -= sub_size;
 		}
-		progress_unlock();
-
 		target->list_size = sub_size;
 		target->remaining = sub_size;
-		pthread_mutex_unlock(&data_ready);
+		target->working = 1;
+		progress_unlock();
+
+		pthread_mutex_lock(&target->mutex);
+		target->data_ready = 1;
+		pthread_mutex_unlock(&target->mutex);
+		pthread_cond_signal(&target->cond);
 
 		if (!sub_size) {
 			pthread_join(target->thread, NULL);
+			pthread_cond_destroy(&target->cond);
+			pthread_mutex_destroy(&target->mutex);
 			active_threads--;
 		}
-	} while (active_threads);
+	}
 }
 
 #else
-- 
1.5.3.6.954.g0c63

Re: [PATCH] threaded pack-objects: Use condition variables for thread communication.

[Peter Baumann]

On Sun, Dec 16, 2007 at 12:18:53AM +0100, Johannes Sixt wrote:
> In the threaded pack-objects code the main thread and the worker threads
> must mutually signal that they have assigned a new pack of work or have
> completed their work, respectively. Previously, the code used mutexes that
> were locked in one thread and unlocked from a different thread, which is
> bogus (and happens to work on Linux).
> 
> Here we rectify the implementation by using condition variables: There is
> one condition variable on which the main thread waits until a thread
> requests new work; and each worker thread has its own condition variable
> on which it waits until it is assigned new work or signaled to terminate.
> 
> As a cleanup, the worker threads are spawned only after the initial work
> packages have been assigned.
> 
> Signed-off-by: Johannes Sixt <johannes.sixt@telecom.at>
> ---
>  builtin-pack-objects.c |  129 +++++++++++++++++++++++++++++------------------
>  1 files changed, 79 insertions(+), 50 deletions(-)
> 
> diff --git a/builtin-pack-objects.c b/builtin-pack-objects.c
> index 7dd0d7f..451e48e 100644
> --- a/builtin-pack-objects.c
> +++ b/builtin-pack-objects.c
> @@ -1594,6 +1594,15 @@ static void find_deltas(struct object_entry **list, unsigned *list_size,
>  
>  #ifdef THREADED_DELTA_SEARCH
>  
> +/*
> + * The main thread waits on the condition that (at least) one of the workers
> + * has stopped working (which is indicated in the .working member of
> + * struct thread_params).
> + * When a work thread has completed its work, it sets .working to 0 and
> + * signals the main thread and waits on the condition that .data_ready
> + * becomes 1.
> + */
> +
>  struct thread_params {
>  	pthread_t thread;
>  	struct object_entry **list;
> @@ -1601,37 +1610,50 @@ struct thread_params {
>  	unsigned remaining;
>  	int window;
>  	int depth;
> +	int working;
> +	int data_ready;
> +	pthread_mutex_t mutex;
> +	pthread_cond_t cond;
>  	unsigned *processed;
>  };
>  
> -static pthread_mutex_t data_request  = PTHREAD_MUTEX_INITIALIZER;
> -static pthread_mutex_t data_ready    = PTHREAD_MUTEX_INITIALIZER;
> -static pthread_mutex_t data_provider = PTHREAD_MUTEX_INITIALIZER;
> -static struct thread_params *data_requester;
> +static pthread_cond_t progress_cond = PTHREAD_COND_INITIALIZER;
>  
>  static void *threaded_find_deltas(void *arg)
>  {
>  	struct thread_params *me = arg;
>  
> -	for (;;) {
> -		pthread_mutex_lock(&data_request);
> -		data_requester = me;
> -		pthread_mutex_unlock(&data_provider);
> -		pthread_mutex_lock(&data_ready);
> -		pthread_mutex_unlock(&data_request);
> -
> -		if (!me->remaining)
> -			return NULL;
> -
> +	while (me->remaining) {
>  		find_deltas(me->list, &me->remaining,
>  			    me->window, me->depth, me->processed);
> +
> +		progress_lock();
> +		me->working = 0;
> +		progress_unlock();
> +		pthread_cond_signal(&progress_cond);

Shouldn't the pthread_cond_signal be inside the lock?
e.g. swap progress_unlock() with pthread_cond_signal(&progress_cond)

> +
> +		/*
> +		 * We must not set ->data_ready before we wait on the
> +		 * condition because the main thread may have set it to 1
> +		 * before we get here. In order to be sure that new
> +		 * work is available if we see 1 in ->data_ready, it
> +		 * was initialized to 0 before this thread was spawned
> +		 * and we reset it to 0 right away.
> +		 */
> +		pthread_mutex_lock(&me->mutex);
> +		while (!me->data_ready)
> +			pthread_cond_wait(&me->cond, &me->mutex);
> +		me->data_ready = 0;
> +		pthread_mutex_unlock(&me->mutex);
>  	}
> +	/* leave ->working 1 so that this doesn't get more work assigned */
> +	return NULL;
>  }

[...]


-Peter

Re: [PATCH] threaded pack-objects: Use condition variables for thread communication.

[Johannes Sixt]

On Sunday 16 December 2007 13:05, Peter Baumann wrote:
> On Sun, Dec 16, 2007 at 12:18:53AM +0100, Johannes Sixt wrote:
> > +
> > +		progress_lock();
> > +		me->working = 0;
> > +		progress_unlock();
> > +		pthread_cond_signal(&progress_cond);
>
> Shouldn't the pthread_cond_signal be inside the lock?
> e.g. swap progress_unlock() with pthread_cond_signal(&progress_cond)

No, that's not necessary. Both ways are correct, but if it's outside the lock 
there is less contention on the mutex (because the waiting thread must 
acquire the mutex lock before it can return from pthread_cond_wait).

-- Hannes

Re: [PATCH] threaded pack-objects: Use condition variables for thread communication.

[Peter Baumann]

On Sun, Dec 16, 2007 at 07:41:37PM +0100, Johannes Sixt wrote:
> On Sunday 16 December 2007 13:05, Peter Baumann wrote:
> > On Sun, Dec 16, 2007 at 12:18:53AM +0100, Johannes Sixt wrote:
> > > +
> > > +		progress_lock();
> > > +		me->working = 0;
> > > +		progress_unlock();
> > > +		pthread_cond_signal(&progress_cond);
> >
> > Shouldn't the pthread_cond_signal be inside the lock?
> > e.g. swap progress_unlock() with pthread_cond_signal(&progress_cond)
> 
> No, that's not necessary. Both ways are correct, but if it's outside the lock 
> there is less contention on the mutex (because the waiting thread must 
> acquire the mutex lock before it can return from pthread_cond_wait).
> 

At least I was told otherwise and [1] backs my knowledge up. Are you
really sure?

-Peter

http://docs.sun.com/app/docs/doc/806-5257/6je9h032r?a=view#sync-53686

Re: [PATCH] threaded pack-objects: Use condition variables for thread communication.

[David Brown]

On Sun, Dec 16, 2007 at 07:41:37PM +0100, Johannes Sixt wrote:
>On Sunday 16 December 2007 13:05, Peter Baumann wrote:
>> On Sun, Dec 16, 2007 at 12:18:53AM +0100, Johannes Sixt wrote:
>> > +
>> > +		progress_lock();
>> > +		me->working = 0;
>> > +		progress_unlock();
>> > +		pthread_cond_signal(&progress_cond);
>>
>> Shouldn't the pthread_cond_signal be inside the lock?
>> e.g. swap progress_unlock() with pthread_cond_signal(&progress_cond)
>
>No, that's not necessary. Both ways are correct, but if it's outside the lock 
>there is less contention on the mutex (because the waiting thread must 
>acquire the mutex lock before it can return from pthread_cond_wait).

The signal needs to be before the unlock if is more than one signaller.
Since normal usage has the signal inside of the lock, most implementations
should do this efficiently.

Dave

[PATCH v2] threaded pack-objects: Use condition variables for thread communication.

[Johannes Sixt]

In the threaded pack-objects code the main thread and the worker threads
must mutually signal that they have assigned a new pack of work or have
completed their work, respectively. Previously, the code used mutexes that
were locked in one thread and unlocked from a different thread, which is
bogus (and happens to work on Linux).

Here we rectify the implementation by using condition variables: There is
one condition variable on which the main thread waits until a thread
requests new work; and each worker thread has its own condition variable
on which it waits until it is assigned new work or signaled to terminate.

As a cleanup, the worker threads are spawned only after the initial work
packages have been assigned.

Signed-off-by: Johannes Sixt <johannes.sixt@telecom.at>
---
On Sunday 16 December 2007 20:00, Peter Baumann wrote:
> On Sun, Dec 16, 2007 at 07:41:37PM +0100, Johannes Sixt wrote:
> > On Sunday 16 December 2007 13:05, Peter Baumann wrote:
> > > On Sun, Dec 16, 2007 at 12:18:53AM +0100, Johannes Sixt wrote:
> > > > +
> > > > +		progress_lock();
> > > > +		me->working = 0;
> > > > +		progress_unlock();
> > > > +		pthread_cond_signal(&progress_cond);
> > >
> > > Shouldn't the pthread_cond_signal be inside the lock?
> > > e.g. swap progress_unlock() with pthread_cond_signal(&progress_cond)
> >
> > No, that's not necessary. Both ways are correct, but if it's outside the
> > lock there is less contention on the mutex (because the waiting thread
> > must acquire the mutex lock before it can return from pthread_cond_wait).
>
> At least I was told otherwise and [1] backs my knowledge up. Are you
> really sure?
>
> -Peter
>
> http://docs.sun.com/app/docs/doc/806-5257/6je9h032r?a=view#sync-53686

Uh, yes, that's a very compelling reason. I'm wrong. Here's an updated patch.

-- Hannes

 builtin-pack-objects.c |  129 +++++++++++++++++++++++++++++------------------
 1 files changed, 79 insertions(+), 50 deletions(-)

diff --git a/builtin-pack-objects.c b/builtin-pack-objects.c
index 7dd0d7f..4715d5f 100644
--- a/builtin-pack-objects.c
+++ b/builtin-pack-objects.c
@@ -1594,6 +1594,15 @@ static void find_deltas(struct object_entry **list, unsigned *list_size,
 
 #ifdef THREADED_DELTA_SEARCH
 
+/*
+ * The main thread waits on the condition that (at least) one of the workers
+ * has stopped working (which is indicated in the .working member of
+ * struct thread_params).
+ * When a work thread has completed its work, it sets .working to 0 and
+ * signals the main thread and waits on the condition that .data_ready
+ * becomes 1.
+ */
+
 struct thread_params {
 	pthread_t thread;
 	struct object_entry **list;
@@ -1601,37 +1610,50 @@ struct thread_params {
 	unsigned remaining;
 	int window;
 	int depth;
+	int working;
+	int data_ready;
+	pthread_mutex_t mutex;
+	pthread_cond_t cond;
 	unsigned *processed;
 };
 
-static pthread_mutex_t data_request  = PTHREAD_MUTEX_INITIALIZER;
-static pthread_mutex_t data_ready    = PTHREAD_MUTEX_INITIALIZER;
-static pthread_mutex_t data_provider = PTHREAD_MUTEX_INITIALIZER;
-static struct thread_params *data_requester;
+static pthread_cond_t progress_cond = PTHREAD_COND_INITIALIZER;
 
 static void *threaded_find_deltas(void *arg)
 {
 	struct thread_params *me = arg;
 
-	for (;;) {
-		pthread_mutex_lock(&data_request);
-		data_requester = me;
-		pthread_mutex_unlock(&data_provider);
-		pthread_mutex_lock(&data_ready);
-		pthread_mutex_unlock(&data_request);
-
-		if (!me->remaining)
-			return NULL;
-
+	while (me->remaining) {
 		find_deltas(me->list, &me->remaining,
 			    me->window, me->depth, me->processed);
+
+		progress_lock();
+		me->working = 0;
+		pthread_cond_signal(&progress_cond);
+		progress_unlock();
+
+		/*
+		 * We must not set ->data_ready before we wait on the
+		 * condition because the main thread may have set it to 1
+		 * before we get here. In order to be sure that new
+		 * work is available if we see 1 in ->data_ready, it
+		 * was initialized to 0 before this thread was spawned
+		 * and we reset it to 0 right away.
+		 */
+		pthread_mutex_lock(&me->mutex);
+		while (!me->data_ready)
+			pthread_cond_wait(&me->cond, &me->mutex);
+		me->data_ready = 0;
+		pthread_mutex_unlock(&me->mutex);
 	}
+	/* leave ->working 1 so that this doesn't get more work assigned */
+	return NULL;
 }
 
 static void ll_find_deltas(struct object_entry **list, unsigned list_size,
 			   int window, int depth, unsigned *processed)
 {
-	struct thread_params *target, p[delta_search_threads];
+	struct thread_params p[delta_search_threads];
 	int i, ret, active_threads = 0;
 
 	if (delta_search_threads <= 1) {
@@ -1639,49 +1661,42 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,
 		return;
 	}
 
-	pthread_mutex_lock(&data_provider);
-	pthread_mutex_lock(&data_ready);
-
-	/* Start work threads. */
+	/* Partition the work amongst work threads. */
 	for (i = 0; i < delta_search_threads; i++) {
+		unsigned sub_size = list_size / (delta_search_threads - i);
+
 		p[i].window = window;
 		p[i].depth = depth;
 		p[i].processed = processed;
-		p[i].remaining = 0;
-		ret = pthread_create(&p[i].thread, NULL,
-				     threaded_find_deltas, &p[i]);
-		if (ret)
-			die("unable to create thread: %s", strerror(ret));
-		active_threads++;
-	}
-
-	/* Then partition the work amongst them. */
-	for (i = 0; i < delta_search_threads; i++) {
-		unsigned sub_size = list_size / (delta_search_threads - i);
-
-		pthread_mutex_lock(&data_provider);
-		target = data_requester;
-		if (!sub_size) {
-			pthread_mutex_unlock(&data_ready);
-			pthread_join(target->thread, NULL);
-			active_threads--;
-			continue;
-		}
+		p[i].working = 1;
+		p[i].data_ready = 0;
+		pthread_mutex_init(&p[i].mutex, NULL);
+		pthread_cond_init(&p[i].cond, NULL);
 
 		/* try to split chunks on "path" boundaries */
 		while (sub_size < list_size && list[sub_size]->hash &&
 		       list[sub_size]->hash == list[sub_size-1]->hash)
 			sub_size++;
 
-		target->list = list;
-		target->list_size = sub_size;
-		target->remaining = sub_size;
-		pthread_mutex_unlock(&data_ready);
+		p[i].list = list;
+		p[i].list_size = sub_size;
+		p[i].remaining = sub_size;
 
 		list += sub_size;
 		list_size -= sub_size;
 	}
 
+	/* Start work threads. */
+	for (i = 0; i < delta_search_threads; i++) {
+		if (!p[i].list_size)
+			continue;
+		ret = pthread_create(&p[i].thread, NULL,
+				     threaded_find_deltas, &p[i]);
+		if (ret)
+			die("unable to create thread: %s", strerror(ret));
+		active_threads++;
+	}
+
 	/*
 	 * Now let's wait for work completion.  Each time a thread is done
 	 * with its work, we steal half of the remaining work from the
@@ -1690,13 +1705,21 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,
 	 * until the remaining object list segments are simply too short
 	 * to be worth splitting anymore.
 	 */
-	do {
+	while (active_threads) {
+		struct thread_params *target = NULL;
 		struct thread_params *victim = NULL;
 		unsigned sub_size = 0;
-		pthread_mutex_lock(&data_provider);
-		target = data_requester;
 
 		progress_lock();
+		for (;;) {
+			for (i = 0; !target && i < delta_search_threads; i++)
+				if (!p[i].working)
+					target = &p[i];
+			if (target)
+				break;
+			pthread_cond_wait(&progress_cond, &progress_mutex);
+		};
+
 		for (i = 0; i < delta_search_threads; i++)
 			if (p[i].remaining > 2*window &&
 			    (!victim || victim->remaining < p[i].remaining))
@@ -1723,17 +1746,23 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,
 			victim->list_size -= sub_size;
 			victim->remaining -= sub_size;
 		}
-		progress_unlock();
-
 		target->list_size = sub_size;
 		target->remaining = sub_size;
-		pthread_mutex_unlock(&data_ready);
+		target->working = 1;
+		progress_unlock();
+
+		pthread_mutex_lock(&target->mutex);
+		target->data_ready = 1;
+		pthread_cond_signal(&target->cond);
+		pthread_mutex_unlock(&target->mutex);
 
 		if (!sub_size) {
 			pthread_join(target->thread, NULL);
+			pthread_cond_destroy(&target->cond);
+			pthread_mutex_destroy(&target->mutex);
 			active_threads--;
 		}
-	} while (active_threads);
+	}
 }
 
 #else
-- 
1.5.3.6.954.g0c63

Re: [PATCH v2] threaded pack-objects: Use condition variables for thread communication.

[Junio C Hamano]

Looks nice and I think the clean-up makes the logic easier to follow as
well.

Re: [PATCH v2] threaded pack-objects: Use condition variables for thread communication.

[Nicolas Pitre]

On Sun, 16 Dec 2007, Johannes Sixt wrote:

> Signed-off-by: Johannes Sixt <johannes.sixt@telecom.at>

One micro nit:

> +		for (;;) {
> +			for (i = 0; !target && i < delta_search_threads; i++)
> +				if (!p[i].working)
> +					target = &p[i];
> +			if (target)
> +				break;
> +			pthread_cond_wait(&progress_cond, &progress_mutex);
> +		};

There's a spurious ; here.

Otherwise:

Acked-by: Nicolas Pitre <nico@cam.org>


Nicolas

Re: [PATCH] threaded pack-objects: Use condition variables for thread communication.

[Dmitry Potapov]

On Sun, Dec 16, 2007 at 08:00:16PM +0100, Peter Baumann wrote:
> On Sun, Dec 16, 2007 at 07:41:37PM +0100, Johannes Sixt wrote:
> > On Sunday 16 December 2007 13:05, Peter Baumann wrote:
> > > On Sun, Dec 16, 2007 at 12:18:53AM +0100, Johannes Sixt wrote:
> > > > +
> > > > +		progress_lock();
> > > > +		me->working = 0;
> > > > +		progress_unlock();
> > > > +		pthread_cond_signal(&progress_cond);
> > >
> > > Shouldn't the pthread_cond_signal be inside the lock?
> > > e.g. swap progress_unlock() with pthread_cond_signal(&progress_cond)
> > 
> > No, that's not necessary. Both ways are correct, but if it's outside the lock 
> > there is less contention on the mutex (because the waiting thread must 
> > acquire the mutex lock before it can return from pthread_cond_wait).
> > 
> 
> At least I was told otherwise and [1] backs my knowledge up. Are you
> really sure?
> 
> -Peter
> 
> http://docs.sun.com/app/docs/doc/806-5257/6je9h032r?a=view#sync-53686

The POSIX standard clearly says that usage of pthread_cond_signal ouside
of the mutex protection is allowed:

===
The pthread_cond_signal() or pthread_cond_broadcast() functions may be
called by a thread whether or not it currently owns the mutex that
threads calling pthread_cond_wait() or pthread_cond_timedwait() have
associated with the condition variable during their waits; however, if
predictable scheduling behaviour is required, then that mutex is locked
by the thread calling pthread_cond_signal() or pthread_cond_broadcast().
===
http://www.opengroup.org/onlinepubs/007908775/xsh/pthread_cond_signal.html

And the argumentation provided by Sun's manual:
> Otherwise, the condition variable could be signaled between the test
> of the associated condition and blocking in pthread_cond_wait(), which
> can cause an infinite wait.

sounds strange to me. Indeed the condition variable could be signaled
between the test and blocking, but the condition itself is false, so the
thread should be blocked anyway. The only effect of signalling outside
of the mutex protection is that the thread blocked on the condition
variable can be waken up while the associated condition is still false,
but it is not a problem.

As to what is more optimal, it is less clear. Because on one side,
signaling under the lock increases contention, but, on the other hand,
it avoids spurious wakeup. So, I suppose it could be implementation
specific.

Dmitry

Re: [PATCH v2] threaded pack-objects: Use condition variables for thread communication.

[Johannes Sixt]

I just discovered a theoretical resource leakage:

> +	/* Partition the work amongst work threads. */
>  	for (i = 0; i < delta_search_threads; i++) {
...
> +		pthread_mutex_init(&p[i].mutex, NULL);
> +		pthread_cond_init(&p[i].cond, NULL);

These two initializations...

>  	}
>  
> +	/* Start work threads. */
> +	for (i = 0; i < delta_search_threads; i++) {
> +		if (!p[i].list_size)
> +			continue;

... should go here because...

> +		ret = pthread_create(&p[i].thread, NULL,
> +				     threaded_find_deltas, &p[i]);
> +		if (ret)
> +			die("unable to create thread: %s", strerror(ret));
> +		active_threads++;
> +	}
...
> +	while (active_threads) {
...
>  		if (!sub_size) {
>  			pthread_join(target->thread, NULL);
> +			pthread_cond_destroy(&target->cond);
> +			pthread_mutex_destroy(&target->mutex);

... we tear down only those for which we actually started a thread.

>  			active_threads--;
>  		}
> +	}

Will send a patch this evening.

-- Hannes

[PATCH] Plug a resource leak in threaded pack-objects code.

[Johannes Sixt]

A mutex and a condition variable is allocated for each thread and torn
down when the thread terminates. However, for certain workloads it can
happen that some threads are actually not started at all. In this case
we would leak the mutex and condition variable. Now we allocate them only
for those threads that are actually started.

Signed-off-by: Johannes Sixt <johannes.sixt@telecom.at>
---
> I just discovered a theoretical resource leakage:
>
> Will send a patch this evening.

Here it is.

-- Hannes

 builtin-pack-objects.c |    4 ++--
 1 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/builtin-pack-objects.c b/builtin-pack-objects.c
index 5765d02..e0ce114 100644
--- a/builtin-pack-objects.c
+++ b/builtin-pack-objects.c
@@ -1670,8 +1670,6 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,
 		p[i].processed = processed;
 		p[i].working = 1;
 		p[i].data_ready = 0;
-		pthread_mutex_init(&p[i].mutex, NULL);
-		pthread_cond_init(&p[i].cond, NULL);
 
 		/* try to split chunks on "path" boundaries */
 		while (sub_size < list_size && list[sub_size]->hash &&
@@ -1690,6 +1688,8 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,
 	for (i = 0; i < delta_search_threads; i++) {
 		if (!p[i].list_size)
 			continue;
+		pthread_mutex_init(&p[i].mutex, NULL);
+		pthread_cond_init(&p[i].cond, NULL);
 		ret = pthread_create(&p[i].thread, NULL,
 				     threaded_find_deltas, &p[i]);
 		if (ret)
-- 
1.5.4.rc0.37.g78e7