[sheepdog] [PATCH v5 1/3] work: protect nr_threads by wi->workers_lock

[Hitoshi Mitake]

Previous protection scheme of wi->nr_thread in work.c was
unclear because wi->startup_lock was also used for protecting it
during workqueue grow/shrink. This patch let work.c protect
wi->nr_thread by the new wi->workers_lock.

Signed-off-by: Hitoshi Mitake <mitake.hitoshi at lab.ntt.co.jp>
---
 lib/work.c |   26 ++++++++++++++++++++------
 1 file changed, 20 insertions(+), 6 deletions(-)

v5:
 - protect wq_need_grow() with workers_lock

diff --git a/lib/work.c b/lib/work.c
index 50b57a5..2969c21 100644
--- a/lib/work.c
+++ b/lib/work.c
@@ -55,6 +55,9 @@ struct worker_info {
 	pthread_mutex_t pending_lock;
 	/* protected by pending_lock */
 	struct work_queue q;
+
+	pthread_mutex_t workers_lock;
+	/* protected by workers_lock */
 	size_t nr_threads;
 
 	/* protected by uatomic primitives */
@@ -217,14 +220,20 @@ static inline uint64_t wq_get_roof(struct worker_info *wi)
 
 static bool wq_need_grow(struct worker_info *wi)
 {
+	bool ret = false;
+
+	pthread_mutex_lock(&wi->workers_lock);
+
 	if (wi->nr_threads < uatomic_read(&wi->nr_workers) &&
 	    wi->nr_threads * 2 <= wq_get_roof(wi)) {
 		wi->tm_end_of_protection = get_msec_time() +
 			WQ_PROTECTION_PERIOD;
-		return true;
+		ret = true;
 	}
 
-	return false;
+	pthread_mutex_unlock(&wi->workers_lock);
+
+	return ret;
 }
 
 /*
@@ -247,18 +256,18 @@ static int create_worker_threads(struct worker_info *wi, size_t nr_threads)
 	pthread_t thread;
 	int ret;
 
-	pthread_mutex_lock(&wi->startup_lock);
+	pthread_mutex_lock(&wi->workers_lock);
 	while (wi->nr_threads < nr_threads) {
 		ret = pthread_create(&thread, NULL, worker_routine, wi);
 		if (ret != 0) {
 			sd_err("failed to create worker thread: %m");
-			pthread_mutex_unlock(&wi->startup_lock);
+			pthread_mutex_unlock(&wi->workers_lock);
 			return -1;
 		}
 		wi->nr_threads++;
 		sd_debug("create thread %s %zu", wi->name, wi->nr_threads);
 	}
-	pthread_mutex_unlock(&wi->startup_lock);
+	pthread_mutex_unlock(&wi->workers_lock);
 
 	return 0;
 }
@@ -323,10 +332,13 @@ static void *worker_routine(void *arg)
 
 		pthread_mutex_lock(&wi->pending_lock);
 		if (wq_need_shrink(wi)) {
+			pthread_mutex_unlock(&wi->pending_lock);
+
+			pthread_mutex_lock(&wi->workers_lock);
 			wi->nr_threads--;
+			pthread_mutex_unlock(&wi->workers_lock);
 
 			trace_clear_tid_map(tid);
-			pthread_mutex_unlock(&wi->pending_lock);
 			pthread_detach(pthread_self());
 			sd_debug("destroy thread %s %d, %zu", wi->name, tid,
 				 wi->nr_threads);
@@ -415,6 +427,7 @@ struct work_queue *create_work_queue(const char *name,
 	pthread_mutex_init(&wi->finished_lock, NULL);
 	pthread_mutex_init(&wi->pending_lock, NULL);
 	pthread_mutex_init(&wi->startup_lock, NULL);
+	pthread_mutex_init(&wi->workers_lock, NULL);
 
 	ret = create_worker_threads(wi, 1);
 	if (ret < 0)
@@ -424,6 +437,7 @@ struct work_queue *create_work_queue(const char *name,
 
 	return &wi->q;
 destroy_threads:
+	pthread_mutex_unlock(&wi->workers_lock);
 	pthread_mutex_unlock(&wi->startup_lock);
 	pthread_cond_destroy(&wi->pending_cond);
 	pthread_mutex_destroy(&wi->pending_lock);
-- 
1.7.10.4