Merge pull request #69713 from BBoozmen/wip-oozmen-77658

rgw/beast: fix shutdown crash from concurrent socket close in ConnectionList::close()

src/rgw/rgw_asio_frontend.cc

@@ -44,6 +44,7 @@
 
 #include "rgw_zone.h"
 
+#include "rgw_asio_frontend_connection.h"
 #include "rgw_asio_frontend_timer.h"
 #include "rgw_dmclock_async_scheduler.h"
 
@@ -57,13 +58,11 @@ namespace http = boost::beast::http;
 namespace ssl = boost::asio::ssl;
 #endif
 
-struct Connection;
-
 using timeout_timer = rgw::basic_timeout_timer<ceph::coarse_mono_clock,
-      boost::asio::any_io_executor, Connection>;
+      boost::asio::any_io_executor, rgw::asio::Connection>;
 
-static constexpr size_t parse_buffer_size = 65536;
-using parse_buffer = boost::beast::flat_static_buffer<parse_buffer_size>;
+static constexpr size_t parse_buffer_size = rgw::asio::parse_buffer_size;
+using parse_buffer = rgw::asio::parse_buffer;
 
 // use mmap/mprotect to allocate 512k coroutine stacks
 auto make_stack_allocator() {
@@ -407,56 +406,8 @@ void handle_connection(boost::asio::io_context& context,
   }
 }
 
-// timeout support requires that connections are reference-counted, because the
-// timeout_handler can outlive the coroutine
-struct Connection : boost::intrusive::list_base_hook<>,
-                    boost::intrusive_ref_counter<Connection>
-{
-  tcp::socket socket;
-  parse_buffer buffer;
-
-  explicit Connection(tcp::socket&& socket) noexcept
-      : socket(std::move(socket)) {}
-
-  void close(boost::system::error_code& ec) {
-    socket.close(ec);
-  }
-
-  tcp::socket& get_socket() { return socket; }
-};
-
-class ConnectionList {
-  using List = boost::intrusive::list<Connection>;
-  List connections;
-  std::mutex mutex;
-
-  void remove(Connection& c) {
-    std::lock_guard lock{mutex};
-    if (c.is_linked()) {
-      connections.erase(List::s_iterator_to(c));
-    }
-  }
- public:
-  class Guard {
-    ConnectionList *list;
-    Connection *conn;
-   public:
-    Guard(ConnectionList *list, Connection *conn) : list(list), conn(conn) {}
-    ~Guard() { list->remove(*conn); }
-  };
-  [[nodiscard]] Guard add(Connection& conn) {
-    std::lock_guard lock{mutex};
-    connections.push_back(conn);
-    return Guard{this, &conn};
-  }
-  void close(boost::system::error_code& ec) {
-    std::lock_guard lock{mutex};
-    for (auto& conn : connections) {
-      conn.socket.close(ec);
-    }
-    connections.clear();
-  }
-};
+using rgw::asio::Connection;
+using rgw::asio::ConnectionList;
 
 namespace dmc = rgw::dmclock;
 class AsioFrontend {

src/rgw/rgw_asio_frontend_connection.h

@@ -0,0 +1,75 @@
+#pragma once
+
+#include <mutex>
+
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/beast/core/flat_static_buffer.hpp>
+#include <boost/intrusive/list.hpp>
+#include <boost/smart_ptr/intrusive_ref_counter.hpp>
+
+namespace rgw::asio {
+
+using tcp = boost::asio::ip::tcp;
+
+static constexpr size_t parse_buffer_size = 65536;
+using parse_buffer = boost::beast::flat_static_buffer<parse_buffer_size>;
+
+// timeout support requires that connections are reference-counted, because the
+// timeout_handler can outlive the coroutine
+struct Connection : boost::intrusive::list_base_hook<>,
+                    boost::intrusive_ref_counter<Connection>
+{
+  tcp::socket socket;
+  parse_buffer buffer;
+
+  explicit Connection(tcp::socket&& socket) noexcept
+      : socket(std::move(socket)) {}
+
+  void close(boost::system::error_code& ec) {
+    socket.close(ec);
+  }
+
+  tcp::socket& get_socket() { return socket; }
+};
+
+class ConnectionList {
+  using List = boost::intrusive::list<Connection>;
+  List connections;
+  mutable std::mutex mutex;
+
+  void remove(Connection& c) {
+    std::lock_guard lock{mutex};
+    if (c.is_linked()) {
+      connections.erase(List::s_iterator_to(c));
+    }
+  }
+ public:
+  class Guard {
+    ConnectionList *list;
+    Connection *conn;
+   public:
+    Guard(ConnectionList *list, Connection *conn) : list(list), conn(conn) {}
+    ~Guard() { list->remove(*conn); }
+  };
+  [[nodiscard]] Guard add(Connection& conn) {
+    std::lock_guard lock{mutex};
+    connections.push_back(conn);
+    return Guard{this, &conn};
+  }
+  size_t size() const {
+    std::lock_guard lock{mutex};
+    return connections.size();
+  }
+  void close(boost::system::error_code& ec) {
+    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);
+    }
+  }
+};
+
+} // namespace rgw::asio

src/test/rgw/CMakeLists.txt

@@ -480,6 +480,10 @@ add_executable(unittest_rgw_async_utils test_rgw_async_utils.cc)
 add_ceph_unittest(unittest_rgw_async_utils)
 target_link_libraries(unittest_rgw_async_utils ${rgw_libs} ${UNITTEST_LIBS})
 
+add_executable(unittest_rgw_asio_frontend_connection test_rgw_asio_frontend_connection.cc)
+add_ceph_unittest(unittest_rgw_asio_frontend_connection)
+target_link_libraries(unittest_rgw_asio_frontend_connection ${rgw_libs} ${UNITTEST_LIBS})
+
 add_executable(unittest_rgw_tag test_rgw_tag.cc)
 add_ceph_unittest(unittest_rgw_tag)
 target_link_libraries(unittest_rgw_tag ${rgw_libs} ${UNITTEST_LIBS})

src/test/rgw/test_rgw_asio_frontend_connection.cc

@@ -0,0 +1,91 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
+// vim: ts=8 sw=2 sts=2 expandtab ft=cpp
+
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright contributors to the Ceph project
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#include "rgw/rgw_asio_frontend_connection.h"
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/write.hpp>
+#include <boost/intrusive_ptr.hpp>
+
+#include <gtest/gtest.h>
+
+namespace asio = boost::asio;
+using tcp = asio::ip::tcp;
+using rgw::asio::Connection;
+using rgw::asio::ConnectionList;
+
+// Testcases for ConnectionList::close() shutdown behavior.
+//
+// Verifies that close() uses socket.cancel() + ::shutdown() rather than
+// socket.close(). Calling socket.close() from outside the socket's strand
+// is a boost::asio thread-safety violation that can SIGSEGV or orphaned
+// pending operations.
+
+// Verify that ConnectionList::close() does NOT close the sockets (fd stays
+// valid, is_open() remains true) but does shut down the transport (writes
+// fail). This ensures we never call socket.close() from outside the strand.
+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());
+}