mtapi_c: fixed handling of detached tasks in conjunction with retaining queues…

mtapi_c: fixed handling of detached tasks in conjunction with retaining queues and multi-instance tasks

mtapi_c/src/embb_mtapi_scheduler_t.c

@@ -213,6 +213,29 @@ embb_mtapi_thread_context_t * embb_mtapi_scheduler_get_current_thread_context(
   return context;
 }
 
+void embb_mtapi_scheduler_finalize_task(
+  embb_mtapi_task_t * task,
+  embb_mtapi_node_t * node,
+  embb_mtapi_queue_t * queue,
+  embb_mtapi_group_t * group) {
+  /* tell queue that a task is done */
+  if (MTAPI_NULL != queue) {
+    embb_mtapi_queue_task_finished(queue);
+  }
+  /* move task to group queue */
+  if (MTAPI_NULL != group) {
+    embb_mtapi_task_queue_push(&group->queue, task);
+  }
+  /* issue task complete callback if set */
+  if (MTAPI_NULL != task->attributes.complete_func) {
+    task->attributes.complete_func(task->handle, MTAPI_NULL);
+  }
+  /* delete task if detached */
+  if (MTAPI_TRUE == task->attributes.is_detached) {
+    embb_mtapi_task_delete(task, node->task_pool);
+  }
+}
+
 mtapi_boolean_t embb_mtapi_scheduler_execute_task(
   embb_mtapi_task_t * task,
   embb_mtapi_node_t * node,
@@ -254,10 +277,7 @@ mtapi_boolean_t embb_mtapi_scheduler_execute_task(
     embb_mtapi_task_context_initialize_with_thread_context_and_task(
       &task_context, thread_context, task);
     if (embb_mtapi_task_execute(task, &task_context)) {
-      /* tell queue that a task is done */
-      if (MTAPI_NULL != local_queue) {
-        embb_mtapi_queue_task_finished(local_queue);
-      }
+      embb_mtapi_scheduler_finalize_task(task, node, local_queue, local_group);
     }
     result = MTAPI_TRUE;
     break;
@@ -276,13 +296,7 @@ mtapi_boolean_t embb_mtapi_scheduler_execute_task(
     task->error_code = MTAPI_ERR_ACTION_CANCELLED;
     if (embb_atomic_fetch_and_add_unsigned_int(
       &task->instances_todo, (unsigned int)-1) == 0) {
-      /* tell queue that a task is done */
-      if (MTAPI_NULL != local_queue) {
-        embb_mtapi_queue_task_finished(local_queue);
-      }
-      if (MTAPI_NULL != local_group) {
-        embb_mtapi_task_queue_push(&local_group->queue, task);
-      }
+      embb_mtapi_scheduler_finalize_task(task, node, local_queue, local_group);
     }
     if (MTAPI_NULL != local_action) {
       embb_atomic_fetch_and_add_int(&local_action->num_tasks, -1);
@@ -301,11 +315,6 @@ mtapi_boolean_t embb_mtapi_scheduler_execute_task(
     break;
   }
 
-  /* issue task complete callback if set */
-  if (MTAPI_NULL != task->attributes.complete_func) {
-    task->attributes.complete_func(task->handle, MTAPI_NULL);
-  }
-
   return result;
 }
 
@@ -388,9 +397,6 @@ int embb_mtapi_scheduler_worker(void * arg) {
       if (embb_mtapi_scheduler_execute_task(task, node, thread_context)) {
         counter = 0;
       }
-      if (MTAPI_TRUE == task->attributes.is_detached) {
-        embb_mtapi_task_delete(task, node->task_pool);
-      }
     } else if (counter < 1024) {
       /* spin and yield for a while before going to sleep */
       embb_thread_yield();

mtapi_c/src/embb_mtapi_scheduler_t.h

@@ -40,6 +40,7 @@ extern "C" {
 /* ---- FORWARD DECLARATIONS ----------------------------------------------- */
 
 #include <embb_mtapi_queue_t_fwd.h>
+#include <embb_mtapi_group_t_fwd.h>
 #include <embb_mtapi_thread_context_t_fwd.h>
 #include <embb_mtapi_task_t_fwd.h>
 #include <embb_mtapi_node_t_fwd.h>
@@ -138,6 +139,16 @@ embb_mtapi_thread_context_t * embb_mtapi_scheduler_get_current_thread_context(
   embb_mtapi_scheduler_t * that);
 
 /**
+ * Processes finished task.
+ * Notifies associated group and queue and deletes task if it is detached.
+ */
+void embb_mtapi_scheduler_finalize_task(
+  embb_mtapi_task_t * task,
+  embb_mtapi_node_t * node,
+  embb_mtapi_queue_t * queue,
+  embb_mtapi_group_t * group);
+
+/**
  * Executes the given task if the thread context is valid.
  * \memberof embb_mtapi_scheduler_struct
  */

mtapi_c/src/embb_mtapi_task_t.c

@@ -139,14 +139,6 @@ mtapi_boolean_t embb_mtapi_task_execute(
   }
 
   if (todo == 1) {
-    /* is task associated with a group? */
-    if (embb_mtapi_group_pool_is_handle_valid(
-      context->thread_context->node->group_pool, that->group)) {
-      embb_mtapi_group_t* local_group =
-        embb_mtapi_group_pool_get_storage_for_handle(
-        context->thread_context->node->group_pool, that->group);
-      embb_mtapi_task_queue_push(&local_group->queue, that);
-    }
     return MTAPI_TRUE;
   } else {
     return MTAPI_FALSE;