IMPALA-5715: (mitigation only) defer destruction of MemTrackers

One potential candidate for the bad MemTracker IMPALA-5715 is one owned
by a join build sink. I haven't found a clear root cause, but we can
reduce the probability of bugs like this by deferring teardown of the
MemTrackers.

This patch defers destruction of the fragment instance, ExecNode,
DataSink and Codegen MemTrackers until query teardown. Instead
MemTracker::Close() is called at the place where the MemTracker
would have been destroyed to check that all memory is released
and enforce that no more memory is consumed. The entire query
MemTracker subtree is then disconnected in whole from the global
tree in QueryState::ReleaseResources(), instead of the MemTrackers
being incrementally disconnected bottom-up.

In cases where the MemTracker is owned by another object, this
required deferring teardown of the owner until query teardown.
E.g. for LlvmCodeGen I added a Close() method to release resources
and deferred calling the destructor.

We want to make this change anyway - see IMPALA-5587.

Testing:
Ran a core ASAN build.

Change-Id: I205abb0076d1ffd08cb93c0f1671c8b81e7fba0f
Reviewed-on: http://gerrit.cloudera.org:8080/7492
Reviewed-by: Tim Armstrong <tarmstrong@cloudera.com>
Tested-by: Impala Public Jenkins

be/src/codegen/llvm-codegen-test.cc

@@ -29,6 +29,7 @@
 #include "util/cpu-info.h"
 #include "util/hash-util.h"
 #include "util/path-builder.h"
+#include "util/scope-exit-trigger.h"
 #include "util/test-info.h"
 
 #include "common/names.h"
@@ -65,6 +66,10 @@ class LlvmCodeGenTest : public testing:: Test {
       ASSERT_OK(object1.Init(unique_ptr<Module>(new Module("Test", object1.context()))));
       ASSERT_OK(object2.Init(unique_ptr<Module>(new Module("Test", object2.context()))));
       ASSERT_OK(object3.Init(unique_ptr<Module>(new Module("Test", object3.context()))));
+
+      object1.Close();
+      object2.Close();
+      object3.Close();
     }
   }
 
@@ -113,6 +118,7 @@ TEST_F(LlvmCodeGenTest, BadIRFile) {
   string module_file = "NonExistentFile.ir";
   scoped_ptr<LlvmCodeGen> codegen;
   EXPECT_FALSE(CreateFromFile(module_file.c_str(), &codegen).ok());
+  codegen->Close();
 }
 
 // IR for the generated linner loop
@@ -236,6 +242,7 @@ TEST_F(LlvmCodeGenTest, ReplaceFnCall) {
   TestLoopFn new_loop_fn2 = reinterpret_cast<TestLoopFn>(new_loop2);
   new_loop_fn2(5);
   EXPECT_EQ(0, jitted_counter);
+  codegen->Close();
 }
 
 // Test function for c++/ir interop for strings.  Function will do:
@@ -325,6 +332,7 @@ TEST_F(LlvmCodeGenTest, StringValue) {
   int32_t* bytes = reinterpret_cast<int32_t*>(&str_val);
   EXPECT_EQ(1, bytes[2]);   // str_val.len
   EXPECT_EQ(0, bytes[3]);   // padding
+  codegen->Close();
 }
 
 // Test calling memcpy intrinsic
@@ -362,6 +370,7 @@ TEST_F(LlvmCodeGenTest, MemcpyTest) {
   test_fn(dst, src, 4);
 
   EXPECT_EQ(memcmp(src, dst, 4), 0);
+  codegen->Close();
 }
 
 // Test codegen for hash
