Merge branch 'development' into embb531_codesonar_fixes

base_c/include/embb/base/c/internal/atomic/atomic_variables.h

@@ -39,7 +39,7 @@
   EMBB_ATOMIC_PARAMETER_ATOMIC_TYPE_SUFFIX) \
   typedef struct \
 { \
-  EMBB_ATOMIC_PARAMETER_TYPE_NATIVE internal_variable; \
+  volatile EMBB_ATOMIC_PARAMETER_TYPE_NATIVE internal_variable; \
 } EMBB_CAT2(embb_atomic_, EMBB_ATOMIC_PARAMETER_ATOMIC_TYPE_SUFFIX);
 
 EMBB_ATOMIC_INTERNAL_DEFINE_VARIABLE(char, char)

base_c/include/embb/base/c/log.h

@@ -28,6 +28,7 @@
 #define EMBB_BASE_C_LOG_H_
 
 #include <embb/base/c/internal/config.h>
+#include <stdarg.h>
 
 /**
  * \defgroup C_LOG Logging
@@ -197,6 +198,13 @@ void embb_log_error(
                                             \c message */
   );
 
+/* function for internal use only */
+void embb_log_write_internal(
+  char const * channel,
+  embb_log_level_t log_level,
+  char const * message,
+  va_list argp);
+
 #ifdef __cplusplus
 }
 #endif

base_c/include/embb/base/c/mutex.h

