@@ -345,7 +345,8 @@ void thrmgr_destroy(threadpool_t *threadpool)

 	pthread_mutex_destroy(&(threadpool->pool_mutex));

 	pthread_cond_destroy(&(threadpool->idle_cond));

-	pthread_cond_destroy(&(threadpool->queueable_cond));

+	pthread_cond_destroy(&(threadpool->queueable_single_cond));

+	pthread_cond_destroy(&(threadpool->queueable_bulk_cond));

 	pthread_cond_destroy(&(threadpool->pool_cond));

 	pthread_attr_destroy(&(threadpool->pool_attr));

 	free(threadpool->single_queue);

@@ -407,7 +408,7 @@ threadpool_t *thrmgr_new(int max_threads, int idle_timeout, int max_queue, void

 		return NULL;

 	}

-	if (pthread_cond_init(&(threadpool->queueable_cond), NULL) != 0) {

+	if (pthread_cond_init(&(threadpool->queueable_single_cond), NULL) != 0) {

 		pthread_cond_destroy(&(threadpool->pool_cond));

 		pthread_mutex_destroy(&(threadpool->pool_mutex));

 		free(threadpool->single_queue);

@@ -416,8 +417,20 @@ threadpool_t *thrmgr_new(int max_threads, int idle_timeout, int max_queue, void

 		return NULL;

 	}

+	if (pthread_cond_init(&(threadpool->queueable_bulk_cond), NULL) != 0) {

+		pthread_cond_destroy(&(threadpool->queueable_single_cond));

+		pthread_cond_destroy(&(threadpool->pool_cond));

+		pthread_mutex_destroy(&(threadpool->pool_mutex));

+		free(threadpool->single_queue);

+		free(threadpool->bulk_queue);

+		free(threadpool);

+		return NULL;

+	}

+

+

 	if (pthread_cond_init(&(threadpool->idle_cond),NULL) != 0)  {

-		pthread_cond_destroy(&(threadpool->queueable_cond));

+		pthread_cond_destroy(&(threadpool->queueable_single_cond));

+		pthread_cond_destroy(&(threadpool->queueable_bulk_cond));

 		pthread_cond_destroy(&(threadpool->pool_cond));

 		pthread_mutex_destroy(&(threadpool->pool_mutex));

 		free(threadpool->single_queue);

@@ -427,7 +440,8 @@ threadpool_t *thrmgr_new(int max_threads, int idle_timeout, int max_queue, void

 	}

 	if (pthread_attr_init(&(threadpool->pool_attr)) != 0) {

-		pthread_cond_destroy(&(threadpool->queueable_cond));

+		pthread_cond_destroy(&(threadpool->queueable_single_cond));

+		pthread_cond_destroy(&(threadpool->queueable_bulk_cond));

 		pthread_cond_destroy(&(threadpool->idle_cond));

 		pthread_cond_destroy(&(threadpool->pool_cond));

 		pthread_mutex_destroy(&(threadpool->pool_mutex));

@@ -438,7 +452,8 @@ threadpool_t *thrmgr_new(int max_threads, int idle_timeout, int max_queue, void

 	}

 	if (pthread_attr_setdetachstate(&(threadpool->pool_attr), PTHREAD_CREATE_DETACHED) != 0) {

-		pthread_cond_destroy(&(threadpool->queueable_cond));

+		pthread_cond_destroy(&(threadpool->queueable_single_cond));

+		pthread_cond_destroy(&(threadpool->queueable_bulk_cond));

 		pthread_attr_destroy(&(threadpool->pool_attr));

 		pthread_cond_destroy(&(threadpool->idle_cond));

 		pthread_cond_destroy(&(threadpool->pool_cond));

@@ -534,8 +549,12 @@ static void stats_destroy(threadpool_t *pool)

 	pthread_mutex_unlock(&pools_lock);

 }

-static inline int thrmgr_contended(threadpool_t *pool)

+static inline int thrmgr_contended(threadpool_t *pool, int bulk)

 {

+    /* don't allow bulk items to exceed 50% of queue, so that

+     * non-bulk items get a chance to be in the queue */

+    if (bulk && pool->bulk_queue->item_count >= pool->queue_max/2)

+	return 1;

     return pool->bulk_queue->item_count + pool->single_queue->item_count

 	+ pool->thr_alive - pool->thr_idle >= pool->queue_max;

 }

@@ -574,9 +593,14 @@ static void *thrmgr_pop(threadpool_t *pool)

 	}

     }

-    if (!thrmgr_contended(pool)) {

-	logg("$THRMGR: queue crossed low threshold -> signaling\n");

-	pthread_cond_signal(&pool->queueable_cond);

+    if (!thrmgr_contended(pool, 0)) {

+	logg("$THRMGR: queue (single) crossed low threshold -> signaling\n");

+	pthread_cond_signal(&pool->queueable_single_cond);

+    }

+

+    if (!thrmgr_contended(pool, 1)) {

+	logg("$THRMGR: queue (bulk) crossed low threshold -> signaling\n");

+	pthread_cond_signal(&pool->queueable_bulk_cond);

     }

     return task;

@@ -667,6 +691,7 @@ static int thrmgr_dispatch_internal(threadpool_t *threadpool, void *user_data, i

 	do {

 	    work_queue_t *queue;

+	    pthread_cond_t *queueable_cond;

 	    int items;

 	    if (threadpool->state != POOL_VALID) {

@@ -674,14 +699,17 @@ static int thrmgr_dispatch_internal(threadpool_t *threadpool, void *user_data, i

 		break;

 	    }

-	    if (bulk)

+	    if (bulk) {

 		queue = threadpool->bulk_queue;

-	    else

+		queueable_cond = &threadpool->queueable_bulk_cond;

+	    } else {

 		queue = threadpool->single_queue;

+		queueable_cond = &threadpool->queueable_single_cond;

+	    }

-	    while (thrmgr_contended(threadpool)) {

+	    while (thrmgr_contended(threadpool, bulk)) {

 		logg("$THRMGR: contended, sleeping\n");

-		pthread_cond_wait(&threadpool->queueable_cond, &threadpool->pool_mutex);

+		pthread_cond_wait(queueable_cond, &threadpool->pool_mutex);

 		logg("$THRMGR: contended, woken\n");

 	    }

@@ -726,7 +754,7 @@ int thrmgr_group_dispatch(threadpool_t *threadpool, jobgroup_t *group, void *use

 	logg("$THRMGR: active jobs for %p: %d\n", group, group->jobs);

 	pthread_mutex_unlock(&group->mutex);

     }

-    if (!(ret = thrmgr_dispatch_internal(threadpool, user_data, 1)) && group) {

+    if (!(ret = thrmgr_dispatch_internal(threadpool, user_data, group ? 1 : 0)) && group) {

 	pthread_mutex_lock(&group->mutex);

 	group->jobs--;

 	logg("$THRMGR: active jobs for %p: %d\n", group, group->jobs);

@@ -62,8 +62,9 @@ typedef struct threadpool_tag {

 	pthread_attr_t pool_attr;

 	pthread_cond_t  idle_cond;

-	pthread_cond_t  queueable_cond;

-	

+	pthread_cond_t  queueable_single_cond;

+	pthread_cond_t  queueable_bulk_cond;

+

 	pool_state_t state;

 	int thr_max;

 	int queue_max;