@@ -377,6 +386,8 @@ TEST_F(LlvmCodeGenTest, HashTest) {
     scoped_ptr<LlvmCodeGen> codegen;
     ASSERT_OK(LlvmCodeGen::CreateImpalaCodegen(runtime_state_, NULL, "test", &codegen));
     ASSERT_TRUE(codegen.get() != NULL);
+    const auto close_codegen =
+        MakeScopeExitTrigger([&codegen]() { codegen->Close(); });
 
     Value* llvm_data1 =
         codegen->CastPtrToLlvmPtr(codegen->ptr_type(), const_cast<char*>(data1));

be/src/codegen/llvm-codegen.cc

@@ -160,6 +160,7 @@ Status LlvmCodeGen::InitializeLlvm(bool load_backend) {
 
   // Initialize the global shared call graph.
   shared_call_graph_.Init(init_codegen->module_);
+  init_codegen->Close();
   return Status::OK();
 }
 
@@ -168,7 +169,7 @@ LlvmCodeGen::LlvmCodeGen(RuntimeState* state, ObjectPool* pool,
   : state_(state),
     id_(id),
     profile_(pool, "CodeGen"),
-    mem_tracker_(new MemTracker(&profile_, -1, "CodeGen", parent_mem_tracker)),
+    mem_tracker_(pool->Add(new MemTracker(&profile_, -1, "CodeGen", parent_mem_tracker))),
     optimizations_enabled_(false),
     is_corrupt_(false),
     is_compiled_(false),
@@ -405,13 +406,20 @@ void LlvmCodeGen::SetupJITListeners() {
 }
 
 LlvmCodeGen::~LlvmCodeGen() {
-  if (memory_manager_ != NULL) mem_tracker_->Release(memory_manager_->bytes_tracked());
-  if (mem_tracker_->parent() != NULL) mem_tracker_->UnregisterFromParent();
-  mem_tracker_.reset();
+  DCHECK(execution_engine_ == nullptr) << "Must Close() before destruction";
+}
+
+void LlvmCodeGen::Close() {
+  if (memory_manager_ != nullptr) {
+    mem_tracker_->Release(memory_manager_->bytes_tracked());
+    memory_manager_ = nullptr;
+  }
+  if (mem_tracker_ != nullptr) mem_tracker_->Close();
 
   // Execution engine executes callback on event listener, so tear down engine first.
   execution_engine_.reset();
   symbol_emitter_.reset();
+  module_ = nullptr;
 }
 
 void LlvmCodeGen::EnableOptimizations(bool enable) {
@@ -1242,6 +1250,7 @@ Status LlvmCodeGen::GetSymbols(const string& file, const string& module_id,
   for (const Function& fn: codegen->module_->functions()) {
     if (fn.isMaterializable()) symbols->insert(fn.getName());
   }
+  codegen->Close();
   return Status::OK();
 }
 

be/src/codegen/llvm-codegen.h

@@ -154,9 +154,12 @@ class LlvmCodeGen {
   static Status CreateImpalaCodegen(RuntimeState* state, MemTracker* parent_mem_tracker,
       const std::string& id, boost::scoped_ptr<LlvmCodeGen>* codegen);
 
-  /// Removes all jit compiled dynamically linked functions from the process.
   ~LlvmCodeGen();
 
+  /// Releases all resources associated with the codegen object. It is invalid to call
+  /// any other API methods after calling close.
+  void Close();
+
   RuntimeProfile* runtime_profile() { return &profile_; }
   RuntimeProfile::Counter* codegen_timer() { return codegen_timer_; }
 
@@ -668,8 +671,9 @@ class LlvmCodeGen {
   RuntimeProfile profile_;
 
   /// MemTracker used for tracking memory consumed by codegen. Connected to a parent
-  /// MemTracker if one was provided during initialization.
-  boost::scoped_ptr<MemTracker> mem_tracker_;
+  /// MemTracker if one was provided during initialization. Owned by the ObjectPool
+  /// provided in the constructor.
+  MemTracker* mem_tracker_;
 
   /// Time spent reading the .ir file from the file system.
   RuntimeProfile::Counter* load_module_timer_;

be/src/exec/data-sink.cc

@@ -192,14 +192,8 @@ void DataSink::Close(RuntimeState* state) {
   ScalarExprEvaluator::Close(output_expr_evals_, state);
   ScalarExpr::Close(output_exprs_);
   if (expr_mem_pool() != nullptr) expr_mem_pool_->FreeAll();
-  if (expr_mem_tracker_ != NULL) {
-    expr_mem_tracker_->UnregisterFromParent();
-    expr_mem_tracker_.reset();
-  }
-  if (mem_tracker_ != NULL) {
-    mem_tracker_->UnregisterFromParent();
-    mem_tracker_.reset();
-  }
+  if (expr_mem_tracker_ != nullptr) expr_mem_tracker_->Close();
+  if (mem_tracker_ != nullptr) mem_tracker_->Close();
   closed_ = true;
 }
 

be/src/exec/exchange-node.cc

@@ -123,7 +123,6 @@ void ExchangeNode::Close(RuntimeState* state) {
   if (is_closed()) return;
   if (less_than_.get() != nullptr) less_than_->Close(state);
   if (stream_recvr_ != nullptr) stream_recvr_->Close();
-  stream_recvr_.reset();
   ScalarExpr::Close(ordering_exprs_);
   ExecNode::Close(state);
 }

be/src/exec/exec-node.cc

@@ -211,11 +211,15 @@ void ExecNode::Close(RuntimeState* state) {
         &buffer_pool_client_, resource_profile_.min_reservation);
     state->exec_env()->buffer_pool()->DeregisterClient(&buffer_pool_client_);
   }
-  if (mem_tracker() != NULL && mem_tracker()->consumption() != 0) {
-    LOG(WARNING) << "Query " << state->query_id() << " may have leaked memory." << endl
-                 << state->instance_mem_tracker()->LogUsage();
-    DCHECK_EQ(mem_tracker()->consumption(), 0)
-        << "Leaked memory." << endl << state->instance_mem_tracker()->LogUsage();
+  if (expr_mem_tracker_ != nullptr) expr_mem_tracker_->Close();
+  if (mem_tracker_ != nullptr) {
+    if (mem_tracker()->consumption() != 0) {
+      LOG(WARNING) << "Query " << state->query_id() << " may have leaked memory." << endl
+                   << state->instance_mem_tracker()->LogUsage();
+      DCHECK_EQ(mem_tracker()->consumption(), 0)
+          << "Leaked memory." << endl << state->instance_mem_tracker()->LogUsage();
+    }
+    mem_tracker_->Close();
   }
 }
 

be/src/exec/nested-loop-join-node.cc

@@ -138,10 +138,7 @@ void NestedLoopJoinNode::Close(RuntimeState* state) {
   if (is_closed()) return;
   ScalarExprEvaluator::Close(join_conjunct_evals_, state);
   ScalarExpr::Close(join_conjuncts_);
-  if (builder_ != NULL) {
-    builder_->Close(state);
-    builder_.reset();
-  }
+  if (builder_ != NULL) builder_->Close(state);
   build_batches_ = NULL;
   if (matching_build_rows_ != NULL) {
     mem_tracker()->Release(matching_build_rows_->MemUsage());

be/src/exprs/expr-codegen-test.cc

@@ -348,6 +348,7 @@ TEST_F(ExprCodegenTest, TestInlineConstFnAttrs) {
   EXPECT_EQ(result.is_null, false);
   EXPECT_EQ(result.val8, 100003);
   CheckFnAttr();
+  codegen->Close();
 }
 
 }

be/src/runtime/bufferpool/reservation-tracker-test.cc

@@ -284,8 +284,8 @@ TEST_F(ReservationTrackerTest, MemTrackerIntegrationTwoLevel) {
   ASSERT_EQ(0, child_mem_tracker2.consumption());
   ASSERT_EQ(0, root_mem_tracker.consumption());
   ASSERT_EQ(0, root_.GetUsedReservation());
-  child_mem_tracker1.UnregisterFromParent();
-  child_mem_tracker2.UnregisterFromParent();
+  child_mem_tracker1.Close();
+  child_mem_tracker2.Close();
 }
 
 TEST_F(ReservationTrackerTest, MemTrackerIntegrationMultiLevel) {
@@ -364,7 +364,7 @@ TEST_F(ReservationTrackerTest, MemTrackerIntegrationMultiLevel) {
 
   for (int i = HIERARCHY_DEPTH - 1; i >= 0; --i) {
     reservations[i].Close();
-    if (i != 0) mem_trackers[i]->UnregisterFromParent();
+    if (i != 0) mem_trackers[i]->Close();
   }
 }
 
@@ -469,12 +469,12 @@ TEST_F(ReservationTrackerTest, TransferReservation) {
   EXPECT_EQ(GRANDPARENT_LIMIT, root_.GetReservation());
 
   child->Close();
-  child_mem_tracker->UnregisterFromParent();
+  child_mem_tracker->Close();
   aunt->Close();
   parent->Close();
-  parent_mem_tracker->UnregisterFromParent();
+  parent_mem_tracker->Close();
   grandparent->Close();
-  grandparent_mem_tracker->UnregisterFromParent();
+  grandparent_mem_tracker->Close();
 }
 }
 

be/src/runtime/coordinator.cc

@@ -144,7 +144,7 @@ Status Coordinator::Exec() {
   lock_guard<mutex> l(lock_);
 
   query_state_ = ExecEnv::GetInstance()->query_exec_mgr()->CreateQueryState(query_ctx_);
-  filter_mem_tracker_.reset(new MemTracker(
+  filter_mem_tracker_ = query_state_->obj_pool()->Add(new MemTracker(
       -1, "Runtime Filter (Coordinator)", query_state_->query_mem_tracker(), false));
 
   InitFragmentStats();
@@ -1090,15 +1090,11 @@ void Coordinator::ReleaseResources() {
     lock_guard<SpinLock> l(filter_lock_);
     for (auto& filter : filter_routing_table_) {
       FilterState* state = &filter.second;
-      state->Disable(filter_mem_tracker_.get());
+      state->Disable(filter_mem_tracker_);
     }
   }
   // This may be NULL while executing UDFs.
-  if (filter_mem_tracker_.get() != nullptr) {
-    // TODO: move this elsewhere, this isn't releasing resources (it's dismantling
-    // control structures)
-    filter_mem_tracker_->UnregisterFromParent();
-  }
+  if (filter_mem_tracker_ != nullptr) filter_mem_tracker_->Close();
   // Need to protect against failed Prepare(), where root_sink() would not be set.
   if (coord_sink_ != nullptr) {
     coord_sink_->CloseConsumer();
@@ -1171,7 +1167,7 @@ void Coordinator::UpdateFilter(const TUpdateFilterParams& params) {
     }
 
     // Filter is complete, and can be released.
-    state->Disable(filter_mem_tracker_.get());
+    state->Disable(filter_mem_tracker_);
     DCHECK(state->bloom_filter() == nullptr);
   }
 
@@ -1196,15 +1192,15 @@ void Coordinator::FilterState::ApplyUpdate(const TUpdateFilterParams& params,
 
   --pending_count_;
   if (params.bloom_filter.always_true) {
-    Disable(coord->filter_mem_tracker_.get());
+    Disable(coord->filter_mem_tracker_);
   } else if (bloom_filter_.get() == nullptr) {
     int64_t heap_space = params.bloom_filter.directory.size();
-    if (!coord->filter_mem_tracker_.get()->TryConsume(heap_space)) {
+    if (!coord->filter_mem_tracker_->TryConsume(heap_space)) {
       VLOG_QUERY << "Not enough memory to allocate filter: "
                  << PrettyPrinter::Print(heap_space, TUnit::BYTES)
                  << " (query: " << coord->query_id() << ")";
       // Disable, as one missing update means a correct filter cannot be produced.
-      Disable(coord->filter_mem_tracker_.get());
+      Disable(coord->filter_mem_tracker_);
     } else {
       bloom_filter_.reset(new TBloomFilter());
       // Workaround for fact that parameters are const& for Thrift RPCs - yet we want to

be/src/runtime/coordinator.h

@@ -249,8 +249,9 @@ class Coordinator { // NOLINT: The member variables could be re-ordered to save
   TRuntimeFilterMode::type filter_mode_;
 
   /// Tracks the memory consumed by runtime filters during aggregation. Child of
-  /// the query mem tracker in 'query_state_' and set in Exec().
-  std::unique_ptr<MemTracker> filter_mem_tracker_;
+  /// the query mem tracker in 'query_state_' and set in Exec(). Stored in
+  /// query_state_->obj_pool() so it has same lifetime as other MemTrackers.
+  MemTracker* filter_mem_tracker_ = nullptr;
 
   /// Object pool owned by the coordinator.
   boost::scoped_ptr<ObjectPool> obj_pool_;

be/src/runtime/data-stream-recvr.cc

@@ -350,8 +350,7 @@ void DataStreamRecvr::Close() {
     sender_queues_[i]->Close();
   }
   merger_.reset();
-  mem_tracker_->UnregisterFromParent();
-  mem_tracker_.reset();
+  mem_tracker_->Close();
 }
 
 DataStreamRecvr::~DataStreamRecvr() {

be/src/runtime/disk-io-mgr.cc

@@ -362,8 +362,7 @@ DiskIoMgr::~DiskIoMgr() {
     delete disk_queues_[i];
   }
 
-  if (free_buffer_mem_tracker_ != nullptr) free_buffer_mem_tracker_->UnregisterFromParent();
-
+  if (free_buffer_mem_tracker_ != nullptr) free_buffer_mem_tracker_->Close();
   if (cached_read_options_ != nullptr) hadoopRzOptionsFree(cached_read_options_);
 }
 

be/src/runtime/initial-reservations.cc

@@ -40,11 +40,11 @@ namespace impala {
 InitialReservations::InitialReservations(ObjectPool* obj_pool,
     ReservationTracker* query_reservation, MemTracker* query_mem_tracker,
     int64_t initial_reservation_total_claims)
-  : remaining_initial_reservation_claims_(initial_reservation_total_claims) {
-  MemTracker* initial_reservation_tracker = obj_pool->Add(
-      new MemTracker(-1, "Unclaimed reservations", query_mem_tracker, false));
+  : initial_reservation_mem_tracker_(obj_pool->Add(
+      new MemTracker(-1, "Unclaimed reservations", query_mem_tracker, false))),
+      remaining_initial_reservation_claims_(initial_reservation_total_claims) {
   initial_reservations_.InitChildTracker(nullptr, query_reservation,
-      initial_reservation_tracker, numeric_limits<int64_t>::max());
+      initial_reservation_mem_tracker_, numeric_limits<int64_t>::max());
 }
 
 Status InitialReservations::Init(
@@ -86,5 +86,6 @@ void InitialReservations::Return(BufferPool::ClientHandle* src, int64_t bytes) {
 
 void InitialReservations::ReleaseResources() {
   initial_reservations_.Close();
+  initial_reservation_mem_tracker_->Close();
 }
 }

be/src/runtime/initial-reservations.h

@@ -70,6 +70,8 @@ class InitialReservations {
   // Return() returns reservations to.
   ReservationTracker initial_reservations_;
 
+  MemTracker* const initial_reservation_mem_tracker_;
+
   /// The total bytes of additional reservations that we expect to be claimed.
   /// initial_reservations_->GetReservation() <= remaining_initial_reservation_claims_.
   int64_t remaining_initial_reservation_claims_;

be/src/runtime/mem-tracker.cc

@@ -28,6 +28,7 @@
 #include "util/debug-util.h"
 #include "util/mem-info.h"
 #include "util/pretty-printer.h"
+#include "util/test-info.h"
 #include "util/uid-util.h"
 
 #include "common/names.h"
@@ -109,8 +110,18 @@ void MemTracker::AddChildTracker(MemTracker* tracker) {
   tracker->child_tracker_it_ = child_trackers_.insert(child_trackers_.end(), tracker);
 }
 
-void MemTracker::UnregisterFromParent() {
-  DCHECK(parent_ != NULL);
+void MemTracker::Close() {
+  if (closed_) return;
+  if (consumption_metric_ == nullptr) {
+    DCHECK_EQ(consumption_->current_value(), 0) << label_ << "\n"
+                                                << GetStackTrace() << "\n"
+                                                << LogUsage("");
+  }
+  closed_ = true;
+}
+
+void MemTracker::CloseAndUnregisterFromParent() {
+  Close();
   lock_guard<SpinLock> l(parent_->child_trackers_lock_);
   parent_->child_trackers_.erase(child_tracker_it_);
   child_tracker_it_ = parent_->child_trackers_.end();
@@ -187,9 +198,10 @@ MemTracker* MemTracker::CreateQueryMemTracker(const TUniqueId& id,
 }
 
 MemTracker::~MemTracker() {
-  DCHECK_EQ(consumption_->current_value(), 0) << label_ << "\n"
-                                              << GetStackTrace() << "\n"
-                                              << LogUsage("");
+  // We should explicitly close MemTrackers in the context of a daemon process. It is ok
+  // if backend tests don't call Close() to make tests more concise.
+  if (TestInfo::is_test()) Close();
+  DCHECK(closed_) << label_;
   delete reservation_counters_.Load();
 }
 

be/src/runtime/mem-tracker.h

@@ -95,8 +95,19 @@ class MemTracker {
 
   ~MemTracker();
 
-  /// Removes this tracker from parent_->child_trackers_.
-  void UnregisterFromParent();
+  /// Closes this MemTracker. After closing it is invalid to consume memory on this
+  /// tracker and the tracker's consumption counter (which may be owned by a
+  /// RuntimeProfile, not this MemTracker) can be safely destroyed. MemTrackers without
+  /// consumption metrics in the context of a daemon must always be closed.
+  /// Idempotent: calling multiple times has no effect.
+  void Close();
+
+  /// Closes the MemTracker and deregisters it from its parent. Can be called before
+  /// destruction to prevent other threads from getting a reference to the MemTracker
+  /// via its parent. Only used to deregister the query-level MemTracker from the
+  /// global hierarchy.
+  /// TODO: IMPALA-3200: this is also used by BufferedBlockMgr, which will be deleted.
+  void CloseAndUnregisterFromParent();
 
   /// Include counters from a ReservationTracker in logs and other diagnostics.
   /// The counters should be owned by the fragment's RuntimeProfile.
@@ -112,6 +123,7 @@ class MemTracker {
 
   /// Increases consumption of this tracker and its ancestors by 'bytes'.
   void Consume(int64_t bytes) {
+    DCHECK(!closed_) << label_;
     if (bytes <= 0) {
       if (bytes < 0) Release(-bytes);
       return;
@@ -121,11 +133,10 @@ class MemTracker {
       RefreshConsumptionFromMetric();
       return;
     }
-    for (std::vector<MemTracker*>::iterator tracker = all_trackers_.begin();
-         tracker != all_trackers_.end(); ++tracker) {
-      (*tracker)->consumption_->Add(bytes);
-      if ((*tracker)->consumption_metric_ == NULL) {
-        DCHECK_GE((*tracker)->consumption_->current_value(), 0);
+    for (MemTracker* tracker : all_trackers_) {
+      tracker->consumption_->Add(bytes);
+      if (tracker->consumption_metric_ == NULL) {
+        DCHECK_GE(tracker->consumption_->current_value(), 0);
       }
     }
   }
@@ -136,11 +147,13 @@ class MemTracker {
   /// to update tracking on a particular mem tracker but the consumption against
   /// the limit recorded in one of its ancestors already happened.
   void ConsumeLocal(int64_t bytes, MemTracker* end_tracker) {
+    DCHECK(!closed_) << label_;
     DCHECK(consumption_metric_ == NULL) << "Should not be called on root.";
-    for (int i = 0; i < all_trackers_.size(); ++i) {
-      if (all_trackers_[i] == end_tracker) return;
-      DCHECK(!all_trackers_[i]->has_limit());
-      all_trackers_[i]->consumption_->Add(bytes);
+    for (MemTracker* tracker : all_trackers_) {
+      if (tracker == end_tracker) return;
+      DCHECK(!tracker->has_limit());
+      DCHECK(!tracker->closed_) << tracker->label_;
+      tracker->consumption_->Add(bytes);
     }
     DCHECK(false) << "end_tracker is not an ancestor";
   }
@@ -155,6 +168,7 @@ class MemTracker {
   /// Returns true if the try succeeded.
   WARN_UNUSED_RESULT
   bool TryConsume(int64_t bytes) {
+    DCHECK(!closed_) << label_;
     if (consumption_metric_ != NULL) RefreshConsumptionFromMetric();
     if (UNLIKELY(bytes <= 0)) return true;
     int i;
@@ -195,6 +209,7 @@ class MemTracker {
 
   /// Decreases consumption of this tracker and its ancestors by 'bytes'.
   void Release(int64_t bytes) {
+    DCHECK(!closed_) << label_;
     if (bytes <= 0) {
       if (bytes < 0) Consume(-bytes);
       return;
@@ -204,30 +219,26 @@ class MemTracker {
       RefreshConsumptionFromMetric();
       return;
     }
-    for (std::vector<MemTracker*>::iterator tracker = all_trackers_.begin();
-         tracker != all_trackers_.end(); ++tracker) {
-      (*tracker)->consumption_->Add(-bytes);
+    for (MemTracker* tracker : all_trackers_) {
+      tracker->consumption_->Add(-bytes);
       /// If a UDF calls FunctionContext::TrackAllocation() but allocates less than the
       /// reported amount, the subsequent call to FunctionContext::Free() may cause the
       /// process mem tracker to go negative until it is synced back to the tcmalloc
       /// metric. Don't blow up in this case. (Note that this doesn't affect non-process
       /// trackers since we can enforce that the reported memory usage is internally
       /// consistent.)
-      if ((*tracker)->consumption_metric_ == NULL) {
-        DCHECK_GE((*tracker)->consumption_->current_value(), 0)
-          << std::endl << (*tracker)->LogUsage();
+      if (tracker->consumption_metric_ == NULL) {
+        DCHECK_GE(tracker->consumption_->current_value(), 0)
+          << std::endl << tracker->LogUsage();
       }
     }
-
-    /// TODO: Release brokered memory?
   }
 
   /// Returns true if a valid limit of this tracker or one of its ancestors is
   /// exceeded.
   bool AnyLimitExceeded() {
-    for (std::vector<MemTracker*>::iterator tracker = limit_trackers_.begin();
-         tracker != limit_trackers_.end(); ++tracker) {
-      if ((*tracker)->LimitExceeded()) return true;
+    for (MemTracker* tracker : limit_trackers_) {
+      if (tracker->LimitExceeded()) return true;
     }
     return false;
   }
@@ -403,6 +414,8 @@ class MemTracker {
   /// if consumption is 0.
   bool log_usage_if_zero_;
 
+  bool closed_ = false;
+
   /// The number of times the GcFunctions were called.
   IntCounter* num_gcs_metric_;
 

be/src/runtime/query-state.cc

@@ -94,8 +94,13 @@ void QueryState::ReleaseResources() {
   // Release any remaining reservation.
   if (initial_reservations_ != nullptr) initial_reservations_->ReleaseResources();
   if (buffer_reservation_ != nullptr) buffer_reservation_->Close();
-  // Avoid dangling reference from the parent of 'query_mem_tracker_'.
-  if (query_mem_tracker_ != nullptr) query_mem_tracker_->UnregisterFromParent();
+  if (query_mem_tracker_ != nullptr) {
+    // No more tracked memory should be used by the query after this point, so we can
+    // close the MemTracker and remove the whole query subtree of MemTrackers from the
+    // global tree. After this point nothing should be touching this query's MemTrackers
+    // and they can be safely destroyed.
+    query_mem_tracker_->CloseAndUnregisterFromParent();
+  }
   if (desc_tbl_ != nullptr) desc_tbl_->ReleaseResources();
   released_resources_ = true;
 }

be/src/runtime/runtime-filter-bank.cc

@@ -228,5 +228,5 @@ void RuntimeFilterBank::Close() {
   closed_ = true;
   obj_pool_.Clear();
   filter_mem_tracker_->Release(memory_allocated_->value());
-  filter_mem_tracker_->UnregisterFromParent();
+  filter_mem_tracker_->Close();
 }

be/src/runtime/runtime-state.cc

@@ -70,9 +70,7 @@ RuntimeState::RuntimeState(QueryState* query_state, const TPlanFragmentCtx& frag
         query_state->query_ctx().utc_timestamp_string))),
     exec_env_(exec_env),
     profile_(obj_pool(), "Fragment " + PrintId(instance_ctx.fragment_instance_id)),
-    instance_buffer_reservation_(new ReservationTracker),
-    is_cancelled_(false),
-    root_node_id_(-1) {
+    instance_buffer_reservation_(new ReservationTracker) {
   Init();
 }
 
@@ -85,10 +83,7 @@ RuntimeState::RuntimeState(
     now_(new TimestampValue(TimestampValue::Parse(qctx.now_string))),
     utc_timestamp_(new TimestampValue(TimestampValue::Parse(qctx.utc_timestamp_string))),
     exec_env_(exec_env),
-    profile_(obj_pool(), "<unnamed>"),
-    instance_buffer_reservation_(nullptr),
-    is_cancelled_(false),
-    root_node_id_(-1) {
+    profile_(obj_pool(), "<unnamed>") {
   if (query_ctx().request_pool.empty()) {
     const_cast<TQueryCtx&>(query_ctx()).request_pool = "test-pool";
   }
@@ -97,7 +92,7 @@ RuntimeState::RuntimeState(
 }
 
 RuntimeState::~RuntimeState() {
-  DCHECK(instance_mem_tracker_ == nullptr) << "Must call ReleaseResources()";
+  DCHECK(released_resources_) << "Must call ReleaseResources()";
 }
 
 void RuntimeState::Init() {
@@ -233,33 +228,30 @@ void RuntimeState::UnregisterReaderContexts() {
 }
 
 void RuntimeState::ReleaseResources() {
-  // TODO: IMPALA-5587: control structures (e.g. MemTrackers) shouldn't be destroyed here.
+  DCHECK(!released_resources_);
   UnregisterReaderContexts();
   if (filter_bank_ != nullptr) filter_bank_->Close();
   if (resource_pool_ != nullptr) {
     exec_env_->thread_mgr()->UnregisterPool(resource_pool_);
   }
-  codegen_.reset(); // Release any memory associated with codegen.
+  // Release any memory associated with codegen.
+  if (codegen_ != nullptr) codegen_->Close();
 
   // Release the reservation, which should be unused at the point.
   if (instance_buffer_reservation_ != nullptr) instance_buffer_reservation_->Close();
 
-  // 'query_mem_tracker()' must be valid as long as 'instance_mem_tracker_' is so
-  // delete 'instance_mem_tracker_' first.
-  // LogUsage() walks the MemTracker tree top-down when the memory limit is exceeded, so
-  // break the link between 'instance_mem_tracker_' and its parent before
-  // 'instance_mem_tracker_' and its children are destroyed.
-  instance_mem_tracker_->UnregisterFromParent();
+  // No more memory should be tracked for this instance at this point.
   if (instance_mem_tracker_->consumption() != 0) {
     LOG(WARNING) << "Query " << query_id() << " may have leaked memory." << endl
                  << instance_mem_tracker_->LogUsage();
   }
-  instance_mem_tracker_.reset();
+  instance_mem_tracker_->Close();
 
   if (local_query_state_.get() != nullptr) {
     // if we created this QueryState, we must call ReleaseResources()
     local_query_state_->ReleaseResources();
   }
+  released_resources_ = true;
 }
 
 const std::string& RuntimeState::GetEffectiveUser() const {

be/src/runtime/runtime-state.h

@@ -302,7 +302,7 @@ class RuntimeState {
   /// Helper to call QueryState::StartSpilling().
   Status StartSpilling(MemTracker* mem_tracker);
 
-  /// Release resources and prepare this object for destruction.
+  /// Release resources and prepare this object for destruction. Can only be called once.
   void ReleaseResources();
 
  private:
@@ -376,7 +376,10 @@ class RuntimeState {
   boost::scoped_ptr<ReservationTracker> instance_buffer_reservation_;
 
   /// if true, execution should stop with a CANCELLED status
-  bool is_cancelled_;
+  bool is_cancelled_ = false;
+
+  /// if true, ReleaseResources() was called.
+  bool released_resources_ = false;
 
   /// Non-OK if an error has occurred and query execution should abort. Used only for
   /// asynchronously reporting such errors (e.g., when a UDF reports an error), so this
@@ -397,7 +400,7 @@ class RuntimeState {
   /// 2) It is different between different fragments, so we do not run into hash
   /// collisions after data partitioning (across fragments). See IMPALA-219 for more
   /// details.
-  PlanNodeId root_node_id_;
+  PlanNodeId root_node_id_ = -1;
 
   /// Manages runtime filters that are either produced or consumed (or both!) by plan
   /// nodes that share this runtime state.