Initial commit

CMakeLists.txt

+cmake_minimum_required(VERSION 3.10)
+
+# set the project name
+project(Eventloop VERSION 0.0.1)
+
+# add the executable
+add_executable(eventloop main.cpp workers.cpp)
+
+set(CMAKE_CXX_FLAGS "-lpthread")
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED True)

main.cpp

+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <iostream>
+#include <mutex>
+#include <ostream>
+#include <ratio>
+#include <thread>
+
+#include "workers.hpp"
+
+auto helloWorld = [](int i) {
+  std::cout << "Task " << i << "\n";
+ };
+
+void workersTest() {
+  jlworkers::Workers workerThread(16);
+
+  workerThread.post([]() {
+    std::cout << "Hello, World!\n";
+  });
+
+  workerThread.start();
+
+  workerThread.post([]() {
+    // Let's delay this a bit
+    std::this_thread::sleep_for(std::chrono::milliseconds(500));
+    std::cout << "Hello again, World!\n";
+  });
+
+  for (int i = 0; i < 10; i++)
+    workerThread.post([i] {
+      std::cout << "Task " << i
+                << ", thread id " << std::this_thread::get_id() <<"\n";
+    });
+
+  std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+  workerThread.stop();
+}
+
+void stuff() {
+    std::atomic_bool wait(true);
+  std::mutex mtx;
+  std::condition_variable cv;
+
+  std::thread t([&wait, &mtx, &cv] {
+    //while (wait);
+    std::unique_lock<std::mutex> lock(mtx);
+    while (wait)
+      cv.wait(lock);
+    
+    std::cout << "Thread finished" << std::endl;
+  });
+  //  std::this_thread::sleep_for(std::chrono::seconds(1));
+  std::this_thread::sleep_for(std::chrono::seconds(1));
+  {
+    std::unique_lock<std::mutex> lock(mtx);
+    wait = false;
+  }
+  
+  cv.notify_all();
+  t.join();
+}
+
+int main(int argc, const char** argv) {
+
+  workersTest();
+  
+  return 0;
+}

workers.cpp

+#include <condition_variable>
+#include <cstdlib>
+#include <functional>
+#include <iostream>
+#include <mutex>
+#include <thread>
+#include <utility>
+
+#include "workers.hpp"
+
+namespace jlworkers {
+  Workers::Workers() {
+    this->m_running = false;
+    this->m_runningThreadCount = 0;
+    this->m_maxThreadCount = 1;
+  }
+
+  Workers::Workers(int numThreads) {
+    this->m_running = false;
+    this->m_runningThreadCount = 0;
+    // Ensure the number of threads is sane
+    this->m_maxThreadCount = numThreads > 0 ? numThreads : 1;
+  }
+  
+  void Workers::post(const std::function<void()>& f) {
+    //std::cout << "Adding task to queue, size before: " << m_queue.size();
+    std::unique_lock<std::mutex> lock(m_runningMutex);
+    this->m_queue.emplace_back(f);
+    m_runningCondition.notify_all();
+    //std::cout << ", size after: " << m_queue.size() << "\n";
+  }
+
+  void Workers::start() {
+    m_running = true;
+    std::cout << "Starting task runner thread\n";
+    m_runnerThread = std::thread([this] {
+      while (m_running) {
+        std::cout << "Queue: " << m_queue.size() << "\n";
+        if (!m_queue.empty()) {
+          if (m_runningThreadCount < m_maxThreadCount) {
+            // Fetch next task in queue
+            auto f = *m_queue.begin();
+            // Keep track of running threads
+            m_runningThreadCount++;
+            m_queue.pop_front();
+            m_workers.emplace_back(std::thread([this, f] {
+              // Run the task, blocking this thread
+              f();
+              std::unique_lock<std::mutex> lock(m_runningMutex);
+              m_runningThreadCount--;
+              m_runningCondition.notify_all();
+            }));
+          }
+        }
+
+        // Wait for any status updates
+        std::cout << "Create lock and wait for update\n";
+        std::unique_lock<std::mutex> lock(m_runningMutex);
+        m_runningCondition.wait(lock);
+      }
+      std::cout << "Queue empty\n";
+      for (; !m_workers.empty(); m_workers.pop_back()) m_workers.back().join();
+    });
+  }
+
+  void Workers::stop() {
+    /* Sohuld we wait for queue to clear, or do we want to stop adding tasks?
+     * Let's do the latter for now..
+     */
+    std::cout << "Stop runner thread; locking mutex\n";
+
+    // Scope this to avoid deadlock
+    {
+      std::unique_lock<std::mutex> lock(this->m_runningMutex);
+      this->m_running = false;
+      std::cout << "Notify threads\n";
+      m_runningCondition.notify_all();
+    }
+    
+    std::cout << "Wait for runner to join\n";
+    m_runnerThread.join();
+    std::cout << "Runner done!\n";
+  }
+}
+

workers.hpp

+#ifndef __WORKERS_HPP__
+#define __WORKERS_HPP__
+
+#include <atomic>
+#include <condition_variable>
+#include <functional>
+#include <list>
+#include <map>
+#include <mutex>
+#include <thread>
+#include <vector>
+
+namespace jlworkers {
+  class Workers {
+
+  private:
+    void m_workerFunc() {
+      
+    };
+    std::map<std::thread, std::atomic_bool> m_workers_status;
+    int m_maxThreadCount;
+    std::atomic_int m_runningThreadCount;
+
+    std::atomic_bool m_running;
+    std::condition_variable m_runningCondition;
+    std::mutex m_runningMutex;
+
+    std::list<std::function<void()>> m_queue;
+    std::vector<std::thread> m_workers;
+
+    std::thread m_runnerThread;
+
+  public:
+    Workers();
+    Workers(int numThreads);
+
+    void post(const std::function<void ()>&);
+    void start();
+    void stop();
+  };
+}
+
+#endif//__WORKERS_HPP__