Sorter uses memory pool

Author: pmenon

src/codegen/operator/order_by_translator.cpp

@@ -191,7 +191,9 @@ OrderByTranslator::OrderByTranslator(const planner::OrderByPlan &plan,
 }
 
 void OrderByTranslator::InitializeQueryState() {
-  sorter_.Init(GetCodeGen(), LoadStatePtr(sorter_id_), compare_func_);
+  auto *sorter_ptr = LoadStatePtr(sorter_id_);
+  auto *exec_ctx_ptr = GetExecutorContextPtr();
+  sorter_.Init(GetCodeGen(), sorter_ptr, exec_ctx_ptr, compare_func_);
 }
 
 void OrderByTranslator::TearDownQueryState() {
@@ -364,7 +366,8 @@ void OrderByTranslator::InitializePipelineState(PipelineContext &pipeline_ctx) {
       pipeline_ctx.IsParallel()) {
     CodeGen &codegen = GetCodeGen();
     auto *sorter_ptr = pipeline_ctx.LoadStatePtr(codegen, thread_sorter_id_);
-    sorter_.Init(codegen, sorter_ptr, compare_func_);
+    auto *exec_ctx_ptr = GetExecutorContextPtr();
+    sorter_.Init(codegen, sorter_ptr, exec_ctx_ptr, compare_func_);
   }
 }
 

src/codegen/sorter.cpp

