This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 2f7775b93 Support RSS trimming for idle bthread stacks (#3668)
2f7775b93 is described below
commit 2f7775b936d487817e805120026fec28e1c9c1a7
Author: Bright Chen <[email protected]>
AuthorDate: Fri Oct 30 00:09:46 2026 +0800
Support RSS trimming for idle bthread stacks (#3668)
* Support RSS trimming for idle bthread stacks
* Disable by default
---
config_brpc.sh | 3 +-
src/bthread/stack.cpp | 126 +++++++-
src/bthread/stack.h | 14 +
src/bthread/stack_inl.h | 14 +-
src/bthread/task_control.cpp | 3 +
src/butil/object_pool.h | 18 ++
src/butil/object_pool_inl.h | 86 +++++-
test/bthread_stack_trim_unittest.cpp | 557 +++++++++++++++++++++++++++++++++++
test/object_pool_unittest.cpp | 174 +++++++++++
9 files changed, 975 insertions(+), 20 deletions(-)
diff --git a/config_brpc.sh b/config_brpc.sh
index e940e486f..b77f81bed 100755
--- a/config_brpc.sh
+++ b/config_brpc.sh
@@ -685,8 +685,9 @@ else
GTEST_HDR=$(find_dir_of_header_or_die gtest/gtest.h)
append_to_output_libs $GTEST_LIB " "
append_to_output_headers $GTEST_HDR " "
- append_to_output_linkings $GTEST_LIB gtest " "
+ # gtest_main depends on gtest, including when all tests are compiled out.
append_to_output_linkings $GTEST_LIB gtest_main " "
+ append_to_output_linkings $GTEST_LIB gtest " "
fi
append_to_output "endif"
diff --git a/src/bthread/stack.cpp b/src/bthread/stack.cpp
index 64608cd5e..2b25e77b0 100644
--- a/src/bthread/stack.cpp
+++ b/src/bthread/stack.cpp
@@ -20,11 +20,12 @@
// Date: Sun Sep 7 22:37:39 CST 2014
#include <unistd.h> // getpagesize
-#include <sys/mman.h> // mmap, munmap, mprotect
+#include <sys/mman.h> // mmap, munmap, mprotect,
madvise
#include <algorithm> // std::max
#include <stdlib.h> // posix_memalign
#include "butil/macros.h" // BAIDU_CASSERT
-#include "butil/memory/singleton_on_pthread_once.h"
+#include "butil/reloadable_flags.h"
+#include "butil/threading/platform_thread.h"
#include "butil/third_party/dynamic_annotations/dynamic_annotations.h" //
RunningOnValgrind
#include "butil/third_party/valgrind/valgrind.h" // VALGRIND_STACK_REGISTER
#include "bvar/passive_status.h"
@@ -37,6 +38,14 @@ DEFINE_int32(stack_size_large, 8388608, "size of large
stacks");
DEFINE_int32(guard_page_size, 4096, "size of guard page, allocate stacks by
malloc if it's 0(not recommended)");
DEFINE_int32(tc_stack_small, 32, "maximum small stacks cached by each thread");
DEFINE_int32(tc_stack_normal, 8, "maximum normal stacks cached by each
thread");
+DEFINE_int32(bthread_stack_trim_idle_s, 0,
+ "Release memory of stacks idle in the global pool for this many "
+ "seconds, 0 disables (Linux only, unsupported under ASan)");
+BUTIL_VALIDATE_GFLAG(bthread_stack_trim_idle_s, butil::NonNegativeInteger);
+DEFINE_int32(bthread_stack_trim_batch_size, 64,
+ "Object budget per idle stack trimming batch; whole chunks may "
+ "exceed it by up to 63 stacks (must be positive)");
+BUTIL_VALIDATE_GFLAG(bthread_stack_trim_batch_size, butil::PositiveInteger);
namespace bthread {
@@ -150,4 +159,117 @@ int* SmallStackClass::stack_size_flag =
&FLAGS_stack_size_small;
int* NormalStackClass::stack_size_flag = &FLAGS_stack_size_normal;
int* LargeStackClass::stack_size_flag = &FLAGS_stack_size_large;
+#if defined(BTHREAD_STACK_TRIM_SUPPORTED)
+
+#if defined(UNIT_TEST)
+// Replace only when no other thread is trimming stacks.
+int (*g_stack_madvise)(void*, size_t, int) = ::madvise;
+#endif
+
+namespace internal {
+
+static butil::static_atomic<int64_t> s_trimmed_count =
BUTIL_STATIC_ATOMIC_INIT(0);
+static int64_t get_trimmed_count(void*) {
+ return s_trimmed_count.load(butil::memory_order_relaxed);
+}
+static bvar::PassiveStatus<int64_t> bvar_trimmed_count(
+ "bthread_stack_trimmed_count", get_trimmed_count, nullptr);
+
+template <typename StackClass>
+static size_t trim_idle_stacks(int64_t idle_us) {
+ typedef typename StackFactory<StackClass>::Wrapper Wrapper;
+ size_t released = 0;
+ int error = 0;
+ auto fn = [&released, &error](Wrapper* w) noexcept {
+ StackStorage& s = w->storage;
+ // malloc owns unguarded stacks. Already processed stacks can return
+ // to the free list when a borrowed chunk is flushed from TLS.
+ if (w->context == nullptr || s.guardsize == 0) {
+ return;
+ }
+ void* low = (char*)s.bottom - s.stacksize;
+#if defined(UNIT_TEST)
+ int rc = g_stack_madvise(low, s.stacksize, MADV_DONTNEED);
+#else
+ int rc = madvise(low, s.stacksize, MADV_DONTNEED);
+#endif
+ // An error can follow partial reclamation across VMAs. Never resume
+ // the old continuation, even on failure; get_stack rebuilds it.
+ w->context = nullptr;
+ if (rc == 0) {
+ ++released;
+ } else {
+ error = errno;
+ }
+ };
+ size_t n = butil::process_idle_objects<Wrapper>(
+ idle_us, FLAGS_bthread_stack_trim_batch_size, fn);
+ if (released != 0) {
+ s_trimmed_count.fetch_add(released, butil::memory_order_relaxed);
+ }
+ if (error != 0) {
+ errno = error;
+ PLOG_EVERY_SECOND(ERROR) << "Fail to madvise idle stacks";
+ }
+ return n;
+}
+
+template <typename StackClass>
+static size_t trim_one_batch() {
+ // Runtime changes take effect at the next batch, including after sleeping.
+ int idle_s = FLAGS_bthread_stack_trim_idle_s;
+ if (idle_s <= 0) {
+ return 0;
+ }
+ size_t n = trim_idle_stacks<StackClass>(
+ static_cast<int64_t>(idle_s) * 1000000);
+ if (n != 0) {
+ // Spread madvise work across batches when the idle backlog is large.
+ butil::PlatformThread::Sleep(butil::TimeDelta::FromMilliseconds(10));
+ }
+ return n;
+}
+
+static void* stack_trim_thread(void*) {
+ butil::PlatformThread::SetNameSimple("bthread_trim");
+ for (;;) {
+ size_t n = trim_one_batch<SmallStackClass>();
+ n += trim_one_batch<NormalStackClass>();
+ n += trim_one_batch<LargeStackClass>();
+ if (n == 0) {
+ butil::PlatformThread::Sleep(butil::TimeDelta::FromSeconds(1));
+ }
+ }
+ return nullptr;
+}
+
+void init_stack_trimming() {
+ static pthread_once_t once = PTHREAD_ONCE_INIT;
+ pthread_once(&once, [] {
+ pthread_attr_t attr;
+ pthread_attr_init(&attr);
+ pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
+ pthread_t tid;
+ int rc = pthread_create(&tid, &attr, stack_trim_thread, nullptr);
+ pthread_attr_destroy(&attr);
+ if (rc != 0) {
+ LOG(ERROR) << "Fail to start stack trimmer: " << berror(rc);
+ }
+ });
+}
+
+#if defined(UNIT_TEST)
+// Synchronous test entries must not overlap background reclamation.
+size_t trim_idle_stacks_for_test(int64_t idle_us) {
+ size_t n = trim_idle_stacks<SmallStackClass>(idle_us);
+ n += trim_idle_stacks<NormalStackClass>(idle_us);
+ n += trim_idle_stacks<LargeStackClass>(idle_us);
+ return n;
+}
+#endif
+
+} // namespace internal
+
+#endif // BTHREAD_STACK_TRIM_SUPPORTED
+
} // namespace bthread
diff --git a/src/bthread/stack.h b/src/bthread/stack.h
index d703566e9..10e1a3fe1 100644
--- a/src/bthread/stack.h
+++ b/src/bthread/stack.h
@@ -26,8 +26,15 @@
#include <gflags/gflags.h> // DECLARE_int32
#include "bthread/types.h"
#include "bthread/context.h" // bthread_fcontext_t
+#include "butil/build_config.h"
+#include "butil/compiler_specific.h"
#include "butil/object_pool.h"
+// Discarding the continuation would leak ASan's saved fake stack.
+#if defined(OS_LINUX) && !defined(BUTIL_USE_ASAN)
+#define BTHREAD_STACK_TRIM_SUPPORTED
+#endif
+
namespace bthread {
struct StackStorage {
@@ -53,6 +60,13 @@ int allocate_stack_storage(StackStorage* s, int stacksize,
int guardsize);
// corresponding allocate_stack_storage() otherwise behavior is undefined.
void deallocate_stack_storage(StackStorage* s);
+#if defined(BTHREAD_STACK_TRIM_SUPPORTED)
+namespace internal {
+// Called from TaskControl::init, never during command-line parsing.
+void init_stack_trimming();
+}
+#endif
+
enum StackType {
STACK_TYPE_MAIN = 0,
STACK_TYPE_PTHREAD = BTHREAD_STACKTYPE_PTHREAD,
diff --git a/src/bthread/stack_inl.h b/src/bthread/stack_inl.h
index 6c313ca36..e60c28968 100644
--- a/src/bthread/stack_inl.h
+++ b/src/bthread/stack_inl.h
@@ -136,7 +136,8 @@ template <typename StackClass> struct StackFactory {
BTHREAD_ASAN_POISON_MEMORY_REGION(storage);
}
~Wrapper() {
- if (context) {
+ // A trimmed stack has no context but still owns its mapping.
+ if (storage.bottom) {
context = nullptr;
// Unpoison to avoid affecting other allocator.
BTHREAD_ASAN_UNPOISON_MEMORY_REGION(storage);
@@ -147,9 +148,18 @@ template <typename StackClass> struct StackFactory {
};
static ContextualStack* get_stack(void (*entry)(intptr_t)) {
- ContextualStack* cs = butil::get_object<Wrapper>(entry);
+ Wrapper* cs = butil::get_object<Wrapper>(entry);
+ if (cs == nullptr) {
+ return nullptr;
+ }
// Marks stack as addressable.
BTHREAD_ASAN_UNPOISON_MEMORY_REGION(cs->storage);
+ // Reclamation may discard the saved continuation, even on failure.
+ // The mapping is kept, so restarting at entry needs no allocation.
+ if (cs->context == nullptr) {
+ cs->context = bthread_make_fcontext(
+ cs->storage.bottom, cs->storage.stacksize, entry);
+ }
return cs;
}
diff --git a/src/bthread/task_control.cpp b/src/bthread/task_control.cpp
index 6da8a4ab8..eeec8d5d1 100644
--- a/src/bthread/task_control.cpp
+++ b/src/bthread/task_control.cpp
@@ -283,6 +283,9 @@ int TaskControl::init(int concurrency) {
LOG(ERROR) << "Fail to get global_timer_thread";
return -1;
}
+#if defined(BTHREAD_STACK_TRIM_SUPPORTED)
+ internal::init_stack_trimming();
+#endif
#ifdef BRPC_BTHREAD_TRACER
if (!_task_tracer.Init()) {
diff --git a/src/butil/object_pool.h b/src/butil/object_pool.h
index 40bd28e7a..f0a66cc32 100644
--- a/src/butil/object_pool.h
+++ b/src/butil/object_pool.h
@@ -23,6 +23,7 @@
#define BUTIL_OBJECT_POOL_H
#include <cstddef> // size_t
+#include <cstdint>
#include "butil/type_traits.h"
// ObjectPool is a derivative class of ResourcePool to allocate and
@@ -109,6 +110,23 @@ template <typename T> ObjectPoolInfo describe_objects() {
return ObjectPool<T>::singleton()->describe_objects();
}
+// Process objects cached globally for at least idle_us (monotonic time).
+// Exclusively borrow whole chunks until at least max_items objects are
selected
+// or no eligible chunks remain, then invoke fn outside the pool lock. The
count
+// may exceed max_items by at most ObjectPool<T>::FREE_CHUNK_NITEM - 1 objects.
+// Use SIZE_MAX to process all eligible global chunks in one call.
+// fn must be noexcept, preserve each object's lifetime/address,
+// and leave it reusable. Borrowers restore any resources released by fn.
+// Processed chunks become cold and are reused only after the regular global
+// free list is empty. They are not processed again until borrowed and returned
+// to the global free list.
+// Return the number of objects passed to fn, regardless of its effects.
+// TLS caches are untouched. Finish all calls before clearing a pool in tests.
+template <typename T, typename Fn>
+inline size_t process_idle_objects(int64_t idle_us, size_t max_items, Fn& fn) {
+ return ObjectPool<T>::singleton()->process_idle_objects(idle_us,
max_items, fn);
+}
+
} // namespace butil
#endif // BUTIL_OBJECT_POOL_H
diff --git a/src/butil/object_pool_inl.h b/src/butil/object_pool_inl.h
index ea78ee5a0..15a95870b 100644
--- a/src/butil/object_pool_inl.h
+++ b/src/butil/object_pool_inl.h
@@ -31,6 +31,7 @@
#include "butil/macros.h" // BAIDU_CACHELINE_ALIGNMENT
#include "butil/scoped_lock.h" // BAIDU_SCOPED_LOCK
#include "butil/thread_local.h" // BAIDU_THREAD_LOCAL
+#include "butil/time.h" // monotonic_time_us
#include "butil/memory/aligned_memory.h" // butil::AlignedMemory
#include "butil/debug/address_annotations.h"
@@ -117,7 +118,11 @@ public:
// Free objects are batched in a FreeChunk before they're added to
// global list(_free_chunks).
typedef ObjectPoolFreeChunk<T, FREE_CHUNK_NITEM> FreeChunk;
- typedef ObjectPoolFreeChunk<T, 0> DynamicFreeChunk;
+ struct GlobalFreeChunk {
+ int64_t push_us;
+ size_t nfree;
+ T* ptrs[0];
+ };
#ifdef BUTIL_USE_ASAN
// According to
https://github.com/google/sanitizers/wiki/AddressSanitizerManualPoisoning ,
// The allocated chunks should start with 8-aligned addresses,
@@ -307,6 +312,44 @@ public:
return (n < FREE_CHUNK_NITEM ? n : FREE_CHUNK_NITEM);
}
+ template <typename Fn>
+ size_t process_idle_objects(int64_t idle_us, size_t max_items, Fn& fn) {
+ static_assert(noexcept(std::declval<Fn&>()(std::declval<T*>())),
+ "idle object callback must be noexcept");
+ if (idle_us < 0 || max_items == 0) {
+ return 0;
+ }
+ std::vector<GlobalFreeChunk*> batch;
+ size_t nchunk = 0;
+ size_t nitem = 0;
+ int64_t deadline = monotonic_time_us() - idle_us;
+ {
+ BAIDU_SCOPED_LOCK(_free_chunks_mutex);
+ while (nchunk < _free_chunks.size() &&
+ nitem < max_items && _free_chunks[nchunk]->push_us <=
deadline) {
+ nitem += _free_chunks[nchunk]->nfree;
+ ++nchunk;
+ }
+ batch.assign(_free_chunks.begin(), _free_chunks.begin() + nchunk);
+ _free_chunks.erase(_free_chunks.begin(), _free_chunks.begin() +
nchunk);
+ publish_nchunks_locked();
+ }
+ for (size_t i = 0; i < nchunk; ++i) {
+ for (size_t j = 0; j < batch[i]->nfree; ++j) {
+ T* obj = batch[i]->ptrs[j];
+ OBJECT_POOL_ASAN_UNPOISON_MEMORY_REGION(obj);
+ fn(obj);
+ OBJECT_POOL_ASAN_POISON_MEMORY_REGION(obj);
+ }
+ }
+ if (nchunk != 0) {
+ BAIDU_SCOPED_LOCK(_free_chunks_mutex);
+ _cold_chunks.insert(_cold_chunks.end(), batch.begin(),
batch.end());
+ publish_nchunks_locked();
+ }
+ return nitem;
+ }
+
// Number of all allocated objects, including being used and free.
ObjectPoolInfo describe_objects() const {
ObjectPoolInfo info;
@@ -355,9 +398,8 @@ public:
}
private:
- ObjectPool() {
+ ObjectPool() : _nchunks(0) {
_free_chunks.reserve(OP_INITIAL_FREE_LIST_SIZE);
- _nfree_chunks.store(0, butil::memory_order_relaxed);
pthread_mutex_init(&_free_chunks_mutex, nullptr);
#if defined(BUTIL_USE_ASAN) && \
!defined(BAIDU_CLEAR_OBJECT_POOL_AFTER_ALL_THREADS_QUIT)
@@ -536,21 +578,28 @@ private:
}
private:
+ void publish_nchunks_locked() {
+ _nchunks.store(_free_chunks.size() + _cold_chunks.size(),
+ butil::memory_order_relaxed);
+ }
+
bool pop_free_chunk(FreeChunk& c) {
// This relaxed load is only a hint that avoids contending on the
- // mutex when the shared free list is empty. The list itself remains
- // protected by _free_chunks_mutex below.
- if (_nfree_chunks.load(butil::memory_order_relaxed) == 0) {
+ // mutex when both shared free lists are empty. Both lists remain
+ // protected by `_free_chunks_mutex` below.
+ if (_nchunks.load(butil::memory_order_relaxed) == 0) {
return false;
}
pthread_mutex_lock(&_free_chunks_mutex);
- if (_free_chunks.empty()) {
+ std::vector<GlobalFreeChunk*>& list =
+ _free_chunks.empty() ? _cold_chunks : _free_chunks;
+ if (list.empty()) {
pthread_mutex_unlock(&_free_chunks_mutex);
return false;
}
- DynamicFreeChunk* p = _free_chunks.back();
- _free_chunks.pop_back();
- _nfree_chunks.fetch_sub(1, butil::memory_order_relaxed);
+ GlobalFreeChunk* p = list.back();
+ list.pop_back();
+ publish_nchunks_locked();
pthread_mutex_unlock(&_free_chunks_mutex);
c.nfree = p->nfree;
memcpy(c.ptrs, p->ptrs, sizeof(*p->ptrs) * p->nfree);
@@ -559,16 +608,21 @@ private:
}
bool push_free_chunk(const FreeChunk& c) {
- DynamicFreeChunk* p = (DynamicFreeChunk*)malloc(
- offsetof(DynamicFreeChunk, ptrs) + sizeof(*c.ptrs) * c.nfree);
+ if (c.nfree == 0) {
+ return true;
+ }
+ GlobalFreeChunk* p = (GlobalFreeChunk*)malloc(
+ offsetof(GlobalFreeChunk, ptrs) + sizeof(*c.ptrs) * c.nfree);
if (!p) {
return false;
}
p->nfree = c.nfree;
memcpy(p->ptrs, c.ptrs, sizeof(*c.ptrs) * c.nfree);
pthread_mutex_lock(&_free_chunks_mutex);
+ // Timestamp under the lock so the list is ordered by idle age.
+ p->push_us = monotonic_time_us();
_free_chunks.push_back(p);
- _nfree_chunks.fetch_add(1, butil::memory_order_relaxed);
+ publish_nchunks_locked();
pthread_mutex_unlock(&_free_chunks_mutex);
return true;
}
@@ -582,9 +636,11 @@ private:
static pthread_mutex_t _change_thread_mutex;
static butil::static_atomic<BlockGroup*>
_block_groups[OP_MAX_BLOCK_NGROUP];
- std::vector<DynamicFreeChunk*> _free_chunks;
+ std::vector<GlobalFreeChunk*> _free_chunks;
+ // Processed idle chunks, reused only when `_free_chunks` is empty.
+ std::vector<GlobalFreeChunk*> _cold_chunks;
+ butil::atomic<size_t> _nchunks;
pthread_mutex_t _free_chunks_mutex;
- butil::static_atomic<size_t> _nfree_chunks;
#ifdef BUTIL_OBJECT_POOL_NEED_FREE_ITEM_NUM
static butil::static_atomic<size_t> _global_nfree;
diff --git a/test/bthread_stack_trim_unittest.cpp
b/test/bthread_stack_trim_unittest.cpp
new file mode 100644
index 000000000..db7547088
--- /dev/null
+++ b/test/bthread_stack_trim_unittest.cpp
@@ -0,0 +1,557 @@
+// 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.
+
+#include <algorithm>
+#include <atomic>
+#include <csignal>
+#include <cstdio>
+#include <cstring>
+#include <cstdlib>
+#include <sys/mman.h>
+#include <thread>
+#include <vector>
+#include <gtest/gtest.h>
+#include "bthread/bthread.h"
+#include "bthread/butex.h"
+#include "bthread/stack.h"
+#include "bthread/task_group.h"
+#include "butil/build_config.h"
+#include "butil/threading/platform_thread.h"
+#include "bvar/variable.h"
+
+DECLARE_int32(bthread_stack_trim_idle_s);
+DECLARE_int32(bthread_stack_trim_batch_size);
+DECLARE_int32(stack_size_small);
+DECLARE_int32(stack_size_normal);
+DECLARE_int32(stack_size_large);
+
+#if defined(BTHREAD_STACK_TRIM_SUPPORTED)
+namespace bthread {
+extern int (*g_stack_madvise)(void*, size_t, int);
+namespace internal {
+size_t trim_idle_stacks_for_test(int64_t idle_us);
+}
+}
+#endif
+
+namespace {
+#if defined(BTHREAD_STACK_TRIM_SUPPORTED)
+
+typedef bthread::StackFactory<bthread::NormalStackClass> NormalFactory;
+
+int64_t counter(const char* name) {
+ return strtoll(bvar::Variable::describe_exposed(name).c_str(), nullptr,
10);
+}
+
+void sleep_ms(int ms) {
+ butil::PlatformThread::Sleep(butil::TimeDelta::FromMilliseconds(ms));
+}
+
+void trim_all() {
+ while (bthread::internal::trim_idle_stacks_for_test(0) != 0) {}
+}
+
+int64_t process_rss_kb() {
+ FILE* file = fopen("/proc/self/smaps_rollup", "r");
+ if (file == nullptr) { file = fopen("/proc/self/smaps", "r"); }
+ if (file == nullptr) {
+ ADD_FAILURE() << "Cannot read process RSS: " << strerror(errno);
+ return -1;
+ }
+ char line[256];
+ long long rss = -1;
+ long long value;
+ while (fgets(line, sizeof(line), file) != nullptr) {
+ if (sscanf(line, "Rss: %lld kB", &value) == 1) {
+ if (rss < 0) { rss = 0; }
+ rss += value;
+ }
+ }
+ fclose(file);
+ EXPECT_GE(rss, 0);
+ return rss;
+}
+
+// Excludes batches held by the trimmer. Use only after synchronous trimming.
+template <typename StackClass>
+size_t untrimmed_global_stacks() {
+ typedef typename bthread::StackFactory<StackClass>::Wrapper Wrapper;
+ auto* pool = butil::ObjectPool<Wrapper>::singleton();
+ BAIDU_SCOPED_LOCK(pool->_free_chunks_mutex);
+ size_t n = 0;
+ for (auto* chunk : pool->_free_chunks) {
+ for (size_t i = 0; i < chunk->nfree; ++i) {
+ n += chunk->ptrs[i]->context != nullptr;
+ }
+ }
+ return n;
+}
+
+size_t resident_pages(bthread::ContextualStack* s) {
+ std::vector<unsigned char> pages(s->storage.stacksize / getpagesize());
+ if (mincore(static_cast<char*>(s->storage.bottom) - s->storage.stacksize,
+ s->storage.stacksize, pages.data()) != 0) {
+ ADD_FAILURE() << "mincore failed: " << strerror(errno);
+ return SIZE_MAX;
+ }
+ return std::count_if(pages.begin(), pages.end(),
+ [](unsigned char p) { return (p & 1) != 0; });
+}
+
+void flush_stacks(bthread::StackType type = bthread::STACK_TYPE_NORMAL) {
+ switch (type) {
+ case bthread::STACK_TYPE_SMALL:
+
butil::clear_objects<bthread::StackFactory<bthread::SmallStackClass>::Wrapper>();
+ break;
+ case bthread::STACK_TYPE_LARGE:
+
butil::clear_objects<bthread::StackFactory<bthread::LargeStackClass>::Wrapper>();
+ break;
+ default:
+ butil::clear_objects<NormalFactory::Wrapper>();
+ break;
+ }
+}
+
+std::vector<bthread::ContextualStack*> cache_stacks(
+ size_t n, bthread::StackType type = bthread::STACK_TYPE_NORMAL) {
+ std::vector<bthread::ContextualStack*> stacks;
+ for (size_t i = 0; i < n; ++i) {
+ auto* s = bthread::get_stack(type, bthread::TaskGroup::task_runner);
+ EXPECT_NE(nullptr, s);
+ if (!s) { break; }
+ // Touch several pages; merely reserving an mmap does not consume RSS.
+ memset(static_cast<char*>(s->storage.bottom) - 16384, 0x5a, 8192);
+ stacks.push_back(s);
+ }
+ for (auto* s : stacks) { bthread::return_stack(s); }
+ flush_stacks(type);
+ return stacks;
+}
+
+struct Burst {
+ int depth;
+ std::atomic<int> ready{0};
+ std::atomic<int> errors{0};
+ butil::atomic<int>* gate =
bthread::butex_create_checked<butil::atomic<int>>();
+ explicit Burst(int stack_depth) : depth(stack_depth) { gate->store(0); }
+ ~Burst() { bthread::butex_destroy(gate); }
+};
+
+NOINLINE void touch_stack(Burst* burst, int depth) {
+ volatile char page[4096];
+ page[0] = static_cast<char>(depth);
+ if (depth != 0) {
+ touch_stack(burst, depth - 1);
+ } else {
+ ++burst->ready;
+ while (burst->gate->load() == 0) {
+ bthread::butex_wait(burst->gate, 0, nullptr);
+ }
+ bthread_usleep(1000);
+ bthread_yield();
+ }
+ if (page[0] != depth) { ++burst->errors; }
+}
+
+bool run_burst(size_t n, int64_t* ready_rss_kb = nullptr,
+ bthread_attr_t attr = BTHREAD_ATTR_NORMAL, int depth = 16) {
+ Burst burst(depth);
+ std::vector<bthread_t> tids;
+ for (size_t i = 0; i < n; ++i) {
+ bthread_t tid;
+ if (bthread_start_background(&tid, &attr,
+ [](void* p) -> void* {
+ auto* burst = static_cast<Burst*>(p);
+ touch_stack(burst, burst->depth);
+ return nullptr;
+ }, &burst) != 0) {
+ break;
+ }
+ tids.push_back(tid);
+ }
+ int64_t deadline = butil::monotonic_time_us() + 10000000;
+ while (burst.ready.load() != static_cast<int>(tids.size()) &&
+ butil::monotonic_time_us() < deadline) {
+ sleep_ms(1);
+ }
+ bool ready = burst.ready.load() == static_cast<int>(n);
+ if (ready_rss_kb != nullptr) { *ready_rss_kb = process_rss_kb(); }
+ burst.gate->store(1);
+ bthread::butex_wake_all(burst.gate);
+ for (bthread_t tid : tids) { bthread_join(tid, nullptr); }
+ return ready && burst.errors.load() == 0;
+}
+
+struct MadviseOverride {
+ explicit MadviseOverride(int (*fn)(void*, size_t, int))
+ : old(bthread::g_stack_madvise) { bthread::g_stack_madvise = fn; }
+ ~MadviseOverride() { bthread::g_stack_madvise = old; }
+ int (*old)(void*, size_t, int);
+};
+
+class StackTrimTest : public ::testing::Test {
+protected:
+ void SetUp() override {
+ FLAGS_bthread_stack_trim_idle_s = 0;
+ flush_stacks();
+ trim_all();
+ }
+ GFLAGS_NAMESPACE::FlagSaver flags;
+};
+
+// Re-exec isolates tests that enable background reclamation.
+class StackTrimDeathTest : public StackTrimTest {
+ void SetUp() override {
+ StackTrimTest::SetUp();
+#ifdef GTEST_FLAG_SET
+ GTEST_FLAG_SET(death_test_style, "threadsafe");
+#else
+ ::testing::GTEST_FLAG(death_test_style) = "threadsafe";
+#endif
+ // Otherwise a failed CHECK only logs and the child still exits with 0.
+ ASSERT_FALSE(GFLAGS_NAMESPACE::SetCommandLineOption(
+ "crash_on_fatal_log", "true").empty());
+ }
+};
+
+TEST_F(StackTrimTest, trim_and_rebuild_all_stack_types) {
+ for (auto type : {bthread::STACK_TYPE_SMALL, bthread::STACK_TYPE_NORMAL,
+ bthread::STACK_TYPE_LARGE}) {
+ auto stacks = cache_stacks(80, type);
+ ASSERT_EQ(80u, stacks.size());
+ std::vector<bthread::StackStorage> storage;
+ for (auto* s : stacks) {
+ ASSERT_GT(resident_pages(s), 0u);
+ storage.push_back(s->storage);
+ }
+ int64_t before = counter("bthread_stack_count");
+ int64_t released_before = counter("bthread_stack_trimmed_count");
+ trim_all();
+ EXPECT_EQ(before, counter("bthread_stack_count"));
+ EXPECT_EQ(released_before + 80,
counter("bthread_stack_trimmed_count"));
+ for (size_t i = 0; i < stacks.size(); ++i) {
+ auto* s = stacks[i];
+ EXPECT_EQ(nullptr, s->context);
+ EXPECT_EQ(storage[i].bottom, s->storage.bottom);
+ EXPECT_EQ(storage[i].stacksize, s->storage.stacksize);
+ EXPECT_EQ(storage[i].guardsize, s->storage.guardsize);
+ EXPECT_EQ(storage[i].valgrind_stack_id,
s->storage.valgrind_stack_id);
+ EXPECT_EQ(0u, resident_pages(s));
+ }
+ // Rebuilding only touches the existing mapping, including its guard.
+ std::vector<bthread::ContextualStack*> reused;
+ for (int i = 0; i < 80; ++i) {
+ auto* s = bthread::get_stack(type,
bthread::TaskGroup::task_runner);
+ ASSERT_NE(nullptr, s);
+ EXPECT_NE(nullptr, s->context);
+ EXPECT_GT(s->storage.guardsize, 0u);
+ EXPECT_EQ(type, s->stacktype);
+ auto found = std::find(stacks.begin(), stacks.end(), s);
+ ASSERT_NE(stacks.end(), found);
+ EXPECT_EQ(storage[found - stacks.begin()].bottom,
s->storage.bottom);
+ reused.push_back(s);
+ }
+ EXPECT_EQ(before, counter("bthread_stack_count"));
+ for (auto* s : reused) { bthread::return_stack(s); }
+ flush_stacks(type);
+ trim_all();
+ }
+}
+
+TEST_F(StackTrimTest, batch_size_updates_on_next_batch) {
+ // LARGE stacks have one object per chunk, so the budget is exact here.
+ ASSERT_EQ(256u, cache_stacks(256, bthread::STACK_TYPE_LARGE).size());
+ int64_t before = counter("bthread_stack_trimmed_count");
+ ASSERT_FALSE(GFLAGS_NAMESPACE::SetCommandLineOption(
+ "bthread_stack_trim_batch_size", "1").empty());
+ EXPECT_EQ(1u, bthread::internal::trim_idle_stacks_for_test(0));
+ EXPECT_EQ(before + 1, counter("bthread_stack_trimmed_count"));
+ ASSERT_FALSE(GFLAGS_NAMESPACE::SetCommandLineOption(
+ "bthread_stack_trim_batch_size", "128").empty());
+ EXPECT_EQ(128u, bthread::internal::trim_idle_stacks_for_test(0));
+ EXPECT_EQ(before + 129, counter("bthread_stack_trimmed_count"));
+ ASSERT_FALSE(GFLAGS_NAMESPACE::SetCommandLineOption(
+ "bthread_stack_trim_batch_size", "256").empty());
+ EXPECT_EQ(127u, bthread::internal::trim_idle_stacks_for_test(0));
+ EXPECT_EQ(before + 256, counter("bthread_stack_trimmed_count"));
+ EXPECT_EQ(0u, bthread::internal::trim_idle_stacks_for_test(0));
+}
+
+TEST_F(StackTrimDeathTest, malloc_stacks_are_not_trimmed) {
+ ASSERT_EXIT({
+ FLAGS_guard_page_size = 0;
+ auto stacks = cache_stacks(80);
+ CHECK_EQ(80u, stacks.size());
+ std::vector<bthread_fcontext_t> contexts;
+ for (auto* s : stacks) { contexts.push_back(s->context); }
+ int64_t before = counter("bthread_stack_count");
+ trim_all();
+ CHECK_EQ(before, counter("bthread_stack_count"));
+ CHECK_EQ(0, counter("bthread_stack_trimmed_count"));
+ for (size_t i = 0; i < stacks.size(); ++i) {
+ CHECK_EQ(0u, stacks[i]->storage.guardsize);
+ CHECK_EQ(contexts[i], stacks[i]->context);
+ CHECK_EQ(0x5a, *(static_cast<unsigned char*>(
+ stacks[i]->storage.bottom) - 16384));
+ }
+ _exit(0);
+ }, ::testing::ExitedWithCode(0), "");
+}
+
+void check_failed_trim(bool partial) {
+ auto stacks = cache_stacks(256);
+ std::vector<bthread::StackStorage> storage;
+ std::vector<size_t> resident;
+ for (auto* s : stacks) {
+ storage.push_back(s->storage);
+ resident.push_back(resident_pages(s));
+ }
+ int64_t before = counter("bthread_stack_count");
+ int64_t released_before = counter("bthread_stack_trimmed_count");
+ {
+ auto fail = [](void*, size_t, int) { errno = EINVAL; return -1; };
+ auto partial_fail = [](void* low, size_t size, int advice) {
+ // Discard the page containing the saved context, then report an
+ // error. An error alone cannot prove that the old context is safe.
+ size_t page = getpagesize();
+ if (madvise(static_cast<char*>(low) + size - page, page, advice)
!= 0) {
+ return -1;
+ }
+ errno = EINVAL;
+ return -1;
+ };
+ MadviseOverride hook(partial ? +partial_fail : +fail);
+ trim_all();
+ }
+ EXPECT_EQ(before, counter("bthread_stack_count"));
+ EXPECT_EQ(released_before, counter("bthread_stack_trimmed_count"));
+ for (size_t i = 0; i < stacks.size(); ++i) {
+ EXPECT_EQ(nullptr, stacks[i]->context);
+ EXPECT_EQ(storage[i].bottom, stacks[i]->storage.bottom);
+ EXPECT_EQ(storage[i].stacksize, stacks[i]->storage.stacksize);
+ EXPECT_EQ(storage[i].guardsize, stacks[i]->storage.guardsize);
+ EXPECT_EQ(storage[i].valgrind_stack_id,
stacks[i]->storage.valgrind_stack_id);
+ EXPECT_EQ(resident[i] - (partial ? 1 : 0), resident_pages(stacks[i]));
+ }
+ EXPECT_EQ(0u, bthread::internal::trim_idle_stacks_for_test(0));
+ ASSERT_EQ(0, bthread_setconcurrency(8));
+ ASSERT_TRUE(run_burst(256)); // Must rebuild even when madvise failed.
+ int64_t deadline = butil::monotonic_time_us() + 2000000;
+ do {
+ trim_all();
+ if (counter("bthread_stack_trimmed_count") > released_before) { break;
}
+ sleep_ms(1);
+ } while (butil::monotonic_time_us() < deadline);
+ EXPECT_GT(counter("bthread_stack_trimmed_count"), released_before);
+}
+
+TEST_F(StackTrimTest, failed_trim_rebuilds_context_and_retries_after_reuse) {
+ check_failed_trim(false);
+}
+
+TEST_F(StackTrimTest,
partial_failure_rebuilds_context_and_retries_after_reuse) {
+ check_failed_trim(true);
+}
+
+TEST_F(StackTrimTest, burst_reclaim_and_second_burst) {
+ const char* stress_count = getenv("BRPC_STACK_TRIM_TEST_BURST");
+ size_t n = stress_count ? strtoul(stress_count, nullptr, 10) : 2048;
+ const char* stress_workers = getenv("BRPC_STACK_TRIM_TEST_WORKERS");
+ int workers = stress_workers ? strtol(stress_workers, nullptr, 10) : 8;
+ const char* stress_type = getenv("BRPC_STACK_TRIM_TEST_TYPE");
+ std::string type = stress_type ? stress_type : "normal";
+ bthread_attr_t attr = BTHREAD_ATTR_NORMAL;
+ int depth = 16;
+ int stack_size = FLAGS_stack_size_normal;
+ int cached_per_worker = FLAGS_tc_stack_normal;
+ auto count_untrimmed = untrimmed_global_stacks<bthread::NormalStackClass>;
+ if (type == "small") {
+ attr = BTHREAD_ATTR_SMALL;
+ depth = 3;
+ stack_size = FLAGS_stack_size_small;
+ cached_per_worker = FLAGS_tc_stack_small;
+ count_untrimmed = untrimmed_global_stacks<bthread::SmallStackClass>;
+ } else if (type == "large") {
+ attr = BTHREAD_ATTR_LARGE;
+ depth = 64;
+ stack_size = FLAGS_stack_size_large;
+ cached_per_worker = 1;
+ count_untrimmed = untrimmed_global_stacks<bthread::LargeStackClass>;
+ } else {
+ ASSERT_EQ("normal", type);
+ }
+ ASSERT_GE(n, 256u);
+ ASSERT_EQ(0, bthread_setconcurrency(workers));
+ int64_t released_before = counter("bthread_stack_trimmed_count");
+ int64_t baseline_rss_kb = process_rss_kb();
+ int64_t ready_rss_kb = 0;
+ ASSERT_TRUE(run_burst(n, &ready_rss_kb, attr, depth));
+ ASSERT_EQ(workers, counter("bthread_worker_count"));
+ int64_t peak = counter("bthread_stack_count");
+ EXPECT_GE(peak, static_cast<int64_t>(n));
+ int64_t before_trim_rss_kb = process_rss_kb();
+ int64_t trim_start_us = butil::monotonic_time_us();
+ int64_t deadline = trim_start_us + 3000000;
+ do {
+ trim_all();
+ if (counter("bthread_stack_trimmed_count") - released_before >=
+ static_cast<int64_t>(n) - workers * cached_per_worker &&
+ count_untrimmed() == 0) { break; }
+ sleep_ms(1);
+ } while (butil::monotonic_time_us() < deadline);
+ EXPECT_GE(counter("bthread_stack_trimmed_count") - released_before,
+ static_cast<int64_t>(n) - workers * cached_per_worker);
+ EXPECT_EQ(0u, count_untrimmed());
+ EXPECT_EQ(peak, counter("bthread_stack_count"));
+ int64_t trim_elapsed_us = butil::monotonic_time_us() - trim_start_us;
+ int64_t after_trim_rss_kb = process_rss_kb();
+ RecordProperty("stack_type", type);
+ RecordProperty("worker_count", std::to_string(workers));
+ RecordProperty("stack_size_kb", std::to_string(stack_size / 1024));
+ RecordProperty("touched_stack_kb", std::to_string((depth + 1) * 4));
+ RecordProperty("peak_stacks", std::to_string(peak));
+ RecordProperty("trimmed_stacks", std::to_string(
+ counter("bthread_stack_trimmed_count") - released_before));
+ RecordProperty("baseline_rss_kb", std::to_string(baseline_rss_kb));
+ RecordProperty("ready_rss_kb", std::to_string(ready_rss_kb));
+ RecordProperty("before_trim_rss_kb", std::to_string(before_trim_rss_kb));
+ RecordProperty("after_trim_rss_kb", std::to_string(after_trim_rss_kb));
+ RecordProperty("trim_elapsed_us", std::to_string(trim_elapsed_us));
+ ASSERT_TRUE(run_burst(n, nullptr, attr, depth));
+ EXPECT_GE(counter("bthread_stack_count"), static_cast<int64_t>(n));
+}
+
+TEST_F(StackTrimTest, reclaim_while_a_second_burst_is_running) {
+ cache_stacks(2048);
+ ASSERT_EQ(0, bthread_setconcurrency(8));
+ std::atomic<bool> stop(false);
+ std::thread trimmer([&] {
+ while (!stop.load()) {
+ bthread::internal::trim_idle_stacks_for_test(0);
+ std::this_thread::yield();
+ }
+ });
+ bool ok = run_burst(2048);
+ stop = true;
+ trimmer.join();
+ ASSERT_TRUE(ok);
+}
+
+TEST_F(StackTrimDeathTest, rebuilt_stack_guard_is_protected) {
+ ASSERT_EXIT({
+ auto stacks = cache_stacks(1);
+ CHECK_EQ(1u, stacks.size());
+ trim_all();
+ CHECK(stacks[0]->context == nullptr);
+ auto* s = NormalFactory::get_stack(bthread::TaskGroup::task_runner);
+ CHECK(s && s->storage.guardsize != 0);
+ *static_cast<volatile char*>(
+ static_cast<void*>(static_cast<char*>(s->storage.bottom) -
+ s->storage.stacksize - 1)) = 1;
+ _exit(0);
+ }, ::testing::KilledBySignal(SIGSEGV), "");
+}
+
+TEST_F(StackTrimTest, destructor_releases_mapping_without_context) {
+ int64_t before = counter("bthread_stack_count");
+ {
+ NormalFactory::Wrapper stack(bthread::TaskGroup::task_runner);
+ ASSERT_NE(nullptr, stack.context);
+ ASSERT_EQ(0, madvise(static_cast<char*>(stack.storage.bottom) -
+ stack.storage.stacksize,
+ stack.storage.stacksize, MADV_DONTNEED));
+ stack.context = nullptr;
+ EXPECT_EQ(before + 1, counter("bthread_stack_count"));
+ }
+ EXPECT_EQ(before, counter("bthread_stack_count"));
+}
+
+void check_background(bool runtime_enable) {
+ // This thread keeps exactly one warm stack in TLS. A producer puts a known
+ // number of other stacks in the global pool; no bthread borrows them.
+ auto* warm = NormalFactory::get_stack(bthread::TaskGroup::task_runner);
+ CHECK(warm != nullptr);
+ bthread_fcontext_t warm_context = warm->context;
+ NormalFactory::return_stack(warm);
+ if (!runtime_enable) {
+ CHECK(!GFLAGS_NAMESPACE::SetCommandLineOption(
+ "bthread_stack_trim_idle_s", "1").empty());
+ }
+ std::thread producer([] { CHECK_EQ(256u, cache_stacks(256).size()); });
+ producer.join();
+ int64_t before = counter("bthread_stack_count");
+ sleep_ms(1200);
+ CHECK_EQ(0, counter("bthread_stack_trimmed_count")); // Not initialized
yet.
+
+ CHECK_EQ(0, bthread_setconcurrency(8));
+ bthread_t tid;
+ CHECK_EQ(0, bthread_start_background(&tid, &BTHREAD_ATTR_PTHREAD,
+ [](void*) -> void* { return nullptr; }, nullptr));
+ CHECK_EQ(0, bthread_join(tid, nullptr));
+ if (runtime_enable) {
+ sleep_ms(1200);
+ CHECK_EQ(before, counter("bthread_stack_count"));
+ CHECK_EQ(0, counter("bthread_stack_trimmed_count"));
+ // No validator callback: the thread must already exist after init.
+ FLAGS_bthread_stack_trim_idle_s = 1;
+ }
+ int64_t deadline = butil::monotonic_time_us() + 10000000;
+ // The counter is published after a batch completes. Waiting for all 256
+ // stacks also accounts for batches already removed from the global list.
+ while (counter("bthread_stack_trimmed_count") < 256 &&
+ butil::monotonic_time_us() < deadline) {
+ sleep_ms(20);
+ }
+ CHECK_EQ(256, counter("bthread_stack_trimmed_count"));
+ CHECK_EQ(before, counter("bthread_stack_count"));
+ // Repeated TLS use stays warm across several idle periods.
+ for (int i = 0; i < 120; ++i) {
+ auto* s = NormalFactory::get_stack(bthread::TaskGroup::task_runner);
+ CHECK_EQ(warm, s);
+ CHECK_EQ(warm_context, s->context);
+ NormalFactory::return_stack(s);
+ sleep_ms(20);
+ }
+ CHECK_EQ(256, counter("bthread_stack_trimmed_count"));
+}
+
+TEST_F(StackTrimDeathTest, startup_enable_waits_for_bthread_initialization) {
+ ASSERT_EXIT({ check_background(false); _exit(0); },
+ ::testing::ExitedWithCode(0), "");
+}
+
+TEST_F(StackTrimDeathTest, runtime_enable_uses_existing_thread) {
+ ASSERT_EXIT({ check_background(true); _exit(0); },
+ ::testing::ExitedWithCode(0), "");
+}
+
+TEST_F(StackTrimTest, reject_negative_idle_timeout) {
+ EXPECT_TRUE(GFLAGS_NAMESPACE::SetCommandLineOption(
+ "bthread_stack_trim_idle_s", "-1").empty());
+}
+
+TEST_F(StackTrimTest, reject_nonpositive_batch_size) {
+ int before = FLAGS_bthread_stack_trim_batch_size;
+ EXPECT_TRUE(GFLAGS_NAMESPACE::SetCommandLineOption(
+ "bthread_stack_trim_batch_size", "0").empty());
+ EXPECT_TRUE(GFLAGS_NAMESPACE::SetCommandLineOption(
+ "bthread_stack_trim_batch_size", "-1").empty());
+ EXPECT_EQ(before, FLAGS_bthread_stack_trim_batch_size);
+}
+
+#endif // BTHREAD_STACK_TRIM_SUPPORTED
+} // namespace
diff --git a/test/object_pool_unittest.cpp b/test/object_pool_unittest.cpp
index 10ad931ab..367bb8a0e 100644
--- a/test/object_pool_unittest.cpp
+++ b/test/object_pool_unittest.cpp
@@ -16,6 +16,8 @@
// under the License.
#include <memory>
+#include <atomic>
+#include <thread>
#include <gtest/gtest.h>
#include "butil/time.h"
#include "butil/macros.h"
@@ -419,3 +421,175 @@ TEST_F(ObjectPoolTest, verify_get) {
clear_objects<int>();
}
} // namespace
+
+namespace {
+std::atomic<int> idle_objects_alive(0);
+size_t idle_chunk_size = 4;
+struct IdleObject {
+ IdleObject() : in_use(false), processed(false) { ++idle_objects_alive; }
+ ~IdleObject() { --idle_objects_alive; }
+ std::atomic<bool> in_use;
+ bool processed;
+};
+}
+
+namespace butil {
+template <> struct ObjectPoolBlockMaxItem<IdleObject> {
+ static const size_t value = 4;
+};
+template <> struct ObjectPoolFreeChunkMaxItem<IdleObject> {
+ static size_t value() { return idle_chunk_size; }
+};
+}
+
+namespace {
+typedef butil::ObjectPool<IdleObject> IdlePool;
+
+class ObjectPoolIdleTest : public ::testing::Test {
+protected:
+ void TearDown() override {
+ butil::clear_objects<IdleObject>();
+ EXPECT_EQ(0, idle_objects_alive.load());
+ idle_chunk_size = 4;
+ }
+ void fill(size_t n) {
+ std::vector<IdleObject*> objects;
+ for (size_t i = 0; i < n; ++i) {
+ objects.push_back(butil::get_object<IdleObject>());
+ }
+ for (IdleObject* obj : objects) {
+ ASSERT_EQ(0, butil::return_object(obj));
+ }
+ }
+ void check_count() {
+ IdlePool* pool = IdlePool::singleton();
+ BAIDU_SCOPED_LOCK(pool->_free_chunks_mutex);
+ EXPECT_EQ(pool->_free_chunks.size() + pool->_cold_chunks.size(),
+ pool->_nchunks.load(butil::memory_order_relaxed));
+ }
+};
+
+TEST_F(ObjectPoolIdleTest, age_budget_and_tls) {
+ fill(13); // Three global chunks and one TLS object.
+ IdlePool* pool = IdlePool::singleton();
+ size_t calls = 0;
+ auto process = [&calls](IdleObject* obj) noexcept {
+ EXPECT_FALSE(obj->in_use.load());
+ EXPECT_FALSE(obj->processed);
+ obj->processed = true;
+ ++calls;
+ };
+ EXPECT_EQ(0u, butil::process_idle_objects<IdleObject>(INT64_MAX, 64,
process));
+ EXPECT_EQ(0u, butil::process_idle_objects<IdleObject>(0, 0, process));
+ {
+ BAIDU_SCOPED_LOCK(pool->_free_chunks_mutex);
+ ASSERT_EQ(3u, pool->_free_chunks.size());
+ EXPECT_LE(pool->_free_chunks[0]->push_us,
pool->_free_chunks[1]->push_us);
+ EXPECT_LE(pool->_free_chunks[1]->push_us,
pool->_free_chunks[2]->push_us);
+ pool->_free_chunks[0]->push_us -= 2000000;
+ }
+ EXPECT_EQ(4u, butil::process_idle_objects<IdleObject>(1000000, 1,
process));
+ EXPECT_EQ(0u, butil::process_idle_objects<IdleObject>(1000000, 64,
process));
+ // Whole chunks may overshoot the budget by fewer than FREE_CHUNK_NITEM.
+ EXPECT_EQ(8u, butil::process_idle_objects<IdleObject>(0, 5, process));
+ EXPECT_EQ(0u, butil::process_idle_objects<IdleObject>(0, 64, process));
+ EXPECT_EQ(12u, calls);
+ IdleObject* local = butil::get_object<IdleObject>();
+ EXPECT_FALSE(local->processed);
+ butil::return_object(local);
+ check_count();
+ // TearDown drains both global lists and destroys every object exactly
once.
+}
+
+TEST_F(ObjectPoolIdleTest, prefer_unprocessed_and_reuse_without_construction) {
+ fill(13);
+ auto process = [](IdleObject* obj) noexcept { obj->processed = true; };
+ EXPECT_EQ(12u, butil::process_idle_objects<IdleObject>(0, 64, process));
+ std::vector<IdleObject*> held;
+ for (int i = 0; i < 5; ++i) {
+ held.push_back(butil::get_object<IdleObject>());
+ held.back()->processed = false;
+ }
+ for (IdleObject* obj : held) { butil::return_object(obj); }
+ held.clear();
+ // The five returned objects (TLS + global) precede the eight processed
ones.
+ for (int i = 0; i < 13; ++i) {
+ IdleObject* obj = butil::get_object<IdleObject>();
+ EXPECT_EQ(i >= 5, obj->processed);
+ held.push_back(obj);
+ }
+ EXPECT_EQ(13, idle_objects_alive.load());
+ for (IdleObject* obj : held) { butil::return_object(obj); }
+ check_count();
+}
+
+TEST_F(ObjectPoolIdleTest, zero_cache_does_not_publish_empty_chunks) {
+ idle_chunk_size = 0;
+ fill(9);
+ auto process = [](IdleObject*) noexcept {};
+ EXPECT_EQ(8u, butil::process_idle_objects<IdleObject>(0, 64, process));
+ EXPECT_EQ(0u, butil::process_idle_objects<IdleObject>(0, 64, process));
+ check_count();
+}
+
+TEST_F(ObjectPoolIdleTest, honors_item_budget_across_many_chunks) {
+ idle_chunk_size = 1;
+ fill(301); // 300 global chunks and one TLS object.
+ size_t calls = 0;
+ auto process = [&calls](IdleObject* obj) noexcept {
+ EXPECT_FALSE(obj->processed);
+ obj->processed = true;
+ ++calls;
+ };
+ EXPECT_EQ(200u, butil::process_idle_objects<IdleObject>(0, 200, process));
+ EXPECT_EQ(200u, calls);
+ check_count();
+ EXPECT_EQ(100u, butil::process_idle_objects<IdleObject>(0, SIZE_MAX,
process));
+ EXPECT_EQ(300u, calls);
+ EXPECT_EQ(0u, butil::process_idle_objects<IdleObject>(0, SIZE_MAX,
process));
+ check_count();
+}
+
+TEST_F(ObjectPoolIdleTest, concurrent_borrow_return_and_process) {
+ fill(1024);
+ std::atomic<bool> stop(false);
+ std::atomic<size_t> calls(0);
+ auto process = [&calls](IdleObject* obj) noexcept {
+ EXPECT_FALSE(obj->in_use.exchange(true));
+ obj->processed = true;
+ EXPECT_TRUE(obj->in_use.exchange(false));
+ ++calls;
+ };
+ std::thread processor([&] {
+ // Run at least once: the workers may finish before this is scheduled.
+ do {
+ butil::process_idle_objects<IdleObject>(0, 32, process);
+ check_count();
+ std::this_thread::yield();
+ } while (!stop.load());
+ });
+ std::vector<std::thread> workers;
+ for (int i = 0; i < 4; ++i) {
+ workers.emplace_back([] {
+ for (int j = 0; j < 1000; ++j) {
+ IdleObject* held[16];
+ for (IdleObject*& obj : held) {
+ obj = butil::get_object<IdleObject>();
+ EXPECT_FALSE(obj->in_use.exchange(true));
+ obj->processed = false;
+ }
+ for (IdleObject* obj : held) {
+ EXPECT_TRUE(obj->in_use.exchange(false));
+ butil::return_object(obj);
+ }
+ }
+ });
+ }
+ for (std::thread& worker : workers) { worker.join(); }
+ stop = true;
+ processor.join();
+ EXPECT_GT(calls.load(), 0u);
+ EXPECT_EQ(1024, idle_objects_alive.load());
+ check_count();
+}
+} // namespace
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
