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 bcaf803f9 Reserve extra bthread workers for blocking event loops
(#3683)
bcaf803f9 is described below
commit bcaf803f9226d1dea69b49be344b75f8a081a0f6
Author: Bright Chen <[email protected]>
AuthorDate: Fri Oct 30 00:10:03 2026 +0800
Reserve extra bthread workers for blocking event loops (#3683)
* Reserve extra bthread workers for blocking event loops
* Abort dispatcher shutdown test on timeout to prevent dangling references
---
docs/cn/bthread_tagged_task_group.md | 5 +-
docs/cn/server.md | 15 +-
docs/en/server.md | 14 +-
src/brpc/event_dispatcher.cpp | 18 +-
src/bthread/bthread.cpp | 99 +++++++----
src/bthread/bthread.h | 31 +++-
src/bthread/fd.cpp | 18 +-
src/bthread/task_control.cpp | 118 ++++++++-----
src/bthread/task_control.h | 68 ++++++--
src/bthread/types.h | 7 +-
test/brpc_event_dispatcher_unittest.cpp | 185 ++++++++++++++++++++
test/bthread_setconcurrency_unittest.cpp | 284 ++++++++++++++++++++++++++++++-
12 files changed, 758 insertions(+), 104 deletions(-)
diff --git a/docs/cn/bthread_tagged_task_group.md
b/docs/cn/bthread_tagged_task_group.md
index 038eaead0..0af08ec3f 100644
--- a/docs/cn/bthread_tagged_task_group.md
+++ b/docs/cn/bthread_tagged_task_group.md
@@ -19,12 +19,13 @@
./echo_client -dummy_port 8889 -server "0.0.0.0:8003" -use_bthread true
```
-FLAGS_bthread_concurrency为所有线程的数，FLAGS_bthread_min_concurrency为所有分组的线程数的下限，FLAGS_event_dispatcher_num为单个分组中事件驱动器的数量。FLAGS_bthread_current_tag为将要修改的分组的tag值，FLAGS_bthread_concurrency_by_tag设置这个分组的线程数。
+FLAGS_bthread_concurrency为所有分组的普通并发数配置，FLAGS_bthread_min_concurrency为启用懒扩容时所有分组的初始普通worker总数，FLAGS_event_dispatcher_num为单个分组中事件驱动器的数量。event
dispatcher跑的是会阻塞 pthread的事件循环，所以每个分组会为它的event_dispatcher_num个event
dispatcher**额外**预留同样多的worker，不从FLAGS_bthread_concurrency里扣。FLAGS_bthread_current_tag为将要修改的分组的tag值，FLAGS_bthread_concurrency_by_tag设置这个分组的普通并发数。每个分组的总并发数等于当前普通并发数与预留并发数之和。懒扩容期间，当前普通并发数可以低于配置值。
+
一般情况应用创建的bthread不需要设置bthread_attr_t的tag字段，创建的bthread会在当前tag上下文中执行；如果希望创建的bthread不在当前tag上下文中执行，可以设置bthread_attr_t的tag字段为希望的值，这么做会对性能有些损失，关键路径上应该避免这么做。
Q：如何动态改变分组线程的数量？
-A：你可以根据你的服务更自由的设计你的每个分组的线程数，启动的时候会根据你设置的 bthread_concurrency 来初始化线程池，如果你设置了
bthread_min_concurrency，那么会根据 bthread_min_concurrency 来设置线程池，对于 server
来说，num_threads 就是该 tag 对应的 worker 数量。可以通过设置 FLAGS_bthread_current_tag 和
FLAGS_bthread_concurrency_by_tag
来改变某个分组的线程数。如果没有设置（相当于没有启用分组，默认值为BTHREAD_TAG_INVALID）,num_threads的含义是所有分组的
worker 总数。
+A：你可以根据你的服务更自由的设计你的每个分组的线程数，启动的时候会根据你设置的 bthread_concurrency 来初始化线程池，如果你设置了
bthread_min_concurrency，那么会根据 bthread_min_concurrency 来设置线程池，对于 server
来说，num_threads 就是该 tag 对应的普通并发数。可以通过设置 FLAGS_bthread_current_tag 和
FLAGS_bthread_concurrency_by_tag
来改变某个分组的线程数。如果没有设置（相当于没有启用分组，默认值为BTHREAD_TAG_INVALID）,num_threads的含义是所有分组的普通并发数配置，不包含预留并发数。
Q：不同分组之间有什么关系吗？
diff --git a/docs/cn/server.md b/docs/cn/server.md
index 697812637..82dbac3da 100644
--- a/docs/cn/server.md
+++ b/docs/cn/server.md
@@ -593,7 +593,20 @@ server的验证是基于连接的。当server收到连接上的第一个请求
注意: ServerOptions.num_threads仅仅是个**提示**。
-你不能认为Server就用了这么多线程，因为进程内的所有Server和Channel会共享线程资源，线程总数是所有ServerOptions.num_threads和-bthread_concurrency中的最大值。比如一个程序内有两个Server，num_threads分别为24和36，bthread_concurrency为16。那么worker线程数为max(24,
36, 16) = 36。这不同于其他RPC实现中往往是加起来。
+你不能认为Server就用了这么多线程，因为进程内的所有Server和Channel会共享线程资源，配置的普通并发数是所有ServerOptions.num_threads和-bthread_concurrency中的最大值。比如一个程序内有两个Server，num_threads分别为24和36，bthread_concurrency为16。那么配置的普通并发数为max(24,
36, 16) = 36。这不同于其他RPC实现中往往是加起来。
+
+注意: event
dispatcher（-event_dispatcher_num个，每个tag一组）和bthread内部的epoll线程跑的是不会退出的事件循环：空闲时等在epoll_wait/kevent上，有事件时就地处理。由于bthread不可抢占，这两种状态下承载它们的worker
pthread都轮不到别的bthread，相当于被独占了。所以它们各自**额外**预留一个worker，不占用上述的最大值。这里区分三个概念：
+
+- **普通并发数**：用于普通bthread的worker数，由`num_threads`/`-bthread_concurrency`配置。
+- **预留并发数**：为event dispatcher、内部epoll等阻塞循环额外增加的worker数。
+- **总并发数**：普通并发数与预留并发数之和；当前运行的worker总数可从bvar `bthread_worker_count`看到。
+
+默认立即创建worker时，总并发数为`max(所有num_threads, -bthread_concurrency) +
预留并发数`。启用`-bthread_min_concurrency`后，普通worker按需增长，当前普通并发数可以低于配置值；预留worker仍立即创建。
+
+普通并发数的配置下限为`BTHREAD_MIN_CONCURRENCY`（4）。内部epoll线程使用独立的预留额度，不计入这个下限。
+
+每个tag的普通worker上限为`BTHREAD_MAX_CONCURRENCY`（1024），预留worker使用独立的`BTHREAD_MAX_RESERVED_CONCURRENCY`（512）累计额度，总容量为1536。普通并发数设为1024后仍可为event
dispatcher和内部epoll线程预留worker，初始化顺序不会改变这两类额度。
+
Channel没有相应的选项，但可以通过选项-bthread_concurrency调整。
diff --git a/docs/en/server.md b/docs/en/server.md
index 1a6318c89..1fa01c92e 100644
--- a/docs/en/server.md
+++ b/docs/en/server.md
@@ -589,7 +589,19 @@ Controlled by `ServerOptions.num_threads` , number of cpu
cores by default(inclu
NOTE: ServerOptions.num_threads is just a **hint**.
-Don't think that Server uses exactly so many workers because all servers and
channels in one process share worker pthreads. Total number of threads is the
maximum of all ServerOptions.num_threads and bthread_concurrency. For example,
a program has 2 servers with num_threads=24 and 36 respectively, and
bthread_concurrency is 16. Then the number of worker pthreads is max (24, 36,
16) = 36, which is different from other RPC implementations which do summations
generally.
+Don't think that Server uses exactly so many workers because all servers and
channels in one process share worker pthreads. The configured normal
concurrency is the maximum of all ServerOptions.num_threads and
bthread_concurrency. For example, a program has 2 servers with num_threads=24
and 36 respectively, and bthread_concurrency is 16. Then the configured normal
concurrency is max (24, 36, 16) = 36, which is different from other RPC
implementations which do summations generally.
+
+NOTE: the event dispatchers (-event_dispatcher_num of them per tag) and
bthread's internal epoll thread run an event loop that never returns: they wait
in epoll_wait/kevent when idle and dispatch the events in place otherwise. As
bthreads are not preemptible, in neither state does the worker pthread carrying
them get to run another bthread, so it is effectively dedicated to them. Each
of them therefore reserves one **extra** worker which is not part of the
maximum above. There are three [...]
+
+- **Normal concurrency**: workers available to normal bthreads, configured by
`num_threads` / `-bthread_concurrency`.
+- **Reserved concurrency**: extra workers created for blocking loops such as
event dispatchers and internal epoll threads.
+- **Total concurrency**: normal concurrency plus reserved concurrency. The
bvar `bthread_worker_count` reports the total number of running workers.
+
+With the default eager worker creation, total concurrency is `max(all
num_threads, -bthread_concurrency) + reserved concurrency`. When
`-bthread_min_concurrency` enables lazy growth, the current normal concurrency
may be lower than its configured value. Reserved workers are still created
eagerly.
+
+The minimum configurable normal concurrency is `BTHREAD_MIN_CONCURRENCY` (4).
The internal epoll thread uses the separate reserved budget and is excluded
from this minimum.
+
+Each tag has separate limits of `BTHREAD_MAX_CONCURRENCY` (1024) normal
workers and `BTHREAD_MAX_RESERVED_CONCURRENCY` (512) reserved workers, for a
total capacity of 1536. Reservations accumulate across calls. Setting normal
concurrency to 1024 still leaves room to reserve workers for event dispatchers
and the internal epoll thread, regardless of initialization order.
Channel does not have a corresponding option, but user can change number of
worker pthreads at client-side by setting gflag -bthread_concurrency.
diff --git a/src/brpc/event_dispatcher.cpp b/src/brpc/event_dispatcher.cpp
index 9a61c90c5..bc9ec6ab8 100644
--- a/src/brpc/event_dispatcher.cpp
+++ b/src/brpc/event_dispatcher.cpp
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-
+#include <stdlib.h> // `atexit`
#include <gflags/gflags.h> // DEFINE_int32
#include "butil/compat.h"
#include "butil/fd_utility.h" // make_close_on_exec
@@ -58,9 +58,23 @@ static void StopAndJoinGlobalDispatchers() {
}
void InitializeGlobalDispatchers() {
+ // A dispatcher may switch bthreads through `bthread_start_urgent()`
+ // while handling events. Waiting in `epoll_wait`/`kevent` blocks its
worker
+ // pthread. Reserve one extra worker per dispatcher to compensate for that
+ // blocked capacity and preserve the normal concurrency configured by
+ // `ServerOptions.num_threads` or `-bthread_concurrency`.
+ // Without reservations, `event_dispatcher_num >= num_threads` can leave
+ // all workers blocked in dispatcher waits, preventing other work.
+ for (int i = 0; i < FLAGS_task_group_ntags; ++i) {
+ bthread_tag_t tag = (BTHREAD_TAG_DEFAULT + i) % FLAGS_task_group_ntags;
+ int rc = bthread_add_reserved_concurrency_by_tag(
+ FLAGS_event_dispatcher_num, tag);
+ CHECK_EQ(0, rc) << "Fail to reserve " << FLAGS_event_dispatcher_num
+ << " workers for event dispatchers of tag=" << tag
+ << ": " << berror(rc);
+ }
g_edisp_read_lantency = new bvar::LatencyRecorder("event_dispatcher_read");
g_edisp_write_lantency = new
bvar::LatencyRecorder("event_dispatcher_write");
-
g_edisp = new EventDispatcher[FLAGS_task_group_ntags *
FLAGS_event_dispatcher_num];
for (int i = 0; i < FLAGS_task_group_ntags; ++i) {
for (int j = 0; j < FLAGS_event_dispatcher_num; ++j) {
diff --git a/src/bthread/bthread.cpp b/src/bthread/bthread.cpp
index 836dec999..bd863a8f6 100644
--- a/src/bthread/bthread.cpp
+++ b/src/bthread/bthread.cpp
@@ -44,21 +44,22 @@ static bool validate_bthread_min_concurrency(const char*,
int32_t val);
static bool validate_bthread_current_tag(const char*, int32_t val);
static bool validate_bthread_concurrency_by_tag(const char*, int32_t val);
-DEFINE_int32(bthread_concurrency, 8 + BTHREAD_EPOLL_THREAD_NUM,
- "Number of pthread workers");
+DEFINE_int32(bthread_concurrency, 8, "Number of pthread workers available "
+ "to normal bthreads.");
BUTIL_VALIDATE_GFLAG(bthread_concurrency, validate_bthread_concurrency);
DEFINE_int32(bthread_min_concurrency, 0,
- "Initial number of pthread workers which will be added on-demand."
+ "Initial number of normal pthread workers, added on-demand."
" The laziness is disabled when this value is non-positive,"
- " and workers will be created eagerly according to
-bthread_concurrency and bthread_setconcurrency(). ");
+ " and workers will be created eagerly according to "
+ "-bthread_concurrency and bthread_setconcurrency(). ");
BUTIL_VALIDATE_GFLAG(bthread_min_concurrency,
validate_bthread_min_concurrency);
DEFINE_int32(bthread_current_tag, BTHREAD_TAG_INVALID, "Set bthread
concurrency for this tag");
BUTIL_VALIDATE_GFLAG(bthread_current_tag, validate_bthread_current_tag);
-DEFINE_int32(bthread_concurrency_by_tag, 8 + BTHREAD_EPOLL_THREAD_NUM,
- "Number of pthread workers of FLAGS_bthread_current_tag");
+DEFINE_int32(bthread_concurrency_by_tag, 8,
+ "Number of normal pthread workers of FLAGS_bthread_current_tag");
BUTIL_VALIDATE_GFLAG(bthread_concurrency_by_tag,
validate_bthread_concurrency_by_tag);
DEFINE_int32(bthread_parking_lot_of_each_tag, 4, "Number of parking lots of
each tag");
@@ -91,17 +92,18 @@ extern void (*g_worker_startfn)();
extern void (*g_tagged_worker_startfn)(bthread_tag_t);
inline TaskControl* get_task_control() {
- return g_task_control;
+ butil::atomic<TaskControl*>* p =
(butil::atomic<TaskControl*>*)&g_task_control;
+ return p->load(butil::memory_order_acquire);
}
inline TaskControl* get_or_new_task_control() {
butil::atomic<TaskControl*>* p =
(butil::atomic<TaskControl*>*)&g_task_control;
- TaskControl* c = p->load(butil::memory_order_consume);
+ TaskControl* c = p->load(butil::memory_order_acquire);
if (c != nullptr) {
return c;
}
BAIDU_SCOPED_LOCK(g_task_control_mutex);
- c = p->load(butil::memory_order_consume);
+ c = p->load(butil::memory_order_acquire);
if (c != nullptr) {
return c;
}
@@ -222,7 +224,7 @@ static int add_workers_for_each_tag(int num) {
int added = 0;
auto c = get_task_control();
for (auto i = 0; i < num; ++i) {
- added += c->add_workers(1, i % FLAGS_task_group_ntags);
+ added += c->add_workers(1, i % FLAGS_task_group_ntags, false);
}
return added;
}
@@ -239,9 +241,9 @@ static bool validate_bthread_min_concurrency(const char*,
int32_t val) {
return true;
}
BAIDU_SCOPED_LOCK(g_task_control_mutex);
- int concurrency = c->concurrency();
+ int concurrency = c->normal_concurrency();
if (val > concurrency) {
- int added = bthread::add_workers_for_each_tag(val - concurrency);
+ int added = add_workers_for_each_tag(val - concurrency);
return added == (val - concurrency);
} else {
return true;
@@ -254,13 +256,13 @@ static bool validate_bthread_current_tag(const char*,
int32_t val) {
} else if (val < BTHREAD_TAG_DEFAULT || val >= FLAGS_task_group_ntags) {
return false;
}
- BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
+ BAIDU_SCOPED_LOCK(g_task_control_mutex);
auto c = get_task_control();
if (c == nullptr) {
- FLAGS_bthread_concurrency_by_tag = 8 + BTHREAD_EPOLL_THREAD_NUM;
+ FLAGS_bthread_concurrency_by_tag = 8;
return true;
}
- FLAGS_bthread_concurrency_by_tag = c->concurrency(val);
+ FLAGS_bthread_concurrency_by_tag = c->normal_concurrency(val);
return true;
}
@@ -321,7 +323,7 @@ struct TidTraits {
static const size_t MAX_ENTRIES = 65536;
static const size_t INIT_GC_SIZE = 65536;
static const bthread_t ID_INIT;
- static bool exists(bthread_t id) { return bthread::TaskGroup::exists(id); }
+ static bool exists(bthread_t id) { return TaskGroup::exists(id); }
};
const bthread_t TidTraits::ID_INIT = INVALID_BTHREAD;
@@ -436,7 +438,7 @@ int bthread_getattr(bthread_t tid, bthread_attr_t* attr) {
return bthread::TaskGroup::get_attr(tid, attr);
}
-int bthread_getconcurrency(void) {
+int bthread_getconcurrency() {
return bthread::FLAGS_bthread_concurrency;
}
@@ -457,10 +459,9 @@ int bthread_setconcurrency(int num) {
}
bthread::TaskControl* c = bthread::get_task_control();
if (c != nullptr) {
- if (num < c->concurrency()) {
- return EPERM;
- } else if (num == c->concurrency()) {
- return 0;
+ int normal_concurrency = c->normal_concurrency();
+ if (num <= normal_concurrency) {
+ return num == normal_concurrency ? 0 : EPERM;
}
}
BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
@@ -474,11 +475,11 @@ int bthread_setconcurrency(int num) {
}
return 0;
}
- if (bthread::FLAGS_bthread_concurrency != c->concurrency()) {
+ if (bthread::FLAGS_bthread_concurrency != c->normal_concurrency()) {
LOG(ERROR) << "CHECK failed: bthread_concurrency="
<< bthread::FLAGS_bthread_concurrency
- << " != tc_concurrency=" << c->concurrency();
- bthread::FLAGS_bthread_concurrency = c->concurrency();
+ << " != tc_normal_concurrency=" << c->normal_concurrency();
+ bthread::FLAGS_bthread_concurrency = c->normal_concurrency();
}
if (num > bthread::FLAGS_bthread_concurrency) {
// Create more workers if needed.
@@ -489,12 +490,11 @@ int bthread_setconcurrency(int num) {
}
int bthread_getconcurrency_by_tag(bthread_tag_t tag) {
- BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
auto c = bthread::get_task_control();
if (c == nullptr) {
return EPERM;
}
- return c->concurrency(tag);
+ return c->normal_concurrency(tag);
}
int bthread_setconcurrency_by_tag(int num, bthread_tag_t tag) {
@@ -509,19 +509,48 @@ int bthread_setconcurrency_by_tag(int num, bthread_tag_t
tag) {
}
auto c = bthread::get_or_new_task_control();
BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
- auto tag_ngroup = c->concurrency(tag);
- auto add = num - tag_ngroup;
+ int normal_concurrency = c->normal_concurrency(tag);
+ int add = num - normal_concurrency;
- if (add >= 0) {
- auto added = c->add_workers(add, tag);
- bthread::FLAGS_bthread_concurrency += added;
- return (add == added ? 0 : EPERM);
- } else {
+ if (add < 0) {
LOG(WARNING) << "Fail to set concurrency by tag: " << tag
- << ", tag concurrency should be larger than old
oncurrency. old concurrency: "
- << tag_ngroup << ", new concurrency: " << num;
+ << ", normal concurrency cannot be reduced. "
+ "Old concurrency: " << normal_concurrency
+ << ", new concurrency: " << num;
return EPERM;
}
+ auto added = c->add_workers(add, tag, false);
+ bthread::FLAGS_bthread_concurrency += added;
+ return (add == added ? 0 : EPERM);
+}
+
+int bthread_add_reserved_concurrency_by_tag(int num, bthread_tag_t tag) {
+ if (num <= 0 || num > BTHREAD_MAX_RESERVED_CONCURRENCY) {
+ LOG(ERROR) << "Invalid reserved concurrency increment=" << num;
+ return EINVAL;
+ }
+ if (tag < BTHREAD_TAG_DEFAULT || tag >= FLAGS_task_group_ntags) {
+ LOG(ERROR) << "Invalid tag=" << tag;
+ return EINVAL;
+ }
+ auto c = bthread::get_or_new_task_control();
+ if (c == nullptr) {
+ return ENOMEM;
+ }
+ BAIDU_SCOPED_LOCK(bthread::g_task_control_mutex);
+ if (num > BTHREAD_MAX_RESERVED_CONCURRENCY - c->reserved_concurrency(tag))
{
+ LOG(ERROR) << "Fail to reserve " << num << " workers of tag=" << tag
+ << ", it already has " << c->reserved_concurrency(tag)
+ << " reserved workers of at most "
+ "BTHREAD_MAX_RESERVED_CONCURRENCY="
+ << BTHREAD_MAX_RESERVED_CONCURRENCY;
+ return EINVAL;
+ }
+ // Create reserved workers eagerly even under `-bthread_min_concurrency`.
+ // Blocking waits such as `epoll_wait` do not trigger lazy growth through
+ // `TaskControl::signal_task()`. Add capacity before those waits can occupy
+ // all workers.
+ return c->add_reserved_workers(num, tag) == num ? 0 : EPERM;
}
int bthread_about_to_quit() {
diff --git a/src/bthread/bthread.h b/src/bthread/bthread.h
index e0d95892c..715921318 100644
--- a/src/bthread/bthread.h
+++ b/src/bthread/bthread.h
@@ -151,21 +151,42 @@ extern int bthread_getattr(bthread_t bt, bthread_attr_t*
attr);
// Functions for scheduling control.
// ---------------------------------------------
-// Get number of worker pthreads
+// We distinguish normal, reserved, and total concurrency.
+// Total concurrency is normal + reserved. The configuration below controls
only
+// normal concurrency.
+// `bthread_add_reserved_concurrency_by_tag()` adds reservations.
+// The bvar `bthread_worker_count` reports the running total.
+//
+// Get the configured normal concurrency. When `-bthread_min_concurrency` is
+// positive, normal workers grow lazily and their current count may be lower.
extern int bthread_getconcurrency(void);
-// Set number of worker pthreads to `num'. After a successful call,
-// bthread_getconcurrency() shall return new set number, but workers may
+// Set normal concurrency to `num`. After a successful call,
+// `bthread_getconcurrency()` shall return new set number, but workers may
// take some time to quit or create.
// NOTE: currently concurrency cannot be reduced after any bthread created.
extern int bthread_setconcurrency(int num);
-// Get number of worker pthreads by tag
+// Get the current normal concurrency of `tag`.
extern int bthread_getconcurrency_by_tag(bthread_tag_t tag);
-// Set number of worker pthreads to `num' for specified tag
+// Set normal concurrency to `num` for `tag`.
extern int bthread_setconcurrency_by_tag(int num, bthread_tag_t tag);
+// Add `num` extra reserved worker pthreads of `tag`.
+//
+// Each successful call increases the reserved concurrency of `tag` by `num`.
+// These workers join the shared scheduler pool to compensate for capacity
+// lost when long-lived bthreads block in system calls such as
+// `epoll_wait(..., -1)`.
+//
+// Call this before starting the blocking bthreads. Reserve one extra worker
+// for each, as reservations accumulate and cannot be released.
+// Returns 0 on success, error code otherwise. Exceeding the reserved limit
+// returns `EINVAL` without adding workers. Thread creation failure may leave a
+// partial reservation and returns `EPERM`.
+extern int bthread_add_reserved_concurrency_by_tag(int num, bthread_tag_t tag);
+
// Yield processor to another bthread.
// Notice that current implementation is not fair, which means that
// even if bthread_yield() is called, suspended threads may still starve.
diff --git a/src/bthread/fd.cpp b/src/bthread/fd.cpp
index b9c35166a..5795b1ddf 100644
--- a/src/bthread/fd.cpp
+++ b/src/bthread/fd.cpp
@@ -139,10 +139,24 @@ public:
}
bthread_attr_t attr = BTHREAD_ATTR_NORMAL;
bthread_attr_set_name(&attr, "EpollThread::run_this");
- if (bthread_start_background(
- &_tid, &attr, EpollThread::run_this, this) != 0) {
+ attr.tag = bthread_self_tag();
+ // `run_this()` can block its worker in `epoll_wait` or `kevent`.
+ // Reserve an extra worker to preserve the normal concurrency
configured
+ // via `-bthread_concurrency` or `ServerOptions.num_threads`.
+ // The reservation happens exactly once per `EpollThread`.
+ int rc = bthread_add_reserved_concurrency_by_tag(1, attr.tag);
+ if (rc != 0) {
+ LOG(WARNING) << "Fail to reserve a worker for EpollThread of tag="
+ << attr.tag << ": " << berror(rc)
+ << ". It will consume the configured concurrency"
+ " instead.";
+ }
+ if (bthread_start_background(&_tid, &attr, run_this, this) != 0) {
close(_epfd);
_epfd = -1;
+ // Not releasing the reservation above: this is fatal anyway, and
an
+ // extra idle worker is harmless compared to the alternative of
+ // exposing an unreserve API that could be misused.
LOG(FATAL) << "Fail to create epoll bthread";
return -1;
}
diff --git a/src/bthread/task_control.cpp b/src/bthread/task_control.cpp
index eeec8d5d1..320f86760 100644
--- a/src/bthread/task_control.cpp
+++ b/src/bthread/task_control.cpp
@@ -19,6 +19,7 @@
// Date: Tue Jul 10 17:40:58 CST 2012
+#include <limits.h> // `INT_MAX`
#include <pthread.h>
#include <set>
#include <regex>
@@ -88,8 +89,9 @@ void run_tagged_worker_startfn(bthread_tag_t tag) {
}
struct WorkerThreadArgs {
- WorkerThreadArgs(TaskControl* _c, bthread_tag_t _t) : c(_c), tag(_t) {}
- TaskControl* c;
+ WorkerThreadArgs(TaskControl* c, bthread_tag_t t)
+ : task_control(c), tag(t) {}
+ TaskControl* task_control;
bthread_tag_t tag;
};
@@ -100,13 +102,13 @@ void* TaskControl::worker_thread(void* arg) {
#endif
auto dummy = static_cast<WorkerThreadArgs*>(arg);
- auto c = dummy->c;
+ auto c = dummy->task_control;
auto tag = dummy->tag;
delete dummy;
run_tagged_worker_startfn(tag);
TaskGroup* g = c->create_group(tag);
- TaskStatistics stat;
+ TaskStatistics stat{};
if (nullptr == g) {
LOG(ERROR) << "Fail to create TaskGroup in pthread=" << pthread_self();
return nullptr;
@@ -115,8 +117,7 @@ void* TaskControl::worker_thread(void* arg) {
g->_tid = pthread_self();
// tag_wid is a per-tag monotonic counter: same-tag workers get 0,1,2,...
// Used both for CPU round-robin affinity and the thread name suffix.
- int tag_wid = c->_tag_next_worker_id[tag].fetch_add(
- 1, butil::memory_order_relaxed);
+ int tag_wid = c->_tag_next_worker_id[tag].fetch_add(1,
butil::memory_order_relaxed);
if (!c->_tag_cpus[tag].empty()) {
const auto& cpus = c->_tag_cpus[tag];
bind_thread_to_cpu(pthread_self(), cpus[tag_wid % cpus.size()]);
@@ -195,7 +196,10 @@ TaskControl::TaskControl()
, _tagged_groups(FLAGS_task_group_ntags)
, _init(false)
, _stop(false)
- , _concurrency(0)
+ , _normal_concurrency(0)
+ , _tagged_normal_concurrency(FLAGS_task_group_ntags)
+ , _reserved_concurrency(0)
+ , _tagged_reserved_concurrency(FLAGS_task_group_ntags)
, _nworkers("bthread_worker_count")
, _pending_time(nullptr)
// Delay exposure of following two vars because they rely on TC which
@@ -236,16 +240,16 @@ int TaskControl::init_ed_priority_queues() {
return 0;
}
-int TaskControl::init(int concurrency) {
- if (_concurrency != 0) {
+int TaskControl::init(int normal_concurrency) {
+ if (_normal_concurrency != 0) {
LOG(ERROR) << "Already initialized";
return -1;
}
- if (concurrency <= 0) {
- LOG(ERROR) << "Invalid concurrency=" << concurrency;
+ if (normal_concurrency <= 0) {
+ LOG(ERROR) << "Invalid normal concurrency=" << normal_concurrency;
return -1;
}
- _concurrency = concurrency;
+ _normal_concurrency = normal_concurrency;
if (!FLAGS_cpu_set.empty()) {
if (parse_cpuset(FLAGS_cpu_set) == -1) {
@@ -257,6 +261,8 @@ int TaskControl::init(int concurrency) {
// task group group by tags
for (int i = 0; i < FLAGS_task_group_ntags; ++i) {
_tagged_ngroup[i].store(0, std::memory_order_relaxed);
+ _tagged_normal_concurrency[i].store(0, std::memory_order_relaxed);
+ _tagged_reserved_concurrency[i].store(0, std::memory_order_relaxed);
auto tag_str = std::to_string(i);
_tagged_nworkers.emplace_back(
std::make_unique<bvar::Adder<int64_t>>("bthread_worker_count",
tag_str));
@@ -294,15 +300,18 @@ int TaskControl::init(int concurrency) {
}
#endif // BRPC_BTHREAD_TRACER
- _workers.resize(_concurrency);
- for (int i = 0; i < _concurrency; ++i) {
- auto arg = new WorkerThreadArgs(this, i % FLAGS_task_group_ntags);
- const int rc = pthread_create(&_workers[i], nullptr, worker_thread,
arg);
+ _workers.resize(normal_concurrency);
+ for (int i = 0; i < normal_concurrency; ++i) {
+ bthread_tag_t tag = i % FLAGS_task_group_ntags;
+ _tagged_normal_concurrency[tag].fetch_add(1,
butil::memory_order_release);
+ auto arg = std::make_unique<WorkerThreadArgs>(this, tag);
+ int rc = pthread_create(&_workers[i], nullptr, worker_thread,
arg.get());
if (rc) {
- delete arg;
PLOG(ERROR) << "Fail to create _workers[" << i << "]";
return -1;
}
+ // The worker owns the object pointed to by `arg` and deletes it
itself.
+ arg.release();
}
_worker_usage_second.expose("bthread_worker_usage");
_switch_per_second.expose("bthread_switch_second");
@@ -325,33 +334,60 @@ int TaskControl::init(int concurrency) {
return 0;
}
-int TaskControl::add_workers(int num, bthread_tag_t tag) {
+int TaskControl::add_workers(int num, bthread_tag_t tag, bool reserved) {
if (num <= 0) {
return 0;
}
+ // The two budgets are independent. `_tagged_groups[tag]` has enough slots
+ // for both to reach their limits, regardless of initialization order.
+ int limit = reserved ? BTHREAD_MAX_RESERVED_CONCURRENCY
+ : BTHREAD_MAX_CONCURRENCY;
+ int used = reserved ? reserved_concurrency(tag)
+ : normal_concurrency(tag);
+ int room = limit - used;
+ if (num > room) {
+ LOG_EVERY_SECOND(WARNING)
+ << "Truncated the # of workers to add for tag=" << tag << " from "
+ << num << " to " << (room > 0 ? room : 0) << ", a tag cannot have"
+ " more than " << limit << (reserved ? " reserved" : " normal")
+ << " workers";
+ num = room;
+ if (num <= 0) {
+ return 0;
+ }
+ }
+ int old_total_concurrency = concurrency();
+ if (num > INT_MAX - old_total_concurrency) {
+ LOG(ERROR) << "Total worker count would overflow";
+ return 0;
+ }
try {
- _workers.resize(_concurrency + num);
+ _workers.resize(old_total_concurrency + num);
} catch (...) {
return 0;
}
- const int old_concurency = _concurrency.load(butil::memory_order_relaxed);
- for (int i = 0; i < num; ++i) {
- // Worker will add itself to _idle_workers, so we have to add
- // _concurrency before create a worker.
- _concurrency.fetch_add(1);
- auto arg = new WorkerThreadArgs(this, tag);
- const int rc = pthread_create(
- &_workers[i + old_concurency], nullptr, worker_thread, arg);
+ auto& global_count = reserved ? _reserved_concurrency :
_normal_concurrency;
+ auto& tag_count = reserved ? _tagged_reserved_concurrency[tag]
+ : _tagged_normal_concurrency[tag];
+ int added = 0;
+ for (; added < num; ++added) {
+ auto arg = std::make_unique<WorkerThreadArgs>(this, tag);
+ int rc = pthread_create(&_workers[old_total_concurrency + added],
+ nullptr, worker_thread, arg.get());
if (rc) {
- delete arg;
- PLOG(WARNING) << "Fail to create _workers[" << i + old_concurency
<< "]";
- _concurrency.fetch_sub(1, butil::memory_order_release);
+ PLOG(WARNING) << "Fail to create _workers["
+ << old_total_concurrency + added << "]";
break;
}
+ arg.release();
+ // Only successful creations count. A reservation never changes the
+ // normal count, and failed creations need no counter rollback.
+ global_count.fetch_add(1, butil::memory_order_release);
+ tag_count.fetch_add(1, butil::memory_order_release);
}
// Cannot fail
- _workers.resize(_concurrency.load(butil::memory_order_relaxed));
- return _concurrency.load(butil::memory_order_relaxed) - old_concurency;
+ _workers.resize(old_total_concurrency + added);
+ return added;
}
TaskGroup* TaskControl::choose_one_group(bthread_tag_t tag) {
@@ -560,10 +596,14 @@ int TaskControl::_add_group(TaskGroup* g, bthread_tag_t
tag) {
g->set_tag(tag);
g->set_pl(&_tagged_pl[tag][butil::fmix64(pthread_numeric_id()) %
_pl_num_of_each_tag]);
size_t ngroup = _tagged_ngroup[tag].load(butil::memory_order_relaxed);
- if (ngroup < (size_t)BTHREAD_MAX_CONCURRENCY) {
- _tagged_groups[tag][ngroup] = g;
- _tagged_ngroup[tag].store(ngroup + 1, butil::memory_order_release);
+ if (ngroup >= _tagged_groups[tag].size()) {
+ LOG(ERROR) << "Fail to add TaskGroup to tag=" << tag << ", already "
+ << ngroup << " groups >= capacity="
+ << _tagged_groups[tag].size();
+ return -1;
}
+ _tagged_groups[tag][ngroup] = g;
+ _tagged_ngroup[tag].store(ngroup + 1, butil::memory_order_release);
mu.unlock();
// See the comments in _destroy_group
// TODO: Not needed anymore since non-worker pthread cannot have TaskGroup
@@ -685,13 +725,15 @@ void TaskControl::signal_task(int num_task, bthread_tag_t
tag) {
start_index = 0;
}
}
+ // Lazy growth only uses the normal budget. The unlocked comparison is a
+ // hint. Recheck the normal count under the mutex before creating a worker.
if (num_task > 0 &&
FLAGS_bthread_min_concurrency > 0 && // test min_concurrency for
performance
- _concurrency.load(butil::memory_order_relaxed) <
FLAGS_bthread_concurrency) {
+ normal_concurrency() < FLAGS_bthread_concurrency) {
// TODO: Reduce this lock
BAIDU_SCOPED_LOCK(g_task_control_mutex);
- if (_concurrency.load(butil::memory_order_acquire) <
FLAGS_bthread_concurrency) {
- add_workers(1, tag);
+ if (normal_concurrency() < FLAGS_bthread_concurrency) {
+ add_workers(1, tag, false);
}
}
}
diff --git a/src/bthread/task_control.h b/src/bthread/task_control.h
index fed5bcf3b..f3cc82ac8 100644
--- a/src/bthread/task_control.h
+++ b/src/bthread/task_control.h
@@ -55,8 +55,8 @@ public:
TaskControl();
~TaskControl();
- // Must be called before using. `nconcurrency' is # of worker pthreads.
- int init(int nconcurrency);
+ // Must be called before using. All initial workers are normal workers.
+ int init(int normal_concurrency);
// Create a TaskGroup in this control.
TaskGroup* create_group(bthread_tag_t tag);
@@ -70,12 +70,36 @@ public:
// Stop and join worker threads in TaskControl.
void stop_and_join();
- // Get # of worker threads.
- int concurrency() const
- { return _concurrency.load(butil::memory_order_acquire); }
+ // Get # of normal workers, excluding reservations. This is the current
+ // count, which may be below `-bthread_concurrency` during lazy growth.
+ int normal_concurrency() const {
+ return _normal_concurrency.load(butil::memory_order_acquire);
+ }
+
+ int normal_concurrency(bthread_tag_t tag) const {
+ return
_tagged_normal_concurrency[tag].load(butil::memory_order_acquire);
+ }
- int concurrency(bthread_tag_t tag) const
- { return _tagged_ngroup[tag].load(butil::memory_order_acquire); }
+ // Get # of extra workers reserved for blocking loops, such as event
+ // dispatchers and epoll threads. See `add_reserved_workers()`.
+ int reserved_concurrency() const {
+ return _reserved_concurrency.load(butil::memory_order_acquire);
+ }
+
+ int reserved_concurrency(bthread_tag_t tag) const {
+ return _tagged_reserved_concurrency[tag].load(
+ butil::memory_order_acquire);
+ }
+
+ // Get total # of worker threads (normal + reserved). Hold
+ // `g_task_control_mutex` when using this total to decide on worker growth.
+ int concurrency() const {
+ return normal_concurrency() + reserved_concurrency();
+ }
+
+ int concurrency(bthread_tag_t tag) const {
+ return normal_concurrency(tag) + reserved_concurrency(tag);
+ }
void print_rq_sizes(std::ostream& os);
@@ -84,9 +108,21 @@ public:
int64_t get_cumulated_switch_count();
int64_t get_cumulated_signal_count();
- // [Not thread safe] Add more worker threads.
- // Return the number of workers actually added, which may be less than
|num|
- int add_workers(int num, bthread_tag_t tag);
+ // [Not thread safe] Add more worker threads under `g_task_control_mutex`.
+ // Count them as reserved when `reserved` is true, or normal otherwise.
+ // Normal and reserved workers have separate per-tag limits:
+ // `BTHREAD_MAX_CONCURRENCY` and `BTHREAD_MAX_RESERVED_CONCURRENCY`.
+ // Return the number of workers added, which may be less than `num`.
+ int add_workers(int num, bthread_tag_t tag, bool reserved);
+
+ // [Not thread safe] Add `num` worker threads of `tag` and mark them as
+ // reserved, so that they don't count towards the concurrency configured
+ // via `bthread_setconcurrency*()`. The accumulated reservation of a tag is
+ // limited to `BTHREAD_MAX_RESERVED_CONCURRENCY`.
+ // Return the number of workers added, which may be less than `num`.
+ int add_reserved_workers(int num, bthread_tag_t tag) {
+ return add_workers(num, tag, true);
+ }
// Choose one TaskGroup (randomly right now).
// If this method is called after init(), it never returns nullptr.
@@ -115,7 +151,9 @@ public:
std::vector<bthread_t> get_living_bthreads();
private:
- typedef std::array<TaskGroup*, BTHREAD_MAX_CONCURRENCY> TaggedGroups;
+ // Keep group storage stable for lock-free readers in `steal_task()`.
+ typedef std::array<TaskGroup*, BTHREAD_MAX_CONCURRENCY +
+ BTHREAD_MAX_RESERVED_CONCURRENCY>
TaggedGroups;
typedef std::array<ParkingLot, BTHREAD_MAX_PARKINGLOT> TaggedParkingLot;
// Add/Remove a TaskGroup.
// Returns 0 on success, -1 otherwise.
@@ -158,7 +196,13 @@ private:
butil::atomic<bool> _init; // if not init, bvar will case coredump
bool _stop;
- butil::atomic<int> _concurrency;
+ // Independent normal and reserved counts, globally and per tag.
+ // After `init()`, only grow under `g_task_control_mutex`. Derive the total
+ // from these two counts instead of maintaining a third set of counters.
+ butil::atomic<int> _normal_concurrency;
+ std::vector<butil::atomic<int>> _tagged_normal_concurrency;
+ butil::atomic<int> _reserved_concurrency;
+ std::vector<butil::atomic<int>> _tagged_reserved_concurrency;
std::vector<pthread_t> _workers;
bvar::Adder<int64_t> _nworkers;
butil::Mutex _pending_time_mutex;
diff --git a/src/bthread/types.h b/src/bthread/types.h
index 14411e5e0..5171e6b6b 100644
--- a/src/bthread/types.h
+++ b/src/bthread/types.h
@@ -151,9 +151,12 @@ static const bthread_attr_t BTHREAD_ATTR_DEBUG = {
static const size_t BTHREAD_EPOLL_THREAD_NUM = 1;
static const bthread_t BTHREAD_ATOMIC_INIT = 0;
-// Min/Max number of work pthreads.
-static const int BTHREAD_MIN_CONCURRENCY = 3 + BTHREAD_EPOLL_THREAD_NUM;
+// Min/Max number of worker pthreads available to normal bthreads.
+// Internal epoll workers are reserved separately.
+static const int BTHREAD_MIN_CONCURRENCY = 4;
static const int BTHREAD_MAX_CONCURRENCY = 1024;
+// Independent per-tag limit for extra workers reserved for blocking loops.
+static const int BTHREAD_MAX_RESERVED_CONCURRENCY = 512;
// Min/max number of ParkingLot.
static const int BTHREAD_MIN_PARKINGLOT = 4;
static const int BTHREAD_MAX_PARKINGLOT = 1024;
diff --git a/test/brpc_event_dispatcher_unittest.cpp
b/test/brpc_event_dispatcher_unittest.cpp
index bf3f18758..1825fe296 100644
--- a/test/brpc_event_dispatcher_unittest.cpp
+++ b/test/brpc_event_dispatcher_unittest.cpp
@@ -19,20 +19,45 @@
// Date: Sun Jul 13 15:04:18 CST 2014
+#include <cstdlib>
#include <pthread.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <gtest/gtest.h>
+#include <gflags/gflags.h>
#include "gperftools_helper.h"
#include "butil/time.h"
#include "butil/macros.h"
#include "butil/fd_utility.h"
#include "butil/memory/scope_guard.h"
#include "bthread/bthread.h"
+#include "bthread/task_control.h"
+#include "brpc/channel.h"
+#include "brpc/controller.h"
#include "brpc/event_dispatcher.h"
+#include "brpc/server.h"
#include "brpc/socket.h"
#include "brpc/details/has_epollrdhup.h"
#include "brpc/versioned_ref_with_id.h"
+#include "echo.pb.h"
+
+DECLARE_int32(task_group_ntags);
+DECLARE_int32(event_dispatcher_num);
+
+namespace bthread {
+extern TaskControl* g_task_control;
+}
+
+// Far more dispatchers than the workers the reserve tests at the end of this
+// file configure, which is exactly the situation that used to deadlock.
+static const int NUM_EVENT_DISPATCHERS = 16;
+
+int main(int argc, char* argv[]) {
+ FLAGS_event_dispatcher_num = NUM_EVENT_DISPATCHERS;
+ testing::InitGoogleTest(&argc, argv);
+ GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true);
+ return RUN_ALL_TESTS();
+}
class EventDispatcherTest : public ::testing::Test{
protected:
@@ -42,6 +67,65 @@ protected:
void TearDown() override {}
};
+void set_threadsafe_death_test_style() {
+ // Re-exec instead of forking a process with live bthread workers.
+ // googletest >= 1.12 built with Abseil (`bazel --define absl=1`) exposes
+ // the flag as an `absl::Flag`, which can only be set via `GTEST_FLAG_SET`.
+#ifdef GTEST_FLAG_SET
+ GTEST_FLAG_SET(death_test_style, "threadsafe");
+#else
+ ::testing::FLAGS_gtest_death_test_style = "threadsafe";
+#endif
+}
+
+TEST(EventDispatcherDeathTest, reservation_failure_uses_fatal_logging) {
+ set_threadsafe_death_test_style();
+#if !BRPC_WITH_GLOG
+ GFLAGS_NAMESPACE::FlagSaver flag_saver;
+ ASSERT_FALSE(GFLAGS_NAMESPACE::SetCommandLineOption(
+ "crash_on_fatal_log", "true").empty());
+#endif
+ ASSERT_DEATH({
+ // Exceed the independent reserved budget. Fatal logging must stop
+ // initialization when the logging backend is configured to crash.
+ FLAGS_event_dispatcher_num = BTHREAD_MAX_RESERVED_CONCURRENCY + 1;
+ brpc::GetGlobalEventDispatcher(0, BTHREAD_TAG_DEFAULT);
+ }, "Fail to reserve .* workers for event dispatchers");
+}
+
+void check_max_concurrency_with_dispatcher(bool dispatcher_first) {
+ FLAGS_event_dispatcher_num = 1;
+ if (dispatcher_first) {
+ brpc::GetGlobalEventDispatcher(0, BTHREAD_TAG_DEFAULT);
+ }
+ ASSERT_EQ(0, bthread_setconcurrency_by_tag(
+ BTHREAD_MAX_CONCURRENCY, BTHREAD_TAG_DEFAULT));
+ if (!dispatcher_first) {
+ brpc::GetGlobalEventDispatcher(0, BTHREAD_TAG_DEFAULT);
+ }
+ ASSERT_EQ(BTHREAD_MAX_CONCURRENCY,
+ bthread_getconcurrency_by_tag(BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(1,
bthread::g_task_control->reserved_concurrency(BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(BTHREAD_MAX_CONCURRENCY + 1,
+ bthread::g_task_control->concurrency(BTHREAD_TAG_DEFAULT));
+
+ bthread_t tid;
+ ASSERT_EQ(0, bthread_start_background(
+ &tid, nullptr, [](void*) -> void* { return nullptr; }, nullptr));
+ ASSERT_EQ(0, bthread_join(tid, nullptr));
+}
+
+TEST(EventDispatcherDeathTest,
max_concurrency_supports_both_initialization_orders) {
+ set_threadsafe_death_test_style();
+ for (bool dispatcher_first : {false, true}) {
+ SCOPED_TRACE(dispatcher_first);
+ ASSERT_EXIT({
+ check_max_concurrency_with_dispatcher(dispatcher_first);
+ _exit(::testing::Test::HasFailure() ? 1 : 0);
+ }, ::testing::ExitedWithCode(0), "");
+ }
+}
+
TEST_F(EventDispatcherTest, has_epollrdhup) {
LOG(INFO) << brpc::has_epollrdhup;
}
@@ -532,3 +616,104 @@ TEST_F(EventDispatcherTest, customize_dispatch_task) {
ASSERT_EQ(nullptr, ptr);
ASSERT_NE(0, EventPipe::Address(id, &ptr));
}
+
+// If those workers were taken out of the concurrency configured by the user, a
+// process with `event_dispatcher_num >= num_threads` would have no worker to
+// run anything else, and would never even be able to quit.
+
+// Must stay below `NUM_EVENT_DISPATCHERS` and above the default
+// `-bthread_concurrency`, as concurrency can only ever grow.
+static const int NUM_THREADS = 10;
+static const int RESERVE_PORT = 8798;
+
+class ReserveEchoServiceImpl : public test::EchoService {
+public:
+ void Echo(google::protobuf::RpcController*,
+ const test::EchoRequest* req,
+ test::EchoResponse* res,
+ google::protobuf::Closure* done) override {
+ brpc::ClosureGuard done_guard(done);
+ res->set_message(req->message());
+ }
+};
+
+struct StopArgs {
+ brpc::Server* server;
+ butil::atomic<bool> done;
+ explicit StopArgs(brpc::Server* s) : server(s), done(false) {}
+};
+
+void* stop_and_join(void* arg) {
+ StopArgs* args = static_cast<StopArgs*>(arg);
+ args->server->Stop(0);
+ args->server->Join();
+ args->done.store(true, butil::memory_order_release);
+ return nullptr;
+}
+
+// Dispatchers need extra workers beyond the configured normal concurrency.
+TEST_F(EventDispatcherTest, dispatchers_do_not_consume_configured_concurrency)
{
+ ASSERT_EQ(NUM_EVENT_DISPATCHERS, FLAGS_event_dispatcher_num);
+ // Force initialization of the global dispatchers.
+ brpc::GetGlobalEventDispatcher(0, BTHREAD_TAG_DEFAULT);
+ ASSERT_TRUE(bthread::g_task_control != nullptr);
+
+ int expected = FLAGS_task_group_ntags * FLAGS_event_dispatcher_num;
+ // The internal `bthread::EpollThread` also reserves a worker. It may not
+ // have started yet, so compare with `>=`.
+ ASSERT_GE(bthread::g_task_control->reserved_concurrency(), expected);
+ // Reserved workers are on top of the configured concurrency, not inside
it.
+ int normal = bthread_getconcurrency();
+ ASSERT_EQ(normal, bthread::g_task_control->normal_concurrency());
+ ASSERT_GE(bthread::g_task_control->concurrency(), normal + expected);
+}
+
+// End-to-end: a server whose `num_threads` is smaller than the number of event
+// dispatchers must still serve requests and, crucially, be able to quit.
+TEST_F(EventDispatcherTest,
server_with_fewer_threads_than_dispatchers_can_quit) {
+ brpc::Server server;
+ ReserveEchoServiceImpl service;
+ ASSERT_EQ(0, server.AddService(&service, brpc::SERVER_DOESNT_OWN_SERVICE));
+ brpc::ServerOptions options;
+ options.num_threads = NUM_THREADS;
+ ASSERT_EQ(0, server.Start(RESERVE_PORT, &options));
+ // `num_threads` workers are available to normal bthreads, on top of the
+ // ones reserved for the dispatchers.
+ ASSERT_EQ(NUM_THREADS, bthread_getconcurrency_by_tag(BTHREAD_TAG_DEFAULT));
+
+ brpc::Channel channel;
+ brpc::ChannelOptions copt;
+ copt.timeout_ms = 3000;
+ butil::EndPoint ep;
+ ASSERT_EQ(0, butil::str2endpoint("127.0.0.1", RESERVE_PORT, &ep));
+ ASSERT_EQ(0, channel.Init(ep, &copt));
+
+ test::EchoService_Stub stub(&channel);
+ for (int i = 0; i < 20; ++i) {
+ brpc::Controller cntl;
+ test::EchoRequest req;
+ test::EchoResponse res;
+ req.set_message("hello");
+ stub.Echo(&cntl, &req, &res, nullptr);
+ ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
+ ASSERT_EQ("hello", res.message());
+ }
+
+ // Previously, `Stop()`/`Join()` could not schedule their bthreads because
+ // every worker was stuck in `epoll_wait`.
+ StopArgs args(&server);
+ pthread_t th;
+ ASSERT_EQ(0, pthread_create(&th, nullptr, stop_and_join, &args));
+ int64_t deadline_us = butil::gettimeofday_us() + 10 * 1000000L;
+ while (!args.done.load(butil::memory_order_acquire) &&
+ butil::gettimeofday_us() < deadline_us) {
+ usleep(10000);
+ }
+ bool done = args.done.load(butil::memory_order_acquire);
+ if (!done) {
+ EXPECT_TRUE(done) << "Server::Stop()/Join() did not finish in 10s, "
+ "no worker available?";
+ std::abort();
+ }
+ ASSERT_EQ(0, pthread_join(th, nullptr));
+}
diff --git a/test/bthread_setconcurrency_unittest.cpp
b/test/bthread_setconcurrency_unittest.cpp
index e07239e87..8e98a0308 100644
--- a/test/bthread_setconcurrency_unittest.cpp
+++ b/test/bthread_setconcurrency_unittest.cpp
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+#include <climits>
+#include <thread>
#include <gtest/gtest.h>
#include <gflags/gflags.h>
#include "butil/atomicops.h"
@@ -37,7 +39,7 @@ void* dummy(void*) {
}
TEST(BthreadTest, setconcurrency) {
- ASSERT_EQ(8 + BTHREAD_EPOLL_THREAD_NUM, (size_t)bthread_getconcurrency());
+ ASSERT_EQ(8, bthread_getconcurrency());
ASSERT_EQ(EINVAL, bthread_setconcurrency(BTHREAD_MIN_CONCURRENCY - 1));
ASSERT_EQ(EINVAL, bthread_setconcurrency(0));
ASSERT_EQ(EINVAL, bthread_setconcurrency(-1));
@@ -167,11 +169,11 @@ TEST(BthreadTest, min_concurrency) {
ASSERT_EQ(0, bthread_setconcurrency(conn + add_conn + 1)); // set max
success
ASSERT_EQ(0, bthread_setconcurrency(conn + add_conn)); // set max success
ASSERT_EQ(conn + add_conn, bthread_getconcurrency());
- ASSERT_EQ(conn, bthread::g_task_control->concurrency());
+ ASSERT_EQ(conn, bthread::g_task_control->normal_concurrency());
ASSERT_EQ(1, set_min_concurrency(conn + 1)); // set min success
ASSERT_EQ(conn + 1, get_min_concurrency());
- ASSERT_EQ(conn + 1, bthread::g_task_control->concurrency());
+ ASSERT_EQ(conn + 1, bthread::g_task_control->normal_concurrency());
std::vector<bthread_t> tids;
for (int i = 0; i < conn; ++i) {
@@ -188,7 +190,7 @@ TEST(BthreadTest, min_concurrency) {
bthread_join(tids[i], nullptr);
}
ASSERT_EQ(conn + add_conn, bthread_getconcurrency());
- ASSERT_EQ(conn + add_conn, bthread::g_task_control->concurrency());
+ ASSERT_EQ(conn + add_conn, bthread::g_task_control->normal_concurrency());
}
int current_tag(int tag) {
@@ -222,4 +224,278 @@ TEST(BthreadTest, concurrency_by_tag) {
ASSERT_EQ(concurrency_by_tag(con + 1), true);
}
+// The tests below only assert on deltas: concurrency can never be reduced, so
+// the tests above leave it at an unspecified value. Keep them last in the file
+// as they make `concurrency()` exceed `bthread_getconcurrency()`.
+
+void initialize_task_control() {
+ ASSERT_TRUE(set_min_concurrency(0));
+ bthread_t tid;
+ ASSERT_EQ(0, bthread_start_background(&tid, nullptr, dummy, nullptr));
+ ASSERT_EQ(0, bthread_join(tid, nullptr));
+ ASSERT_NE(nullptr, bthread::g_task_control);
+}
+
+TEST(BthreadTest, reserve_concurrency_invalid_args) {
+ ASSERT_NO_FATAL_FAILURE(initialize_task_control());
+ ASSERT_EQ(0, bthread_add_reserved_concurrency_by_tag(
+ 1, BTHREAD_TAG_DEFAULT));
+ int total = bthread::g_task_control->concurrency();
+ int reserved = bthread::g_task_control->reserved_concurrency();
+ ASSERT_EQ(EINVAL, bthread_add_reserved_concurrency_by_tag(
+ 0, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(EINVAL, bthread_add_reserved_concurrency_by_tag(
+ -1, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(EINVAL, bthread_add_reserved_concurrency_by_tag(
+ 1, BTHREAD_TAG_DEFAULT - 1));
+ // Only one tag by default.
+ ASSERT_EQ(EINVAL, bthread_add_reserved_concurrency_by_tag(
+ 1, BTHREAD_TAG_DEFAULT + 1));
+ ASSERT_EQ(EINVAL, bthread_add_reserved_concurrency_by_tag(
+ BTHREAD_MAX_RESERVED_CONCURRENCY + 1,
BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(EINVAL, bthread_add_reserved_concurrency_by_tag(
+ INT_MAX, BTHREAD_TAG_DEFAULT));
+ // A valid individual request must also fit the accumulated reservation.
+ int room = BTHREAD_MAX_RESERVED_CONCURRENCY -
+ bthread::g_task_control->reserved_concurrency(BTHREAD_TAG_DEFAULT);
+ ASSERT_GT(room, 0);
+ ASSERT_EQ(EINVAL, bthread_add_reserved_concurrency_by_tag(
+ room + 1, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(total, bthread::g_task_control->concurrency());
+ ASSERT_EQ(reserved, bthread::g_task_control->reserved_concurrency());
+}
+
+// Reserved workers are extra: they raise the number of worker pthreads but not
+// the concurrency configured by the user.
+TEST(BthreadTest, reserve_concurrency_is_not_counted) {
+ ASSERT_NO_FATAL_FAILURE(initialize_task_control());
+ int normal = bthread_getconcurrency();
+ int total = bthread::g_task_control->concurrency();
+ int reserved = bthread::g_task_control->reserved_concurrency();
+ ASSERT_EQ(normal, bthread::g_task_control->normal_concurrency());
+ ASSERT_EQ(normal + reserved, total);
+ // Only one tag by default, so the per-tag counters must match the totals.
+ ASSERT_EQ(normal, bthread_getconcurrency_by_tag(BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(reserved, bthread::g_task_control->reserved_concurrency(
+ BTHREAD_TAG_DEFAULT));
+
+ int k = 3;
+ ASSERT_EQ(0, bthread_add_reserved_concurrency_by_tag(
+ k, BTHREAD_TAG_DEFAULT));
+ // Reservations update only the reserved count. The total is its sum with
+ // the normal count. These values hold as soon as `add_workers()` returns,
+ // without waiting for workers to register their `TaskGroup`.
+ ASSERT_EQ(total + k, bthread::g_task_control->concurrency());
+ ASSERT_EQ(reserved + k, bthread::g_task_control->reserved_concurrency());
+ ASSERT_EQ(reserved + k, bthread::g_task_control->reserved_concurrency(
+ BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(normal, bthread_getconcurrency());
+ ASSERT_EQ(normal, bthread::g_task_control->normal_concurrency());
+ ASSERT_EQ(normal, bthread_getconcurrency_by_tag(BTHREAD_TAG_DEFAULT));
+}
+
+// Asking for `N` workers must yield `N` runnable ones no matter whether the
+// reservation happens before or after the request.
+TEST(BthreadTest, reserve_concurrency_is_order_independent) {
+ ASSERT_NO_FATAL_FAILURE(initialize_task_control());
+ int k = 2;
+ int delta = 10;
+
+ // reserve -> setconcurrency
+ int normal = bthread_getconcurrency();
+ int reserved = bthread::g_task_control->reserved_concurrency();
+ ASSERT_EQ(0, bthread_add_reserved_concurrency_by_tag(
+ k, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(0, bthread_setconcurrency(normal + delta));
+ ASSERT_EQ(normal + delta, bthread_getconcurrency());
+ ASSERT_EQ(reserved + k, bthread::g_task_control->reserved_concurrency());
+ ASSERT_EQ(normal + delta, bthread::g_task_control->normal_concurrency());
+ ASSERT_EQ(normal + delta + reserved + k,
+ bthread::g_task_control->concurrency());
+
+ // setconcurrency -> reserve, same outcome.
+ normal = bthread_getconcurrency();
+ reserved = bthread::g_task_control->reserved_concurrency();
+ ASSERT_EQ(0, bthread_setconcurrency(normal + delta));
+ ASSERT_EQ(0, bthread_add_reserved_concurrency_by_tag(
+ k, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(normal + delta, bthread_getconcurrency());
+ ASSERT_EQ(reserved + k, bthread::g_task_control->reserved_concurrency());
+ ASSERT_EQ(normal + delta, bthread::g_task_control->normal_concurrency());
+ ASSERT_EQ(normal + delta + reserved + k,
+ bthread::g_task_control->concurrency());
+}
+
+// Reloading `-bthread_concurrency` with the same value must not make the
worker
+// count creep upwards by the reserved amount each time.
+TEST(BthreadTest, reserve_concurrency_no_creep_on_flag_reload) {
+ ASSERT_NO_FATAL_FAILURE(initialize_task_control());
+ ASSERT_EQ(0, bthread_add_reserved_concurrency_by_tag(
+ 2, BTHREAD_TAG_DEFAULT));
+ int normal = bthread_getconcurrency();
+ int total = bthread::g_task_control->concurrency();
+ for (int i = 0; i < 3; ++i) {
+ std::stringstream ss;
+ ss << normal;
+ ASSERT_FALSE(GFLAGS_NAMESPACE::SetCommandLineOption(
+ "bthread_concurrency", ss.str().c_str()).empty());
+ ASSERT_EQ(normal, bthread_getconcurrency());
+ ASSERT_EQ(total, bthread::g_task_control->concurrency());
+ }
+}
+
+TEST(BthreadTest, setconcurrency_by_tag_no_creep) {
+ ASSERT_NO_FATAL_FAILURE(initialize_task_control());
+ int target = bthread_getconcurrency_by_tag(BTHREAD_TAG_DEFAULT) + 5;
+ ASSERT_EQ(0, bthread_setconcurrency_by_tag(target, BTHREAD_TAG_DEFAULT));
+ int total = bthread::g_task_control->concurrency();
+ for (int i = 0; i < 3; ++i) {
+ ASSERT_EQ(0, bthread_setconcurrency_by_tag(target,
BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(target, bthread_getconcurrency_by_tag(BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(total, bthread::g_task_control->concurrency());
+ }
+}
+
+TEST(BthreadTest, reserve_concurrency_during_setconcurrency) {
+ ASSERT_NO_FATAL_FAILURE(initialize_task_control());
+ int normal = bthread_getconcurrency();
+ int reserved = bthread::g_task_control->reserved_concurrency();
+ butil::atomic<bool> stop(false);
+ butil::atomic<int> ready(0);
+ butil::atomic<int> failures(0);
+ int nsetters = 8;
+ int nreserve = 128;
+ std::vector<std::thread> setters;
+ for (int i = 0; i < nsetters; ++i) {
+ setters.emplace_back([&]() {
+ ready.fetch_add(1, butil::memory_order_release);
+ while (!stop.load(butil::memory_order_relaxed)) {
+ if (bthread_setconcurrency(normal) != 0) {
+ failures.fetch_add(1, butil::memory_order_relaxed);
+ }
+ }
+ });
+ }
+ while (ready.load(butil::memory_order_acquire) != nsetters) {
+ usleep(1000);
+ }
+ int rc = bthread_add_reserved_concurrency_by_tag(
+ nreserve, BTHREAD_TAG_DEFAULT);
+ stop.store(true, butil::memory_order_relaxed);
+ for (auto& setter : setters) {
+ setter.join();
+ }
+ ASSERT_EQ(0, rc);
+ ASSERT_EQ(0, failures.load(butil::memory_order_relaxed));
+ ASSERT_EQ(normal, bthread_getconcurrency());
+ ASSERT_EQ(normal, bthread::g_task_control->normal_concurrency());
+ ASSERT_EQ(reserved + nreserve,
+ bthread::g_task_control->reserved_concurrency());
+}
+
+// Keep this last, it maxes out the normal concurrency of the default tag.
+TEST(BthreadTest, concurrency_limit_excludes_reserved_workers) {
+ ASSERT_NO_FATAL_FAILURE(initialize_task_control());
+ ASSERT_EQ(0, bthread_add_reserved_concurrency_by_tag(
+ 1, BTHREAD_TAG_DEFAULT));
+ int reserved =
+ bthread::g_task_control->reserved_concurrency(BTHREAD_TAG_DEFAULT);
+ ASSERT_GT(reserved, 0);
+ // Reserving first must leave the entire normal budget available.
+ ASSERT_EQ(0, bthread_setconcurrency_by_tag(
+ BTHREAD_MAX_CONCURRENCY, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(BTHREAD_MAX_CONCURRENCY + reserved,
+ bthread::g_task_control->concurrency(BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(BTHREAD_MAX_CONCURRENCY,
+ bthread_getconcurrency_by_tag(BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(0, bthread_setconcurrency(BTHREAD_MAX_CONCURRENCY));
+ // Filling the normal budget first must still allow reservations.
+ ASSERT_EQ(0, bthread_add_reserved_concurrency_by_tag(
+ 1, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(BTHREAD_MAX_CONCURRENCY + reserved + 1,
+ bthread::g_task_control->concurrency(BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(BTHREAD_MAX_CONCURRENCY,
+ bthread_getconcurrency_by_tag(BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(EINVAL, bthread_setconcurrency_by_tag(
+ BTHREAD_MAX_CONCURRENCY + 1, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(BTHREAD_MAX_CONCURRENCY + reserved + 1,
+ bthread::g_task_control->concurrency(BTHREAD_TAG_DEFAULT));
+}
+
+void check_reserved_concurrency_limit() {
+ ASSERT_NO_FATAL_FAILURE(initialize_task_control());
+ int normal = bthread_getconcurrency();
+ ASSERT_EQ(0, bthread_add_reserved_concurrency_by_tag(
+ BTHREAD_MAX_RESERVED_CONCURRENCY - 1, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(0, bthread_add_reserved_concurrency_by_tag(
+ 1, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(BTHREAD_MAX_RESERVED_CONCURRENCY,
+ bthread::g_task_control->reserved_concurrency(
+ BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(normal, bthread_getconcurrency_by_tag(BTHREAD_TAG_DEFAULT));
+
+ // A full reserved budget must not prevent adding normal workers.
+ ASSERT_EQ(0, bthread_setconcurrency(normal + 1));
+ ASSERT_EQ(normal + 1, bthread_getconcurrency_by_tag(BTHREAD_TAG_DEFAULT));
+ int total = bthread::g_task_control->concurrency();
+ ASSERT_EQ(normal + 1 + BTHREAD_MAX_RESERVED_CONCURRENCY, total);
+ ASSERT_EQ(EINVAL, bthread_add_reserved_concurrency_by_tag(
+ 1, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(total, bthread::g_task_control->concurrency());
+ ASSERT_EQ(BTHREAD_MAX_RESERVED_CONCURRENCY,
+ bthread::g_task_control->reserved_concurrency(
+ BTHREAD_TAG_DEFAULT));
+}
+
+void set_threadsafe_death_test_style() {
+#ifdef GTEST_FLAG_SET
+ GTEST_FLAG_SET(death_test_style, "threadsafe");
+#else
+ ::testing::FLAGS_gtest_death_test_style = "threadsafe";
+#endif
+}
+
+TEST(BthreadConcurrencyDeathTest, reserved_limit_is_independent) {
+ // Isolate the full reservation so it cannot consume later tests' budgets.
+ set_threadsafe_death_test_style();
+ ASSERT_EXIT({
+ check_reserved_concurrency_limit();
+ _exit(::testing::Test::HasFailure() ? 1 : 0);
+ }, ::testing::ExitedWithCode(0), "");
+}
+
+void check_minimum_concurrency_with_reservation(bool lazy) {
+ // Blocking loops reserve workers on top of the four normal workers.
+ ASSERT_EQ(0, bthread_setconcurrency(4));
+ ASSERT_EQ(EINVAL, bthread_setconcurrency(3));
+ ASSERT_TRUE(set_min_concurrency(lazy ? 4 : 0));
+ ASSERT_FALSE(set_min_concurrency(3));
+ ASSERT_EQ(lazy ? 4 : 0, get_min_concurrency());
+
+ // The first reservation initializes the scheduler at the minimum.
+ ASSERT_EQ(0, bthread_add_reserved_concurrency_by_tag(
+ 1, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(0, bthread_setconcurrency_by_tag(4, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(EINVAL, bthread_setconcurrency_by_tag(3, BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(4, bthread_getconcurrency());
+ ASSERT_EQ(4, bthread_getconcurrency_by_tag(BTHREAD_TAG_DEFAULT));
+ ASSERT_EQ(1, bthread::g_task_control->reserved_concurrency());
+ ASSERT_EQ(5, bthread::g_task_control->concurrency());
+
+ bthread_t tid;
+ ASSERT_EQ(0, bthread_start_background(&tid, nullptr, dummy, nullptr));
+ ASSERT_EQ(0, bthread_join(tid, nullptr));
+}
+
+TEST(BthreadConcurrencyDeathTest,
minimum_concurrency_excludes_reserved_workers) {
+ set_threadsafe_death_test_style();
+ for (bool lazy : {false, true}) {
+ SCOPED_TRACE(lazy);
+ ASSERT_EXIT({
+ check_minimum_concurrency_with_reservation(lazy);
+ _exit(::testing::Test::HasFailure() ? 1 : 0);
+ }, ::testing::ExitedWithCode(0), "");
+ }
+}
+
} // namespace
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
