WIP commit

Threads/Inc/threads_actor.hpp

@@ -18,12 +18,46 @@
 #include "threads_primitivethread.hpp"
 #include "threads_signal.hpp"
 #include "threads_timer.hpp"
+#include "threads_actor_thread.hpp"
 
 namespace Threads
 {
 
+class IActor
+{
+public:
+  IActor() = delete;
+  IActor(ActorThread* thread)
+  {
+    moveToThread(thread);
+  }
+
+  virtual ~IActor()
+  { }
+
+  virtual void init()
+  { }
+
+  virtual void doWork() = 0;
+  virtual bool hasWork() = 0;
+
+  void moveToThread(ActorThread* thread)
+  {
+    assert(thread);
+
+    if (thread_)
+      thread_->removeActor_(this);
+
+    thread->acceptActor_(this);
+    thread_ = thread;
+  }
+
+protected:
+  ActorThread* thread_ = nullptr;
+};
+
 template<typename CRTP, typename... Ts>
-class Actor
+class Actor : public IActor
 {
 public:
   using value_type = std::variant<Ts...>;
@@ -31,8 +65,8 @@ public:
   static_assert(std::is_default_constructible_v<value_type>, "Ts must be default constructible.");
   static_assert(std::is_trivially_copyable_v<value_type>, "Ts must be trivially copyable.");
 
-  Actor(const std::uint32_t queue_size, const char* name, const osPriority_t priority, const std::uint32_t stack_size)
+  Actor(const std::uint32_t queue_size, ActorThread* thread)
-      : thread_(&Actor<CRTP, Ts...>::threadHandler_, this, name, priority, stack_size),
+      : IActor(thread),
         msg_queue_(xQueueCreate(queue_size, sizeof(value_type))), signal_(this)
   {
     assert(msg_queue_);
@@ -40,6 +74,23 @@ public:
 
   virtual ~Actor() {}
 
+  void doWork() override
+  {
+    value_type data{};
+
+    const BaseType_t got = xQueueReceive(msg_queue_, &data, 0);
+
+    if (got == pdTRUE)
+    {
+      std::visit(Visitor_{this}, data);
+    }
+  }
+
+  bool hasWork() override
+  {
+    return uxQueueMessagesWaiting(msg_queue_) > 0;
+  }
+
   template<typename T>
   struct Signal : Signallable<T>
   {
@@ -84,11 +135,7 @@ public:
     return &signal_;
   }
 
-protected:
-  virtual void init() = 0;
-
 private:
-  PrimitiveThread thread_;
   QueueHandle_t msg_queue_;
 
   struct Visitor_
@@ -108,44 +155,17 @@ private:
     }
   };
 
-  void operator()()
-  {
-    init();
-
-    value_type data{};
-
-    while (true)
-    {
-      const BaseType_t got = xQueueReceive(msg_queue_, &data, portMAX_DELAY);
-
-      if (got == pdTRUE)
-      {
-        std::visit(Visitor_{this}, data);
-      }
-    }
-  }
-
-  static void threadHandler_(void* d)
-  {
-    auto* dd = static_cast<Actor<CRTP, Ts...>*>(d);
-    (*dd)();
-  }
-
   Signal<value_type> signal_;
 };
 
 template<typename CRTP>
