From: Lumir Sliva <61183145+lumir-sliva@users.noreply.github.com> Date: Fri, 24 Apr 2026 11:01:23 +0000 (+0200) Subject: rgw/beast: dispatch connection close on strand during pause X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=refs%2Fpull%2F68920%2Fhead;p=ceph.git rgw/beast: dispatch connection close on strand during pause AsioFrontend::pause() (and stop()) call ConnectionList::close() from the realm reloader thread. That invokes socket.close() on every live connection from a thread that isn't the connection's asio strand, so the reactor deregisters each socket's descriptor_state racing against the coroutine that handles the connection. If a keep-alive connection is starting its next async_read_some() on its strand at the same time, epoll_reactor::start_op() dereferences the descriptor_state that close has already freed and radosgw segfaults in an io_context_pool thread. This matches the race pattern fixed for timeout_timer in 86ad0b891b0: any caller that touches socket state from outside the strand needs to be serialized with the coroutine. Store the coroutine's strand executor on Connection and dispatch the close onto that strand in ConnectionList::close(). The lambda captures an intrusive_ptr to Connection so the object lives until the close actually runs, since connections.clear() unlinks the list entry right after dispatch. Tracker: https://tracker.ceph.com/issues/76094 Signed-off-by: Lumir Sliva <61183145+lumir-sliva@users.noreply.github.com> Update `unittest_rgw_asio_frontend` to run the `io_context` and execute in a coroutine to match the environment in RGW. Signed-off-by: Adam C. Emerson --- diff --git a/src/rgw/rgw_asio_frontend.cc b/src/rgw/rgw_asio_frontend.cc index 8148001ef3c..96c20548723 100644 --- a/src/rgw/rgw_asio_frontend.cc +++ b/src/rgw/rgw_asio_frontend.cc @@ -1174,7 +1174,7 @@ void AsioFrontend::on_accept(Listener& l, tcp::socket stream) #endif boost::asio::spawn(make_strand(context), std::allocator_arg, make_stack_allocator(), [this, s=std::move(stream), ssl_ctx] (boost::asio::yield_context yield) mutable { - auto conn = boost::intrusive_ptr{new Connection(std::move(s))}; + auto conn = boost::intrusive_ptr{new Connection(std::move(s), yield.get_executor())}; auto c = connections.add(*conn); // wrap the tcp stream in an ssl stream boost::asio::ssl::stream stream{conn->socket, *ssl_ctx}; @@ -1209,7 +1209,7 @@ void AsioFrontend::on_accept(Listener& l, tcp::socket stream) #endif // WITH_RADOSGW_BEAST_OPENSSL boost::asio::spawn(make_strand(context), std::allocator_arg, make_stack_allocator(), [this, s=std::move(stream)] (boost::asio::yield_context yield) mutable { - auto conn = boost::intrusive_ptr{new Connection(std::move(s))}; + auto conn = boost::intrusive_ptr{new Connection(std::move(s), yield.get_executor())}; auto c = connections.add(*conn); auto timeout = timeout_timer{yield.get_executor(), request_timeout, conn}; boost::system::error_code ec; @@ -1254,7 +1254,7 @@ void AsioFrontend::stop() } // close all connections - connections.close(ec); + connections.close(); pause_mutex.cancel(); } @@ -1280,7 +1280,7 @@ void AsioFrontend::pause() const bool graceful_stop{ g_ceph_context->_conf->rgw_graceful_stop }; if (!graceful_stop) { // close all connections so outstanding requests fail quickly - connections.close(ec); + connections.close(); } // pause and wait until outstanding requests complete diff --git a/src/rgw/rgw_asio_frontend_connection.h b/src/rgw/rgw_asio_frontend_connection.h index d06ef98849a..0f9956fd099 100644 --- a/src/rgw/rgw_asio_frontend_connection.h +++ b/src/rgw/rgw_asio_frontend_connection.h @@ -2,9 +2,12 @@ #include +#include +#include #include #include #include +#include #include namespace rgw::asio { @@ -21,12 +24,19 @@ struct Connection : boost::intrusive::list_base_hook<>, { tcp::socket socket; parse_buffer buffer; + // executor of the handle_connection coroutine's strand. used so that close() + // from outside the strand (e.g. frontend pause) serializes with coroutine + // operations on the socket instead of racing the reactor + boost::asio::any_io_executor strand; + Connection(tcp::socket&& socket, + boost::asio::any_io_executor strand) noexcept + : socket(std::move(socket)), strand(std::move(strand)) {} - explicit Connection(tcp::socket&& socket) noexcept - : socket(std::move(socket)) {} - - void close(boost::system::error_code& ec) { - socket.close(ec); + void close() { + boost::asio::dispatch(strand, [c = boost::intrusive_ptr{this}] { + boost::system::error_code ec; + c->socket.close(ec); + }); } tcp::socket& get_socket() { return socket; } @@ -60,16 +70,21 @@ class ConnectionList { std::lock_guard lock{mutex}; return connections.size(); } - void close(boost::system::error_code& ec) { + void close() { std::lock_guard lock{mutex}; for (auto& conn : connections) { - // cancel pending reactor operations which delivers operation_aborted - // to completion handlers and then shutdown the transport so any - // subsequent I/O attempts fail immediately. - conn.socket.cancel(ec); - conn.socket.shutdown(tcp::socket::shutdown_both, ec); + // dispatch operations to the connection's strand so it serializes with + // the handle_connection coroutine. + boost::asio::dispatch( + conn.strand, [c = boost::intrusive_ptr{&conn}] { + boost::system::error_code ec; + // cancel pending reactor operations which delivers operation_aborted + // to completion handlers and then shutdown the transport so any + // subsequent I/O attempts fail immediately. + c->socket.cancel(ec); + c->socket.shutdown(tcp::socket::shutdown_both, ec); + }); } } }; - } // namespace rgw::asio diff --git a/src/test/rgw/test_rgw_asio_frontend_connection.cc b/src/test/rgw/test_rgw_asio_frontend_connection.cc index 079723f586d..c88cfb391ae 100644 --- a/src/test/rgw/test_rgw_asio_frontend_connection.cc +++ b/src/test/rgw/test_rgw_asio_frontend_connection.cc @@ -17,6 +17,8 @@ #include #include +#include +#include #include #include @@ -40,52 +42,64 @@ using rgw::asio::ConnectionList; TEST(BeastFrontendShutdown, CloseShutdownsButDoesNotCloseSocket) { asio::io_context ioctx; - ConnectionList connections; - - tcp::acceptor acceptor(ioctx, tcp::endpoint(tcp::v4(), 0)); - acceptor.listen(3); - - tcp::socket c0(ioctx), c1(ioctx), c2(ioctx); - c0.connect(acceptor.local_endpoint()); - auto s0 = acceptor.accept(); - c1.connect(acceptor.local_endpoint()); - auto s1 = acceptor.accept(); - c2.connect(acceptor.local_endpoint()); - auto s2 = acceptor.accept(); - - boost::intrusive_ptr conns[] = { - new Connection(std::move(s0)), - new Connection(std::move(s1)), - new Connection(std::move(s2)), - }; - - { - auto g0 = connections.add(*conns[0]); - auto g1 = connections.add(*conns[1]); - auto g2 = connections.add(*conns[2]); - - ASSERT_EQ(3u, connections.size()); - - boost::system::error_code ec; - connections.close(ec); - - // After close(): sockets must still be "open" from boost::asio's - // perspective (fd not released, descriptor_data not zeroed). - // socket.close() would have set is_open()=false and fd=-1, which is the - // thread-safety violation that causes SIGSEGV. - for (auto& conn : conns) { - EXPECT_TRUE(conn->socket.is_open()); - EXPECT_GE(conn->socket.native_handle(), 0); - } - - // But the transport is shut down — writes must fail - for (auto& conn : conns) { - char buf[] = "test"; - asio::write(conn->socket, asio::buffer(buf), ec); - EXPECT_TRUE(ec.failed()) - << "Write should fail after shutdown"; - } - } // guards destroyed here — connections removed from list - - EXPECT_EQ(0u, connections.size()); + asio::spawn( + ioctx.get_executor(), + [&](asio::yield_context yield) -> void { + ConnectionList connections; + + tcp::acceptor acceptor(ioctx, tcp::endpoint(tcp::v4(), 0)); + acceptor.listen(3); + + tcp::socket c0(ioctx), c1(ioctx), c2(ioctx); + c0.async_connect(acceptor.local_endpoint(), yield); + auto s0 = acceptor.async_accept(yield); + c1.async_connect(acceptor.local_endpoint(), yield); + auto s1 = acceptor.async_accept(yield); + c2.async_connect(acceptor.local_endpoint(), yield); + auto s2 = acceptor.async_accept(yield); + + boost::intrusive_ptr conns[] = { + new Connection( + std::move(s0), asio::make_strand(yield.get_executor())), + new Connection( + std::move(s1), asio::make_strand(yield.get_executor())), + new Connection( + std::move(s2), asio::make_strand(yield.get_executor())), + }; + + { + auto g0 = connections.add(*conns[0]); + auto g1 = connections.add(*conns[1]); + auto g2 = connections.add(*conns[2]); + + ASSERT_EQ(3u, connections.size()); + + connections.close(); + + // After close(): sockets must still be "open" from boost::asio's + // perspective (fd not released, descriptor_data not zeroed). + // socket.close() would have set is_open()=false and fd=-1, which is the + // thread-safety violation that causes SIGSEGV. + for (auto& conn : conns) { + EXPECT_TRUE(conn->socket.is_open()); + EXPECT_GE(conn->socket.native_handle(), 0); + } + + // But the transport is shut down — writes must fail + for (auto& conn : conns) { + boost::system::error_code ec; + char buf[] = "test"; + asio::async_write(conn->socket, asio::buffer(buf), yield[ec]); + EXPECT_TRUE(ec.failed()) << "Write should fail after shutdown"; + } + } // guards destroyed here — connections removed from list + + EXPECT_EQ(0u, connections.size()); + }, + [](std::exception_ptr eptr) { + if (eptr) { + std::rethrow_exception(eptr); + } + }); + ioctx.run(); }