chenBright commented on code in PR #3683:
URL: https://github.com/apache/brpc/pull/3683#discussion_r4396368691
##########
src/brpc/event_dispatcher.cpp:
##########
@@ -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);
Review Comment:
The shared 512-worker limit is intentional. CHECK_EQ follows the existing
dispatcher Start() policy and does not unconditionally abort under default
butil logging. Termination depends on the logging backend and
crash_on_fatal_log. A flag validator cannot account for earlier reservations or
thread-creation failures. Replacing this with a warning would bypass the
configured fatal-check policy and could start dispatchers without enough
workers. I’m keeping the current behavior.
##########
src/bthread/fd.cpp:
##########
@@ -139,10 +139,24 @@ class EpollThread {
}
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();
Review Comment:
This preserves the existing placement. With BTHREAD_TAG_INVALID,
bthread_start_background() takes the thread-local path when called from a
bthread, and TaskGroup::start_background() uses the current group's tag. Only
the non-worker path defaults to BTHREAD_TAG_DEFAULT. Setting attr.tag =
bthread_self_tag() makes that selection explicit so the reservation is charged
to the same tag. Forcing tag 0 would change the existing behavior
##########
test/brpc_event_dispatcher_unittest.cpp:
##########
@@ -532,3 +615,100 @@ 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);
+ }
+ ASSERT_TRUE(args.done.load(butil::memory_order_acquire))
+ << "Server::Stop()/Join() did not finish in 10s, no worker available?";
+ ASSERT_EQ(0, pthread_join(th, nullptr));
Review Comment:
Addressed in e5dc6713e3354996bff006eafbf3d662579bb33e.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
