Merge pull request #48 from stefanfred/play

Further CPU and GPU impementation alingment

include/builders/internal_memory_builder_partitioned_phf.hpp

@@ -35,6 +35,7 @@ struct internal_memory_builder_partitioned_phf {
         m_bucketer.init(num_partitions);
         m_offsets.resize(num_partitions + 1);
         m_builders.resize(num_partitions);
+        m_num_buckets_per_partition = compute_num_buckets(config.avg_partition_size, config.lambda);
 
         std::vector<std::vector<typename hasher_type::hash_type>> partitions(num_partitions);
         for (auto& partition : partitions) partition.reserve(1.1 * config.avg_partition_size);
@@ -72,12 +73,7 @@ struct internal_memory_builder_partitioned_phf {
 
         auto partition_config = config;
         partition_config.seed = m_seed;
-
-        const uint64_t num_buckets_single_phf = compute_num_buckets(num_keys, config.lambda);
-        const uint64_t num_buckets_per_partition =
-            std::ceil(static_cast<double>(num_buckets_single_phf) / num_partitions);
-        m_num_buckets_per_partition = num_buckets_per_partition;
-        partition_config.num_buckets = num_buckets_per_partition;
+        partition_config.num_buckets = m_num_buckets_per_partition;
         if (config.verbose_output) {
             std::cout << "num_buckets_per_partition = " << partition_config.num_buckets
                       << std::endl;

include/dense_partitioned_phf.hpp

@@ -21,7 +21,7 @@ struct dense_partitioned_phf {
     }
 
     template <typename Builder>
-    double build(Builder& builder, build_configuration const& /* config */)  //
+    double build(Builder& builder, build_configuration const&  config)  //
     {
         auto start = clock_type::now();
 
@@ -40,7 +40,7 @@ struct dense_partitioned_phf {
         const uint64_t increment = m_table_size / num_partitions;
         m_offsets.encode(offsets.begin(), offsets.size(), increment);
         m_pilots.encode(builder.interleaving_pilots_iterator_begin(), num_partitions,
-                        num_buckets_per_partition);
+                        num_buckets_per_partition, config.num_threads);
         if constexpr (needsFreeArray) {
             assert(builder.free_slots().size() == m_table_size - m_num_keys);
             m_free_slots.encode(builder.free_slots().data(), m_table_size - m_num_keys);
@@ -93,8 +93,7 @@ struct dense_partitioned_phf {
     }
 
     size_t num_bits_for_mapper() const {
-        return m_partitioner.num_bits() + m_bucketer.num_bits() + m_offsets.num_bits() +
-               m_free_slots.num_bits();
+        return m_partitioner.num_bits() + m_bucketer.num_bits()  + m_offsets.num_bits() + (needsFreeArray ? m_free_slots.num_bits() : 0);
     }
 
     size_t num_bits() const {
@@ -118,7 +117,8 @@ struct dense_partitioned_phf {
         visitor.visit(m_bucketer);
         visitor.visit(m_pilots);
         visitor.visit(m_offsets);
-        visitor.visit(m_free_slots);
+        if(needsFreeArray)
+            visitor.visit(m_free_slots);
     }
 
 private:

include/encoders/dense_encoders.hpp

@@ -2,6 +2,7 @@
 
 #include <vector>
 #include <cassert>
+#include <thread>
 
 namespace pthash {
 
@@ -53,7 +54,7 @@ struct mono_interleaved {
     template <typename Iterator>
     void encode(Iterator begin,                            //
                 const uint64_t num_partitions,             //
-                const uint64_t num_buckets_per_partition)  //
+                const uint64_t num_buckets_per_partition, const uint64_t /*num_threads*/)  //
     {
         m_num_partitions = num_partitions;
         m_encoder.encode(begin, num_partitions * num_buckets_per_partition);
@@ -92,11 +93,36 @@ struct multi_interleaved {
     template <typename Iterator>
     void encode(Iterator begin,                            //
                 const uint64_t num_partitions,             //
-                const uint64_t num_buckets_per_partition)  //
+                const uint64_t num_buckets_per_partition, const uint64_t num_threads)  //
     {
         m_encoders.resize(num_buckets_per_partition);
-        for (uint64_t i = 0; i != num_buckets_per_partition; ++i) {
-            m_encoders[i].encode(begin + i * num_partitions, num_partitions);
+        if(num_threads==1) {
+            for (uint64_t i = 0; i != num_buckets_per_partition; ++i) {
+                m_encoders[i].encode(begin + i * num_partitions, num_partitions);
+            }
+        } else {
+            auto exe = [&](uint64_t beginEncoder, uint64_t endEncoder) {
+                for (; beginEncoder != endEncoder; ++beginEncoder) {
+                    m_encoders[beginEncoder].encode(begin + beginEncoder * num_partitions, num_partitions);
+                }
+            };
+
+            std::vector<std::thread> threads(num_threads);
+            uint64_t currentEncoder = 0;
+            uint64_t i = 0;
+            const uint64_t enc_per_thread =
+                (num_buckets_per_partition + num_threads - 1) / num_threads;
+            while (currentEncoder < num_buckets_per_partition) {
+                uint64_t endEncoder = currentEncoder + enc_per_thread;
+                if (endEncoder > num_buckets_per_partition) endEncoder = num_buckets_per_partition;
+                threads[i] = std::thread(exe, currentEncoder, endEncoder);
+                currentEncoder = endEncoder;
+                i++;
+            }
+            for (auto& t : threads) {
+                if (t.joinable()) t.join();
+            }
+
         }
     }
 
@@ -129,16 +155,29 @@ struct dual_interleaved {
     template <typename Iterator>
     void encode(Iterator begin,                            //
                 const uint64_t num_partitions,             //
-                const uint64_t num_buckets_per_partition)  //
+                const uint64_t num_buckets_per_partition, const uint64_t num_threads)  //
     {
         m_front_size = num_buckets_per_partition * (static_cast<double>(numerator) / denominator);
-        m_front.encode(begin, num_partitions, m_front_size);
-        m_back.encode(begin + m_front_size * num_partitions, num_partitions,
-                      num_buckets_per_partition - m_front_size);
+        if(num_threads == 1) {
+            if(m_front_size > 0) m_front.encode(begin, num_partitions, m_front_size, 1);
+            if(num_buckets_per_partition - m_front_size > 0) m_back.encode(begin + m_front_size * num_partitions, num_partitions,
+                          num_buckets_per_partition - m_front_size, 1);
+        } else {
+            uint64_t m_front_threads =
+                (num_threads * m_front_size + num_buckets_per_partition - 1) /
+                num_buckets_per_partition;
+            auto exe = [&]() {
+                if(m_front_size > 0) m_front.encode(begin, num_partitions, m_front_size, m_front_threads);
+            };
+            std::thread frontThread = std::thread(exe);
+            if(num_buckets_per_partition - m_front_size > 0) m_back.encode(begin + m_front_size * num_partitions, num_partitions,
+                          num_buckets_per_partition - m_front_size, num_threads - m_front_threads);
+            if (frontThread.joinable()) frontThread.join();
+        }
     }
 
     static std::string name() {
-        return Front::name() + "-" + Back::name();
+        return Front::name() + "-" + Back::name() + "-" + std::to_string(static_cast<double>(numerator)/denominator);
     }
 
     size_t num_bits() const {