working with light testing

Author: etwest

CMakeLists.txt

@@ -81,8 +81,8 @@ add_library(GutterTree
   include/buffer_flusher.h
   src/standalone_gutters.cpp
   include/standalone_gutters.h
-  src/cache_guttering.cpp
-  include/cache_guttering.h
+  src/pht.cpp
+  include/pht.h
   src/numa_pht.cpp
   include/numa_pht.h
   include/types.h)

experiment/cache_exp.cpp

@@ -3,7 +3,7 @@
 #include <chrono>
 #include <atomic>
 #include <fstream>
-#include "../include/cache_guttering.h"
+#include "../include/pht.h"
 
 static bool shutdown = false;
 static constexpr uint32_t prime = 100000007;
@@ -52,7 +52,7 @@ static void run_randomized(const int nodes, const unsigned long updates, const u
               .gutter_bytes(gutter_size)
               .wq_batch_per_elm(wq_batch);
 
-  CacheGuttering *gutters = new CacheGuttering(nodes, num_workers, nthreads, conf);
+  PipelineHyperTree *gutters = new PipelineHyperTree(nodes, num_workers, nthreads, conf);
 
   // create queriers
 #ifndef EARLY_EXIT
@@ -130,7 +130,7 @@ static void run_test(const int nodes, const unsigned long updates, const unsigne
               .num_flushers(2)
               .gutter_bytes(32 * 1024)
               .wq_batch_per_elm(8);
-  CacheGuttering *gutters = new CacheGuttering(nodes, num_workers, nthreads, conf);
+  PipelineHyperTree *gutters = new PipelineHyperTree(nodes, num_workers, nthreads, conf);
 
   // create queriers
 #ifndef EARLY_EXIT

include/numa_pht.h

@@ -0,0 +1,84 @@
+
+#include "guttering_system.h"
+#include "pht.h"
+/**
+ * The idea of NumaPHT is to limit the amount of tree space that each thread touches without
+ * giving up on performance. If updates are uniformly distributed then this is easy, simply divide
+ * the gutter space among the threads evenly and allow each to operate independently.
+ * 
+ * However, if updates are not distributed uniformly, or arrive in a non-uniform order, then this
+ * strategy will reduce performance as only some of the threads will be performing significant
+ * work.
+ * 
+ * TODO: Solve this problem at some point I guess
+ */
+class NumaPHT : public PipelineHyperTree {
+ private:
+  const node_id_t num_gutters;
+
+  const size_t num_inserters;
+  const size_t num_trees;
+  const size_t tree_bits;
+  const size_t vert_bits;
+
+  constexpr static size_t thread_buffer_size = 2048;
+
+  struct ThreadBuffer {
+    size_t size = 0;
+    size_t capacity;
+    update_t *buffer = nullptr;
+
+    ThreadBuffer() : capacity(thread_buffer_size), buffer(new update_t[capacity]) {}
+    ThreadBuffer(ThreadBuffer&& oth) : size(oth.size), capacity(oth.capacity), buffer(oth.buffer) {
+      oth.size = 0;
+      oth.capacity = 0;
+      oth.buffer = nullptr;
+    }
+    ThreadBuffer& operator=(ThreadBuffer&& oth) {
+      size = oth.size;
+      capacity = oth.capacity;
+      buffer = oth.buffer;
+
+      oth.size = 0;
+      oth.capacity = 0;
+      oth.buffer = nullptr;
+
+      return *this;
+    }
+
+    ~ThreadBuffer() { if (buffer != nullptr) delete[] buffer; }
+  };
+
+  // thread buffers for its roots and updates it wants to route to other trees
+  ThreadBuffer *thread_buffers;
+
+  WorkQueue<ThreadBuffer> **tree_queues;
+ public:
+  /**
+   * NumaPHT constructor
+   * 
+   * @param num_gutters     number of gutters to construct
+   * @param num_consumers   number of threads that will be removing leaf gutters from the tree
+   * @param num_inserters   number of threads that will be performing updates
+   * @param num_trees       number of independent trees to divide the gutters into
+   */
+  NumaPHT(node_id_t num_gutters, size_t num_consumers, size_t num_inserters, size_t num_trees,
+          GutteringConfiguration conf);
+
+  ~NumaPHT();
+
+  insert_ret_t insert(const update_t &upd, size_t thr_id);
+
+  insert_ret_t batch_insert(const update_t *batch, size_t num_updates, size_t thr_id);
+
+  insert_ret_t process_stream_upd_batch(const GraphStreamUpdate *batch, size_t num_updates,
+                                        size_t thr_id);
+
+  insert_ret_t force_flush();
+
+  insert_ret_t insert(const update_t &upd) {
+    insert(upd, 0);
+  }
+
+  thread_local static size_t thr_insert_num;
+};

include/cache_guttering.h include/pht.hinclude/cache_guttering.h renamed to include/pht.h

@@ -12,7 +12,8 @@ constexpr int log2_constexpr(size_t num) {
   return power;
 }
 
-class PipelineHyperTree {
+
+class PipelineHyperTree : public GutteringSystem {
  private:
   const size_t inserters;
   const node_id_t num_nodes;
@@ -252,16 +253,14 @@ class PipelineHyperTree {
   friend class InsertThread;
 
   std::vector<InsertThread> insert_threads; // vector of InsertThreads
-  VertexBatchQueue &wq;
  public:
   /**
    * Constructs a new guttering systems using a tree like structure for cache efficiency.
    * @param nodes       number of nodes in the graph.
    * @param workers     the number of workers which will be removing batches
    * @param inserters   the number of inserter buffers
    */
-  PipelineHyperTree(node_id_t nodes, size_t inserters, GutteringConfiguration &conf,
-                    VertexBatchQueue &wq);
+  PipelineHyperTree(node_id_t num_gutters, size_t num_consumers, size_t num_inserters, GutteringConfiguration conf);
 
   ~PipelineHyperTree();
 
@@ -313,59 +312,4 @@ class PipelineHyperTree {
    */
   void print_r_to_l(node_id_t src);
   void print_fanouts();
-
-  // number of batches per work queue element
-  const size_t wq_batch_per_elm;
-  const size_t leaf_gutter_size;
-};
-
-// The CacheGuttering class adds the GutteringSystem base class to the PipelineHyperTree
-class CacheGuttering : public GutteringSystem {
- private:
-  PipelineHyperTree pht;
- public:
-  CacheGuttering(node_id_t nodes, size_t workers, size_t inserters, GutteringConfiguration conf)
-      : GutteringSystem(nodes, workers, conf), pht(nodes, inserters, conf, wq){};
-
-
-  /**
-   * Puts an update into the data structure.
-   * @param upd the edge update.
-   * @param which, which thread is inserting this update
-   * @return nothing.
-   */
-  insert_ret_t insert(const update_t &upd, size_t which) override { 
-    pht.insert(upd, which);
-  }
-
-  insert_ret_t batch_insert(const update_t *batch, size_t num_updates, size_t which) override {
-    pht.batch_insert(batch, num_updates, which);
-  }
-
-  insert_ret_t process_stream_upd_batch(const GraphStreamUpdate *batch, size_t num_updates,
-                                        size_t which) override {
-    pht.process_stream_upd_batch(batch, num_updates, which);
-  }
-
-  // pure virtual functions don't like default params, so default to 'which' of 0
-  insert_ret_t insert(const update_t &upd) { pht.insert(upd); }
-
-  /**
-   * Flushes all pending buffers. When this function returns there are no more updates in the
-   * guttering system
-   * @return nothing.
-   */
-  flush_ret_t force_flush() {
-    pht.force_flush();
-  }
-
-  /**
-   * Set the "offset" for incoming edges. That is, if we set an offset of x, an incoming edge
-   * {i,j} will be stored internally as an edge {i - x, j}. Use only for integration with
-   * distributed guttering. If you don't know what that means, don't use this function!
-   * 
-   * @param offset 
-   * @return a reference to the parent PipelineHyperTree object.
-   */
-  void set_offset(node_id_t offset) { pht.set_offset(offset); }
 };

include/work_queue.h

@@ -63,6 +63,9 @@ class WorkQueue {
     consumer_list_lock.unlock();
   }
 
+  WorkQueue(WorkQueue &) = delete;
+  WorkQueue& operator=(WorkQueue &) = delete;
+
   /**
    * TODO: Rewrite this description
    * Initialize the queue pointers to point at actual data instead of nullptrs
@@ -89,12 +92,16 @@ class WorkQueue {
    * @param push_data   the data the user wants to add to the queue. When this function returns,
    *                    this reference will hold the data that was in the "empty" queue node it replaced
    */
-  void push(T &push_data) {
+  bool push(T &push_data, bool block = true) {
     std::unique_lock<std::mutex> lk(producer_list_lock);
-    producer_condition.wait(lk, [this]{return !full();});
+    producer_condition.wait(lk, [this, block]{return !full() || !block;});
 
     // printf("WQ: Push:\n");
     // print();
+    if (full()) {
+      lk.unlock();
+      return false;
+    }
 
     // remove head from produce_list
     DataNode *node = producer_list;
@@ -110,18 +117,19 @@ class WorkQueue {
     consumer_list = node;
     consumer_list_lock.unlock();
     consumer_condition.notify_one();
+    return true;
   }
 
   /**
    * Get data from the queue for processing
    * @param data   where to place the Data
    * @return  true if we were able to get good data, false otherwise
    */
-  bool pop(DataNode *&data) {
+  bool pop(DataNode *&data, bool block = true) {
     // wait while queue is empty
     // printf("waiting to peek\n");
     std::unique_lock<std::mutex> lk(consumer_list_lock);
-    consumer_condition.wait(lk, [this]{return !empty() || non_block;});
+    consumer_condition.wait(lk, [this, block]{return !empty() || !block || non_block;});
 
     // printf("WQ: Peek\n");
     // print();