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 <aemerson@redhat.com>

diff --git a/src/rgw/rgw_asio_frontend.cc b/src/rgw/rgw_asio_frontend.cc
index 8148001ef3c1a603a4cb7f04f4e3f44079564b60..96c205487231b281ef1c02e242cb4c864be4c02f 100644 (file)
--- 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<tcp::socket&> 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 d06ef98849a27010b51e98531c502c50a2254f25..0f9956fd099975acd22094f38196166e2828de1b 100644 (file)
--- a/src/rgw/rgw_asio_frontend_connection.h
+++ b/src/rgw/rgw_asio_frontend_connection.h
@@ -2,9 +2,12 @@
 
 #include <mutex>
 
+#include <boost/asio/any_io_executor.hpp>
+#include <boost/asio/dispatch.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/beast/core/flat_static_buffer.hpp>
 #include <boost/intrusive/list.hpp>
+#include <boost/smart_ptr/intrusive_ptr.hpp>
 #include <boost/smart_ptr/intrusive_ref_counter.hpp>
 
 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<Connection>{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<Connection>{&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 079723f586d072ecde50fca31b67880ccc69e562..c88cfb391ae54931109edd22283ec652ca0e4d20 100644 (file)
--- 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 <boost/asio/io_context.hpp>
 #include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/strand.hpp>
+#include <boost/asio/spawn.hpp>
 #include <boost/asio/write.hpp>
 #include <boost/intrusive_ptr.hpp>
 
@@ -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<Connection> 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<Connection> 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();
 }