[WIP] Support Raco User-defined Aggregates

applications/join/Aggregates.hpp

@@ -1,13 +1,38 @@
 #pragma once
+#include <algorithm>
 
 namespace Aggregates {
   template < typename State, typename UV >
-    State SUM(State sofar, UV nextval) {
+    State SUM(const State& sofar, const UV& nextval) {
       return sofar + nextval;
     }
 
   template < typename State, typename UV >
-    State COUNT(State sofar, UV nextval) {
+    State COUNT(const State& sofar, const UV& nextval) {
       return sofar + 1;
     }
+
+  // keep MAX macro from being used here
+#pragma push_macro("MAX")
+#undef MAX
+  template <typename State, typename UV >
+    State MAX(const State& sofar, const UV& nextval) {
+      return std::max(sofar, nextval); 
+    }
+#pragma pop_macro("MAX")
+  // keep MIN macro from being used here
+#pragma push_macro("MIN")
+#undef MIN
+  template <typename State, typename UV >
+    State MIN(const State& sofar, const UV& nextval) {
+      return std::min(sofar, nextval); 
+    }
+#pragma pop_macro("MIN")
+
+
+  template <typename N>
+    N Zero() {
+      return 0;
+    }
+
 }

applications/join/CMakeLists.txt

@@ -7,6 +7,7 @@ set(QUERYLIB_SOURCES
   MatchesDHT.hpp
   MatchesDHT.cpp
   DoubleDHT.hpp
+  DoubleDHT.cpp
   Hypercube.hpp
   Hypercube.cpp
   local_graph.cpp
@@ -23,17 +24,27 @@ set(QUERYLIB_SOURCES
   stats.cpp
   MapReduce.cpp
   MapReduce.hpp
-  HashJoin.hpp
-  HashJoin.cpp
   Aggregates.hpp
   DHT_symmetric.hpp
+  DHT_symmetric.cpp
+  DHT_symmetric_generic.hpp
+  Operators.hpp
 )
+#FIXME: these MapReduce Hash joins belong in the above sources
+  #HashJoin.hpp
+  #HashJoin.cpp
+
+
 set(QUERYIO_SOURCES
   relation_io.hpp
   relation_io.cpp
+  block_distribution.hpp
+  block_distribution.cpp
   Tuple.hpp
   Tuple.cpp
   relation.hpp
+  json/json.h
+  jsoncpp.cpp
 )
 
 set(INCLUDE_DIRS
@@ -98,7 +109,6 @@ endforeach()
 # exe targets for generated query codes
 foreach(query ${GENERATED_SOURCES})
   get_filename_component(query_name ${query}, NAME_WE)
-  add_grappa_application(${query_name}.exe ${query})
 
   # Raco C++ environment
   set(RACO_DIR "$ENV{RACO_HOME}")
@@ -110,6 +120,8 @@ foreach(query ${GENERATED_SOURCES})
       message(FATAL_ERROR "Undefined RACO_HOME environment variable, required
       for generated queries applications/join/grappa_*.cpp ")
   endif()
+
+  add_grappa_application(${query_name}.exe ${query})
 
   target_link_libraries(${query_name}.exe generator querylib queryio racoc)
   list(APPEND GENERATED_EXES "${query_name}.exe")
@@ -151,5 +163,7 @@ macro(add_check test_cpp nnode ppn target)
   add_dependencies( check-all-${target}-compile-only ${test})
 endmacro()
 
+# add test for IO
 add_check(Relation_io_tests.cpp 2 1 pass)
+file(COPY splits_test DESTINATION .)
 

applications/join/DHT_symmetric.cpp

@@ -0,0 +1,5 @@
+//#include "DHT_symmetric.hpp"
+#include "Metrics.hpp"
+
+GRAPPA_DEFINE_METRIC(SimpleMetric<uint64_t>, dht_inserts, 0);
+GRAPPA_DEFINE_METRIC(SimpleMetric<uint64_t>, dht_partition_inserts, 0);

applications/join/DHT_symmetric.hpp

@@ -7,38 +7,37 @@
 #include <Metrics.hpp>
 
 #include <cmath>
+#include <unordered_map>
 
 
-//GRAPPA_DECLARE_METRIC(MaxMetric<uint64_t>, max_cell_length);
-GRAPPA_DECLARE_METRIC(SimpleMetric<uint64_t>, hash_tables_size);
-GRAPPA_DECLARE_METRIC(SummarizingMetric<uint64_t>, hash_tables_lookup_steps);
-
+GRAPPA_DECLARE_METRIC(SimpleMetric<uint64_t>, dht_inserts);
+GRAPPA_DECLARE_METRIC(SimpleMetric<uint64_t>, dht_partition_inserts);
 
 // for naming the types scoped in DHT_symmetric
-#define DHT_symmetric_TYPE(type) typename DHT_symmetric<K,V,HF>::type
-#define DHT_symmetric_T DHT_symmetric<K,V,HF>
+#define DHT_symmetric_TYPE(type) typename DHT_symmetric<K,V,Hash>::type
+#define DHT_symmetric_T DHT_symmetric<K,V,Hash>
 
 // Hash table for joins
 // * allows multiple copies of a Key
 // * lookups return all Key matches
-template <typename K, typename V, uint64_t (*HF)(K)> 
+template <typename K, typename V, typename Hash> 
 class DHT_symmetric {
 
   private:
     // private members
     GlobalAddress< DHT_symmetric_T > self;
-    std::unordered_map<K, V> * local_map;
+    std::unordered_map<K, V, Hash > * local_map;
     size_t partitions;
 
-    uint64_t computeIndex( K key ) {
-      return HF(key) % partitions;
+    size_t computeIndex( K key ) {
+      return Hash()(key) % partitions;
     }
 
     // for creating local DHT_symmetric
     DHT_symmetric( GlobalAddress<DHT_symmetric_T> self ) 
       : self(self)
       , partitions(Grappa::cores())
-      , local_map(new std::unordered_map<K,V>())
+      , local_map(new std::unordered_map<K,V,Hash>())
       {}
 
   public:
@@ -55,30 +54,44 @@ class DHT_symmetric {
       return object;
     }
 
-    template< GlobalCompletionEvent * GCE, typename UV, V (*UpF)(V oldval, UV incVal), V Init, SyncMode S = SyncMode::Async >
+    template< typename UV, V (*UpF)(const V& oldval, const UV& incVal), V (*Init)(void) >
+  void update_partition( K key, UV val ) {
+      std::pair<K,V> entry(key, Init());
+
+      auto res = this->local_map->insert(entry); auto resIt = res.first; //auto resNew = res.second;
+
+      // perform the update in place
+      resIt->second = UpF(resIt->second, val);
+      dht_partition_inserts++;
+  }
+
+    template< GlobalCompletionEvent * GCE, typename UV, V (*UpF)(const V& oldval, const UV& incVal), V (*Init)(void), SyncMode S = SyncMode::Async >
     void update( K key, UV val ) {
-      uint64_t index = computeIndex( key );
+      auto index = computeIndex( key );
       auto target = this->self;
 
       Grappa::delegate::call<S,GCE>(index, [key, val, target]() {   
         // inserts initial value only if the key is not yet present
-        std::pair<K,V> entry(key, Init);
+        std::pair<K,V> entry(key, Init());
+
         auto res = target->local_map->insert(entry); auto resIt = res.first; //auto resNew = res.second;
 
         // perform the update in place
         resIt->second = UpF(resIt->second, val);
+        dht_inserts++;
       });
     }
 
     template< GlobalCompletionEvent * GCE, SyncMode S = SyncMode::Async >
     void insert_unique( K key, V val ) {
-      uint64_t index = computeIndex( key );
+      auto index = computeIndex( key );
       auto target = this->self;
 
       Grappa::delegate::call<S,GCE>(index, [key, val, target]() {   
         // inserts initial value only if the key is not yet present
         std::pair<K,V> entry(key, val);
         target->local_map->insert(entry); 
+        dht_inserts++;
       });
     }
 
@@ -89,15 +102,29 @@ class DHT_symmetric {
           // TODO: cannot use forall_here because unordered_map->begin() is a forward iterator (std::advance is O(n))
           // TODO: for now the serial loop is only performant if the continuation code is also in CPS
           // TODO: best solution is a forall_here where loop decomposition is just linear continuation instead of divide and conquer
+
+          int64_t iter_count = 0;
           auto m = target->local_map;
-          for (auto it = m->begin(); it != m->end(); it++) {
+          for (auto it = m->begin(); it != m->end(); it++, iter_count++) {
             // continuation takes a mapping
             f(*it);
+
+            // get the same effect as a forall_here that would have linear decomposition: 
+            // specifically that there is at least one yield per FLAGS_loop_theshold iterations.
+            if (iter_count == FLAGS_loop_threshold) {
+              iter_count = 0;
+              Grappa::yield();
+            }
           }
       }); 
       // TODO GCE->wait(); // block until all tasks are done
     }
 
+    std::unordered_map<K,V,Hash> * get_local_map() {
+      return local_map;
+    }
+
+
 
 } GRAPPA_BLOCK_ALIGNED;
 

applications/join/DHT_symmetric_generic.hpp

@@ -0,0 +1,119 @@
+#pragma once
+
+#include <Grappa.hpp>
+#include <GlobalAllocator.hpp>
+#include <GlobalCompletionEvent.hpp>
+#include <ParallelLoop.hpp>
+#include <Metrics.hpp>
+
+#include <cmath>
+#include <unordered_map>
+
+
+GRAPPA_DECLARE_METRIC(SimpleMetric<uint64_t>, dht_inserts);
+
+// for naming the types scoped in DHT_symmetric_generic
+#define DHT_symmetric_generic_TYPE(type) typename DHT_symmetric_generic<K,V,UV,Hash>::type
+#define DHT_symmetric_generic_T DHT_symmetric_generic<K,V,UV,Hash>
+
+// TODO: The functionality of this class is covered by DHT_symmetric already.
+//       The only difference is that update_f/init_f are template parameters
+//       versus instance parameters. We can just add these instance parameters
+//       to DHT_symmetric and have two versions of update() method.
+
+// Hash table for joins
+// * allows multiple copies of a Key
+// * lookups return all Key matches
+template <typename K, typename V, typename UV, typename Hash> 
+class DHT_symmetric_generic {
+  public:
+    typedef V (*update_f)(const V& oldval, const UV& incVal);
+    typedef V (*init_f)(void);
+
+  private:
+    // private members
+    GlobalAddress< DHT_symmetric_generic_T > self;
+    std::unordered_map<K, V, Hash > * local_map;
+    size_t partitions;
+
+    update_f UpF;
+    init_f Init;
+
+    size_t computeIndex( K key ) {
+      return Hash()(key) % partitions;
+    }
+
+    // for creating local DHT_symmetric_generic
+    DHT_symmetric_generic( GlobalAddress<DHT_symmetric_generic_T> self, update_f upf, init_f initf ) 
+      : self(self)
+      , UpF(upf)
+      , Init(initf)
+      , partitions(Grappa::cores())
+      , local_map(new std::unordered_map<K,V,Hash>())
+      {}
+
+  public:
+    // for static construction
+    DHT_symmetric_generic( ) {}
+
+    static GlobalAddress<DHT_symmetric_generic_T> create_DHT_symmetric( update_f upf, init_f initf ) {
+      auto object = Grappa::symmetric_global_alloc<DHT_symmetric_generic_T>();
+
+      Grappa::on_all_cores( [object, upf, initf] {
+        new(object.pointer()) DHT_symmetric_generic_T(object, upf, initf);
+      });
+
+      return object;
+    }
+
+    void update_partition(K key, UV val) {
+      std::pair<K,V> entry(key, Init());
+
+      auto res = this->local_map->insert(entry); auto resIt = res.first; //auto resNew = res.second;
+
+      // perform the update in place
+      resIt->second = this->UpF(resIt->second, val);
+      dht_partition_inserts++;
+    }
+
+    template< GlobalCompletionEvent * GCE, SyncMode S = SyncMode::Async >
+    void update( K key, UV val ) {
+      auto index = computeIndex( key );
+      auto target = this->self;
+
+      Grappa::delegate::call<S,GCE>(index, [key, val, target]() {   
+        // inserts initial value only if the key is not yet present
+        std::pair<K,V> entry(key, target->Init());
+
+        auto res = target->local_map->insert(entry); auto resIt = res.first; //auto resNew = res.second;
+
+        // perform the update in place
+        resIt->second = target->UpF(resIt->second, val);
+        dht_inserts++;
+      });
+    }
+
+    template < GlobalCompletionEvent * GCE, typename CF >
+    void forall_entries( CF f ) {
+      auto target = this->self;
+      Grappa::on_all_cores([target, f] {
+          // TODO: cannot use forall_here because unordered_map->begin() is a forward iterator (std::advance is O(n))
+          // TODO: for now the serial loop is only performant if the continuation code is also in CPS
+          // TODO: best solution is a forall_here where loop decomposition is just linear continuation instead of divide and conquer
+          auto m = target->local_map;
+          for (auto it = m->begin(); it != m->end(); it++) {
+            // continuation takes a mapping
+            f(*it);
+          }
+      }); 
+      // TODO GCE->wait(); // block until all tasks are done
+    }
+
+    std::unordered_map<K,V,Hash> * get_local_map() {
+      return local_map;
+    }
+
+
+
+} GRAPPA_BLOCK_ALIGNED;
+

applications/join/DoubleDHT.cpp

@@ -0,0 +1,5 @@
+#include <Metrics.hpp>
+
+GRAPPA_DEFINE_METRIC(SimpleMetric<uint64_t>, hash_matches_iterator_cell_single_misses, 0);
+GRAPPA_DEFINE_METRIC(SimpleMetric<uint64_t>, hash_matches_iterator_cell_hits, 0);
+GRAPPA_DEFINE_METRIC(SimpleMetric<uint64_t>, hash_matches_iterator_cell_both_misses, 0);