Merge branch 'embb413_network_plugin_improvements' into development

d83a8571 · Marcus Winter · a7e94398 · a5832fc1 · d83a8571 · d83a8571
Commit d83a8571 authored Apr 19, 2016 by Marcus Winter
8 changed files
--- a/mtapi_c/src/embb_mtapi_scheduler_t.c
+++ b/mtapi_c/src/embb_mtapi_scheduler_t.c
@@ -40,6 +40,7 @@
 #include <embb_mtapi_action_t.h>
 #include <embb_mtapi_alloc.h>
 #include <embb_mtapi_queue_t.h>
+#include <embb_mtapi_group_t.h>
 /* ---- CLASS MEMBERS ------------------------------------------------------ */
@@ -293,6 +294,8 @@ int embb_mtapi_scheduler_worker(void * arg) {
    /* check if there was work */
    if (MTAPI_NULL != task) {
      embb_mtapi_queue_t * local_queue = MTAPI_NULL;
+      embb_mtapi_group_t * local_group = MTAPI_NULL;
+      embb_mtapi_action_t * local_action = MTAPI_NULL;
      /* is task associated with a queue? */
      if (embb_mtapi_queue_pool_is_handle_valid(
@@ -302,6 +305,21 @@ int embb_mtapi_scheduler_worker(void * arg) {
            node->queue_pool, task->queue);
      }
+      /* is task associated with a group? */
+      if (embb_mtapi_group_pool_is_handle_valid(
+        node->group_pool, task->group)) {
+        local_group =
+          embb_mtapi_group_pool_get_storage_for_handle(
+            node->group_pool, task->group);
+      }
+      if (embb_mtapi_action_pool_is_handle_valid(
+        node->action_pool, task->action)) {
+        local_action =
+          embb_mtapi_action_pool_get_storage_for_handle(
+            node->action_pool, task->action);
+      }
      switch (embb_atomic_load_int(&task->state)) {
      case MTAPI_TASK_SCHEDULED:
      /* multi-instance task, another instance might be running */
@@ -328,7 +346,7 @@ int embb_mtapi_scheduler_worker(void * arg) {
        break;
      case MTAPI_TASK_CANCELLED:
-        /* set return value to canceled */
+        /* set return value to cancelled */
        task->error_code = MTAPI_ERR_ACTION_CANCELLED;
        if (embb_atomic_fetch_and_add_unsigned_int(
          &task->instances_todo, (unsigned int)-1) == 0) {
@@ -336,6 +354,12 @@ int embb_mtapi_scheduler_worker(void * arg) {
          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);
+          }
+        }
+        if (MTAPI_NULL != local_action) {
+          embb_atomic_fetch_and_add_int(&local_action->num_tasks, -1);
        }
        break;

--- a/mtapi_c/src/embb_mtapi_task_t.c
+++ b/mtapi_c/src/embb_mtapi_task_t.c
@@ -501,7 +501,6 @@ void mtapi_task_cancel(
    if (embb_mtapi_task_pool_is_handle_valid(node->task_pool, task)) {
      embb_mtapi_task_t* local_task =
        embb_mtapi_task_pool_get_storage_for_handle(node->task_pool, task);
-      embb_mtapi_task_set_state(local_task, MTAPI_TASK_CANCELLED);
      /* call plugin action cancel function */
      if (embb_mtapi_action_pool_is_handle_valid(
@@ -511,8 +510,14 @@ void mtapi_task_cancel(
          node->action_pool, local_task->action);
        if (local_action->is_plugin_action) {
          local_action->plugin_task_cancel_function(task, &local_status);
+        } else {
+          embb_mtapi_task_set_state(local_task, MTAPI_TASK_CANCELLED);
+          local_task->error_code = MTAPI_ERR_ACTION_CANCELLED;
+          local_status = MTAPI_SUCCESS;
        }
      } else {
+        embb_mtapi_task_set_state(local_task, MTAPI_TASK_CANCELLED);
+        local_task->error_code = MTAPI_ERR_ACTION_CANCELLED;
        local_status = MTAPI_SUCCESS;
      }
    } else {

--- a/mtapi_plugins_c/mtapi_network_c/src/embb_mtapi_network.c
+++ b/mtapi_plugins_c/mtapi_network_c/src/embb_mtapi_network.c
--- a/mtapi_plugins_c/mtapi_network_c/src/embb_mtapi_network_buffer.c
+++ b/mtapi_plugins_c/mtapi_network_c/src/embb_mtapi_network_buffer.c
@@ -28,9 +28,10 @@
 #include <embb/base/c/memory_allocation.h>
 #include <string.h>
-void embb_mtapi_network_buffer_initialize(
+int embb_mtapi_network_buffer_initialize(
  embb_mtapi_network_buffer_t * that,
  int capacity) {
+  int result = 1;
  that->position = 0;
  that->size = 0;
  that->data = (char*)embb_alloc((size_t)capacity);
@@ -38,7 +39,9 @@ void embb_mtapi_network_buffer_initialize(
    that->capacity = capacity;
  } else {
    that->capacity = 0;
+    result = 0;
  }
+  return result;
 }
 void embb_mtapi_network_buffer_finalize(

--- a/mtapi_plugins_c/mtapi_network_c/src/embb_mtapi_network_buffer.h
+++ b/mtapi_plugins_c/mtapi_network_c/src/embb_mtapi_network_buffer.h
@@ -43,7 +43,7 @@ struct embb_mtapi_network_buffer_struct {
 typedef struct embb_mtapi_network_buffer_struct embb_mtapi_network_buffer_t;
-void embb_mtapi_network_buffer_initialize(
+int embb_mtapi_network_buffer_initialize(
  embb_mtapi_network_buffer_t * that,
  int capacity
 );

--- a/mtapi_plugins_c/mtapi_network_c/src/embb_mtapi_network_socket.c
+++ b/mtapi_plugins_c/mtapi_network_c/src/embb_mtapi_network_socket.c
@@ -116,7 +116,8 @@ int embb_mtapi_network_socket_connect(
  if (SOCKET_ERROR == connect(that->handle, (struct sockaddr *)&addr,
    sizeof(addr))) {
 #ifdef _WIN32
-    if (WSAEWOULDBLOCK != WSAGetLastError())
+    int err = WSAGetLastError();
+    if (WSAEWOULDBLOCK != err)
 #else
    if (EAGAIN != errno)
 #endif

--- a/mtapi_plugins_c/mtapi_network_c/test/embb_mtapi_network_test_task.cc
+++ b/mtapi_plugins_c/mtapi_network_c/test/embb_mtapi_network_test_task.cc
@@ -61,13 +61,52 @@ static void test(
  }
 }
+static void cancel_test(
+  void const * /*arguments*/,
+  mtapi_size_t /*arguments_size*/,
+  void * /*result_buffer*/,
+  mtapi_size_t /*result_buffer_size*/,
+  void const * /*node_local_data*/,
+  mtapi_size_t /*node_local_data_size*/,
+  mtapi_task_context_t * context) {
+  mtapi_status_t status;
+  while (true) {
+    mtapi_task_state_t state = mtapi_context_taskstate_get(context, &status);
+    if (status != MTAPI_SUCCESS) {
+      break;
+    } else {
+      if (state == MTAPI_TASK_CANCELLED) {
+        break;
+      }
+    }
+  }
+}
 NetworkTaskTest::NetworkTaskTest() {
-  CreateUnit("mtapi network task test").Add(&NetworkTaskTest::TestBasic, this);
+  CreateUnit("mtapi network task test")
+    .Add(&NetworkTaskTest::TestBasic, this);
 }
 void NetworkTaskTest::TestBasic() {
  mtapi_status_t status;
+  mtapi_initialize(
+    NETWORK_DOMAIN,
+    NETWORK_LOCAL_NODE,
+    MTAPI_NULL,
+    MTAPI_NULL,
+    &status);
+  MTAPI_CHECK_STATUS(status);
+  TestSimple();
+  TestCancel();
+  mtapi_finalize(&status);
+  MTAPI_CHECK_STATUS(status);
+}
+void NetworkTaskTest::TestSimple() {
+  mtapi_status_t status;
  mtapi_job_hndl_t job;
  mtapi_task_hndl_t task;
  mtapi_action_hndl_t network_action, local_action;
@@ -81,14 +120,6 @@ void NetworkTaskTest::TestBasic() {
    arguments[ii + kElements] = static_cast<float>(ii);
  }
-  mtapi_initialize(
-    NETWORK_DOMAIN,
-    NETWORK_LOCAL_NODE,
-    MTAPI_NULL,
-    MTAPI_NULL,
-    &status);
-  MTAPI_CHECK_STATUS(status);
  mtapi_network_plugin_initialize("127.0.0.1", 12345, 5,
    kElements * 4 * 3 + 32, &status);
  MTAPI_CHECK_STATUS(status);
@@ -139,7 +170,68 @@ void NetworkTaskTest::TestBasic() {
  mtapi_network_plugin_finalize(&status);
  MTAPI_CHECK_STATUS(status);
+}
+void NetworkTaskTest::TestCancel() {
+  mtapi_status_t status;
+  mtapi_job_hndl_t job;
+  mtapi_task_hndl_t task;
+  mtapi_action_hndl_t network_action, local_action;
+  float argument = 1.0f;
+  float result;
+  mtapi_network_plugin_initialize("127.0.0.1", 12345, 5,
+    4 * 3 + 32, &status);
+  MTAPI_CHECK_STATUS(status);
-  mtapi_finalize(&status);
+  float node_remote = 1.0f;
+  local_action = mtapi_action_create(
+    NETWORK_REMOTE_JOB,
+    cancel_test,
+    &node_remote, sizeof(float),
+    MTAPI_DEFAULT_ACTION_ATTRIBUTES,
+    &status);
+  MTAPI_CHECK_STATUS(status);
+  network_action = mtapi_network_action_create(
+    NETWORK_DOMAIN,
+    NETWORK_LOCAL_JOB,
+    NETWORK_REMOTE_JOB,
+    "127.0.0.1", 12345,
+    &status);
+  MTAPI_CHECK_STATUS(status);
+  status = MTAPI_ERR_UNKNOWN;
+  job = mtapi_job_get(NETWORK_LOCAL_JOB, NETWORK_DOMAIN, &status);
+  MTAPI_CHECK_STATUS(status);
+  task = mtapi_task_start(
+    MTAPI_TASK_ID_NONE,
+    job,
+    &argument, sizeof(float),
+    &result, sizeof(float),
+    MTAPI_DEFAULT_TASK_ATTRIBUTES,
+    MTAPI_GROUP_NONE,
+    &status);
+  MTAPI_CHECK_STATUS(status);
+  mtapi_task_wait(task, 1, &status);
+  PT_ASSERT_EQ(status, MTAPI_TIMEOUT);
+  mtapi_task_cancel(task, &status);
+  MTAPI_CHECK_STATUS(status);
+  mtapi_task_wait(task, MTAPI_INFINITE, &status);
+  PT_ASSERT_NE(status, MTAPI_TIMEOUT);
+  PT_ASSERT_EQ(status, MTAPI_ERR_ACTION_CANCELLED);
+  mtapi_action_delete(network_action, MTAPI_INFINITE, &status);
+  MTAPI_CHECK_STATUS(status);
+  mtapi_action_delete(local_action, MTAPI_INFINITE, &status);
+  MTAPI_CHECK_STATUS(status);
+  mtapi_network_plugin_finalize(&status);
  MTAPI_CHECK_STATUS(status);
 }
--- a/mtapi_plugins_c/mtapi_network_c/test/embb_mtapi_network_test_task.h
+++ b/mtapi_plugins_c/mtapi_network_c/test/embb_mtapi_network_test_task.h
@@ -35,6 +35,9 @@ class NetworkTaskTest : public partest::TestCase {
 private:
  void TestBasic();
+  void TestSimple();
+  void TestCancel();
 };
 #endif // MTAPI_PLUGINS_C_MTAPI_NETWORK_C_TEST_EMBB_MTAPI_NETWORK_TEST_TASK_H_