httpd: support scheduling group for TLS handshakes

Add set_new_connection_scheduling_group_cb() to http_server, allowing
callers to provide a callback that returns the scheduling group in
which TLS handshakes should run.  When the callback is configured and
the listener uses TLS, the server performs an explicit early handshake
(via tls::get_protocol_version) in the designated scheduling group
before switching back to the default group for request processing.

When no callback is set, behaviour is unchanged: the TLS handshake
happens lazily inside conn->process().

The per-connection work is moved into a new do_process_connection()
member coroutine (called from the non-coroutine try_with_gate lambda
in do_accept_one) so that scheduling-group switching, the early
handshake, and conn->process() can all be expressed with co_await.

Includes a test that verifies, on a TLS connection, that the callback
is invoked, the handshake runs in the configured scheduling group,
and the request handler runs in the default scheduling group.

Author: ScyllaPiotr

include/seastar/http/httpd.hh

@@ -21,6 +21,7 @@
 
 #pragma once
 
+#include <functional>
 #include <limits>
 #include <cctype>
 #include <vector>
@@ -37,6 +38,7 @@
 #include <seastar/http/routes.hh>
 #include <seastar/net/tls.hh>
 #include <seastar/core/shared_ptr.hh>
+#include <seastar/core/scheduling.hh>
 
 namespace seastar {
 
@@ -136,6 +138,11 @@ class http_server {
     bool _generate_date_header = true;
     gate _task_gate;
     std::optional<net::keepalive_params> _keepalive_params;
+    // Returns scheduling group for TLS handshakes. Consulted only for TLS listeners.
+    std::function<scheduling_group()> _new_connection_scheduling_group_cb;
+    // Hook invoked after switching to the TLS handshake scheduling group, right before the handshake.
+    // Aimed for testing if the scheduling group was really changed.
+    std::function<void()> _on_tls_handshake_sg_switch;
 public:
     routes _routes;
     using connection = seastar::httpd::connection;
@@ -199,6 +206,8 @@ public:
     /// When set to false the periodic date-update timer is also stopped.
     void set_generate_date_header(bool b);
 
+    void set_new_connection_scheduling_group_cb(std::function<scheduling_group()> fn);
+
     future<> listen(socket_address addr, server_credentials_ptr credentials);
     future<> listen(socket_address addr, listen_options lo, server_credentials_ptr credentials);
     future<> listen(socket_address addr, listen_options lo);
@@ -219,6 +228,7 @@ public:
     static sstring http_date();
 private:
     future<> do_accept_one(int which, bool with_tls);
+    future<> do_process_connection(connected_socket conn_fd, socket_address remote_address, bool tls);
     boost::intrusive::list<connection> _connections;
     friend class seastar::httpd::connection;
     friend class http_server_tester;
@@ -229,6 +239,9 @@ public:
     static std::vector<server_socket>& listeners(http_server& server) {
         return server._listeners;
     }
+    static void set_on_tls_handshake_sg_switch(http_server& server, std::function<void()> fn) {
+        server._on_tls_handshake_sg_switch = std::move(fn);
+    }
 };
 
 /*

src/http/httpd.cc

@@ -47,6 +47,7 @@
 #include <seastar/util/short_streams.hh>
 #include <seastar/util/log.hh>
 #include <seastar/util/string_utils.hh>
+#include <seastar/coroutine/switch_to.hh>
 
 
 using namespace std::chrono_literals;
@@ -401,6 +402,10 @@ void http_server::set_generate_date_header(bool b) {
     }
 }
 
+void http_server::set_new_connection_scheduling_group_cb(std::function<scheduling_group()> fn) {
+    _new_connection_scheduling_group_cb = std::move(fn);
+}
+
 future<> http_server::listen(socket_address addr, listen_options lo,
             server_credentials_ptr listener_credentials) {
     if (listener_credentials) {
@@ -476,16 +481,46 @@ future<> http_server::do_accept_one(int which, bool tls) {
         ar.connection.set_keepalive(true);
         ar.connection.set_keepalive_parameters(_keepalive_params.value());
     }
-    auto local_address = ar.connection.local_address();
-    auto conn = std::make_unique<connection>(*this, std::move(ar.connection),
-            std::move(ar.remote_address), std::move(local_address), tls);
-    (void)try_with_gate(_task_gate, [conn = std::move(conn)]() mutable {
-        return conn->process().handle_exception([conn = std::move(conn)] (std::exception_ptr ex) {
-            hlogger.error("request error: {}", ex);
-        });
+    // The lambda passed to try_with_gate must not be a coroutine,
+    // because try_with_gate invokes it but does not store it: a coroutine
+    // lambda would be destroyed while its frame is still running.
+    (void)try_with_gate(_task_gate,
+            [this, conn_fd = std::move(ar.connection),
+             remote_address = std::move(ar.remote_address), tls]() mutable {
+        return do_process_connection(std::move(conn_fd), std::move(remote_address), tls);
     }).handle_exception_type([] (const gate_closed_exception& e) {});
 }
 
+// Named member coroutine for per-connection processing, called from the
+// non-coroutine lambda in try_with_gate inside do_accept_one(). Parameters
+// are passed by value so they live safely in the coroutine frame.
+future<> http_server::do_process_connection(connected_socket conn_fd, socket_address remote_address, bool tls) {
+    auto local_address = conn_fd.local_address();
+    // When a scheduling group callback is configured, perform an early
+    // TLS handshake in the designated scheduling group. Without the
+    // callback, the handshake happens lazily inside conn->process().
+    if (tls && _new_connection_scheduling_group_cb) {
+        co_await coroutine::switch_to(_new_connection_scheduling_group_cb());
+        if (_on_tls_handshake_sg_switch) {
+            _on_tls_handshake_sg_switch();
+        }
+        try {
+            co_await seastar::tls::get_protocol_version(conn_fd);
+        } catch (...) {
+            hlogger.debug("TLS handshake failed: {}", std::current_exception());
+            co_return;
+        }
+        co_await coroutine::switch_to(default_scheduling_group());
+    }
+    auto conn = std::make_unique<connection>(*this, std::move(conn_fd),
+            std::move(remote_address), std::move(local_address), tls);
+    try {
+        co_await conn->process();
+    } catch (...) {
+        hlogger.error("request error: {}", std::current_exception());
+    }
+}
+
 uint64_t http_server::total_connections() const {
     return _total_connections;
 }

tests/unit/httpd_test.cc

@@ -17,6 +17,7 @@
 #include <seastar/core/do_with.hh>
 #include <seastar/core/future.hh>
 #include <seastar/core/loop.hh>
+#include <seastar/core/scheduling.hh>
 #include <seastar/core/when_all.hh>
 #include <seastar/core/units.hh>
 #include <seastar/testing/test_case.hh>
@@ -2530,3 +2531,105 @@ SEASTAR_TEST_CASE(test_http_reply_formatting) {
 
     return make_ready_future<>();
 }
+
+// Verify that when set_new_connection_scheduling_group_cb() is configured,
+// the TLS handshake runs in the designated scheduling group and the
+// request handler runs in the default scheduling group.
+SEASTAR_TEST_CASE(test_new_connection_scheduling_group) {
+    return seastar::async([] {
+        auto sg = create_scheduling_group("test-httpd-sg", 100).get();
+        std::exception_ptr ex;
+
+        try {
+            bool callback_invoked = false;
+            scheduling_group observed_handler_sg = sg; // intentionally wrong initial value
+            scheduling_group observed_handshake_sg; // default SG; should become sg
+
+            static const char test_cert[] =
+                "-----BEGIN CERTIFICATE-----\n"
+                "MIIBzDCCAXOgAwIBAgIULUjD6YNpUtkaibrj8Z5pMg/bQlEwCgYIKoZIzj0EAwIw\n"
+                "FDESMBAGA1UEAwwJbG9jYWxob3N0MCAXDTI2MDMxOTA4Mjg0NloYDzIxMjYwMjIz\n"
+                "MDgyODQ2WjAUMRIwEAYDVQQDDAlsb2NhbGhvc3QwWTATBgcqhkjOPQIBBggqhkjO\n"
+                "PQMBBwNCAATsDbUqgqgR3llGWbVnz3jcY16/ZUWZT4kaIneoRbjTlMgr/Z5cBM2H\n"
+                "ZaN7zY6yuLFPkf/yRNCki9Q565EqqIzRo4GgMIGdMB0GA1UdDgQWBBTDUmIk5BxS\n"
+                "nWiOWNucItd2PWCL3jBPBgNVHSMESDBGgBTDUmIk5BxSnWiOWNucItd2PWCL3qEY\n"
+                "pBYwFDESMBAGA1UEAwwJbG9jYWxob3N0ghQtSMPpg2lS2RqJuuPxnmkyD9tCUTAP\n"
+                "BgNVHRMBAf8EBTADAQH/MBoGA1UdEQQTMBGCCWxvY2FsaG9zdIcEfwAAATAKBggq\n"
+                "hkjOPQQDAgNHADBEAiBVsdCtlAEz0bqR73UViNXOKln5KCEon77KdjKHsDI1gAIg\n"
+                "da4opFU4dPGt3kKOe2UNTIxre0rdIuQmv4sidCTBfx4=\n"
+                "-----END CERTIFICATE-----\n";
+            static const char test_key[] =
+                "-----BEGIN PRIVATE KEY-----\n"
+                "MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgPJLOJNZ40LFhA4h4\n"
+                "DD+iARBzBUVEHY18F04UNcm/FNihRANCAATsDbUqgqgR3llGWbVnz3jcY16/ZUWZ\n"
+                "T4kaIneoRbjTlMgr/Z5cBM2HZaN7zY6yuLFPkf/yRNCki9Q565EqqIzR\n"
+                "-----END PRIVATE KEY-----\n";
+
+            auto server_creds = ::make_shared<tls::server_credentials>(::make_shared<tls::dh_params>());
+            server_creds->set_x509_key(
+                    tls::blob(test_cert, sizeof(test_cert) - 1),
+                    tls::blob(test_key, sizeof(test_key) - 1),
+                    tls::x509_crt_format::PEM);
+
+            tls::credentials_builder client_builder;
+            client_builder.set_x509_trust(tls::blob(test_cert, sizeof(test_cert) - 1), tls::x509_crt_format::PEM);
+            auto client_creds = client_builder.build_certificate_credentials();
+
+            auto addr = socket_address(ipv4_addr("127.0.0.1", 0));
+            listen_options lo;
+            lo.reuse_address = true;
+            lo.set_fixed_cpu(this_shard_id());
+
+            http_server server("test");
+            server.set_new_connection_scheduling_group_cb([sg, &callback_invoked] {
+                callback_invoked = true;
+                return sg;
+            });
+            http_server_tester::set_on_tls_handshake_sg_switch(server, [&observed_handshake_sg] {
+                observed_handshake_sg = current_scheduling_group();
+            });
+            server._routes.put(GET, "/test", new function_handler([&observed_handler_sg](const_req req) {
+                observed_handler_sg = current_scheduling_group();
+                return "";
+            }, "txt"));
+
+            server.listen(addr, lo, server_creds).get();
+            auto actual_addr = http_server_tester::listeners(server)[0].local_address();
+
+            // Run the client in a separate fiber so the reactor can
+            // interleave client and server TLS handshake progress.
+            // The server's get_protocol_version() forces an early
+            // handshake that needs the client to send ClientHello
+            // concurrently.
+            future<> client = seastar::async([&] {
+                auto c_socket = tls::connect(client_creds, actual_addr,
+                        tls::tls_options{.server_name = sstring("localhost")}).get();
+                auto input = c_socket.input();
+                auto output = c_socket.output();
+                auto close_in = deferred_close(input);
+                auto close_out = deferred_close(output);
+
+                output.write(sstring("GET /test HTTP/1.1\r\nHost: test\r\n\r\n")).get();
+                output.flush().get();
+                auto resp = input.read().get();
+                BOOST_REQUIRE_NE(resp.size(), 0u);
+                BOOST_REQUIRE_NE(std::string(resp.get(), resp.size()).find("200 OK"), std::string::npos);
+            });
+
+            client.get();
+
+            BOOST_REQUIRE(callback_invoked);
+            BOOST_REQUIRE(observed_handler_sg == default_scheduling_group());
+            BOOST_REQUIRE(observed_handshake_sg == sg);
+
+            server.stop().get();
+        } catch (...) {
+            ex = std::current_exception();
+        }
+
+        destroy_scheduling_group(sg).get();
+        if (ex) {
+            std::rethrow_exception(std::move(ex));
+        }
+    });
+}