-class Actor<CRTP, void>
+class Actor<CRTP, void> : public IActor
 {
 public:
   using value_type = void;
 
-  Actor(const std::uint32_t queue_size, const char* name, const osPriority_t priority, const std::uint32_t stack_size)
+  Actor(const std::uint32_t queue_size, ActorThread* thread)
-      : thread_([](void* d) {
+      : IActor(thread),
-          auto* dd = static_cast<Actor<CRTP, value_type>*>(d);
-          (*dd)();
-        },
-                this, name, priority, stack_size),
         msg_queue_(xQueueCreate(queue_size, sizeof(int))), signal_(*this)
   {
     assert(msg_queue_);
@@ -153,27 +173,11 @@ public:
 
   virtual ~Actor() {}
 
-  Signallable<value_type>* getSignal()
+  void doWork() override
   {
-    return &signal_;
+    int data{};
-  }
 
-protected:
+    const BaseType_t got = xQueueReceive(msg_queue_, &data, 0);
-  virtual void init() = 0;
-
-private:
-  PrimitiveThread thread_;
-  QueueHandle_t msg_queue_;
-
-  void operator()()
-  {
-    init();
-
-    int data;
-
-    while (true)
-    {
-      const BaseType_t got = xQueueReceive(msg_queue_, &data, portMAX_DELAY);
 
     if (got == pdTRUE)
     {
@@ -181,8 +185,20 @@ private:
       crtp->onSignal();
     }
   }
+
+  bool hasWork() override
+  {
+    return uxQueueMessagesWaiting(msg_queue_) > 0;
   }
 
+  Signallable<value_type>* getSignal()
+  {
+    return &signal_;
+  }
+
+private:
+  QueueHandle_t msg_queue_;
+
   friend struct Signal_;
 
   struct Signal_ : Signallable<value_type>

Threads/Inc/threads_actor_thread.hpp

@@ -0,0 +1,43 @@
+/*
+ * threads_actor_thread.hpp
+ *
+ *  Created on: Sep 19, 2021
+ *      Author: erki
+ */
+
+#ifndef THREADS_ACTOR_THREAD_HPP_
+#define THREADS_ACTOR_THREAD_HPP_
+
+#include <threads_primitivethread.hpp>
+
+namespace Threads
+{
+
+class IActor;
+
+class ActorThread : PrimitiveThread
+{
+public:
+  ActorThread() = delete;
+  ActorThread(const ActorThread&) = delete;
+  ActorThread(ActorThread&&) = delete;
+
+  ActorThread(const char* name, const osPriority_t priority, const std::uint32_t stack_size)
+    : PrimitiveThread(&ActorThread::run_, this, name, priority, stack_size)
+  { }
+
+private:
+  friend class IActor;
+
+  static void run_(void* data);
+
+  void acceptActor_(IActor* actor);
+  void removeActor_(IActor* actor);
+
+  std::array<IActor*, 32> actors_;
+};
+
+}
+
+
+#endif /* THREADS_ACTOR_THREAD_HPP_ */

Threads/Src/threads_actor_thread.cpp

@@ -0,0 +1,69 @@
+/*
+ * threads_actor_thread.cpp
+ *
+ *  Created on: Sep 25, 2021
+ *      Author: erki
+ */
+
+#include "threads_actor_thread.hpp"
+
+#include <algorithm>
+
+#include "threads_actor.hpp"
+
+namespace Threads
+{
+
+void ActorThread::run_(void* data)
+{
+  auto* t = reinterpret_cast<ActorThread*>(data);
+
+  for (IActor* p : t->actors_)
+  {
+    if (p)
+      p->init();
+  }
+
+  while (true)
+  {
+    bool remainingWork = false;
+
+    for (IActor* p : t->actors_)
+    {
+      if (p && p->hasWork())
+      {
+        p->doWork();
+        remainingWork = remainingWork || p->hasWork();
+        t->yield();
+      }
+    }
+
+    if (!remainingWork)
+    {
+      const bool unused = t->notifyWait(portMAX_DELAY);
+      (void)unused;
+    }
+  }
+}
+
+void ActorThread::acceptActor_(IActor* actor)
+{
+  auto it = std::find(actors_.begin(), actors_.end(), nullptr);
+
+  if (it == actors_.end())
+    return;
+
+  *it = actor;
+}
+
+void ActorThread::removeActor_(IActor* actor)
+{
+  auto it = std::find(actors_.begin(), actors_.end(), actor);
+
+  if (it == actors_.end())
+    return;
+
+  *it = nullptr;
+}
+
+}