Copilot commented on code in PR #3683:
URL: https://github.com/apache/brpc/pull/3683#discussion_r4395935082
##########
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:
Reservation failure at dispatcher startup aborts the process through
`CHECK_EQ`. This happens when `-event_dispatcher_num` exceeds
`BTHREAD_MAX_RESERVED_CONCURRENCY` (512) per tag, or when earlier reservations
already use the shared per-tag budget. Configurations that started before this
PR now crash at startup, while the `EpollThread` reservation in
`src/bthread/fd.cpp:147` only logs a warning. Either validate
`event_dispatcher_num` against the limit with `BUTIL_VALIDATE_GFLAG` where the
flag is defined (`src/bthread/task_control.cpp:60`), or use the same warning
fallback as `fd.cpp`.
##########
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:
The internal epoll thread now runs on the tag of the first bthread that
calls `bthread_fd_wait()`. Before this PR, `BTHREAD_ATTR_NORMAL` left the tag
unset, and `bthread_start_background()` mapped that to `BTHREAD_TAG_DEFAULT`,
so the shared loop always ran on the default tag. Because this one loop serves
fds from all tags, the chosen tag's workers are now pinned to it. Pin the
default tag as in the suggested fix, or keep caller-tag placement and document
it in the PR description and docs.
##########
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:
If the 10-second wait times out, the test returns while the `stop_and_join`
thread still uses `args` and `server`, which are stack variables. The thread is
then left with dangling references during teardown. Abort on timeout instead of
unwinding, and join only after the thread has finished.
--
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]
