[Bf-blender-cvs] [0c451bc530] unlock_task_scheduler: Attempt to address nearly-starving cases.

[Bastien Montagne]

Commit: 0c451bc530c9b7dd5431f8c2b5203c6f37339aa6
Author: Bastien Montagne
Date:   Fri Dec 23 10:50:45 2016 +0100
Branches: unlock_task_scheduler
https://developer.blender.org/rB0c451bc530c9b7dd5431f8c2b5203c6f37339aa6

Attempt to address nearly-starving cases.

Idea here is to reduce number of threads a pool is allowed to work on,
in case it does not get tasks quickly enough.

This does not seem to be really great result (have to only do the checks
once every 200 tasks pushed to avoid too much overhead), but I cannot
reproduce that nearly-starving case here so far. @sergey, curious if it
gives any difference on your 12cores with 14_03_G?

===================================================================

M	source/blender/blenlib/BLI_task.h
M	source/blender/blenlib/intern/task.c

===================================================================

diff --git a/source/blender/blenlib/BLI_task.h b/source/blender/blenlib/BLI_task.h
index d27bf4dad2..0de0dec770 100644
--- a/source/blender/blenlib/BLI_task.h
+++ b/source/blender/blenlib/BLI_task.h
@@ -97,7 +97,7 @@ void BLI_task_pool_work_and_wait(TaskPool *pool);
 void BLI_task_pool_cancel(TaskPool *pool);
 
 /* set number of threads allowed to be used by this pool */
-void BLI_pool_set_num_threads(TaskPool *pool, int num_threads);
+void BLI_pool_set_num_threads(TaskPool *pool, size_t num_threads_max);
 
 /* for worker threads, test if canceled */
 bool BLI_task_pool_canceled(TaskPool *pool);
diff --git a/source/blender/blenlib/intern/task.c b/source/blender/blenlib/intern/task.c
index fd14b7becd..4d8becf8e0 100644
--- a/source/blender/blenlib/intern/task.c
+++ b/source/blender/blenlib/intern/task.c
@@ -35,8 +35,6 @@
 #include "BLI_task.h"
 #include "BLI_threads.h"
 
-#include "PIL_time.h"
-
 #include "atomic_ops.h"
 
 /* Define this to enable some detailed statistic print. */
@@ -112,6 +110,7 @@ struct TaskPool {
 	TaskScheduler *scheduler;
 
 	size_t num;
+	size_t num_threads_max;
 	size_t num_threads;
 	size_t currently_running_tasks;
 
@@ -301,8 +300,10 @@ BLI_INLINE bool task_find(
 				continue;
 			}
 
-			if (atomic_add_and_fetch_z(&current_pool->currently_running_tasks, 1) <= current_pool->num_threads ||
-			    is_main || current_pool->num_threads == 0)
+			/* Order is important, we do not want to increase currently_running_tasks if we are in main thread
+			 * (run_and_wait)! */
+			if (is_main ||
+			    atomic_add_and_fetch_z(&current_pool->currently_running_tasks, 1) <= current_pool->num_threads)
 			{
 				*task = current_task;
 				found_task = true;
@@ -583,7 +584,9 @@ static TaskPool *task_pool_create_ex(TaskScheduler *scheduler, void *userdata, c
 
 	pool->scheduler = scheduler;
 	pool->num = 0;
-	pool->num_threads = 0;
+	/* Do not use BLI_task_scheduler_num_threads(scheduler) here, we want number of workers, without main thred! */
+	pool->num_threads_max = scheduler->num_threads;
+	pool->num_threads = pool->num_threads_max;
 	pool->currently_running_tasks = 0;
 	pool->do_cancel = false;
 	pool->run_in_background = is_background;
@@ -670,12 +673,14 @@ void BLI_task_pool_free(TaskPool *pool)
 
 	BLI_end_threaded_malloc();
 }
-
+#include "PIL_time_utildefines.h"
 static void task_pool_push(
         TaskPool *pool, TaskRunFunction run, void *taskdata,
         bool free_taskdata, TaskFreeFunction freedata, TaskPriority priority,
         int thread_id)
 {
+	static int i = 1;
+
 	Task *task = task_alloc(pool, thread_id);
 
 	task->run = run;
@@ -685,6 +690,23 @@ static void task_pool_push(
 	task->pool = pool;
 
 	task_scheduler_push(pool->scheduler, task, priority);
+#if 1
+	if ((i++ % 200) == 0) {
+		const size_t min_threads = 1;  /* At least one worker. */
+//		printf("%s: %lu, %lu  ->  ", __func__, pool->num, pool->num_threads);
+		if (pool->num < pool->num_threads / 2 && pool->num_threads > min_threads) {
+			if (atomic_sub_and_fetch_z(&pool->num_threads, 1) < min_threads) {
+				pool->num_threads = min_threads;
+			}
+		}
+		else if (pool->num > pool->num_threads && pool->num_threads < pool->num_threads_max) {
+			if (atomic_add_and_fetch_z(&pool->num_threads, 1) > pool->num_threads_max) {
+				pool->num_threads = pool->num_threads_max;
+			}
+		}
+//		printf("%lu\n", pool->num_threads);
+	}
+#endif
 }
 
 void BLI_task_pool_push_ex(
@@ -728,7 +750,7 @@ void BLI_task_pool_work_and_wait(TaskPool *pool)
 			/* delete task */
 			task_free(pool, task, 0);
 
-			atomic_sub_and_fetch_z(&pool->currently_running_tasks, 1);
+//			atomic_sub_and_fetch_z(&pool->currently_running_tasks, 1);
 
 			/* notify pool task was done */
 			task_pool_num_decrease(pool, 1);
@@ -745,10 +767,13 @@ void BLI_task_pool_work_and_wait(TaskPool *pool)
 	}
 }
 
-void BLI_pool_set_num_threads(TaskPool *pool, int num_threads)
+void BLI_pool_set_num_threads(TaskPool *pool, size_t num_threads_max)
 {
-	/* NOTE: Don't try to modify threads while tasks are running! */
-	pool->num_threads = num_threads;
+	pool->num_threads_max = num_threads_max;
+
+	for (size_t num_threads_old = pool->num_threads;
+	     atomic_cas_z(&pool->num_threads, num_threads_old, min_ii(num_threads_old, num_threads_max)) != num_threads_old;
+	     num_threads_old = pool->num_threads);
 }
 
 void BLI_task_pool_cancel(TaskPool *pool)