You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2026/06/01 00:56:50 UTC
(impala) 01/02: IMPALA-14863: Fixes Potential OOM During Row Batch Processing
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 12e79a22ef438c3b515ba0218e8e002b3b234cc6
Author: jasonmfehr <jf...@cloudera.com>
AuthorDate: Wed Apr 29 13:24:18 2026 -0700
IMPALA-14863: Fixes Potential OOM During Row Batch Processing
Impala checks at the end of every row batch to determine if the query
or process memory limit has been exceeded. If the batch processing
requires evaluating expressions that allocate memory for their
results (such as UPPER() or LOWER()), that memory is allocated
without checking any memory limits. If a large number of these
expression evaluations are performed during a single row batch,
memory will be consumed until the process memory limit is reached.
At that point, the executor begins failing other queries with cannot
allocate memory errors as they request memory. However, the row batch
processing continues to run since it does no memory limit checking
during batch processing.
This patch determines a limit for the amount of memory that conjunct
expression evaluation results can consume as a percentage of the
lowest soft memory limit for the query execution. Before evaluating
conjuncts for each join probe row, the expression results memory pool
size is cleared if its allocated memory has exceeded this threshold.
Clearing the memory pool does not result in memory being freed but
instead marks all memory in the pool as available for re-use.
Testing accomplished by a new custom cluster test that replicates
the query that revealed this issue.
Regression functionality testing performed by running existing test
suite. Regression performance testing performed by running AB test
job.
Generated-by: Github Copilot (GPT-5.3-Codex)
Change-Id: Ic54b5c39e1388681275681f22e61b27728dba5af
Reviewed-on: http://gerrit.cloudera.org:8080/24159
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/codegen/llvm-codegen-test.cc | 54 ++++++++++-
be/src/codegen/llvm-codegen.cc | 19 +++-
be/src/codegen/llvm-codegen.h | 4 +
be/src/exec/partitioned-hash-join-node-ir.cc | 10 ++
be/src/exec/partitioned-hash-join-node.cc | 20 ++++
be/src/exec/partitioned-hash-join-node.h | 9 ++
be/src/exec/partitioned-hash-join-node.inline.h | 16 ++++
tests/custom_cluster/test_conjuncts.py | 122 ++++++++++++++++++++++++
8 files changed, 251 insertions(+), 3 deletions(-)
diff --git a/be/src/codegen/llvm-codegen-test.cc b/be/src/codegen/llvm-codegen-test.cc
index 8fb707488..86266e639 100644
--- a/be/src/codegen/llvm-codegen-test.cc
+++ b/be/src/codegen/llvm-codegen-test.cc
@@ -41,6 +41,8 @@ using std::unique_ptr;
namespace impala {
+typedef void (*TestLoopFn)(int, int64_t*);
+
class CodegenFnPtrTest : public testing:: Test {
protected:
typedef int (*FnPtr)(int);
@@ -234,7 +236,6 @@ llvm::Function* CodegenInnerLoop(
TEST_F(LlvmCodeGenTest, ReplaceFnCall) {
const string loop_call_name("_Z21DefaultImplementationPl");
const string loop_name("_Z8TestLoopiPl");
- typedef void (*TestLoopFn)(int, int64_t*);
string module_file;
PathBuilder::GetFullPath("llvm-ir/test-loop.bc", &module_file);
@@ -309,6 +310,57 @@ TEST_F(LlvmCodeGenTest, ReplaceFnCall) {
codegen->Close();
}
+// This test loads a precompiled IR file (compiled from testdata/llvm/test-loop.cc).
+// The test contains two functions, an outer loop function and an inner loop function.
+// The outer loop calls the inner loop function.
+// The test will
+// 1. Create a LlvmCodegen object from the precompiled file.
+// 2. Replaces the inner loop with with an implementation that increments the counter
+// each time it is called.
+// 3. Remove the inner loop call from the outer loop.
+// 4. Compile and run all loops and make sure the inner loop is not called.
+TEST_F(LlvmCodeGenTest, RemoveFnCall) {
+ const string loop_call_name("_Z21DefaultImplementationPl");
+ const string loop_name("_Z8TestLoopiPl");
+
+ string module_file;
+ PathBuilder::GetFullPath("llvm-ir/test-loop.bc", &module_file);
+
+ // Part 1: Load the module and make sure everything is loaded correctly.
+ scoped_ptr<LlvmCodeGen> codegen;
+ ASSERT_OK(CreateFromFile(module_file.c_str(), &codegen));
+ EXPECT_TRUE(codegen.get() != nullptr);
+
+ llvm::Function* loop = codegen->GetFunction(loop_name, false);
+ EXPECT_TRUE(loop != nullptr);
+ EXPECT_EQ(loop->arg_size(), 2);
+
+ // Part 2: Replace the default inner loop implementation that does not modify counter
+ // with a function that increments the counter each time it is called.
+ llvm::Function* jitted_loop_call = CodegenInnerLoop(codegen.get(), 1);
+ llvm::Function* jitted_loop = codegen->CloneFunction(loop);
+ int num_replaced =
+ codegen->ReplaceCallSites(jitted_loop, jitted_loop_call, loop_call_name);
+ EXPECT_EQ(1, num_replaced);
+ EXPECT_TRUE(VerifyFunction(codegen.get(), jitted_loop));
+
+ // Part 3: Remove the inner loop call from the outer loop.
+ EXPECT_EQ(1, codegen->RemoveCallSites(jitted_loop, "JittedInnerLoop"));
+
+ // Part 4: compile all the function and run it.
+ CodegenFnPtr<TestLoopFn> new_loop;
+ AddFunctionToJit(codegen.get(), jitted_loop, &new_loop);
+ ASSERT_OK(LlvmCodeGenTest::FinalizeModule(codegen.get()));
+ ASSERT_TRUE(new_loop.load() != nullptr);
+
+ int64_t counter = 0;
+ TestLoopFn new_loop_fn = new_loop.load();
+ new_loop_fn(5, &counter);
+ EXPECT_EQ(0, counter);
+
+ codegen->Close();
+}
+
// Test function for c++/ir interop for strings. Function will do:
// int StringTest(StringValue* strval) {
// strval->ptr[0] = 'A';
diff --git a/be/src/codegen/llvm-codegen.cc b/be/src/codegen/llvm-codegen.cc
index 0e3cd0ccd..90cdff6a2 100644
--- a/be/src/codegen/llvm-codegen.cc
+++ b/be/src/codegen/llvm-codegen.cc
@@ -991,8 +991,8 @@ Status LlvmCodeGen::LoadFunction(const TFunction& fn, const string& symbol,
int LlvmCodeGen::ReplaceCallSites(
llvm::Function* caller, llvm::Function* new_fn, const string& target_name) {
DCHECK(!is_compiled_);
- DCHECK(caller->getParent() == module_);
DCHECK(caller != NULL);
+ DCHECK(caller->getParent() == module_);
DCHECK(new_fn != NULL);
DCHECK(find(handcrafted_functions_.begin(), handcrafted_functions_.end(), new_fn)
== handcrafted_functions_.end()
@@ -1012,8 +1012,8 @@ int LlvmCodeGen::ReplaceCallSites(
int LlvmCodeGen::ReplaceCallSitesWithValue(
llvm::Function* caller, llvm::Value* replacement, const string& target_name) {
DCHECK(!is_compiled_);
- DCHECK(caller->getParent() == module_);
DCHECK(caller != NULL);
+ DCHECK(caller->getParent() == module_);
DCHECK(replacement != NULL);
vector<llvm::CallInst*> call_sites;
@@ -1032,6 +1032,21 @@ int LlvmCodeGen::ReplaceCallSitesWithBoolConst(llvm::Function* caller, bool cons
return ReplaceCallSitesWithValue(caller, replacement, target_name);
}
+int LlvmCodeGen::RemoveCallSites(llvm::Function* caller, const std::string& target_name) {
+ DCHECK(!is_compiled_);
+ DCHECK(caller != NULL);
+ DCHECK(caller->getParent() == module_);
+
+ vector<llvm::CallInst*> call_sites;
+ FindCallSites(caller, target_name, &call_sites);
+
+ for (llvm::CallInst* call_instr : call_sites) {
+ call_instr->eraseFromParent();
+ }
+
+ return call_sites.size();
+}
+
int LlvmCodeGen::InlineConstFnAttrs(const FunctionContext::TypeDesc& ret_type,
const vector<FunctionContext::TypeDesc>& arg_types, llvm::Function* fn) {
int replaced = 0;
diff --git a/be/src/codegen/llvm-codegen.h b/be/src/codegen/llvm-codegen.h
index 0450d06ba..db5a3c206 100644
--- a/be/src/codegen/llvm-codegen.h
+++ b/be/src/codegen/llvm-codegen.h
@@ -410,6 +410,10 @@ class LlvmCodeGen {
int ReplaceCallSitesWithValue(llvm::Function* caller, llvm::Value* replacement,
const std::string& target_name);
+ /// Remove calls to functions with a name of 'target_name' from caller. The return value
+ /// is the number of calls removed.
+ int RemoveCallSites(llvm::Function* caller, const std::string& target_name);
+
/// This function replaces calls to FunctionContextImpl::GetConstFnAttr() with constants
/// derived from 'return_type', 'arg_types' and the runtime state 'state_'. Please note
/// that this function only replaces call instructions inside 'fn' so to replace the
diff --git a/be/src/exec/partitioned-hash-join-node-ir.cc b/be/src/exec/partitioned-hash-join-node-ir.cc
index e8ba92b8e..674bb4442 100644
--- a/be/src/exec/partitioned-hash-join-node-ir.cc
+++ b/be/src/exec/partitioned-hash-join-node-ir.cc
@@ -49,6 +49,8 @@ bool IR_ALWAYS_INLINE PartitionedHashJoinNode::ProcessProbeRowInnerJoin(
TupleRow* matched_build_row = hash_tbl_iterator_.GetRow();
DCHECK(matched_build_row != NULL);
+ ClearExprResultsPool(num_conjuncts, num_other_join_conjuncts);
+
// Create an output row with all probe/build tuples and evaluate the
// non-equi-join conjuncts.
CreateOutputRow(out_row, current_probe_row_, matched_build_row);
@@ -81,6 +83,8 @@ bool IR_ALWAYS_INLINE PartitionedHashJoinNode::ProcessProbeRowRightSemiJoins(
DCHECK(matched_build_row != NULL);
if (hash_tbl_iterator_.IsMatched()) continue;
+ ClearExprResultsPool(num_conjuncts, num_other_join_conjuncts);
+
// Evaluate the non-equi-join conjuncts against a temp row assembled from all
// build and probe tuples.
if (num_other_join_conjuncts > 0) {
@@ -120,6 +124,9 @@ bool IR_ALWAYS_INLINE PartitionedHashJoinNode::ProcessProbeRowLeftSemiJoins(
for (; !hash_tbl_iterator_.AtEnd(); hash_tbl_iterator_.NextDuplicate()) {
TupleRow* matched_build_row = hash_tbl_iterator_.GetRow();
DCHECK(matched_build_row != NULL);
+
+ ClearExprResultsPool(num_conjuncts, num_other_join_conjuncts);
+
// Evaluate the non-equi-join conjuncts against a temp row assembled from all
// build and probe tuples.
if (num_other_join_conjuncts > 0) {
@@ -184,6 +191,9 @@ bool IR_ALWAYS_INLINE PartitionedHashJoinNode::ProcessProbeRowOuterJoins(
for (; !hash_tbl_iterator_.AtEnd(); hash_tbl_iterator_.NextDuplicate()) {
TupleRow* matched_build_row = hash_tbl_iterator_.GetRow();
DCHECK(matched_build_row != NULL);
+
+ ClearExprResultsPool(num_conjuncts, num_other_join_conjuncts);
+
// Create an output row with all probe/build tuples and evaluate the
// non-equi-join conjuncts.
CreateOutputRow(out_row, current_probe_row_, matched_build_row);
diff --git a/be/src/exec/partitioned-hash-join-node.cc b/be/src/exec/partitioned-hash-join-node.cc
index 249e9f662..2b2a65469 100644
--- a/be/src/exec/partitioned-hash-join-node.cc
+++ b/be/src/exec/partitioned-hash-join-node.cc
@@ -20,9 +20,12 @@
#include <functional>
#include <numeric>
#include <sstream>
+#include <gflags/gflags.h>
#include <gutil/strings/substitute.h>
#include "codegen/llvm-codegen.h"
+#include "common/logging.h"
+#include "gen-cpp/Metrics_types.h"
#include "exec/blocking-join-node.inline.h"
#include "exec/exec-node-util.h"
#include "exec/exec-node.inline.h"
@@ -37,12 +40,18 @@
#include "runtime/runtime-state.h"
#include "util/cyclic-barrier.h"
#include "util/debug-util.h"
+#include "util/pretty-printer.h"
#include "util/runtime-profile-counters.h"
#include "gen-cpp/PlanNodes_types.h"
#include "common/names.h"
+DEFINE_double_hidden(result_pool_mem_multiplier, 0.5, "");
+DEFINE_validator(result_pool_mem_multiplier, [](const char* flagname, double val) {
+ return val > 0.0 && val <= 1.0;
+});
+
static const string PREPARE_FOR_READ_FAILED_ERROR_MSG =
"Failed to acquire initial read buffer for stream in hash join node $0. Reducing "
"query concurrency or increasing the memory limit may help this query to complete "
@@ -115,9 +124,13 @@ PartitionedHashJoinNode::PartitionedHashJoinNode(RuntimeState* state,
probe_exprs_(pnode.probe_exprs_),
other_join_conjuncts_(pnode.other_join_conjuncts_),
hash_table_config_(*pnode.hash_table_config_),
+ expr_results_mem_limit_(FLAGS_result_pool_mem_multiplier
+ * state->query_mem_tracker()->GetLowestLimit(MemLimit::SOFT)),
process_probe_batch_fn_(pnode.process_probe_batch_fn_),
process_probe_batch_fn_level0_(pnode.process_probe_batch_fn_level0_) {
memset(hash_tbls_, 0, sizeof(HashTable*) * PARTITION_FANOUT);
+ VLOG(2) << "Conjunct expression results pool memory limit: "
+ << PrettyPrinter::Print(expr_results_mem_limit_, TUnit::BYTES);
}
PartitionedHashJoinNode::~PartitionedHashJoinNode() {
@@ -1480,6 +1493,13 @@ Status PartitionedHashJoinPlanNode::CodegenProcessProbeBatch(
RETURN_IF_ERROR(ExecNode::CodegenEvalConjuncts(codegen, conjuncts_,
&eval_conjuncts_fn));
+ // Remove call to ClearExprResultsPool if no join conjuncts exist.
+ if (conjuncts_.empty() && other_join_conjuncts_.empty()) {
+ int removed =
+ codegen->RemoveCallSites(process_probe_batch_fn, "ClearExprResultsPool");
+ DCHECK_EQ(removed, 1);
+ }
+
// Replace all call sites with codegen version
int replaced = codegen->ReplaceCallSites(process_probe_batch_fn, eval_row_fn,
"EvalProbeRow");
diff --git a/be/src/exec/partitioned-hash-join-node.h b/be/src/exec/partitioned-hash-join-node.h
index c7e7b8901..c725418cd 100644
--- a/be/src/exec/partitioned-hash-join-node.h
+++ b/be/src/exec/partitioned-hash-join-node.h
@@ -546,6 +546,11 @@ class PartitionedHashJoinNode : public BlockingJoinNode {
std::string NodeDebugString() const;
+ /// Checks if the expression results memory pool has exceeded the memory limit defined
+ /// by 'expr_results_mem_limit_'. If it has, clears the mem pool.
+ inline void ClearExprResultsPool(const int num_conjuncts,
+ const int num_other_join_conjuncts);
+
RuntimeState* runtime_state_;
/// Our equi-join predicates "<lhs> = <rhs>" are separated into
@@ -562,6 +567,10 @@ class PartitionedHashJoinNode : public BlockingJoinNode {
/// an instance of the HashTableCtx in Prepare(). Not Owned.
const HashTableConfig& hash_table_config_;
+ /// Memory limit for the conjunct expression results pool used during building probe
+ /// batches.
+ const int64_t expr_results_mem_limit_;
+
/// Used for hash-related functionality, such as evaluating rows and calculating hashes.
/// This owns the evaluators for the build and probe expressions used during insertion
/// and probing of the hash tables.
diff --git a/be/src/exec/partitioned-hash-join-node.inline.h b/be/src/exec/partitioned-hash-join-node.inline.h
index a53b40efe..aad6e6724 100644
--- a/be/src/exec/partitioned-hash-join-node.inline.h
+++ b/be/src/exec/partitioned-hash-join-node.inline.h
@@ -32,6 +32,22 @@ inline void PartitionedHashJoinNode::ResetForProbe() {
ht_ctx_->expr_values_cache()->Reset();
}
+// Must not be inlined during codegen since calls to this function from ProcessProbeBatch
+// are removed by codegen if there are no conjuncts.
+//
+// The actual function calls are in the ProcessProbeRow* functions which are inlined into
+// ProcessProbeBatch, thus the codegen of ProcessProbeBatch will be able to eliminate the
+// calls to this function.
+inline void IR_NO_INLINE PartitionedHashJoinNode::ClearExprResultsPool(
+ const int num_conjuncts, const int num_other_join_conjuncts) {
+ DCHECK(expr_results_pool_.get() != nullptr);
+ DCHECK(expr_results_mem_limit_ > 0);
+ if (num_conjuncts + num_other_join_conjuncts > 0 &&
+ UNLIKELY(expr_results_pool_->total_allocated_bytes() > expr_results_mem_limit_)) {
+ expr_results_pool_->Clear();
+ }
+}
+
}
#endif
diff --git a/tests/custom_cluster/test_conjuncts.py b/tests/custom_cluster/test_conjuncts.py
new file mode 100644
index 000000000..99bccdef2
--- /dev/null
+++ b/tests/custom_cluster/test_conjuncts.py
@@ -0,0 +1,122 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import absolute_import, division, print_function
+import pytest
+
+from time import sleep
+
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+from tests.common.impala_connection import RUNNING
+from tests.common.skip import SkipIfExploration
+from tests.util.retry import retry
+
+
+@SkipIfExploration.is_not_exhaustive()
+class TestConjuncts(CustomClusterTestSuite):
+
+ @pytest.mark.stress
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(cluster_size=1)
+ def test_create_cache_many_tables(self, unique_database):
+ """Replicates the situation behund IMPALA-14863 where processing a row batch can cause
+ the memory usage of a fragment instance on an executor to consume up to the process
+ memory limit. Must be a custom cluster test because the fragment instances from the
+ bad query continues to run for a long time after the query is cancelled."""
+ self.client.execute("CREATE TABLE {}.test_left_fact (product_sk BIGINT,"
+ "drv_event_type STRING)".format(unique_database))
+ self.client.execute("CREATE TABLE {}.test_right_cte (product_sk BIGINT,"
+ "product_type STRING)".format(unique_database))
+
+ # Insert 1,024 rows into the Left Fact table (Exactly 1 Impala Batch). All rows share
+ # the same join key (999).
+ self.client.execute("""WITH ten AS (
+ SELECT 0 AS n UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3 UNION ALL
+ SELECT 4 UNION ALL SELECT 5 UNION ALL SELECT 6 UNION ALL SELECT 7 UNION ALL
+ SELECT 8 UNION ALL SELECT 9
+ )
+ INSERT INTO {}.test_left_fact
+ SELECT 999 AS product_sk, 'fail_event' AS drv_event_type FROM ten a
+ CROSS JOIN ten b CROSS JOIN ten c LIMIT 1024""".format(unique_database))
+
+ # Insert 1,000,000 rows into the Right CTE table. All 1 Million rows share the exact
+ # same join key (999).
+ self.client.execute("""WITH ten AS (
+ SELECT 0 AS n UNION ALL SELECT 1 UNION ALL SELECT 2 UNION ALL SELECT 3 UNION ALL
+ SELECT 4 UNION ALL SELECT 5 UNION ALL SELECT 6 UNION ALL SELECT 7 UNION ALL
+ SELECT 8 UNION ALL SELECT 9
+ )
+ INSERT INTO {}.test_right_cte
+ SELECT 999 AS product_sk, 'wrong_type' AS product_type FROM ten a
+ CROSS JOIN ten b CROSS JOIN ten c CROSS JOIN ten d CROSS JOIN ten e
+ CROSS JOIN ten f""".format(unique_database))
+
+ # Set a very low memory limit that will quickly be exceeded by the join conjuncts.
+ self.client.set_configuration_option("mem_limit_executors", "185mb")
+
+ # Run a query that will cause the join conjuncts to exceed the memory limit.
+ handle = self.client.execute_async("""SELECT STRAIGHT_JOIN count(1)
+ FROM {0}.test_left_fact f
+ LEFT JOIN {0}.test_right_cte p2
+ ON f.product_sk = p2.product_sk
+ WHERE (
+ UPPER(f.drv_event_type) LIKE '%_CONF'
+ OR
+ (
+ CASE
+ WHEN UPPER(p2.product_type) = 'COMMODITY-SWAP'
+ THEN UPPER(CONCAT(f.drv_event_type, '_SWAP_1'))
+ WHEN UPPER(p2.product_type) = 'METAL-FUTURE'
+ THEN UPPER(CONCAT(f.drv_event_type, '_METAL_2'))
+ WHEN UPPER(p2.product_type) = 'ENERGY-FUTURE'
+ THEN UPPER(CONCAT(f.drv_event_type, '_ENERGY_3'))
+ WHEN UPPER(p2.product_type) = 'AGRICULTURAL-FUTURE'
+ THEN UPPER(CONCAT(f.drv_event_type, '_AGRI_4'))
+ WHEN UPPER(p2.product_type) = 'SOFT-FUTURE'
+ THEN UPPER(CONCAT(f.drv_event_type, '_SOFT_5'))
+ WHEN UPPER(p2.product_type) = 'OTHER-FUTURE'
+ THEN UPPER(CONCAT(f.drv_event_type, '_OTHER_6'))
+ WHEN UPPER(p2.product_type) = 'METAL-OPTION'
+ THEN UPPER(CONCAT(f.drv_event_type, '_MOPT_7'))
+ WHEN UPPER(p2.product_type) = 'ENERGY-OPTION'
+ THEN UPPER(CONCAT(f.drv_event_type, '_EOPT_8'))
+ WHEN UPPER(p2.product_type) = 'SPREAD-OPTION'
+ THEN UPPER(CONCAT(f.drv_event_type, '_SOPT_9'))
+ WHEN UPPER(p2.product_type) = 'COMMODITY-INDEX'
+ THEN UPPER(CONCAT(f.drv_event_type, '_INDEX_10'))
+ ELSE UPPER(CONCAT(UPPER(f.drv_event_type), '_UNKNOWN'))
+ END = 'COMMODITY-FORWARD'
+ )
+ )""".format(unique_database))
+ self.client.wait_for_impala_state(handle, RUNNING, 10)
+
+ # Wait for the query to consume more than 270mb of memory indicating it has started
+ # evaluating the join conjuncts.
+ def __wait_for_270mb():
+ return self.get_metric("memory.total-used") > 270 * 1024 * 1024
+ retry(__wait_for_270mb, 10, 1, 1)
+ mem_before = self.get_metric("memory.total-used")
+
+ # Give the query some time to continue to run.
+ sleep(10)
+
+ # Assert the memory usage did not increase more than 25mb during the sleep. If the
+ # expression results pool is not being cleared properly, the memory usage will
+ # continue to grow as more rows are processed.
+ mem_after = self.get_metric("memory.total-used")
+ assert mem_after - mem_before < 25 * 1024 * 1024, "Memory usage increased during " \
+ "sleep. Before: {} After: {}".format(mem_before, mem_after)