@@ -179,6 +179,9 @@ int embb_spin_init(
 /**
  * Spins until the spinlock can be locked and locks it.
  *
+ * \note This method yields the current thread in regular,
+ *       implementation-defined intervals.
+ *
  * \pre \c spinlock is initialized \n
  * \post If successful, \c spinlock is locked.
  * \return EMBB_SUCCESS if spinlock could be locked. \n

base_c/src/mutex.c

@@ -25,6 +25,7 @@
  */
 
 #include <embb/base/c/mutex.h>
+#include <embb/base/c/thread.h>
 #include <assert.h>
 
 #include <embb/base/c/internal/unused.h>
@@ -125,10 +126,15 @@ int embb_spin_init(embb_spinlock_t* spinlock) {
 
 int embb_spin_lock(embb_spinlock_t* spinlock) {
   int expected = 0;
+  int spins = 1;
 
   // try to swap the
   while (0 == embb_atomic_compare_and_swap_int(
     &spinlock->atomic_spin_variable_, &expected, 1)) {
+    if (0 == (spins & 1023)) {
+      embb_thread_yield();
+    }
+    spins++;
     // reset expected, as CAS might change it...
     expected = 0;
   }

base_cpp/include/embb/base/mutex.h

@@ -192,6 +192,9 @@ class Spinlock {
   /**
    * Waits until the spinlock can be locked and locks it.
    *
+   * \note This method yields the current thread in regular,
+   *       implementation-defined intervals.
+   *
    * \pre The spinlock is not locked by the current thread.
    * \post The spinlock is locked.
    * \threadsafe

base_cpp/src/log.cc

@@ -27,14 +27,6 @@
 #include <embb/base/log.h>
 #include <embb/base/c/internal/unused.h>
 
-#include <cstdarg>
-
-extern "C" void embb_log_write_internal(
-  char const * channel,
-  embb_log_level_t log_level,
-  char const * message,
-  va_list argp);
-
 namespace embb {
 namespace base {
 

base_cpp/test/log_test.cc

@@ -55,10 +55,18 @@ void LogTest::Test() {
   Log::SetLogLevel(EMBB_LOG_LEVEL_TRACE);
   logged_message = null;
   Log::Trace("chn", test_msg);
+#ifdef EMBB_DEBUG
   PT_EXPECT(0 == strcmp(logged_message, "[chn] - [TRACE] hello"));
+#else
+  PT_EXPECT_EQ(null, logged_message);
+#endif
   logged_message = null;
   Log::Info("chn", test_msg);
+#ifdef EMBB_DEBUG
   PT_EXPECT(0 == strcmp(logged_message, "[chn] - [INFO ] hello"));
+#else
+  PT_EXPECT_EQ(null, logged_message);
+#endif
   logged_message = null;
   Log::Warning("chn", test_msg);
   PT_EXPECT(0 == strcmp(logged_message, "[chn] - [WARN ] hello"));
@@ -72,7 +80,11 @@ void LogTest::Test() {
   PT_EXPECT_EQ(null, logged_message);
   logged_message = null;
   Log::Info("chn", test_msg);
+#ifdef EMBB_DEBUG
   PT_EXPECT(0 == strcmp(logged_message, "[chn] - [INFO ] hello"));
+#else
+  PT_EXPECT_EQ(null, logged_message);
+#endif
   logged_message = null;
   Log::Warning("chn", test_msg);
   PT_EXPECT(0 == strcmp(logged_message, "[chn] - [WARN ] hello"));

dataflow_cpp/include/embb/dataflow/internal/process.h

@@ -102,8 +102,9 @@ class Process< Slices, Serial, Inputs<Slices, I1, I2, I3, I4, I5>,
   }
 
   template <typename T>
-  void operator >> (T & target) {
+  T & operator >> (T & target) {
     GetOutput<0>() >> target.template GetInput<0>();
+    return target;
   }
 
   virtual void OnClock(int clock) {

dataflow_cpp/include/embb/dataflow/internal/select.h

@@ -105,8 +105,9 @@ class Select
   }
 
   template <typename T>
-  void operator >> (T & target) {
+  T & operator >> (T & target) {
     GetOutput<0>() >> target.template GetInput<0>();
+    return target;
   }
 
   virtual void OnClock(int clock) {

dataflow_cpp/include/embb/dataflow/internal/source.h

@@ -83,8 +83,9 @@ class Source< Slices, Outputs<Slices, O1, O2, O3, O4, O5> >
   }
 
   template <typename T>
-  void operator >> (T & target) {
+  T & operator >> (T & target) {
     GetOutput<0>() >> target.template GetInput<0>();
+    return target;
   }
 
  private:

dataflow_cpp/include/embb/dataflow/internal/source_executor.h

@@ -53,7 +53,9 @@ class SourceExecutor< Outputs<Slices, O1> > {
     Outputs<Slices, O1> & outputs) {
     O1 o1;
     bool result = function_(o1);
-    outputs.template Get<0>().Send(Signal<O1>(clock, o1));
+    if (result) {
+      outputs.template Get<0>().Send(Signal<O1>(clock, o1));
+    }
     return result;
   }
 
@@ -78,8 +80,10 @@ class SourceExecutor< Outputs<Slices, O1, O2> > {
     O1 o1;
     O2 o2;
     bool result = function_(o1, o2);
-    outputs.template Get<0>().Send(Signal<O1>(clock, o1));
-    outputs.template Get<1>().Send(Signal<O2>(clock, o2));
+    if (result) {
+      outputs.template Get<0>().Send(Signal<O1>(clock, o1));
+      outputs.template Get<1>().Send(Signal<O2>(clock, o2));
+    }
     return result;
   }
 
@@ -106,9 +110,11 @@ class SourceExecutor< Outputs<Slices, O1, O2, O3> > {
     O2 o2;
     O3 o3;
     bool result = function_(o1, o2, o3);
-    outputs.template Get<0>().Send(Signal<O1>(clock, o1));
-    outputs.template Get<1>().Send(Signal<O2>(clock, o2));
-    outputs.template Get<2>().Send(Signal<O3>(clock, o3));
+    if (result) {
+      outputs.template Get<0>().Send(Signal<O1>(clock, o1));
+      outputs.template Get<1>().Send(Signal<O2>(clock, o2));
+      outputs.template Get<2>().Send(Signal<O3>(clock, o3));
+    }
     return result;
   }
 
@@ -137,10 +143,12 @@ class SourceExecutor< Outputs<Slices, O1, O2, O3, O4> > {
     O3 o3;
     O4 o4;
     bool result = function_(o1, o2, o3, o4);
-    outputs.template Get<0>().Send(Signal<O1>(clock, o1));
-    outputs.template Get<1>().Send(Signal<O2>(clock, o2));
-    outputs.template Get<2>().Send(Signal<O3>(clock, o3));
-    outputs.template Get<3>().Send(Signal<O4>(clock, o4));
+    if (result) {
+      outputs.template Get<0>().Send(Signal<O1>(clock, o1));
+      outputs.template Get<1>().Send(Signal<O2>(clock, o2));
+      outputs.template Get<2>().Send(Signal<O3>(clock, o3));
+      outputs.template Get<3>().Send(Signal<O4>(clock, o4));
+    }
     return result;
   }
 
@@ -172,11 +180,13 @@ class SourceExecutor< Outputs<Slices, O1, O2, O3, O4, O5> > {
     O4 o4;
     O5 o5;
     bool result = function_(o1, o2, o3, o4, o5);
-    outputs.template Get<0>().Send(Signal<O1>(clock, o1));
-    outputs.template Get<1>().Send(Signal<O2>(clock, o2));
-    outputs.template Get<2>().Send(Signal<O3>(clock, o3));
-    outputs.template Get<3>().Send(Signal<O4>(clock, o4));
-    outputs.template Get<4>().Send(Signal<O5>(clock, o5));
+    if (result) {
+      outputs.template Get<0>().Send(Signal<O1>(clock, o1));
+      outputs.template Get<1>().Send(Signal<O2>(clock, o2));
+      outputs.template Get<2>().Send(Signal<O3>(clock, o3));
+      outputs.template Get<3>().Send(Signal<O4>(clock, o4));
+      outputs.template Get<4>().Send(Signal<O5>(clock, o5));
+    }
     return result;
   }
 

dataflow_cpp/include/embb/dataflow/internal/switch.h

@@ -103,8 +103,9 @@ class Switch
   }
 
   template <typename T>
-  void operator >> (T & target) {
+  T & operator >> (T & target) {
     GetOutput<0>() >> target.template GetInput<0>();
+    return target;
   }
 
   virtual void OnClock(int clock) {

dataflow_cpp/test/dataflow_cpp_test_simple.cc

@@ -56,12 +56,16 @@ embb::base::Atomic<int> source_counter;
 int source_array[TEST_COUNT];
 
 bool sourceFunc(int & out) {
-  out = source_counter;
+  if (source_counter < TEST_COUNT) {
+    out = source_counter;
 
-  source_array[source_counter] = out;
-  source_counter++;
+    source_array[source_counter] = out;
+    source_counter++;
 
-  return source_counter < TEST_COUNT;
+    return true;
+  } else {
+    return false;
+  }
 }
 
 embb::base::Atomic<int> pred_counter;
@@ -188,11 +192,14 @@ void SimpleTest::TestBasic() {
 
     source.GetOutput<0>() >> sw.GetInput<1>();
 
-    source.GetOutput<0>() >> pred.GetInput<0>();
-    pred.GetOutput<0>() >> sw.GetInput<0>();
+    // connection chain representing the commented single connections below
+    source >> pred >> sw >> filter;
+
+    //source.GetOutput<0>() >> pred.GetInput<0>();
+    //pred.GetOutput<0>() >> sw.GetInput<0>();
     pred.GetOutput<0>() >> sel.GetInput<0>();
 
-    sw.GetOutput<0>() >> filter.GetInput<0>();
+    //sw.GetOutput<0>() >> filter.GetInput<0>();
     filter.GetOutput<0>() >> sel.GetInput<1>();
 
     constant.GetOutput<0>() >> mult.GetInput<0>();

doc/examples/dataflow/dataflow_connect-snippet.h

-  read >> replace;
-  replace >> write;
+  read >> replace >> write;

doc/examples/dataflow/dataflow_producer-snippet.h

@@ -3,10 +3,14 @@ class Producer {
  public:
   explicit Producer(int seed) : seed_(seed), count_(4) {}
   bool Run(T& x) {
-    // produce a new value x
-    x = SimpleRand(seed_);
-    count_--;
-    return count_ >= 0;
+    if (count_ >= 0) {
+      // produce a new value x
+      x = SimpleRand(seed_);
+      count_--;
+      return true;
+    } else {
+      return false;
+    }
   }
 
  private:

doc/examples/dataflow/dataflow_source_function-snippet.h

 bool SourceFunction(std::string & str) {
-  std::getline(file, str);
-  return !file.eof();
+  if (!file.eof()) {
+    std::getline(file, str);
+    return true;
+  } else {
+    return false;
+  }
 }

mtapi_c/src/embb_mtapi_scheduler_t.c

@@ -398,7 +398,8 @@ mtapi_boolean_t embb_mtapi_scheduler_wait_for_task(
     node->scheduler);
 
   /* now wait and schedule new tasks if we are on a worker */
-  mtapi_task_state_t task_state = embb_atomic_load_int(&task->state);
+  mtapi_task_state_t task_state =
+    (mtapi_task_state_t)embb_atomic_load_int(&task->state);
   while (
     (MTAPI_TASK_SCHEDULED == task_state) ||
     (MTAPI_TASK_RUNNING == task_state) ||
@@ -418,7 +419,7 @@ mtapi_boolean_t embb_mtapi_scheduler_wait_for_task(
       node,
       context);
 
-    task_state = embb_atomic_load_int(&task->state);
+    task_state = (mtapi_task_state_t)embb_atomic_load_int(&task->state);
   }
 
   return MTAPI_TRUE;

mtapi_c/src/embb_mtapi_task_context_t.c

@@ -187,7 +187,8 @@ mtapi_task_state_t mtapi_context_taskstate_get(
         &(task_context->thread_context->tss_id));
 
     if (local_context == task_context->thread_context) {
-      task_state = embb_atomic_load_int(&task_context->task->state);
+      task_state = (mtapi_task_state_t)embb_atomic_load_int(
+        &task_context->task->state);
       local_status = MTAPI_SUCCESS;
     } else {
       local_status = MTAPI_ERR_CONTEXT_OUTOFCONTEXT;

mtapi_plugins_c/mtapi_opencl_c/src/embb_mtapi_opencl.c

@@ -174,31 +174,40 @@ static void opencl_task_start(
 
         err = clSetKernelArg(opencl_action->kernel, 0, sizeof(cl_mem),
           (const void*)&opencl_task->arguments);
-        err = clSetKernelArg(opencl_action->kernel, 1, sizeof(cl_int),
+        err |= clSetKernelArg(opencl_action->kernel, 1, sizeof(cl_int),
           (const void*)&opencl_task->arguments_size);
 
-        err = clSetKernelArg(opencl_action->kernel, 2, sizeof(cl_mem),
+        err |= clSetKernelArg(opencl_action->kernel, 2, sizeof(cl_mem),
           (const void*)&opencl_task->result_buffer);
-        err = clSetKernelArg(opencl_action->kernel, 3, sizeof(cl_int),
+        err |= clSetKernelArg(opencl_action->kernel, 3, sizeof(cl_int),
           (const void*)&opencl_task->result_buffer_size);
 
-        err = clEnqueueWriteBuffer(plugin->command_queue,
+        err |= clEnqueueWriteBuffer(plugin->command_queue,
           opencl_task->arguments, CL_FALSE, 0,
           (size_t)opencl_task->arguments_size, local_task->arguments,
           0, NULL, NULL);
-        err = clEnqueueNDRangeKernel(plugin->command_queue,
-          opencl_action->kernel, 1, NULL,
-          &global_work_size, &opencl_action->local_work_size, 0, NULL, NULL);
-        err = clEnqueueReadBuffer(plugin->command_queue,
-          opencl_task->result_buffer, CL_FALSE, 0,
-          (size_t)opencl_task->result_buffer_size, local_task->result_buffer,
-          0, NULL, &opencl_task->kernel_finish_event);
-        err = clSetEventCallback(opencl_task->kernel_finish_event,
-          CL_COMPLETE, opencl_task_complete, opencl_task);
-        err = clFlush(plugin->command_queue);
-
-        embb_mtapi_task_set_state(local_task, MTAPI_TASK_RUNNING);
-        local_status = MTAPI_SUCCESS;
+
+        if (CL_SUCCESS == err) {
+          embb_mtapi_task_set_state(local_task, MTAPI_TASK_RUNNING);
+
+          err |= clEnqueueNDRangeKernel(plugin->command_queue,
+            opencl_action->kernel, 1, NULL,
+            &global_work_size, &opencl_action->local_work_size, 0, NULL, NULL);
+          err |= clEnqueueReadBuffer(plugin->command_queue,
+            opencl_task->result_buffer, CL_FALSE, 0,
+            (size_t)opencl_task->result_buffer_size, local_task->result_buffer,
+            0, NULL, &opencl_task->kernel_finish_event);
+          err |= clSetEventCallback(opencl_task->kernel_finish_event,
+            CL_COMPLETE, opencl_task_complete, opencl_task);
+        }
+
+        err |= clFlush(plugin->command_queue);
+        if (CL_SUCCESS != err) {
+          embb_mtapi_task_set_state(local_task, MTAPI_TASK_ERROR);
+          local_status = MTAPI_ERR_ACTION_FAILED;
+        } else {
+          local_status = MTAPI_SUCCESS;
+        }
       }
     }
   }

tasks_cpp/include/embb/tasks/node.h

@@ -29,6 +29,7 @@
 
 #include <list>
 #include <embb/base/core_set.h>
+#include <embb/base/mutex.h>
 #include <embb/mtapi/c/mtapi.h>
 #include <embb/tasks/action.h>
 #include <embb/tasks/task.h>
@@ -233,6 +234,8 @@ class Node {
   mtapi_action_hndl_t action_handle_;
   std::list<Queue*> queues_;
   std::list<Group*> groups_;
+  embb::base::Spinlock queue_lock_;
+  embb::base::Spinlock group_lock_;
 };
 
 } // namespace tasks

tasks_cpp/src/execution_policy.cc

@@ -35,7 +35,7 @@ namespace tasks {
 
 ExecutionPolicy::ExecutionPolicy() :
     priority_(DefaultPriority) {
-#if MTAPI_CPP_AUTOMATIC_INITIALIZE
+#if TASKS_CPP_AUTOMATIC_INITIALIZE
   Node::GetInstance(); // MTAPI has to be initialized
 #endif
   mtapi_status_t status;
@@ -48,7 +48,7 @@ ExecutionPolicy::ExecutionPolicy() :
 
 ExecutionPolicy::ExecutionPolicy(bool initial_affinity, mtapi_uint_t priority)
 :priority_(priority) {
-#if MTAPI_CPP_AUTOMATIC_INITIALIZE
+#if TASKS_CPP_AUTOMATIC_INITIALIZE
   Node::GetInstance(); // MTAPI has to be initialized
 #endif
   mtapi_status_t status;
@@ -62,7 +62,7 @@ ExecutionPolicy::ExecutionPolicy(bool initial_affinity, mtapi_uint_t priority)
 
 ExecutionPolicy::ExecutionPolicy(mtapi_uint_t priority)
 :priority_(priority) {
-#if MTAPI_CPP_AUTOMATIC_INITIALIZE
+#if TASKS_CPP_AUTOMATIC_INITIALIZE
   Node::GetInstance(); // MTAPI has to be initialized
 #endif
   mtapi_status_t status;
@@ -75,7 +75,7 @@ ExecutionPolicy::ExecutionPolicy(mtapi_uint_t priority)
 
 ExecutionPolicy::ExecutionPolicy(bool initial_affinity)
 :priority_(DefaultPriority) {
-#if MTAPI_CPP_AUTOMATIC_INITIALIZE
+#if TASKS_CPP_AUTOMATIC_INITIALIZE
   Node::GetInstance(); // MTAPI has to be initialized
 #endif
   mtapi_status_t status;

tasks_cpp/src/node.cc

@@ -31,6 +31,7 @@
 
 #include <embb/base/memory_allocation.h>
 #include <embb/base/exceptions.h>
+#include <embb/base/thread.h>
 #include <embb/tasks/tasks.h>
 #if TASKS_CPP_AUTOMATIC_INITIALIZE
 #include <embb/base/mutex.h>
@@ -237,7 +238,11 @@ void Node::Finalize() {
 
 Group & Node::CreateGroup() {
   Group * group = embb::base::Allocation::New<Group>();
+  while (!group_lock_.TryLock(1024)) {
+      embb::base::Thread::CurrentYield();
+  }
   groups_.push_back(group);
+  group_lock_.Unlock();
   return *group;
 }
 
@@ -252,7 +257,11 @@ void Node::DestroyGroup(Group & group) {
 
 Queue & Node::CreateQueue(mtapi_uint_t priority, bool ordered) {
   Queue * queue = embb::base::Allocation::New<Queue>(priority, ordered);
+  while (!queue_lock_.TryLock(1024)) {
+      embb::base::Thread::CurrentYield();
+  }
   queues_.push_back(queue);
+  queue_lock_.Unlock();
   return *queue;
 }