added consumer

cmake/main.cpp

@@ -13,6 +13,19 @@
 #include "util/dataIo.h"
 #include "blasJoin/blasJoin.h"
 
+#ifndef COUNT_ONLY
+size_t consumer_count;
+
+int consumer(boost::lockfree::queue<join_pair> &queue)
+{
+    join_pair jp;
+    while (queue.pop(jp)){
+        #pragma omp atomic write
+            consumer_count= consumer_count +1;
+        // printf("%lu-%lu\n", jp.p1, jp.p2);
+    }
+}
+#endif
 
 int main(int argc, char** argv) {
     CounterBin hwcounters;
@@ -45,6 +58,14 @@ int main(int argc, char** argv) {
 
     blasJoin( x, N, D, EPS, threads, blocksize, &joinCounts, &hwcounters, queue);
 
+#ifndef COUNT_ONLY
+    // if we materialize with a non-blocking linked list, then joincounts are zero
+    #pragma omp parallel for
+    for ( int i = 0 ; i < threads ; i++ ){
+        consumer(queue);
+    }
+    joinCounts = consumer_count;
+#endif
 
 #pragma omp parallel
 {