Global MPMC Queue

src/thread_pool_pppcsr/thread_pool_pppcsr.cpp

@@ -19,7 +19,7 @@ using namespace std;
  */
 ThreadPoolPPPCSR::ThreadPoolPPPCSR(const int NUM_OF_THREADS, bool lock_search, uint32_t init_num_nodes,
                                    int partitions_per_domain, bool use_numa)
-    : tasks(NUM_OF_THREADS),
+    : tasks(),
       finished(false),
       available_nodes(std::min(numa_max_node() + 1, NUM_OF_THREADS)),
       indeces(available_nodes, 0),
@@ -52,17 +52,14 @@ ThreadPoolPPPCSR::ThreadPoolPPPCSR(const int NUM_OF_THREADS, bool lock_search, u
 // Finishes when finished is set to true and there are no outstanding tasks
 template <bool isMasterThread>
 void ThreadPoolPPPCSR::execute(const int thread_id) {
-  cout << "Thread " << thread_id << " has " << tasks[thread_id].size() << " tasks, runs on domain "
-       << threadToDomain[thread_id] << endl;
   if (numa_available() >= 0) {
     numa_run_on_node(threadToDomain[thread_id]);
   }
   int registered = -1;
 
-  while (!tasks[thread_id].empty() || (!isMasterThread && !finished)) {
-    if (!tasks[thread_id].empty()) {
-      task t = tasks[thread_id].front();
-      tasks[thread_id].pop();
+  while (!tasks.empty() || (!isMasterThread && !finished)) {
+    if (!tasks.empty()) {
+      task t = tasks.front();
 
       int currentPar = pcsr->get_partiton(t.src);
 
@@ -95,25 +92,19 @@ void ThreadPoolPPPCSR::execute(const int thread_id) {
 // Submit an update for edge {src, target} to thread with number thread_id
 void ThreadPoolPPPCSR::submit_add(int thread_id, int src, int target) {
   (void)thread_id;
-  auto par = pcsr->get_partiton(src) / partitions_per_domain;
-  auto index = (indeces[par]++) % numThreadsDomain[par];
-  tasks[firstThreadDomain[par] + index].push(task{true, false, src, target});
+  tasks.push(task{true, false, src, target});
 }
 
 // Submit a delete edge task for edge {src, target} to thread with number thread_id
 void ThreadPoolPPPCSR::submit_delete(int thread_id, int src, int target) {
   (void)thread_id;
-  auto par = pcsr->get_partiton(src) / partitions_per_domain;
-  auto index = (indeces[par]++) % numThreadsDomain[par];
-  tasks[firstThreadDomain[par] + index].push(task{false, false, src, target});
+  tasks.push(task{false, false, src, target});
 }
 
 // Submit a read neighbourhood task for vertex src to thread with number thread_id
 void ThreadPoolPPPCSR::submit_read(int thread_id, int src) {
   (void)thread_id;
-  auto par = pcsr->get_partiton(src) / partitions_per_domain;
-  auto index = (indeces[par]++) % numThreadsDomain[par];
-  tasks[firstThreadDomain[par] + index].push(task{false, true, src, src});
+  tasks.push(task{false, true, src, src});
 }
 
 // starts a new number of threads

src/thread_pool_pppcsr/thread_pool_pppcsr.h

@@ -8,6 +8,7 @@
 #include <vector>
 
 #include "../pppcsr/PPPCSR.h"
+#include "../utility/concurrentqueue.h"
 #include "task.h"
 
 using namespace std;
@@ -30,7 +31,7 @@ class ThreadPoolPPPCSR {
 
  private:
   vector<thread> thread_pool;
-  vector<queue<task>> tasks;
+  moodycamel::ConcurrentQueue<task> tasks;
   chrono::steady_clock::time_point s;
   chrono::steady_clock::time_point end;
   std::atomic_bool finished;