World: split Emerger into its own class

src/World.cpp

@@ -10,6 +10,7 @@
 #include "Game.hpp"
 #include "Universe.hpp"
 #include "util/Log.hpp"
+#include "world/Emerger.hpp"
 
 namespace diggler {
 
@@ -21,67 +22,12 @@ static const char *TAG = "World";
 World::World(Game *G, WorldId id, bool remote) :
   G(G), id(id), isRemote(remote) {
   // TODO: emerger thread setting, default to std::thread::hardware_concurrency()
-  for (int i=0; i < 2; ++i) {
-    emergerThreads.emplace_back(&World::emergerProc, this, i);
-  }
+  emerger.reset(new world::Emerger);
 }
 
 World::~World() {
-  emergerRun = false;
-  emergerCondVar.notify_all();
-  for (std::thread &t : emergerThreads) {
-    if (t.joinable())
-      t.join();
-  }
-  std::unique_lock<std::mutex> lk(emergeQueueMutex);
 }
 
-void World::addToEmergeQueue(ChunkRef &cr) {
-  { std::unique_lock<std::mutex> lk(emergeQueueMutex);
-    emergeQueue.emplace(cr);
-  }
-  emergerCondVar.notify_one();
-}
-void World::addToEmergeQueue(ChunkWeakRef &cwr) {
-  { std::unique_lock<std::mutex> lk(emergeQueueMutex);
-    emergeQueue.emplace(cwr);
-  }
-  emergerCondVar.notify_one();
-}
-
-void World::emergerProc(int emergerId) {
-  ChunkRef c;
-  emergerRun = true;
-  while (emergerRun) {
-    { std::unique_lock<std::mutex> lk(emergeQueueMutex);
-      if (emergeQueue.size() == 0) {
-        // No more chunks to emerge, wait for more
-        emergerCondVar.wait(lk);
-        if (emergeQueue.size() == 0) // Weird threading shenanigans
-          continue;
-        if (!emergerRun)
-          break;
-      }
-      c = emergeQueue.front().lock();
-      emergeQueue.pop();
-      if (!c) {
-        continue; // Chunk was not referenced anymore
-      }
-    }
-
-    // TODO: loading
-    auto genStart = std::chrono::high_resolution_clock::now();
-    CaveGenerator::GenConf gc;
-    CaveGenerator::Generate(c->getWorld(), gc, c);
-    auto genEnd = std::chrono::high_resolution_clock::now();
-    auto genDelta = std::chrono::duration_cast<std::chrono::milliseconds>(genEnd - genStart);
-    glm::ivec3 cp = c->getWorldChunkPos();
-    Log(Verbose, TAG) << "Map gen for " << id << '.' << cp.x << ',' << cp.y << ',' << cp.z <<
-      " took " << genDelta.count() << "ms, thread #" << emergerId;
-
-    c.reset(); // Release ref ownership
-  }
-}
 
 ChunkRef World::getNewEmptyChunk(int cx, int cy, int cz) {
   ChunkRef c = std::make_shared<Chunk>(G, G->U->getWorld(id), cx, cy, cz);
@@ -99,7 +45,7 @@ ChunkRef World::getChunk(int cx, int cy, int cz, bool load) {
   }
   if (load) {
     ChunkRef c = getNewEmptyChunk(cx, cy, cz);
-    addToEmergeQueue(c);
+    emerger->queue(c);
     return c;
   }
   return ChunkRef();

src/World.hpp

@@ -4,6 +4,7 @@
 #include <condition_variable>
 #include <functional>
 #include <map>
+#include <memory>
 #include <mutex>
 #include <queue>
 #include <thread>
@@ -23,6 +24,10 @@ class InMessage;
 class OutMessage;
 }
 
+namespace world {
+class Emerger;
+}
+
 using WorldId = int;
 struct WorldChunkMapSorter {
   constexpr bool operator()(const glm::ivec3 &lhs, const glm::ivec3 &rhs) const {
@@ -49,14 +54,7 @@ private:
 
   Game *G;
 
-  std::queue<ChunkWeakRef> emergeQueue;
-  std::mutex emergeQueueMutex;
-  void addToEmergeQueue(ChunkRef&);
-  void addToEmergeQueue(ChunkWeakRef&);
-  bool emergerRun;
-  std::vector<std::thread> emergerThreads;
-  std::condition_variable emergerCondVar;
-  void emergerProc(int);
+  std::unique_ptr<world::Emerger> emerger;
 
 public:
   using WorldChunkMap::size;

src/util/Worker.hpp

@@ -0,0 +1,61 @@
+#ifndef DIGGLER_UTIL_WORKER
+#define DIGGLER_UTIL_WORKER
+
+#include <atomic>
+#include <mutex>
+#include <thread>
+
+namespace diggler {
+namespace util {
+
+template<typename>
+class WorkerPoolBase;
+
+template<typename TaskT>
+class Worker {
+public:
+  using WorkerPoolBaseT = WorkerPoolBase<TaskT>;
+  friend WorkerPoolBaseT;
+
+  using Task = TaskT;
+
+private:
+  std::atomic_bool keepRunning;
+
+protected:
+  unsigned id;
+  std::thread thread;
+
+  void run(WorkerPoolBaseT &pool) {
+    keepRunning.store(true, std::memory_order_release);
+    while (keepRunning.load(std::memory_order_relaxed)) {
+      std::unique_lock<std::recursive_mutex> lk(pool.taskQueueMutex);
+      if (pool.taskQueue.size() == 0) {
+        // No more items to process, wait for more
+        pool.taskQueueCondVar.wait(lk);
+        if (!keepRunning.load(std::memory_order_relaxed))
+          break;
+        if (pool.taskQueue.size() == 0)  // Spurious wake-up
+          continue;
+      }
+      Task task(std::move(pool.taskQueue.front()));
+      pool.taskQueue.pop();
+      lk.unlock();
+      processTask(std::move(task));
+    }
+  }
+
+public:
+  Worker(unsigned id) :
+    id(id) {
+  }
+
+  virtual ~Worker() {}
+
+  virtual void processTask(Task&&) = 0;
+};
+
+}
+}
+
+#endif /* DIGGLER_UTIL_WORKER */

src/util/WorkerPool.hpp

@@ -0,0 +1,97 @@
+#ifndef DIGGLER_UTIL_WORKER_POOL
+#define DIGGLER_UTIL_WORKER_POOL
+
+#include <atomic>
+#include <condition_variable>
+#include <list>
+#include <mutex>
+#include <queue>
+#include <type_traits>
+#include <utility>
+
+#include "Worker.hpp"
+
+namespace diggler {
+namespace util {
+
+template<typename TaskT>
+class WorkerPoolBase {
+public:
+  using Task = TaskT;
+
+protected:
+  friend Worker<Task>;
+
+  std::queue<Task> taskQueue;
+  std::recursive_mutex taskQueueMutex;
+  std::condition_variable_any taskQueueCondVar;
+
+  void addTask(const Task &task) {
+    {
+      std::unique_lock<std::recursive_mutex> lk(taskQueueMutex);
+      taskQueue.push(task);
+    }
+    taskQueueCondVar.notify_one();
+  }
+
+  void addTask(Task &&task) {
+    {
+      std::unique_lock<std::recursive_mutex> lk(taskQueueMutex);
+      taskQueue.push(std::move(task));
+    }
+    taskQueueCondVar.notify_one();
+  }
+
+  void startWorkerThread(Worker<Task> &worker) {
+    if (!worker.thread.joinable()) {
+      worker.thread = std::thread([](WorkerPoolBase *self, Worker<Task> &worker) {
+        worker.run(*self);
+      }, this, std::ref(worker));
+    }
+  }
+
+  void markWorkerStop(Worker<Task> &worker) {
+    worker.keepRunning.store(false, std::memory_order_release);
+  }
+
+  void stopWorkerThread(Worker<Task> &worker) {
+    if (worker.thread.joinable()) {
+      worker.thread.join();
+    }
+  }
+};
+
+template<class WorkerT, typename = std::enable_if_t<
+    std::is_base_of<Worker<typename WorkerT::Task>, WorkerT>::value>>
+class WorkerPool : public WorkerPoolBase<typename WorkerT::Task> {
+public:
+  using Task = typename WorkerT::Task;
+
+protected:
+  friend WorkerT;
+
+  std::list<WorkerT> workers;
+
+public:
+  WorkerPool(unsigned nWorkers = 1) {
+    for (unsigned i = 0; i < nWorkers; ++i) {
+      workers.emplace_back(i);
+      this->startWorkerThread(workers.back());
+    }
+  }
+
+  ~WorkerPool() {
+    for (WorkerT &worker : workers) {
+      this->markWorkerStop(worker);
+    }
+    this->taskQueueCondVar.notify_all();
+    for (WorkerT &worker : workers) {
+      this->stopWorkerThread(worker);
+    }
+  }
+};
+
+}
+}
+
+#endif /* DIGGLER_UTIL_WORKER_POOL */

src/world/CMakeLists.txt

@@ -0,0 +1,4 @@
+set(CSD ${CMAKE_CURRENT_SOURCE_DIR})
+diggler_add_sources(
+  ${CSD}/Emerger.cpp
+)

src/world/Emerger.cpp

@@ -0,0 +1,34 @@
+#include "Emerger.hpp"
+
+#include "../CaveGenerator.hpp"
+#include "../util/Log.hpp"
+
+namespace diggler {
+
+using Util::Log;
+using namespace Util::Logging::LogLevels;
+
+namespace world {
+
+Emerger::Worker::~Worker() {
+}
+
+void Emerger::Worker::processTask(Task &&cwr) {
+  ChunkRef c = cwr.lock();
+  if (!c) {  // Chunk was not referenced anymore
+    return;
+  }
+
+  // TODO: loading
+  auto genStart = std::chrono::high_resolution_clock::now();
+  CaveGenerator::GenConf gc;
+  CaveGenerator::Generate(c->getWorld(), gc, c);
+  auto genEnd = std::chrono::high_resolution_clock::now();
+  auto genDelta = std::chrono::duration_cast<std::chrono::milliseconds>(genEnd - genStart);
+  glm::ivec3 cp = c->getWorldChunkPos();
+  Log(Verbose, "Emerger") << "Map gen for " << c->getWorld()->id << '.' <<
+    cp.x << ',' << cp.y << ',' << cp.z << " took " << genDelta.count() << "ms, thread #" << id;
+}
+
+}
+}

src/world/Emerger.hpp

@@ -0,0 +1,40 @@
+#ifndef DIGGLER_WORLD_EMERGER_HPP
+#define DIGGLER_WORLD_EMERGER_HPP
+
+#include "../Chunk.hpp"
+#include "../util/WorkerPool.hpp"
+
+namespace diggler {
+namespace world {
+
+class Emerger final {
+private:
+  class Worker : public util::Worker<ChunkWeakRef> {
+  public:
+    using util::Worker<ChunkWeakRef>::Worker;
+
+    ~Worker();
+
+    void processTask(Task&&) final override;
+  };
+
+  class Pool : public util::WorkerPool<Worker> {
+  public:
+    inline void queue(const ChunkWeakRef &c) {
+      addTask(c);
+    }
+  } pool;
+
+public:
+  inline void queue(const ChunkRef &c) {
+    pool.queue(c);
+  }
+  inline void queue(const ChunkWeakRef &c) {
+    pool.queue(c);
+  }
+};
+
+}
+}
+
+#endif /* DIGGLER_WORLD_EMERGER_HPP */