Improve multithreading partitioning

Author: Tae Hyoun Park

include/queryosity/dataset_partition.h

@@ -16,16 +16,18 @@ namespace queryosity {
 
 namespace dataset {
 
+using entry_t = unsigned long long;
+
 using partition_t = std::vector<part_t>;
 
+using slot_t = unsigned int;
+
 namespace partition {
 
 partition_t align(std::vector<partition_t> const &partitions);
 
 partition_t truncate(partition_t const &parts, long long nentries_max);
 
-partition_t merge(partition_t const &parts, unsigned int nslots_max);
-
 } // namespace partition
 
 } // namespace dataset
@@ -34,12 +36,12 @@ partition_t merge(partition_t const &parts, unsigned int nslots_max);
 
 inline queryosity::dataset::partition_t queryosity::dataset::partition::align(
     std::vector<partition_t> const &partitions) {
-  std::map<unsigned long long, unsigned int> edge_counts;
+  std::map<entry_t, unsigned int> edge_counts;
   const unsigned int num_vectors = partitions.size();
 
   // Count appearances of each edge
   for (const auto &vec : partitions) {
-    std::map<unsigned long long, bool>
+    std::map<entry_t, bool>
         seen_edges; // Ensure each edge is only counted once per vector
     for (const auto &p : vec) {
       if (seen_edges.find(p.first) == seen_edges.end()) {
@@ -54,74 +56,22 @@ inline queryosity::dataset::partition_t queryosity::dataset::partition::align(
   }
 
   // Filter edges that appear in all vectors
-  std::vector<unsigned long long> aligned_edges;
+  std::vector<entry_t> aligned_edges;
   for (const auto &pair : edge_counts) {
     if (pair.second == num_vectors) {
       aligned_edges.push_back(pair.first);
     }
   }
 
   // Create aligned vector of pairs
-  std::vector<std::pair<unsigned long long, unsigned long long>> aligned_ranges;
+  std::vector<std::pair<entry_t, entry_t>> aligned_ranges;
   for (size_t i = 0; i < aligned_edges.size() - 1; ++i) {
     aligned_ranges.emplace_back(aligned_edges[i], aligned_edges[i + 1]);
   }
 
   return aligned_ranges;
 }
 
-inline queryosity::dataset::partition_t queryosity::dataset::partition::merge(
-    queryosity::dataset::partition_t const &parts, unsigned int nslots_max) {
-
-  // no merging needed
-  if (nslots_max >= static_cast<unsigned int>(parts.size()))
-    return parts;
-
-  assert(!parts.empty() && nslots_max > 0);
-
-  partition_t parts_merged;
-
-  const unsigned int total_size = parts.back().second - parts.front().first;
-  const unsigned int size_per_slot = total_size / nslots_max;
-  const unsigned int extra_size = total_size % nslots_max;
-
-  unsigned int current_start = parts[0].first;
-  unsigned int current_end = current_start;
-  unsigned int accumulated_size = 0;
-  unsigned int nslots_created = 0;
-
-  for (const auto &part : parts) {
-    unsigned int part_size = part.second - part.first;
-    // check if another part can be added
-    if (accumulated_size + part_size >
-            size_per_slot + (nslots_created < extra_size ? 1 : 0) &&
-        nslots_created < nslots_max - 1) {
-      // add the current range if adding next part will exceed the average size
-      parts_merged.emplace_back(current_start, current_end);
-      current_start = current_end;
-      accumulated_size = 0;
-      ++nslots_created;
-    }
-
-    // add part size to the current slot
-    accumulated_size += part_size;
-    current_end += part_size;
-
-    // handle the last slot differently to include all remaining parts
-    if (nslots_created == nslots_max - 1) {
-      parts_merged.emplace_back(current_start, parts.back().second);
-      break; // All parts have been processed
-    }
-  }
-
-  // ensure we have exactly nslots_max slots
-  if (static_cast<unsigned int>(parts_merged.size()) < nslots_max) {
-    parts_merged.emplace_back(current_start, parts.back().second);
-  }
-
-  return parts_merged;
-}
-
 inline queryosity::dataset::partition_t
 queryosity::dataset::partition::truncate(
     queryosity::dataset::partition_t const &parts, long long nentries_max) {

include/queryosity/dataset_player.h

@@ -15,8 +15,7 @@ class player : public query::experiment {
 
 public:
   void play(std::vector<std::unique_ptr<source>> const &sources, double scale,
-            unsigned int slot, unsigned long long begin,
-            unsigned long long end);
+            slot_t slot, std::vector<part_t> const& parts);
 };
 
 } // namespace dataset
@@ -27,55 +26,56 @@ class player : public query::experiment {
 
 inline void queryosity::dataset::player::play(
     std::vector<std::unique_ptr<source>> const &sources, double scale,
-    unsigned int slot, unsigned long long begin, unsigned long long end) {
+    slot_t slot, std::vector<part_t> const& parts) {
 
   // apply dataset scale in effect for all queries
   for (auto const &qry : m_queries) {
     qry->apply_scale(scale);
   }
 
-  // initialize
-  for (auto const &ds : sources) {
-    ds->initialize(slot, begin, end);
-  }
-  for (auto const &col : m_columns) {
-    col->initialize(slot, begin, end);
-  }
-  for (auto const &sel : m_selections) {
-    sel->initialize(slot, begin, end);
-  }
-  for (auto const &qry : m_queries) {
-    qry->initialize(slot, begin, end);
-  }
-
-  // execute
-  for (auto entry = begin; entry < end; ++entry) {
+  // traverse each part
+  for (auto const& part : parts) {
+    // initialize
     for (auto const &ds : sources) {
-      ds->execute(slot, entry);
+      ds->initialize(slot, part.first, part.second);
     }
     for (auto const &col : m_columns) {
-      col->execute(slot, entry);
+      col->initialize(slot, part.first, part.second);
     }
     for (auto const &sel : m_selections) {
-      sel->execute(slot, entry);
+      sel->initialize(slot, part.first, part.second);
     }
     for (auto const &qry : m_queries) {
-      qry->execute(slot, entry);
+      qry->initialize(slot, part.first, part.second);
+    }
+    // execute
+    for (auto entry = part.first; entry < part.second; ++entry) {
+      for (auto const &ds : sources) {
+        ds->execute(slot, entry);
+      }
+      for (auto const &col : m_columns) {
+        col->execute(slot, entry);
+      }
+      for (auto const &sel : m_selections) {
+        sel->execute(slot, entry);
+      }
+      for (auto const &qry : m_queries) {
+        qry->execute(slot, entry);
+      }
+    }
+    // finalize (in reverse order)
+    for (auto const &qry : m_queries) {
+      qry->finalize(slot);
+    }
+    for (auto const &sel : m_selections) {
+      sel->finalize(slot);
+    }
+    for (auto const &col : m_columns) {
+      col->finalize(slot);
+    }
+    for (auto const &ds : sources) {
+      ds->finalize(slot);
     }
-  }
-
-  // finalize (in reverse order)
-  for (auto const &qry : m_queries) {
-    qry->finalize(slot);
-  }
-  for (auto const &sel : m_selections) {
-    sel->finalize(slot);
-  }
-  for (auto const &col : m_columns) {
-    col->finalize(slot);
-  }
-  for (auto const &ds : sources) {
-    ds->finalize(slot);
   }
 
   // clear out queries (should not be re-played)

include/queryosity/dataset_processor.h

@@ -104,33 +104,40 @@ inline void queryosity::dataset::processor::process(
   }
 
   // 2. partition dataset(s)
-  std::vector<std::vector<std::pair<unsigned long long, unsigned long long>>>
-      partitions;
+  // 2.1 get partition from each dataset source
+  std::vector<partition_t > partitions_from_sources;
   for (auto const &ds : sources) {
-    auto partition = ds->partition();
-    if (partition.size())
-      partitions.push_back(partition);
+    auto partition_from_source = ds->partition();
+    if (partition_from_source.size())
+      partitions_from_sources.push_back(std::move(partition_from_source));
   }
-  if (!partitions.size()) {
-    throw std::logic_error("no valid dataset partition implemented");
+  if (!partitions_from_sources.size()) {
+    throw std::runtime_error("no valid dataset partition implemented");
   }
-  // find common denominator partition
-  auto partition = dataset::partition::align(partitions);
-  // truncate entries to row limit
-  partition = dataset::partition::truncate(partition, nrows);
-  // merge partition to concurrency limit
-  partition = dataset::partition::merge(partition, nslots);
-  // match processor & partition parallelism
-  this->downsize(partition.size());
+  // 2.2 find common denominator partition
+  const auto partition_aligned = dataset::partition::align(partitions_from_sources);
+  // 2.3 truncate entries to row limit
+  const auto partition_truncated = dataset::partition::truncate(partition_aligned, nrows);
+  // 2.3 distribute partition amongst threads
+  std::vector<partition_t> partitions_for_slots(nslots); 
+  auto nparts_remaining = partition_truncated.size();
+  const auto nparts = nparts_remaining;
+  while (nparts_remaining) {
+    for( unsigned int islot=0; islot<nslots ; ++islot) {
+      partitions_for_slots[islot].push_back(std::move(partition_truncated[nparts-(nparts_remaining--)]));
+      if (!nparts_remaining) break;
+    }
+  } 
+  // todo: can intel tbb distribute slots during parallel processing?
 
   // 3. run event loop
   this->run(
       [&sources,
        scale](dataset::player *plyr, unsigned int slot,
-              std::pair<unsigned long long, unsigned long long> part) {
-        plyr->play(sources, scale, slot, part.first, part.second);
+              std::vector<std::pair<unsigned long long, unsigned long long>> const& parts) {
+        plyr->play(sources, scale, slot, parts);
       },
-      m_player_ptrs, m_range_slots, partition);
+      m_player_ptrs, m_range_slots, partitions_for_slots);
 
   // 4. exit event loop
   for (auto const &ds : sources) {