test: Add perf test for http client

The test opens a server socket that reads socket up until double CRLF
and then responds back with "HTTP/1.1 200 OK host: test" line in a loop.
Then runs http client against it measuring time it takes to perform the
given amount of requests. It uses in-memory loopback sockets (seastar
wrapper around seastar::queue).

Signed-off-by: Pavel Emelyanov <[email protected]>

tests/perf/CMakeLists.txt

@@ -88,3 +88,7 @@ seastar_add_test (allocator
 
 seastar_add_test (container
   SOURCES container_perf.cc)
+
+seastar_add_test (http_client
+  SOURCES http_client_perf.cc
+  NO_SEASTAR_PERF_TESTING_LIBRARY)

tests/perf/http_client_perf.cc

@@ -0,0 +1,143 @@
+/*
+ * This file is open source software, licensed to you under the terms
+ * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
+ * distributed with this work for additional information regarding copyright
+ * ownership.  You may not use this file except in compliance with the License.
+ *
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+/*
+ * Copyright (C) 2024 ScyllaDB
+ */
+
+#include <seastar/core/seastar.hh>
+#include <seastar/core/app-template.hh>
+#include <seastar/core/sstring.hh>
+#include <seastar/core/when_all.hh>
+#include <seastar/http/client.hh>
+#include <seastar/http/request.hh>
+#include <../../tests/unit/loopback_socket.hh>
+#include <fmt/printf.h>
+#include <string>
+
+using namespace seastar;
+using namespace std::chrono_literals;
+
+class server {
+    seastar::server_socket _ss;
+    seastar::connected_socket _cs;
+    seastar::input_stream<char> _in;
+    seastar::output_stream<char> _out;
+    sstring _req;
+
+    future<> do_echo() {
+        return _in.read().then([this] (temporary_buffer<char> buf) {
+            if (buf.empty()) {
+                return make_ready_future<>();
+            }
+
+            _req += sstring(buf.get(), buf.size());
+            if (_req.ends_with("\r\n\r\n")) {
+                sstring r200("HTTP/1.1 200 OK\r\nHost: test\r\n\r\n");
+                return _out.write(r200).then([this] {
+                    return _out.flush().then([this] {
+                        _req = "";
+                        return do_echo();
+                    });
+                });
+            }
+
+            return do_echo();
+        });
+    }
+
+public:
+    server(loopback_connection_factory& lcf) : _ss(lcf.get_server_socket()) {}
+    future<> serve() {
+        return _ss.accept().then([this] (seastar::accept_result ar) {
+            _cs = std::move(ar.connection);
+            _in = _cs.input();
+            _out = _cs.output();
+            return do_echo();
+        });
+    }
+};
+
+class loopback_http_factory : public http::experimental::connection_factory {
+    loopback_socket_impl lsi;
+public:
+    explicit loopback_http_factory(loopback_connection_factory& f) : lsi(f) {}
+    virtual future<connected_socket> make(abort_source* as) override {
+        return lsi.connect(socket_address(ipv4_addr()), socket_address(ipv4_addr()));
+    }
+};
+
+class client {
+    seastar::http::experimental::client _cln;
+    const unsigned _warmup_limit;
+    const unsigned _limit;
+
+    future<> make_requests(unsigned nr) {
+        auto req = http::request::make("GET", "test", "/test");
+        return _cln.make_request(std::move(req), [] (const http::reply& rep, input_stream<char>&& in) {
+            return make_ready_future<>();
+        }, http::reply::status_type::ok).then([this, nr] {
+            if (nr <= 1) {
+                return make_ready_future<>();
+            }
+            return make_requests(nr - 1);
+        });
+    }
+public:
+    client(loopback_connection_factory& lcf, unsigned ops, unsigned warmup)
+            : _cln(std::make_unique<loopback_http_factory>(lcf))
+            , _warmup_limit(warmup)
+            , _limit(ops)
+    {}
+    future<> work() {
+        fmt::print("Warming up with {} requests\n", _warmup_limit);
+        return make_requests(_warmup_limit).then([this] {
+            fmt::print("Warmup finished, making {} requests\n", _limit);
+            auto start = std::chrono::steady_clock::now();
+            auto mallocs = memory::stats().mallocs();
+            return make_requests(_limit).then([this, start, mallocs] {
+                auto delta = std::chrono::steady_clock::now() - start;
+                auto allocs = memory::stats().mallocs() - mallocs;
+                fmt::print("Made {} requests in {}s, {} mallocs\n", _limit, std::chrono::duration_cast<std::chrono::duration<double>>(delta).count(), allocs);
+            });
+        }).finally([this] {
+            return _cln.close();
+        });
+    }
+};
+
+int main(int ac, char** av) {
+    app_template at;
+    namespace bpo = boost::program_options;
+    at.add_options()
+            ("total-ops", bpo::value<unsigned>()->default_value(1000000), "Total requests to make")
+            ("warmup-ops", bpo::value<unsigned>()->default_value(10000), "Requests to warm up")
+            ;
+    return at.run(ac, av, [&at] {
+        auto total_ops = at.configuration()["total-ops"].as<unsigned>();
+        auto warmup_ops = at.configuration()["warmup-ops"].as<unsigned>();
+        return do_with(loopback_connection_factory(1), [total_ops, warmup_ops] (auto& lcf) {
+            auto sf = do_with(server(lcf), [] (auto& srv) {
+                return srv.serve();
+            });
+            auto cf = do_with(client(lcf, total_ops, warmup_ops), [] (auto& cln) {
+                return cln.work();
+            });
+            return when_all(std::move(sf), std::move(cf)).discard_result();
+        });
+    });
+}