@@ -35,9 +35,11 @@ Sorter::Sorter(CodeGen &codegen, const std::vector<type::Type> &row_desc) {
 }
 
 void Sorter::Init(CodeGen &codegen, llvm::Value *sorter_ptr,
+                  llvm::Value *executor_ctx,
                   llvm::Value *comparison_func) const {
   auto *tuple_size = codegen.Const32(storage_format_.GetStorageSize());
-  codegen.Call(SorterProxy::Init, {sorter_ptr, comparison_func, tuple_size});
+  codegen.Call(SorterProxy::Init,
+               {sorter_ptr, executor_ctx, comparison_func, tuple_size});
 }
 
 void Sorter::Append(CodeGen &codegen, llvm::Value *sorter_ptr,

src/codegen/util/sorter.cpp

@@ -18,15 +18,16 @@
 
 #include "common/synchronization/count_down_latch.h"
 #include "common/timer.h"
-#include "storage/backend_manager.h"
 #include "threadpool/mono_queue_pool.h"
 
 namespace peloton {
 namespace codegen {
 namespace util {
 
-Sorter::Sorter(ComparisonFunction func, uint32_t tuple_size)
-    : cmp_func_(func),
+Sorter::Sorter(::peloton::type::AbstractPool &memory, ComparisonFunction func,
+               uint32_t tuple_size)
+    : memory_(memory),
+      cmp_func_(func),
       tuple_size_(tuple_size),
       buffer_pos_(nullptr),
       buffer_end_(nullptr),
@@ -39,12 +40,11 @@ Sorter::Sorter(ComparisonFunction func, uint32_t tuple_size)
 
 Sorter::~Sorter() {
   uint64_t total_alloc = 0;
-  auto &backend_manager = storage::BackendManager::GetInstance();
   for (const auto &iter : blocks_) {
     void *block = iter.first;
     total_alloc += iter.second;
     PELOTON_ASSERT(block != nullptr);
-    backend_manager.Release(BackendType::MM, block);
+    memory_.Free(block);
   }
   buffer_pos_ = buffer_end_ = nullptr;
   tuples_start_ = tuples_end_ = nullptr;
@@ -54,9 +54,9 @@ Sorter::~Sorter() {
             tuples_.size(), blocks_.size(), total_alloc / 1024.0);
 }
 
-void Sorter::Init(Sorter &sorter, ComparisonFunction func,
-                  uint32_t tuple_size) {
-  new (&sorter) Sorter(func, tuple_size);
+void Sorter::Init(Sorter &sorter, executor::ExecutorContext &exec_ctx,
+                  ComparisonFunction func, uint32_t tuple_size) {
+  new (&sorter) Sorter(*exec_ctx.GetPool(), func, tuple_size);
 }
 
 void Sorter::Destroy(Sorter &sorter) { sorter.~Sorter(); }
@@ -254,9 +254,7 @@ void Sorter::MakeRoomForNewTuple() {
   LOG_TRACE("Allocating block of size %.2lf KB ...", next_alloc_size_ / 1024.0);
 
   // We need to allocate another block
-  void *block = storage::BackendManager::GetInstance().Allocate(
-      BackendType::MM, next_alloc_size_);
-  PELOTON_ASSERT(block != nullptr);
+  void *block = memory_.Allocate(next_alloc_size_);
   blocks_.emplace_back(block, next_alloc_size_);
 
   // Setup new buffer boundaries

src/include/codegen/proxy/sorter_proxy.h

@@ -21,7 +21,8 @@ namespace codegen {
 PROXY(Sorter) {
   // clang-format off
   DECLARE_MEMBER(0,
-                 char[sizeof(void *) +              // comparison function
+                 char[sizeof(void *) +              // abstract pool reference
+                      sizeof(void *) +              // comparison function
                       sizeof(uint32_t) +            // tuple size
                       sizeof(char *) +              // buffer start
                       sizeof(char *) +              // buffer end

src/include/codegen/sorter.h

@@ -35,7 +35,7 @@ class Sorter {
    * @brief Initialize the given sorter instance with the comparison function
    */
   void Init(CodeGen &codegen, llvm::Value *sorter_ptr,
-            llvm::Value *comparison_func) const;
+            llvm::Value *executor_ctx, llvm::Value *comparison_func) const;
 
   /**
    * @brief Append the given tuple into the sorter instance

src/include/codegen/util/sorter.h

@@ -6,7 +6,7 @@
 //
 // Identification: src/include/codegen/util/sorter.h
 //
-// Copyright (c) 2015-2017, Carnegie Mellon University Database Group
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
 //
 //===----------------------------------------------------------------------===//
 
@@ -32,58 +32,83 @@ namespace util {
  * This class only supports forward iteration since this is all that's required.
  * Additionally, Sorter does not serialize elements into its memory space.
  * Instead, it allocates space for incoming tuples on demand and returns a
- * pointer to the call, relying on her to serialize into the space. This assumes
- * well-behaved callers. We **could** accept a Serializer type as part of
- * Init(..), but we don't need it at this moment.
+ * pointer to the call, relying on her to serialize into the space.
  */
 class Sorter {
  private:
   // We allocate 4KB of buffer space upon initialization
   static constexpr uint64_t kInitialBufferSize = 4 * 1024;
 
  public:
-  typedef int (*ComparisonFunction)(const char *left_tuple,
-                                    const char *right_tuple);
+  using ComparisonFunction = int (*)(const char *left_tuple,
+                                     const char *right_tuple);
 
-  /// Constructor
-  Sorter(ComparisonFunction func, uint32_t tuple_size);
+  /**
+   * Constructor to create and setup this sorter instance.
+   *
+   * TODO(pmenon): Right now, Sorters assume that all tuples that are inserted
+   * have the same size. If we choose to perform null-suppression or any
+   * compression, this is no longer true.
+   *
+   * @param memory A memory pool that is used for all allocations during the
+   * lifetime of the sorter.
+   * @param func The comparison function used to compare two tuples stored in
+   * this sorter
+   * @param tuple_size The size of the tuples stored in this sorter
+   */
+  Sorter(::peloton::type::AbstractPool &memory, ComparisonFunction func,
+         uint32_t tuple_size);
 
-  /// Destructor
+  /**
+   * Destructor. This destructor cleans up returns all memory it has allocated
+   * back to the memory pool that injected at construction.
+   */
   ~Sorter();
 
   /**
-   * @brief This static function initializes the given sorter instance with the
-   * given comparison function and assumes all input tuples have the given size.
+   * This static function initializes the given sorter instance with the given
+   * comparison function and assumes all input tuples have the given size. This
+   * method is used from codegen to invoke the constructor of the sorter
+   * sorter instance.
    *
    * @param sorter The sorter instance we are initializing
    * @param func The comparison function used during sort
    * @param tuple_size The size of the tuple in bytes
    */
-  static void Init(Sorter &sorter, ComparisonFunction func,
-                   uint32_t tuple_size);
+  static void Init(Sorter &sorter, executor::ExecutorContext &ctx,
+                   ComparisonFunction func, uint32_t tuple_size);
 
   /**
-   * @brief Cleans up all resources maintained by the given sorter instance
+   * Cleans up all resources maintained by the given sorter instance. This
+   * method is used from codegen to invoke the destructor of a sorter instance.
+   *
+   * @param sorter The sorter instance we're destroying the resources from
    */
   static void Destroy(Sorter &sorter);
 
   /**
-   * @brief Allocate space for a new input tuple in this sorter. It is assumed
-   * that the size of the tuple is equivalent to the tuple size provided when
-   * this sorter was initialized.
+   * Allocate space for a new input tuple in this sorter. It is assumed that the
+   * size of the tuple is equivalent to the tuple size provided when this sorter
+   * was initialized.
    *
    * @return A pointer to a memory space large enough to store one tuple
    */
   char *StoreInputTuple();
 
   /**
-   * @brief Sort all tuples stored in this sorter.
+   * Sort all tuples stored in this sorter instance. This is a single-threaded
+   * synchronous call.
    */
   void Sort();
 
   /**
-   * @brief Perform a parallel sort of all sorter instances stored in the thread
-   * states object.
+   * Perform a parallel sort of all sorter instances stored in the thread states
+   * object.
+   *
+   * @param thread_states The states object where all the sorter instances are
+   * stored.
+   * @param sorter_offset The offset in each thread state where the sorters are
+   * stored.
    */
   void SortParallel(
       const executor::ExecutorContext::ThreadStates &thread_states,
@@ -95,17 +120,25 @@ class Sorter {
   ///
   //////////////////////////////////////////////////////////////////////////////
 
-  /// Return the number of tuples
+  /** Return the number tuples stored in this sorter instance */
   uint64_t NumTuples() const { return tuples_.size(); }
 
-  /// Iterators
+  /** Iterators */
   std::vector<char *>::iterator begin() { return tuples_.begin(); }
   std::vector<char *>::iterator end() { return tuples_.end(); }
 
  private:
+  /**
+   * Allocate room for a new tuple. If room is already available, return
+   * immediately. If room has to be made, allocate a block of memory from the
+   * memory pool.
+   */
   void MakeRoomForNewTuple();
 
  private:
+  // The memory pool where this sorter sources memory from
+  ::peloton::type::AbstractPool &memory_;
+
   // The comparison function
   ComparisonFunction cmp_func_;
 

test/codegen/sorter_test.cpp

@@ -6,7 +6,7 @@
 //
 // Identification: test/codegen/sorter_test.cpp
 //
-// Copyright (c) 2015-17, Carnegie Mellon University Database Group
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
 //
 //===----------------------------------------------------------------------===//
 
@@ -37,11 +37,6 @@ static int CompareTuplesForAscending(const char *a, const char *b) {
 
 class SorterTest : public PelotonTest {
  public:
-  // The sorter instance
-  codegen::util::Sorter sorter_;
-
-  SorterTest() : sorter_(CompareTuplesForAscending, sizeof(TestTuple)) {}
-
   static void LoadSorter(codegen::util::Sorter &sorter, uint64_t num_inserts) {
     std::random_device r;
     std::default_random_engine e(r());
@@ -72,30 +67,37 @@ class SorterTest : public PelotonTest {
   }
 
   void TestSort(uint64_t num_tuples_to_insert = 100) {
-    // Time this stuff
-    Timer<std::ratio<1, 1000>> timer;
-    timer.Start();
+    ::peloton::type::EphemeralPool pool;
 
-    // Load the sorter
-    LoadSorter(sorter_, num_tuples_to_insert);
+    {
+      codegen::util::Sorter sorter{pool, CompareTuplesForAscending,
+                                   sizeof(TestTuple)};
 
-    timer.Stop();
-    LOG_INFO("Loading %" PRId64 " tuples into sort took %.2f ms",
-             num_tuples_to_insert, timer.GetDuration());
-    timer.Reset();
-    timer.Start();
+      // Time this stuff
+      Timer<std::ratio<1, 1000>> timer;
+      timer.Start();
 
-    // Sort
-    sorter_.Sort();
+      // Load the sorter
+      LoadSorter(sorter, num_tuples_to_insert);
 
-    timer.Stop();
-    LOG_INFO("Sorting %" PRId64 " tuples took %.2f ms", num_tuples_to_insert,
-             timer.GetDuration());
+      timer.Stop();
+      LOG_INFO("Loading %" PRId64 " tuples into sort took %.2f ms",
+               num_tuples_to_insert, timer.GetDuration());
+      timer.Reset();
+      timer.Start();
+
+      // Sort
+      sorter.Sort();
+
+      timer.Stop();
+      LOG_INFO("Sorting %" PRId64 " tuples took %.2f ms", num_tuples_to_insert,
+               timer.GetDuration());
 
-    // Check sorted results
-    CheckSorted(sorter_, true);
+      // Check sorted results
+      CheckSorted(sorter, true);
 
-    EXPECT_EQ(num_tuples_to_insert, sorter_.NumTuples());
+      EXPECT_EQ(num_tuples_to_insert, sorter.NumTuples());
+    }
   }
 };
 
@@ -110,6 +112,7 @@ TEST_F(SorterTest, BenchmarkSorter) {
 }
 
 TEST_F(SorterTest, ParallelSortTest) {
+  // A fake executor context associated to no transaction
   executor::ExecutorContext ctx(nullptr);
 
   uint32_t num_threads = 4;
@@ -127,31 +130,35 @@ TEST_F(SorterTest, ParallelSortTest) {
   for (uint32_t i = 0; i < num_threads; i++) {
     auto *sorter = reinterpret_cast<codegen::util::Sorter *>(
         thread_states.AccessThreadState(i));
-    codegen::util::Sorter::Init(*sorter, CompareTuplesForAscending,
+    codegen::util::Sorter::Init(*sorter, ctx, CompareTuplesForAscending,
                                 sizeof(TestTuple));
     LoadSorter(*sorter, ntuples_per_sorter);
   }
 
-  Timer<std::milli> timer;
-  timer.Start();
+  {
+    codegen::util::Sorter main_sorter{*ctx.GetPool(), CompareTuplesForAscending,
+                                      sizeof(TestTuple)};
+    Timer<std::milli> timer;
+    timer.Start();
 
-  // Sort parallel
-  sorter_.SortParallel(thread_states, 0);
+    // Sort parallel
+    main_sorter.SortParallel(thread_states, 0);
 
-  timer.Stop();
-  LOG_INFO("Parallel sort took: %.2lf ms", timer.GetDuration());
+    timer.Stop();
+    LOG_INFO("Parallel sort took: %.2lf ms", timer.GetDuration());
 
-  // Check main sorter is sorted
-  CheckSorted(sorter_, true);
+    // Check main sorter is sorted
+    CheckSorted(main_sorter, true);
 
-  // Check result size
-  EXPECT_EQ(num_tuples, sorter_.NumTuples());
+    // Check result size
+    EXPECT_EQ(num_tuples, main_sorter.NumTuples());
 
-  // Clean up
-  for (uint32_t i = 0; i < num_threads; i++) {
-    auto *sorter = reinterpret_cast<codegen::util::Sorter *>(
-        thread_states.AccessThreadState(i));
-    codegen::util::Sorter::Destroy(*sorter);
+    // Clean up
+    for (uint32_t i = 0; i < num_threads; i++) {
+      auto *sorter = reinterpret_cast<codegen::util::Sorter *>(
+          thread_states.AccessThreadState(i));
+      codegen::util::Sorter::Destroy(*sorter);
+    }
   }
 }