Reserve extra bthread workers for blocking event loops

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：不同分组之间有什么关系吗？
 

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调整。
 

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 concurrency counts:
+
+- **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.
 

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) {

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() {