Copilot commented on code in PR #3631:
URL: https://github.com/apache/brpc/pull/3631#discussion_r4345022984
##########
src/brpc/policy/mongo_protocol.cpp:
##########
@@ -212,15 +239,13 @@ void ProcessMongoRequest(InputMessageBase* msg_base) {
const Server::MethodProperty *mp =
server_accessor.FindMethodPropertyByFullName(
srv_des->method(0)->full_name());
- MongoContextMessage* context_msg =
- dynamic_cast<MongoContextMessage*>(socket->parsing_context());
- if (nullptr == context_msg) {
+ if (nullptr == msg->context) {
LOG(WARNING) << "socket context wasn't set correctly";
return;
}
SendMongoResponse* mongo_done = new SendMongoResponse(server);
- mongo_done->cntl.set_mongo_session_data(context_msg->context());
+ mongo_done->cntl.set_mongo_session_data(msg->context.get());
Review Comment:
`MongoInputMessage` correctly retains an `intrusive_ptr<MongoContext>`, but
passing `msg->context.get()` into the controller risks reintroducing a
use-after-free if `Controller::set_mongo_session_data()` stores only a raw
pointer and the response path uses it after `ProcessMongoRequest()` returns
(when `msg` is destroyed and releases the last ref). A concrete fix is to
extend the context lifetime to at least the lifetime of `mongo_done` (e.g.,
store a `butil::intrusive_ptr<MongoContext>` member inside `SendMongoResponse`
/ controller-private state and assign from `msg->context`, then pass the raw
pointer derived from that retained ref).
##########
test/brpc_socket_unittest.cpp:
##########
@@ -2130,6 +2131,88 @@ TEST_F(SocketTest, shutdown_write) {
}
}
+// A parsing context taking a while to be described, during which the bthread
+// parsing messages of the socket may reset the context.
+class SlowDescribedContext : public brpc::Destroyable, public
brpc::Describable {
+public:
+ SlowDescribedContext(bthread::CountdownEvent* describing,
+ butil::atomic<int>* ndestroyed)
+ : _describing(describing)
+ , _ndestroyed(ndestroyed)
+ , _name("SlowDescribedContext") {}
+
+ void Describe(std::ostream& os, const brpc::DescribeOptions&) const
override {
+ _describing->signal();
+ // Not bthread_usleep(): DebugSocket() describes under a pthread mutex.
+ usleep(100000);
+ os << _name;
+ }
+
+ void Destroy() override {
+ _ndestroyed->fetch_add(1);
+ delete this;
+ }
+
+private:
+ bthread::CountdownEvent* _describing;
+ butil::atomic<int>* _ndestroyed;
+ std::string _name;
+};
+
+struct DebugSocketArg {
+ brpc::SocketId id;
+ bthread::CountdownEvent* describing;
+ std::string description;
+};
+
+void* DescribeSocket(void* void_arg) {
+ DebugSocketArg* arg = static_cast<DebugSocketArg*>(void_arg);
+ std::ostringstream os;
+ brpc::Socket::DebugSocket(os, arg->id);
+ arg->description = os.str();
+ return nullptr;
+}
+
+void* ResetParsingContextWhileDescribing(void* void_arg) {
+ DebugSocketArg* arg = static_cast<DebugSocketArg*>(void_arg);
+ EXPECT_EQ(0, arg->describing->wait());
+ brpc::SocketUniquePtr s;
+ EXPECT_EQ(0, brpc::Socket::Address(arg->id, &s));
+ // What InputMessenger does before trying other protocols.
+ s->reset_parsing_context(nullptr);
+ return nullptr;
+}
Review Comment:
This function runs on a `pthread` and uses gtest assertions (`EXPECT_EQ`).
gtest assertions from non-test threads can be unreliable (failures may not be
reported deterministically depending on gtest configuration). Prefer capturing
results into shared state (e.g., atomic/flags or an int error code) and
asserting in the main test thread after `pthread_join`, or use `ASSERT_*` only
in the main thread.
##########
test/brpc_socket_unittest.cpp:
##########
@@ -2130,6 +2131,88 @@ TEST_F(SocketTest, shutdown_write) {
}
}
+// A parsing context taking a while to be described, during which the bthread
+// parsing messages of the socket may reset the context.
+class SlowDescribedContext : public brpc::Destroyable, public
brpc::Describable {
+public:
+ SlowDescribedContext(bthread::CountdownEvent* describing,
+ butil::atomic<int>* ndestroyed)
+ : _describing(describing)
+ , _ndestroyed(ndestroyed)
+ , _name("SlowDescribedContext") {}
+
+ void Describe(std::ostream& os, const brpc::DescribeOptions&) const
override {
+ _describing->signal();
+ // Not bthread_usleep(): DebugSocket() describes under a pthread mutex.
+ usleep(100000);
Review Comment:
`usleep` requires `<unistd.h>`; if it’s not already transitively included in
this test TU, this will fail to compile on stricter toolchains. Add an explicit
include for `<unistd.h>` (or replace with an existing project utility sleep API
that is already included in this file).
##########
test/brpc_socket_unittest.cpp:
##########
@@ -2130,6 +2131,88 @@ TEST_F(SocketTest, shutdown_write) {
}
}
+// A parsing context taking a while to be described, during which the bthread
+// parsing messages of the socket may reset the context.
+class SlowDescribedContext : public brpc::Destroyable, public
brpc::Describable {
+public:
+ SlowDescribedContext(bthread::CountdownEvent* describing,
+ butil::atomic<int>* ndestroyed)
+ : _describing(describing)
+ , _ndestroyed(ndestroyed)
+ , _name("SlowDescribedContext") {}
+
+ void Describe(std::ostream& os, const brpc::DescribeOptions&) const
override {
+ _describing->signal();
+ // Not bthread_usleep(): DebugSocket() describes under a pthread mutex.
+ usleep(100000);
+ os << _name;
+ }
+
+ void Destroy() override {
+ _ndestroyed->fetch_add(1);
+ delete this;
+ }
+
+private:
+ bthread::CountdownEvent* _describing;
+ butil::atomic<int>* _ndestroyed;
+ std::string _name;
+};
+
+struct DebugSocketArg {
+ brpc::SocketId id;
+ bthread::CountdownEvent* describing;
+ std::string description;
+};
+
+void* DescribeSocket(void* void_arg) {
+ DebugSocketArg* arg = static_cast<DebugSocketArg*>(void_arg);
+ std::ostringstream os;
+ brpc::Socket::DebugSocket(os, arg->id);
+ arg->description = os.str();
+ return nullptr;
+}
+
+void* ResetParsingContextWhileDescribing(void* void_arg) {
+ DebugSocketArg* arg = static_cast<DebugSocketArg*>(void_arg);
+ EXPECT_EQ(0, arg->describing->wait());
+ brpc::SocketUniquePtr s;
+ EXPECT_EQ(0, brpc::Socket::Address(arg->id, &s));
+ // What InputMessenger does before trying other protocols.
+ s->reset_parsing_context(nullptr);
+ return nullptr;
+}
+
+TEST_F(SocketTest, debug_socket_keeps_parsing_context_alive) {
+ brpc::SocketId id;
+ brpc::SocketOptions options;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ brpc::SocketUniquePtr s;
+ ASSERT_EQ(0, brpc::Socket::Address(id, &s));
+
+ bthread::CountdownEvent describing(1);
+ butil::atomic<int> ndestroyed(0);
+ s->reset_parsing_context(new SlowDescribedContext(&describing,
&ndestroyed));
+
+ DebugSocketArg arg = { id, &describing, "" };
+ bthread_t describer;
+ ASSERT_EQ(0, bthread_start_background(&describer, nullptr, DescribeSocket,
&arg));
+ // Reset in a pthread: signaling a bthread in Describe() would switch to
+ // it on the worker of the describer, which holds a pthread mutex.
+ pthread_t resetter;
+ ASSERT_EQ(0, pthread_create(&resetter, nullptr,
+ ResetParsingContextWhileDescribing, &arg));
+ ASSERT_EQ(0, bthread_join(describer, nullptr));
+ ASSERT_EQ(0, pthread_join(resetter, nullptr));
+
+ ASSERT_NE(std::string::npos, arg.description.find(
+ "parsing_context=SlowDescribedContext{SlowDescribedContext}"))
+ << arg.description;
Review Comment:
This assertion hard-codes a large, formatting-sensitive substring of
`DebugSocket()` output. Small unrelated formatting changes to `/sockets/<id>`
could break the test. Consider relaxing it to validate only the essential
invariants (e.g., that `parsing_context=` is present and includes
`SlowDescribedContext`, and/or that `ndestroyed==1` and
`s->parsing_context()==nullptr`), which still proves the lifetime/locking
behavior without coupling to exact formatting.
--
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]
