Fix parallel hash joins to merge thread local hash tables in parallel. Also fixed memory leaks by properly cleaning up after pipelines.

Author: pmenon

src/codegen/code_context.cpp

@@ -142,16 +142,14 @@ class InstructionCounts : public llvm::ModulePass {
   }
 
   void DumpStats() const {
-#ifndef NDEBUG
-    LOG_DEBUG("# functions: %" PRId64 " (%" PRId64
+    LOG_INFO("# functions: %" PRId64 " (%" PRId64
               " external), # blocks: %" PRId64 ", # instructions: %" PRId64,
               func_count_, external_func_count_, basic_block_count_,
               total_inst_counts_);
     for (const auto iter : counts_) {
       const char *inst_name = llvm::Instruction::getOpcodeName(iter.first);
-      LOG_DEBUG("↳ %s: %" PRId64, inst_name, iter.second);
+      LOG_INFO("↳ %s: %" PRId64, inst_name, iter.second);
     }
-#endif
   }
 
  private:

src/codegen/hash_table.cpp

@@ -163,8 +163,10 @@ void HashTable::BuildLazy(CodeGen &codegen, llvm::Value *ht_ptr) const {
 }
 
 void HashTable::ReserveLazy(CodeGen &codegen, llvm::Value *ht_ptr,
-                            llvm::Value *thread_states) const {
-  codegen.Call(HashTableProxy::BuildLazy, {ht_ptr, thread_states});
+                            llvm::Value *thread_states,
+                            uint32_t ht_state_offset) const {
+  codegen.Call(HashTableProxy::ReserveLazy,
+               {ht_ptr, thread_states, codegen.Const32(ht_state_offset)});
 }
 
 void HashTable::MergeLazyUnfinished(CodeGen &codegen, llvm::Value *global_ht,

src/codegen/operator/hash_join_translator.cpp

@@ -343,7 +343,7 @@ void HashJoinTranslator::Consume(ConsumerContext &context,
 }
 
 // The given row is coming from the left child. Insert into hash table
-void HashJoinTranslator::ConsumeFromLeft(ConsumerContext &,
+void HashJoinTranslator::ConsumeFromLeft(ConsumerContext &ctx,
                                          RowBatch::Row &row) const {
   CodeGen &codegen = GetCodeGen();
 
@@ -363,61 +363,69 @@ void HashJoinTranslator::ConsumeFromLeft(ConsumerContext &,
   }
 #endif
 
+  llvm::Value *ht_ptr = nullptr;
+  if (ctx.GetPipeline().IsParallel()) {
+    ht_ptr = ctx.GetPipelineContext()->LoadStatePtr(codegen, hash_table_tl_id_);
+  } else {
+    ht_ptr = LoadStatePtr(hash_table_id_);
+  }
+
   // Insert tuples from the left side into the hash table
   InsertLeft insert_left{left_value_storage_, vals};
-  hash_table_.InsertLazy(codegen, LoadStatePtr(hash_table_id_), hash, key,
-                         insert_left);
+  hash_table_.InsertLazy(codegen, ht_ptr, hash, key, insert_left);
 
+  // Update bloom filter, if enabled
   if (GetJoinPlan().IsBloomFilterEnabled()) {
-    // Insert tuples into the bloom filter if enabled
     bloom_filter_.Add(codegen, LoadStatePtr(bloom_filter_id_), key);
   }
 }
 
-void HashJoinTranslator::RegisterPipelineState(PipelineContext &context) {
-  if (context.IsParallel()) {
-    hash_table_tl_id_ =
-        context.RegisterState("localHT", HashTableProxy::GetType(GetCodeGen()));
+void HashJoinTranslator::RegisterPipelineState(PipelineContext &pipeline_ctx) {
+  if (pipeline_ctx.IsParallel() && IsLeftPipeline(pipeline_ctx.GetPipeline())) {
+    hash_table_tl_id_ = pipeline_ctx.RegisterState(
+        "localHT", HashTableProxy::GetType(GetCodeGen()));
   }
 }
 
-void HashJoinTranslator::InitializePipelineState(PipelineContext &context) {
-  if (context.IsParallel() && IsLeftPipeline(context.GetPipeline())) {
+void HashJoinTranslator::InitializePipelineState(
+    PipelineContext &pipeline_ctx) {
+  if (pipeline_ctx.IsParallel() && IsLeftPipeline(pipeline_ctx.GetPipeline())) {
     CodeGen &codegen = GetCodeGen();
     hash_table_.Init(codegen, GetExecutorContextPtr(),
-                     context.LoadStatePtr(codegen, hash_table_tl_id_));
+                     pipeline_ctx.LoadStatePtr(codegen, hash_table_tl_id_));
   }
 }
 
-void HashJoinTranslator::FinishPipeline(PipelineContext &context) {
-  // We only need to do post-pipeline processing work in the left pipeline
-  if (context.GetPipeline() != left_pipeline_) {
-    return;
-  }
-
-  llvm::Value *global_ht_ptr = LoadStatePtr(hash_table_id_);
-
-  if (!context.IsParallel()) {
-    // Build the hash table over the lazily inserted tuples
-    hash_table_.BuildLazy(GetCodeGen(), global_ht_ptr);
-  } else {
+void HashJoinTranslator::FinishPipeline(PipelineContext &pipeline_ctx) {
+  if (IsLeftPipeline(pipeline_ctx.GetPipeline())) {
     CodeGen &codegen = GetCodeGen();
-
-    llvm::Value *local_ht_ptr =
-        context.LoadStatePtr(codegen, hash_table_tl_id_);
-
-    // First size the global hash table
-    hash_table_.ReserveLazy(codegen, global_ht_ptr, GetThreadStatesPtr());
-
-    // Then merge each local table in parallel
-    hash_table_.MergeLazyUnfinished(codegen, global_ht_ptr, local_ht_ptr);
+    llvm::Value *global_ht_ptr = LoadStatePtr(hash_table_id_);
+    if (!pipeline_ctx.IsParallel()) {
+      // Build the hash table over the lazily inserted tuples
+      hash_table_.BuildLazy(codegen, global_ht_ptr);
+    } else {
+      // First size the global hash table
+      hash_table_.ReserveLazy(
+          codegen, global_ht_ptr, GetThreadStatesPtr(),
+          pipeline_ctx.GetEntryOffset(codegen, hash_table_tl_id_));
+
+      // Then merge each local table in parallel
+      PipelineContext::LoopOverStates loop_states{pipeline_ctx};
+      loop_states.DoParallel([this, &pipeline_ctx, &codegen](
+          UNUSED_ATTRIBUTE llvm::Value *thread_state) {
+        llvm::Value *global_ht_ptr = LoadStatePtr(hash_table_id_);
+        llvm::Value *local_ht_ptr =
+            pipeline_ctx.LoadStatePtr(codegen, hash_table_tl_id_);
+        hash_table_.MergeLazyUnfinished(codegen, global_ht_ptr, local_ht_ptr);
+      });
+    }
   }
 }
 
-void HashJoinTranslator::TearDownPipelineState(PipelineContext &context) {
-  if (context.IsParallel() && IsLeftPipeline(context.GetPipeline())) {
+void HashJoinTranslator::TearDownPipelineState(PipelineContext &pipeline_ctx) {
+  if (pipeline_ctx.IsParallel() && IsLeftPipeline(pipeline_ctx.GetPipeline())) {
     CodeGen &codegen = GetCodeGen();
-    auto *local_ht_ptr = context.LoadStatePtr(codegen, hash_table_tl_id_);
+    auto *local_ht_ptr = pipeline_ctx.LoadStatePtr(codegen, hash_table_tl_id_);
     hash_table_.Destroy(codegen, local_ht_ptr);
   }
 }
@@ -434,7 +442,7 @@ void HashJoinTranslator::ConsumeFromRight(ConsumerContext &context,
     llvm::Value *contains = bloom_filter_.Contains(
         GetCodeGen(), LoadStatePtr(bloom_filter_id_), key);
 
-    lang::If is_valid_row(GetCodeGen(), contains);
+    lang::If is_valid_row{GetCodeGen(), contains};
     {
       // For each tuple that passes the bloom filter, probe the hash table
       // to eliminate the false positives.

src/codegen/pipeline.cpp

@@ -18,11 +18,25 @@
 #include "codegen/consumer_context.h"
 #include "codegen/lang/loop.h"
 #include "codegen/proxy/executor_context_proxy.h"
+#include "codegen/proxy/runtime_functions_proxy.h"
 #include "settings/settings_manager.h"
 
 namespace peloton {
 namespace codegen {
 
+namespace {
+
+std::string CreateUniqueFunctionName(Pipeline &pipeline,
+                                     const std::string &prefix) {
+  CompilationContext &compilation_ctx = pipeline.GetCompilationContext();
+  CodeContext &cc = compilation_ctx.GetCodeGen().GetCodeContext();
+  return StringUtil::Format("_%" PRId64 "_pipeline_%u_%s_%s", cc.GetID(),
+                            pipeline.GetId(), prefix.c_str(),
+                            pipeline.ConstructPipelineName().c_str());
+}
+
+}  // namespace
+
 ////////////////////////////////////////////////////////////////////////////////
 ///
 /// LoopOverStates
@@ -36,30 +50,83 @@ void PipelineContext::LoopOverStates::Do(
     const std::function<void(llvm::Value *)> &body) const {
   auto &compilation_ctx = ctx_.GetPipeline().GetCompilationContext();
   auto &exec_consumer = compilation_ctx.GetExecutionConsumer();
-  auto *thread_states = exec_consumer.GetThreadStatesPtr(compilation_ctx);
+  auto &codegen = compilation_ctx.GetCodeGen();
 
-  CodeGen &codegen = compilation_ctx.GetCodeGen();
+  llvm::Value *thread_states =
+      exec_consumer.GetThreadStatesPtr(compilation_ctx);
 
   llvm::Value *num_threads =
       codegen.Load(ThreadStatesProxy::num_threads, thread_states);
   llvm::Value *state_size =
       codegen.Load(ThreadStatesProxy::state_size, thread_states);
-  llvm::Value *states = codegen.Load(ThreadStatesProxy::states, thread_states);
 
-  llvm::Value *state_end = codegen->CreateInBoundsGEP(
-      states, {codegen->CreateMul(num_threads, state_size)});
+  llvm::Value *states = codegen.Load(ThreadStatesProxy::states, thread_states);
+  states = codegen->CreatePointerCast(states, codegen.CharPtrType());
 
-  llvm::Value *loop_cond = codegen->CreateICmpNE(states, state_end);
-  lang::Loop state_loop{codegen, loop_cond, {{"threadState", states}}};
+  llvm::Value *tid = codegen.Const32(0);
+  llvm::Value *loop_cond = codegen->CreateICmpNE(tid, num_threads);
+  lang::Loop state_loop{codegen, loop_cond, {{"tid", tid}}};
   {
-    // Pull out state in this iteration
-    llvm::Value *curr_state = state_loop.GetLoopVar(0);
+    // Pull out state for current TID
+    tid = state_loop.GetLoopVar(0);
+    llvm::Value *offset = codegen->CreateMul(tid, state_size);
+
+    llvm::Value *raw_ptr = codegen->CreateInBoundsGEP(states, {offset});
+    llvm::Value *state = codegen->CreatePointerCast(
+        raw_ptr, ctx_.GetThreadStateType()->getPointerTo());
+
     // Invoke caller
-    body(curr_state);
+    body(state);
+
     // Wrap up
-    states = codegen->CreateInBoundsGEP(states, {state_size});
-    state_loop.LoopEnd(codegen->CreateICmpNE(states, state_end), {states});
+    tid = codegen->CreateAdd(tid, codegen.Const32(1));
+    state_loop.LoopEnd(codegen->CreateICmpNE(tid, num_threads), {tid});
+  }
+}
+
+void PipelineContext::LoopOverStates::DoParallel(
+    const std::function<void(llvm::Value *)> &body) const {
+  Pipeline &pipeline = ctx_.GetPipeline();
+  CompilationContext &comp_ctx = pipeline.GetCompilationContext();
+  QueryState &query_state = comp_ctx.GetQueryState();
+  CodeGen &codegen = comp_ctx.GetCodeGen();
+
+  auto name = CreateUniqueFunctionName(pipeline, "loopThreadState");
+
+  std::vector<FunctionDeclaration::ArgumentInfo> args = {
+      {"queryState", query_state.GetType()->getPointerTo()},
+      {"threadState", ctx_.GetThreadStateType()->getPointerTo()}};
+  FunctionDeclaration decl{codegen.GetCodeContext(), name,
+                           FunctionDeclaration::Visibility::Internal,
+                           codegen.VoidType(), args};
+  FunctionBuilder func{codegen.GetCodeContext(), decl};
+  {
+    // Pull out arguments
+    auto *thread_state_ptr = func.GetArgumentByPosition(1);
+
+    // Setup access to the thread state
+    PipelineContext::ScopedStateAccess state_access{
+        ctx_, func.GetArgumentByPosition(1)};
+
+    // Execute function body
+    body(thread_state_ptr);
+
+    // Finish
+    func.ReturnAndFinish();
   }
+
+  // Invoke the per-state dispatch function
+
+  std::vector<llvm::Value *> dispatch_args = {
+      // The (void*) query state
+      codegen->CreatePointerCast(codegen.GetState(), codegen.VoidPtrType()),
+      // The (ThreadStates &) thread states
+      comp_ctx.GetExecutionConsumer().GetThreadStatesPtr(comp_ctx),
+      // The function
+      codegen->CreatePointerCast(
+          func.GetFunction(),
+          proxy::TypeBuilder<void (*)(void *, void *)>::GetType(codegen))};
+  codegen.Call(RuntimeFunctionsProxy::ExecutePerState, dispatch_args);
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -150,6 +217,12 @@ uint32_t PipelineContext::GetEntryOffset(CodeGen &codegen,
   return static_cast<uint32_t>(codegen.ElementOffset(state_type, state_id));
 }
 
+bool PipelineContext::HasState() const {
+  PELOTON_ASSERT(thread_state_type_ != nullptr &&
+                 "Cannot query state components until it has been finalized");
+  return state_components_.size() > 1;
+}
+
 bool PipelineContext::IsParallel() const { return pipeline_.IsParallel(); }
 
 Pipeline &PipelineContext::GetPipeline() { return pipeline_; }
@@ -283,19 +356,6 @@ uint32_t Pipeline::GetTranslatorStage(
 ///
 ////////////////////////////////////////////////////////////////////////////////
 
-namespace {
-
-std::string CreateUniqueFunctionName(Pipeline &pipeline,
-                                     const std::string &prefix) {
-  CompilationContext &compilation_ctx = pipeline.GetCompilationContext();
-  CodeContext &cc = compilation_ctx.GetCodeGen().GetCodeContext();
-  return StringUtil::Format("_%" PRId64 "_pipeline_%u_%s_%s", cc.GetID(),
-                            pipeline.GetId(), prefix.c_str(),
-                            pipeline.ConstructPipelineName().c_str());
-}
-
-}  // namespace
-
 std::string Pipeline::ConstructPipelineName() const {
   std::vector<std::string> parts;
   for (auto riter = pipeline_.rbegin(), rend = pipeline_.rend(); riter != rend;
@@ -354,8 +414,8 @@ void Pipeline::InitializePipeline(PipelineContext &pipeline_ctx) {
       {"queryState", query_state.GetType()->getPointerTo()},
       {"threadState", pipeline_ctx.GetThreadStateType()->getPointerTo()}};
 
-  FunctionDeclaration init_decl(cc, func_name, visibility, ret_type, args);
-  FunctionBuilder init_func(cc, init_decl);
+  FunctionDeclaration init_decl{cc, func_name, visibility, ret_type, args};
+  FunctionBuilder init_func{cc, init_decl};
   {
     PipelineContext::ScopedStateAccess state_access{
         pipeline_ctx, init_func.GetArgumentByPosition(1)};
@@ -386,7 +446,11 @@ void Pipeline::CompletePipeline(PipelineContext &pipeline_ctx) {
     return;
   }
 
-  // Loop over all states
+  if (!pipeline_ctx.HasState()) {
+    return;
+  }
+
+  // Loop over all states to allow operators to clean up components
   PipelineContext::LoopOverStates loop_state{pipeline_ctx};
   loop_state.Do([this, &pipeline_ctx](llvm::Value *thread_state) {
     PipelineContext::ScopedStateAccess state_access{pipeline_ctx, thread_state};
@@ -429,8 +493,7 @@ void Pipeline::Run(
   InitializePipeline(pipeline_ctx);
 
   // Generate pipeline
-  DoRun(pipeline_ctx, dispatch_func, dispatch_args, pipeline_arg_types,
-        body);
+  DoRun(pipeline_ctx, dispatch_func, dispatch_args, pipeline_arg_types, body);
 
   // Finish
   CompletePipeline(pipeline_ctx);
@@ -460,8 +523,8 @@ void Pipeline::DoRun(
   }
 
   // The main function
-  FunctionDeclaration declaration(cc, func_name, visibility, ret_type, args);
-  FunctionBuilder func(cc, declaration);
+  FunctionDeclaration declaration{cc, func_name, visibility, ret_type, args};
+  FunctionBuilder func{cc, declaration};
   {
     auto *query_state = func.GetArgumentByPosition(0);
     auto *thread_state = func.GetArgumentByPosition(1);

src/codegen/proxy/runtime_functions_proxy.cpp

@@ -32,6 +32,7 @@ DEFINE_METHOD(peloton::codegen, RuntimeFunctions, GetTileGroup);
 DEFINE_METHOD(peloton::codegen, RuntimeFunctions, GetTileGroupLayout);
 DEFINE_METHOD(peloton::codegen, RuntimeFunctions, FillPredicateArray);
 DEFINE_METHOD(peloton::codegen, RuntimeFunctions, ExecuteTableScan);
+DEFINE_METHOD(peloton::codegen, RuntimeFunctions, ExecutePerState);
 DEFINE_METHOD(peloton::codegen, RuntimeFunctions, ThrowDivideByZeroException);
 DEFINE_METHOD(peloton::codegen, RuntimeFunctions, ThrowOverflowException);