[Bf-blender-cvs] [4a3dc6233e6] temp-compositor-scheduling: Fix: Loosing Workerthreads During Execution.

[Jeroen Bakker]

Commit: 4a3dc6233e64e8461fdc0368052d355f8710678c
Author: Jeroen Bakker
Date:   Fri Apr 2 09:35:13 2021 +0200
Branches: temp-compositor-scheduling
https://developer.blender.org/rB4a3dc6233e64e8461fdc0368052d355f8710678c

Fix: Loosing Workerthreads During Execution.

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

M	source/blender/compositor/COM_defines.h
M	source/blender/compositor/intern/COM_ExecutionSystem.cc
M	source/blender/compositor/intern/COM_WorkScheduler.cc

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

diff --git a/source/blender/compositor/COM_defines.h b/source/blender/compositor/COM_defines.h
index 590234d0562..9e3f99cd49d 100644
--- a/source/blender/compositor/COM_defines.h
+++ b/source/blender/compositor/COM_defines.h
@@ -83,18 +83,18 @@ enum class CompositorPriority {
 
 enum class eSchedulingMode {
   /**
-   * Scheduling mode where inputs (render layers/images) are scheduled and when finished the next
-   * work will be scheduled where all] inputs are finished.
+   * Input based scheduling mode. Scheduling starts from input nodes. When a work package is
+   * finished, other work packages are scheduled where all inputs have been executed..
    */
   InputToOutput,
 
   /**
    * Scheduling mode where outputs are scheduled when all its inputs have been completed. When
-   * inputs aren't completed it tries to schedule these inputs.
+   * inputs aren't completed it tries to schedule these inputs recursivly.
    */
   OutputToInput,
 };
-static constexpr eSchedulingMode COM_SCHEDULING_MODE = eSchedulingMode::InputToOutput;
+static constexpr eSchedulingMode COM_SCHEDULING_MODE = eSchedulingMode::OutputToInput;
 
 /**
  * \brief the execution state of a chunk in an ExecutionGroup
diff --git a/source/blender/compositor/intern/COM_ExecutionSystem.cc b/source/blender/compositor/intern/COM_ExecutionSystem.cc
index 88bb313f095..cb2c05c5217 100644
--- a/source/blender/compositor/intern/COM_ExecutionSystem.cc
+++ b/source/blender/compositor/intern/COM_ExecutionSystem.cc
@@ -270,19 +270,42 @@ void ExecutionSystem::execute()
   }
 }
 
+static bool is_completed(Vector<ExecutionGroup *> &groups)
+{
+  for (ExecutionGroup *group : groups) {
+    for (WorkPackage &work_package : group->get_work_packages()) {
+      if (work_package.state != eChunkExecutionState::Executed) {
+        return false;
+      }
+    }
+  }
+  return true;
+}
+
+static void wait_for_completion(Vector<ExecutionGroup *> &groups)
+{
+  /* Todo: check for break! */
+  while (!is_completed(groups)) {
+    PIL_sleep_ms(100);
+  }
+}
+
 void ExecutionSystem::execute_groups(CompositorPriority priority)
 {
   switch (COM_SCHEDULING_MODE) {
     case eSchedulingMode::InputToOutput: {
       const bNodeTree *bnodetree = this->m_context.getbNodeTree();
+      Vector<ExecutionGroup *> groups;
       for (ExecutionGroup *execution_group : m_groups) {
         if (execution_group->get_flags().is_output &&
             execution_group->getRenderPriority() == priority) {
           execution_group->set_btree(bnodetree);
           mark_priority(execution_group->get_work_packages(), priority);
+          groups.append(execution_group);
         }
       }
       schedule_root_work_packages(m_groups);
+      wait_for_completion(groups);
       break;
     }
     case eSchedulingMode::OutputToInput: {
diff --git a/source/blender/compositor/intern/COM_WorkScheduler.cc b/source/blender/compositor/intern/COM_WorkScheduler.cc
index 209e39255af..2fdca10e67d 100644
--- a/source/blender/compositor/intern/COM_WorkScheduler.cc
+++ b/source/blender/compositor/intern/COM_WorkScheduler.cc
@@ -66,8 +66,15 @@ constexpr bool COM_is_opencl_enabled()
   return COM_threading_model() != ThreadingModel::SingleThreaded;
 }
 
+enum class eWorkSchedulerState {
+  Idle,
+  Running,
+  Stopping,
+};
+
 static ThreadLocal(CPUDevice *) g_thread_device;
 static struct {
+  eWorkSchedulerState state = eWorkSchedulerState::Idle;
   struct {
     /** \brief list of all CPUDevices. for every hardware thread an instance of CPUDevice is
      * created
@@ -114,13 +121,14 @@ static void CL_CALLBACK clContextError(const char *errinfo,
 
 static void *thread_execute_gpu(void *data)
 {
-  Device *device = (Device *)data;
+  OpenCLDevice *device = (OpenCLDevice *)data;
   WorkPackage *work;
-
-  while ((work = (WorkPackage *)BLI_thread_queue_pop(g_work_scheduler.opencl.queue))) {
-    device->execute(work);
+  while (g_work_scheduler.state == eWorkSchedulerState::Running) {
+    work = (WorkPackage *)BLI_thread_queue_pop(g_work_scheduler.opencl.queue);
+    if (work) {
+      device->execute(work);
+    }
   }
-
   return nullptr;
 }
 
@@ -318,10 +326,12 @@ static void *threading_model_queue_execute(void *data)
   CPUDevice *device = (CPUDevice *)data;
   WorkPackage *work;
   BLI_thread_local_set(g_thread_device, device);
-  while ((work = (WorkPackage *)BLI_thread_queue_pop(g_work_scheduler.queue.queue))) {
-    device->execute(work);
+  while (g_work_scheduler.state == eWorkSchedulerState::Running) {
+    work = (WorkPackage *)BLI_thread_queue_pop(g_work_scheduler.queue.queue);
+    if (work) {
+      device->execute(work);
+    }
   }
-
   return nullptr;
 }
 
@@ -432,6 +442,10 @@ static void threading_model_task_stop()
 
 void WorkScheduler::schedule(WorkPackage *package)
 {
+  BLI_assert(g_work_scheduler.state != eWorkSchedulerState::Idle);
+  if (g_work_scheduler.state == eWorkSchedulerState::Stopping) {
+    return;
+  }
   // TODO: race condition..... we should add a mutex to the work package.
   if (package->state != eChunkExecutionState::NotScheduled) {
     return;
@@ -464,6 +478,9 @@ void WorkScheduler::schedule(WorkPackage *package)
 
 void WorkScheduler::start(CompositorContext &context)
 {
+  BLI_assert(g_work_scheduler.state == eWorkSchedulerState::Idle);
+  g_work_scheduler.state = eWorkSchedulerState::Running;
+
   if (COM_is_opencl_enabled()) {
     opencl_start(context);
   }
@@ -485,6 +502,7 @@ void WorkScheduler::start(CompositorContext &context)
 
 void WorkScheduler::finish()
 {
+  BLI_assert(g_work_scheduler.state == eWorkSchedulerState::Running);
   if (COM_is_opencl_enabled()) {
     opencl_finish();
   }
@@ -506,6 +524,8 @@ void WorkScheduler::finish()
 
 void WorkScheduler::stop()
 {
+  BLI_assert(g_work_scheduler.state == eWorkSchedulerState::Running);
+  g_work_scheduler.state = eWorkSchedulerState::Stopping;
   if (COM_is_opencl_enabled()) {
     opencl_stop();
   }
@@ -523,6 +543,7 @@ void WorkScheduler::stop()
       threading_model_task_stop();
       break;
   }
+  g_work_scheduler.state = eWorkSchedulerState::Idle;
 }
 
 bool WorkScheduler::has_gpu_devices()
@@ -535,6 +556,7 @@ bool WorkScheduler::has_gpu_devices()
 
 void WorkScheduler::initialize(bool use_opencl, int num_cpu_threads)
 {
+  g_work_scheduler.state = eWorkSchedulerState::Idle;
   if (COM_is_opencl_enabled()) {
     opencl_initialize(use_opencl);
   }
@@ -556,6 +578,7 @@ void WorkScheduler::initialize(bool use_opencl, int num_cpu_threads)
 
 void WorkScheduler::deinitialize()
 {
+  BLI_assert(g_work_scheduler.state == eWorkSchedulerState::Idle);
   if (COM_is_opencl_enabled()) {
     opencl_deinitialize();
   }