summaryrefslogtreecommitdiffstats
path: root/src/seastar/tests/unit/rpc_test.cc
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--src/seastar/tests/unit/rpc_test.cc1199
1 files changed, 1199 insertions, 0 deletions
diff --git a/src/seastar/tests/unit/rpc_test.cc b/src/seastar/tests/unit/rpc_test.cc
new file mode 100644
index 000000000..0b729ada1
--- /dev/null
+++ b/src/seastar/tests/unit/rpc_test.cc
@@ -0,0 +1,1199 @@
+/*
+ * 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) 2016 ScyllaDB
+ */
+
+#include "loopback_socket.hh"
+#include <seastar/rpc/rpc.hh>
+#include <seastar/rpc/rpc_types.hh>
+#include <seastar/rpc/lz4_compressor.hh>
+#include <seastar/rpc/lz4_fragmented_compressor.hh>
+#include <seastar/rpc/multi_algo_compressor_factory.hh>
+#include <seastar/testing/test_case.hh>
+#include <seastar/testing/thread_test_case.hh>
+#include <seastar/testing/test_runner.hh>
+#include <seastar/core/thread.hh>
+#include <seastar/core/sleep.hh>
+#include <seastar/core/distributed.hh>
+#include <seastar/core/loop.hh>
+#include <seastar/util/defer.hh>
+#include <seastar/util/log.hh>
+
+using namespace seastar;
+
+struct serializer {
+};
+
+template <typename T, typename Output>
+inline
+void write_arithmetic_type(Output& out, T v) {
+ static_assert(std::is_arithmetic<T>::value, "must be arithmetic type");
+ return out.write(reinterpret_cast<const char*>(&v), sizeof(T));
+}
+
+template <typename T, typename Input>
+inline
+T read_arithmetic_type(Input& in) {
+ static_assert(std::is_arithmetic<T>::value, "must be arithmetic type");
+ T v;
+ in.read(reinterpret_cast<char*>(&v), sizeof(T));
+ return v;
+}
+
+template <typename Output>
+inline void write(serializer, Output& output, int32_t v) { return write_arithmetic_type(output, v); }
+template <typename Output>
+inline void write(serializer, Output& output, uint32_t v) { return write_arithmetic_type(output, v); }
+template <typename Output>
+inline void write(serializer, Output& output, int64_t v) { return write_arithmetic_type(output, v); }
+template <typename Output>
+inline void write(serializer, Output& output, uint64_t v) { return write_arithmetic_type(output, v); }
+template <typename Output>
+inline void write(serializer, Output& output, double v) { return write_arithmetic_type(output, v); }
+template <typename Input>
+inline int32_t read(serializer, Input& input, rpc::type<int32_t>) { return read_arithmetic_type<int32_t>(input); }
+template <typename Input>
+inline uint32_t read(serializer, Input& input, rpc::type<uint32_t>) { return read_arithmetic_type<uint32_t>(input); }
+template <typename Input>
+inline uint64_t read(serializer, Input& input, rpc::type<uint64_t>) { return read_arithmetic_type<uint64_t>(input); }
+template <typename Input>
+inline uint64_t read(serializer, Input& input, rpc::type<int64_t>) { return read_arithmetic_type<int64_t>(input); }
+template <typename Input>
+inline double read(serializer, Input& input, rpc::type<double>) { return read_arithmetic_type<double>(input); }
+
+template <typename Output>
+inline void write(serializer, Output& out, const sstring& v) {
+ write_arithmetic_type(out, uint32_t(v.size()));
+ out.write(v.c_str(), v.size());
+}
+
+template <typename Input>
+inline sstring read(serializer, Input& in, rpc::type<sstring>) {
+ auto size = read_arithmetic_type<uint32_t>(in);
+ sstring ret = uninitialized_string(size);
+ in.read(ret.data(), size);
+ return ret;
+}
+
+using test_rpc_proto = rpc::protocol<serializer>;
+using make_socket_fn = std::function<seastar::socket ()>;
+
+struct rpc_loopback_error_injector : public loopback_error_injector {
+ int _x = 0;
+ bool server_rcv_error() override {
+ return _x++ >= 50;
+ }
+};
+
+class rpc_socket_impl : public ::net::socket_impl {
+ promise<connected_socket> _p;
+ bool _connect;
+ loopback_socket_impl _socket;
+ rpc_loopback_error_injector _error_injector;
+public:
+ rpc_socket_impl(loopback_connection_factory& factory, bool connect, bool inject_error)
+ : _connect(connect),
+ _socket(factory, inject_error ? &_error_injector : nullptr) {
+ }
+ virtual future<connected_socket> connect(socket_address sa, socket_address local, transport proto = transport::TCP) override {
+ return _connect ? _socket.connect(sa, local, proto) : _p.get_future();
+ }
+ virtual void set_reuseaddr(bool reuseaddr) override {}
+ virtual bool get_reuseaddr() const override { return false; };
+ virtual void shutdown() override {
+ if (_connect) {
+ _socket.shutdown();
+ } else {
+ _p.set_exception(std::make_exception_ptr(std::system_error(ECONNABORTED, std::system_category())));
+ }
+ }
+};
+
+struct rpc_test_config {
+ rpc::resource_limits resource_limits = {};
+ rpc::server_options server_options = {};
+ bool connect = true;
+ bool inject_error = false;
+};
+
+template<typename MsgType = int>
+class rpc_test_env {
+ struct rpc_test_service {
+ test_rpc_proto _proto;
+ test_rpc_proto::server _server;
+ std::vector<MsgType> _handlers;
+
+ rpc_test_service() = delete;
+ explicit rpc_test_service(const rpc_test_config& cfg, loopback_connection_factory& lcf)
+ : _proto(serializer())
+ , _server(_proto, cfg.server_options, lcf.get_server_socket(), cfg.resource_limits)
+ { }
+
+ test_rpc_proto& proto() {
+ return _proto;
+ }
+
+ test_rpc_proto::server& server() {
+ return _server;
+ }
+
+ future<> stop() {
+ return parallel_for_each(_handlers, [this] (auto t) {
+ return proto().unregister_handler(t);
+ }).finally([this] {
+ return server().stop();
+ });
+ }
+
+ template<typename Func>
+ auto register_handler(MsgType t, scheduling_group sg, Func func) {
+ _handlers.emplace_back(t);
+ return proto().register_handler(t, sg, std::move(func));
+ }
+
+ future<> unregister_handler(MsgType t) {
+ auto it = std::find(_handlers.begin(), _handlers.end(), t);
+ assert(it != _handlers.end());
+ _handlers.erase(it);
+ return proto().unregister_handler(t);
+ }
+ };
+
+ rpc_test_config _cfg;
+ loopback_connection_factory _lcf;
+ std::unique_ptr<sharded<rpc_test_service>> _service;
+
+public:
+ rpc_test_env() = delete;
+ explicit rpc_test_env(rpc_test_config cfg)
+ : _cfg(cfg), _service(std::make_unique<sharded<rpc_test_service>>())
+ {
+ }
+
+ using test_fn = std::function<future<> (rpc_test_env<MsgType>& env)>;
+ static future<> do_with(rpc_test_config cfg, test_fn&& func) {
+ return seastar::do_with(rpc_test_env(cfg), [func] (rpc_test_env<MsgType>& env) {
+ return env.start().then([&env, func] {
+ return func(env);
+ }).finally([&env] {
+ return env.stop();
+ });
+ });
+ }
+
+ using thread_test_fn = std::function<void (rpc_test_env<MsgType>& env)>;
+ static future<> do_with_thread(rpc_test_config cfg, thread_test_fn&& func) {
+ return do_with(std::move(cfg), [func] (rpc_test_env<MsgType>& env) {
+ return seastar::async([&env, func] {
+ func(env);
+ });
+ });
+ }
+
+ using thread_test_fn_with_client = std::function<void (rpc_test_env<MsgType>& env, test_rpc_proto::client& cl)>;
+ static future<> do_with_thread(rpc_test_config cfg, rpc::client_options co, thread_test_fn_with_client&& func) {
+ return do_with(std::move(cfg), [func, co = std::move(co)] (rpc_test_env<MsgType>& env) {
+ return seastar::async([&env, func, co = std::move(co)] {
+ test_rpc_proto::client cl(env.proto(), co, env.make_socket(), ipv4_addr());
+ auto stop = defer([&] { cl.stop().get(); });
+ func(env, cl);
+ });
+ });
+ }
+
+ static future<> do_with_thread(rpc_test_config cfg, thread_test_fn_with_client&& func) {
+ return do_with_thread(std::move(cfg), rpc::client_options(), std::move(func));
+ }
+
+ auto make_socket() {
+ return seastar::socket(std::make_unique<rpc_socket_impl>(_lcf, _cfg.connect, _cfg.inject_error));
+ };
+
+ test_rpc_proto& proto() {
+ return local_service().proto();
+ }
+
+ test_rpc_proto::server& server() {
+ return local_service().server();
+ }
+
+ template<typename Func>
+ future<> register_handler(MsgType t, scheduling_group sg, Func func) {
+ return _service->invoke_on_all([t, func = std::move(func), sg] (rpc_test_service& s) mutable {
+ s.register_handler(t, sg, std::move(func));
+ });
+ }
+
+ template<typename Func>
+ future<> register_handler(MsgType t, Func func) {
+ return register_handler(t, scheduling_group(), std::move(func));
+ }
+
+ future<> unregister_handler(MsgType t) {
+ return _service->invoke_on_all([t] (rpc_test_service& s) mutable {
+ return s.unregister_handler(t);
+ });
+ }
+
+private:
+ rpc_test_service& local_service() {
+ return _service->local();
+
+ }
+
+ future<> start() {
+ return _service->start(std::cref(_cfg), std::ref(_lcf));
+ }
+
+ future<> stop() {
+ return _service->stop().then([this] {
+ return _lcf.destroy_all_shards();
+ });
+ }
+};
+
+struct cfactory : rpc::compressor::factory {
+ mutable int use_compression = 0;
+ const sstring name;
+ cfactory(sstring name_ = "LZ4") : name(std::move(name_)) {}
+ const sstring& supported() const override {
+ return name;
+ }
+ class mylz4 : public rpc::lz4_compressor {
+ sstring _name;
+ public:
+ mylz4(const sstring& n) : _name(n) {}
+ sstring name() const override {
+ return _name;
+ }
+ };
+ std::unique_ptr<rpc::compressor> negotiate(sstring feature, bool is_server) const override {
+ if (feature == name) {
+ use_compression++;
+ return std::make_unique<mylz4>(name);
+ } else {
+ return nullptr;
+ }
+ }
+};
+
+SEASTAR_TEST_CASE(test_rpc_connect) {
+ std::vector<future<>> fs;
+
+ for (auto i = 0; i < 2; i++) {
+ for (auto j = 0; j < 4; j++) {
+ auto factory = std::make_unique<cfactory>();
+ rpc::server_options so;
+ rpc::client_options co;
+ if (i == 1) {
+ so.compressor_factory = factory.get();
+ }
+ if (j & 1) {
+ co.compressor_factory = factory.get();
+ }
+ co.send_timeout_data = j & 2;
+ rpc_test_config cfg;
+ cfg.server_options = so;
+ auto f = rpc_test_env<>::do_with_thread(cfg, co, [] (rpc_test_env<>& env, test_rpc_proto::client& c1) {
+ env.register_handler(1, [](int a, int b) {
+ return make_ready_future<int>(a+b);
+ }).get();
+ auto sum = env.proto().make_client<int (int, int)>(1);
+ auto result = sum(c1, 2, 3).get0();
+ BOOST_REQUIRE_EQUAL(result, 2 + 3);
+ }).handle_exception([] (auto ep) {
+ BOOST_FAIL("No exception expected");
+ }).finally([factory = std::move(factory), i, j = j & 1] {
+ if (i == 1 && j == 1) {
+ BOOST_REQUIRE_EQUAL(factory->use_compression, 2);
+ } else {
+ BOOST_REQUIRE_EQUAL(factory->use_compression, 0);
+ }
+ });
+ fs.emplace_back(std::move(f));
+ }
+ }
+ return when_all(fs.begin(), fs.end()).discard_result();
+}
+
+SEASTAR_TEST_CASE(test_rpc_connect_multi_compression_algo) {
+ auto factory1 = std::make_unique<cfactory>();
+ auto factory2 = std::make_unique<cfactory>("LZ4NEW");
+ rpc::server_options so;
+ rpc::client_options co;
+ static rpc::multi_algo_compressor_factory server({factory1.get(), factory2.get()});
+ static rpc::multi_algo_compressor_factory client({factory2.get(), factory1.get()});
+ so.compressor_factory = &server;
+ co.compressor_factory = &client;
+ rpc_test_config cfg;
+ cfg.server_options = so;
+ return rpc_test_env<>::do_with_thread(cfg, co, [] (rpc_test_env<>& env, test_rpc_proto::client& c1) {
+ env.register_handler(1, [](int a, int b) {
+ return make_ready_future<int>(a+b);
+ }).get();
+ auto sum = env.proto().make_client<int (int, int)>(1);
+ auto result = sum(c1, 2, 3).get0();
+ BOOST_REQUIRE_EQUAL(result, 2 + 3);
+ }).finally([factory1 = std::move(factory1), factory2 = std::move(factory2)] {
+ BOOST_REQUIRE_EQUAL(factory1->use_compression, 0);
+ BOOST_REQUIRE_EQUAL(factory2->use_compression, 2);
+ });
+}
+
+SEASTAR_TEST_CASE(test_rpc_connect_abort) {
+ rpc_test_config cfg;
+ cfg.connect = false;
+ return rpc_test_env<>::do_with_thread(cfg, [] (rpc_test_env<>& env) {
+ test_rpc_proto::client c1(env.proto(), {}, env.make_socket(), ipv4_addr());
+ env.register_handler(1, []() { return make_ready_future<>(); }).get();
+ auto f = env.proto().make_client<void ()>(1);
+ c1.stop().get0();
+ try {
+ f(c1).get0();
+ BOOST_REQUIRE(false);
+ } catch (...) {}
+ });
+}
+
+SEASTAR_TEST_CASE(test_rpc_cancel) {
+ using namespace std::chrono_literals;
+ return rpc_test_env<>::do_with_thread(rpc_test_config(), [] (rpc_test_env<>& env, test_rpc_proto::client& c1) {
+ bool rpc_executed = false;
+ int good = 0;
+ promise<> handler_called;
+ future<> f_handler_called = handler_called.get_future();
+ env.register_handler(1, [&rpc_executed, &handler_called] {
+ handler_called.set_value(); rpc_executed = true; return sleep(1ms);
+ }).get();
+ auto call = env.proto().make_client<void ()>(1);
+ rpc::cancellable cancel;
+ auto f = call(c1, cancel);
+ // cancel send side
+ cancel.cancel();
+ try {
+ f.get();
+ } catch(rpc::canceled_error&) {
+ good += !rpc_executed;
+ };
+ f = call(c1, cancel);
+ // cancel wait side
+ f_handler_called.then([&cancel] {
+ cancel.cancel();
+ }).get();
+ try {
+ f.get();
+ } catch(rpc::canceled_error&) {
+ good += 10*rpc_executed;
+ };
+ BOOST_REQUIRE_EQUAL(good, 11);
+ });
+}
+
+SEASTAR_TEST_CASE(test_message_to_big) {
+ rpc_test_config cfg;
+ cfg.resource_limits = {0, 1, 100};
+ return rpc_test_env<>::do_with_thread(cfg, [] (rpc_test_env<>& env, test_rpc_proto::client& c) {
+ bool good = true;
+ env.register_handler(1, [&] (sstring payload) mutable {
+ good = false;
+ }).get();
+ auto call = env.proto().make_client<void (sstring)>(1);
+ try {
+ call(c, uninitialized_string(101)).get();
+ good = false;
+ } catch(std::runtime_error& err) {
+ } catch(...) {
+ good = false;
+ }
+ BOOST_REQUIRE_EQUAL(good, true);
+ });
+}
+
+struct stream_test_result {
+ bool client_source_closed = false;
+ bool server_source_closed = false;
+ bool sink_exception = false;
+ bool sink_close_exception = false;
+ bool source_done_exception = false;
+ bool server_done_exception = false;
+ bool client_stop_exception = false;
+ int server_sum = 0;
+};
+
+future<stream_test_result> stream_test_func(rpc_test_env<>& env, bool stop_client, bool expect_connection_error = false) {
+ return seastar::async([&env, stop_client, expect_connection_error] {
+ stream_test_result r;
+ test_rpc_proto::client c(env.proto(), {}, env.make_socket(), ipv4_addr());
+ future<> server_done = make_ready_future();
+ env.register_handler(1, [&](int i, rpc::source<int> source) {
+ BOOST_REQUIRE_EQUAL(i, 666);
+ auto sink = source.make_sink<serializer, sstring>();
+ auto sink_loop = seastar::async([sink] () mutable {
+ for (auto i = 0; i < 100; i++) {
+ sink("seastar").get();
+ sleep(std::chrono::milliseconds(1)).get();
+ }
+ }).finally([sink] () mutable {
+ return sink.flush();
+ }).finally([sink] () mutable {
+ return sink.close();
+ }).finally([sink] {});
+
+ auto source_loop = seastar::async([source, &r] () mutable {
+ while (!r.server_source_closed) {
+ auto data = source().get0();
+ if (data) {
+ r.server_sum += std::get<0>(*data);
+ } else {
+ r.server_source_closed = true;
+ try {
+ // check that reading after eos does not crash
+ // and throws correct exception
+ source().get();
+ } catch (rpc::stream_closed& ex) {
+ // expected
+ } catch (...) {
+ BOOST_FAIL("wrong exception on reading from a stream after eos");
+ }
+ }
+ }
+ });
+ server_done = when_all_succeed(std::move(sink_loop), std::move(source_loop)).discard_result();
+ return sink;
+ }).get();
+ auto call = env.proto().make_client<rpc::source<sstring> (int, rpc::sink<int>)>(1);
+ auto x = [&] {
+ try {
+ return c.make_stream_sink<serializer, int>(env.make_socket()).get0();
+ } catch (...) {
+ c.stop().get();
+ throw;
+ }
+ };
+ auto sink = x();
+ auto source = call(c, 666, sink).get0();
+ auto source_done = seastar::async([&] {
+ while (!r.client_source_closed) {
+ auto data = source().get0();
+ if (data) {
+ BOOST_REQUIRE_EQUAL(std::get<0>(*data), "seastar");
+ } else {
+ r.client_source_closed = true;
+ }
+ }
+ });
+ auto check_exception = [] (auto f) {
+ try {
+ f.get();
+ } catch (...) {
+ return true;
+ }
+ return false;
+ };
+ future<> stop_client_future = make_ready_future();
+ // With a connection error sink() will eventually fail, but we
+ // cannot guarantee when.
+ int max = expect_connection_error ? std::numeric_limits<int>::max() : 101;
+ for (int i = 1; i < max; i++) {
+ if (stop_client && i == 50) {
+ // stop client while stream is in use
+ stop_client_future = c.stop();
+ }
+ sleep(std::chrono::milliseconds(1)).get();
+ r.sink_exception = check_exception(sink(i));
+ if (r.sink_exception) {
+ break;
+ }
+ }
+ r.sink_close_exception = check_exception(sink.close());
+ r.source_done_exception = check_exception(std::move(source_done));
+ r.server_done_exception = check_exception(std::move(server_done));
+ r.client_stop_exception = check_exception(!stop_client ? c.stop() : std::move(stop_client_future));
+ return r;
+ });
+}
+
+SEASTAR_TEST_CASE(test_stream_simple) {
+ rpc::server_options so;
+ so.streaming_domain = rpc::streaming_domain_type(1);
+ rpc_test_config cfg;
+ cfg.server_options = so;
+ return rpc_test_env<>::do_with(cfg, [] (rpc_test_env<>& env) {
+ return stream_test_func(env, false).then([] (stream_test_result r) {
+ BOOST_REQUIRE(r.client_source_closed);
+ BOOST_REQUIRE(r.server_source_closed);
+ BOOST_REQUIRE(r.server_sum == 5050);
+ BOOST_REQUIRE(!r.sink_exception);
+ BOOST_REQUIRE(!r.sink_close_exception);
+ BOOST_REQUIRE(!r.source_done_exception);
+ BOOST_REQUIRE(!r.server_done_exception);
+ BOOST_REQUIRE(!r.client_stop_exception);
+ });
+ });
+}
+
+SEASTAR_TEST_CASE(test_stream_stop_client) {
+ rpc::server_options so;
+ so.streaming_domain = rpc::streaming_domain_type(1);
+ rpc_test_config cfg;
+ cfg.server_options = so;
+ return rpc_test_env<>::do_with(cfg, [] (rpc_test_env<>& env) {
+ return stream_test_func(env, true).then([] (stream_test_result r) {
+ BOOST_REQUIRE(!r.client_source_closed);
+ BOOST_REQUIRE(!r.server_source_closed);
+ BOOST_REQUIRE(r.sink_exception);
+ BOOST_REQUIRE(r.sink_close_exception);
+ BOOST_REQUIRE(r.source_done_exception);
+ BOOST_REQUIRE(r.server_done_exception);
+ BOOST_REQUIRE(!r.client_stop_exception);
+ });
+ });
+}
+
+
+SEASTAR_TEST_CASE(test_stream_connection_error) {
+ rpc::server_options so;
+ so.streaming_domain = rpc::streaming_domain_type(1);
+ rpc_test_config cfg;
+ cfg.server_options = so;
+ cfg.inject_error = true;
+ return rpc_test_env<>::do_with(cfg, [] (rpc_test_env<>& env) {
+ return stream_test_func(env, false, true).then([] (stream_test_result r) {
+ BOOST_REQUIRE(!r.client_source_closed);
+ BOOST_REQUIRE(!r.server_source_closed);
+ BOOST_REQUIRE(r.sink_exception);
+ BOOST_REQUIRE(r.sink_close_exception);
+ BOOST_REQUIRE(r.source_done_exception);
+ BOOST_REQUIRE(r.server_done_exception);
+ BOOST_REQUIRE(!r.client_stop_exception);
+ });
+ });
+}
+
+SEASTAR_TEST_CASE(test_rpc_scheduling) {
+ return rpc_test_env<>::do_with_thread(rpc_test_config(), [] (rpc_test_env<>& env, test_rpc_proto::client& c1) {
+ auto sg = create_scheduling_group("rpc", 100).get0();
+ env.register_handler(1, sg, [] () {
+ return make_ready_future<unsigned>(internal::scheduling_group_index(current_scheduling_group()));
+ }).get();
+ auto call_sg_id = env.proto().make_client<unsigned ()>(1);
+ auto id = call_sg_id(c1).get0();
+ BOOST_REQUIRE(id == internal::scheduling_group_index(sg));
+ });
+}
+
+SEASTAR_THREAD_TEST_CASE(test_rpc_scheduling_connection_based) {
+ auto sg1 = create_scheduling_group("sg1", 100).get0();
+ auto sg1_kill = defer([&] { destroy_scheduling_group(sg1).get(); });
+ auto sg2 = create_scheduling_group("sg2", 100).get0();
+ auto sg2_kill = defer([&] { destroy_scheduling_group(sg2).get(); });
+ rpc::resource_limits limits;
+ limits.isolate_connection = [sg1, sg2] (sstring cookie) {
+ auto sg = current_scheduling_group();
+ if (cookie == "sg1") {
+ sg = sg1;
+ } else if (cookie == "sg2") {
+ sg = sg2;
+ }
+ rpc::isolation_config cfg;
+ cfg.sched_group = sg;
+ return cfg;
+ };
+ rpc_test_config cfg;
+ cfg.resource_limits = limits;
+ rpc_test_env<>::do_with_thread(cfg, [sg1, sg2] (rpc_test_env<>& env) {
+ rpc::client_options co1;
+ co1.isolation_cookie = "sg1";
+ test_rpc_proto::client c1(env.proto(), co1, env.make_socket(), ipv4_addr());
+ rpc::client_options co2;
+ co2.isolation_cookie = "sg2";
+ test_rpc_proto::client c2(env.proto(), co2, env.make_socket(), ipv4_addr());
+ env.register_handler(1, [] {
+ return make_ready_future<unsigned>(internal::scheduling_group_index(current_scheduling_group()));
+ }).get();
+ auto call_sg_id = env.proto().make_client<unsigned ()>(1);
+ unsigned id;
+ id = call_sg_id(c1).get0();
+ BOOST_REQUIRE(id == internal::scheduling_group_index(sg1));
+ id = call_sg_id(c2).get0();
+ BOOST_REQUIRE(id == internal::scheduling_group_index(sg2));
+ c1.stop().get();
+ c2.stop().get();
+ }).get();
+}
+
+SEASTAR_THREAD_TEST_CASE(test_rpc_scheduling_connection_based_compatibility) {
+ auto sg1 = create_scheduling_group("sg1", 100).get0();
+ auto sg1_kill = defer([&] { destroy_scheduling_group(sg1).get(); });
+ auto sg2 = create_scheduling_group("sg2", 100).get0();
+ auto sg2_kill = defer([&] { destroy_scheduling_group(sg2).get(); });
+ rpc::resource_limits limits;
+ limits.isolate_connection = [sg1, sg2] (sstring cookie) {
+ auto sg = current_scheduling_group();
+ if (cookie == "sg1") {
+ sg = sg1;
+ } else if (cookie == "sg2") {
+ sg = sg2;
+ }
+ rpc::isolation_config cfg;
+ cfg.sched_group = sg;
+ return cfg;
+ };
+ rpc_test_config cfg;
+ cfg.resource_limits = limits;
+ rpc_test_env<>::do_with_thread(cfg, [sg1, sg2] (rpc_test_env<>& env) {
+ rpc::client_options co1;
+ co1.isolation_cookie = "sg1";
+ test_rpc_proto::client c1(env.proto(), co1, env.make_socket(), ipv4_addr());
+ rpc::client_options co2;
+ co2.isolation_cookie = "sg2";
+ test_rpc_proto::client c2(env.proto(), co2, env.make_socket(), ipv4_addr());
+ // An old client, that doesn't have an isolation cookie
+ rpc::client_options co3;
+ test_rpc_proto::client c3(env.proto(), co3, env.make_socket(), ipv4_addr());
+ // A server that uses sg1 if the client is old
+ env.register_handler(1, sg1, [] () {
+ return make_ready_future<unsigned>(internal::scheduling_group_index(current_scheduling_group()));
+ }).get();
+ auto call_sg_id = env.proto().make_client<unsigned ()>(1);
+ unsigned id;
+ id = call_sg_id(c1).get0();
+ BOOST_REQUIRE(id == internal::scheduling_group_index(sg1));
+ id = call_sg_id(c2).get0();
+ BOOST_REQUIRE(id == internal::scheduling_group_index(sg2));
+ id = call_sg_id(c3).get0();
+ BOOST_REQUIRE(id == internal::scheduling_group_index(sg1));
+ c1.stop().get();
+ c2.stop().get();
+ c3.stop().get();
+ }).get();
+}
+
+void test_compressor(std::function<std::unique_ptr<seastar::rpc::compressor>()> compressor_factory) {
+ using namespace seastar::rpc;
+
+ auto linearize = [&] (const auto& buffer) {
+ return seastar::visit(buffer.bufs,
+ [] (const temporary_buffer<char>& buf) {
+ return buf.clone();
+ },
+ [&] (const std::vector<temporary_buffer<char>>& bufs) {
+ auto buf = temporary_buffer<char>(buffer.size);
+ auto dst = buf.get_write();
+ for (auto& b : bufs) {
+ dst = std::copy_n(b.get(), b.size(), dst);
+ }
+ return buf;
+ }
+ );
+ };
+
+ auto split_buffer = [&] (temporary_buffer<char> b, size_t chunk_size) {
+ std::vector<temporary_buffer<char>> bufs;
+ auto src = b.get();
+ auto n = b.size();
+ while (n) {
+ auto this_chunk = std::min(chunk_size, n);
+ bufs.emplace_back(this_chunk);
+ std::copy_n(src, this_chunk, bufs.back().get_write());
+ src += this_chunk;
+ n -= this_chunk;
+ }
+ return bufs;
+ };
+
+ auto clone = [&] (const auto& buffer) {
+ auto c = std::decay_t<decltype(buffer)>();
+ c.size = buffer.size;
+ c.bufs = seastar::visit(buffer.bufs,
+ [] (const temporary_buffer<char>& buf) -> decltype(c.bufs) {
+ return buf.clone();
+ },
+ [] (const std::vector<temporary_buffer<char>>& bufs) -> decltype(c.bufs) {
+ std::vector<temporary_buffer<char>> c;
+ c.reserve(bufs.size());
+ for (auto& b : bufs) {
+ c.emplace_back(b.clone());
+ }
+ return c;
+ }
+ );
+ return c;
+ };
+
+ auto compressor = compressor_factory();
+
+ std::vector<std::tuple<sstring, size_t, snd_buf>> inputs;
+
+ auto& eng = testing::local_random_engine;
+ auto dist = std::uniform_int_distribution<char>();
+
+ auto snd = snd_buf(1);
+ *snd.front().get_write() = 'a';
+ inputs.emplace_back("one byte, no headroom", 0, std::move(snd));
+
+ snd = snd_buf(1);
+ *snd.front().get_write() = 'a';
+ inputs.emplace_back("one byte, 128k of headroom", 128 * 1024, std::move(snd));
+
+ auto gen_fill = [&](size_t s, sstring msg, std::optional<size_t> split = {}) {
+ auto buf = temporary_buffer<char>(s);
+ std::fill_n(buf.get_write(), s, 'a');
+
+ auto snd = snd_buf();
+ snd.size = s;
+ if (split) {
+ snd.bufs = split_buffer(buf.clone(), *split);
+ } else {
+ snd.bufs = buf.clone();
+ }
+ inputs.emplace_back(msg, 0, std::move(snd));
+ };
+
+ gen_fill(16 * 1024, "single 16 kB buffer of \'a\'");
+
+ auto gen_rand = [&](size_t s, sstring msg, std::optional<size_t> split = {}) {
+ auto buf = temporary_buffer<char>(s);
+ std::generate_n(buf.get_write(), s, [&] { return dist(eng); });
+
+ auto snd = snd_buf();
+ snd.size = s;
+ if (split) {
+ snd.bufs = split_buffer(buf.clone(), *split);
+ } else {
+ snd.bufs = buf.clone();
+ }
+ inputs.emplace_back(msg, 0, std::move(snd));
+ };
+
+ gen_rand(16 * 1024, "single 16 kB buffer of random");
+
+ auto gen_text = [&](size_t s, sstring msg, std::optional<size_t> split = {}) {
+ static const std::string_view text = "The quick brown fox wants bananas for his long term health but sneaks bacon behind his wife's back. ";
+
+ auto buf = temporary_buffer<char>(s);
+ size_t n = 0;
+ while (n < s) {
+ auto rem = std::min(s - n, text.size());
+ std::copy(text.data(), text.data() + rem, buf.get_write() + n);
+ n += rem;
+ }
+
+ auto snd = snd_buf();
+ snd.size = s;
+ if (split) {
+ snd.bufs = split_buffer(buf.clone(), *split);
+ } else {
+ snd.bufs = buf.clone();
+ }
+ inputs.emplace_back(msg, 0, std::move(snd));
+ };
+
+
+ for (auto s : { 1, 4, 8 }) {
+ for (auto ss : { 32, 64, 128, 48, 56, 246, 511 }) {
+ gen_fill(s * 1024 * 1024, format("{} MB buffer of \'a\' split into {} kB - {}", s, ss, ss), ss * 1024 - ss);
+ gen_fill(s * 1024 * 1024, format("{} MB buffer of \'a\' split into {} kB", s, ss), ss * 1024);
+ gen_rand(s * 1024 * 1024, format("{} MB buffer of random split into {} kB", s, ss), ss * 1024);
+
+ gen_fill(s * 1024 * 1024 + 1, format("{} MB + 1B buffer of \'a\' split into {} kB", s, ss), ss * 1024);
+ gen_rand(s * 1024 * 1024 + 1, format("{} MB + 1B buffer of random split into {} kB", s, ss), ss * 1024);
+ }
+
+ for (auto ss : { 128, 246, 511, 3567, 2*1024, 8*1024 }) {
+ gen_fill(s * 1024 * 1024, format("{} MB buffer of \'a\' split into {} B", s, ss), ss);
+ gen_rand(s * 1024 * 1024, format("{} MB buffer of random split into {} B", s, ss), ss);
+ gen_text(s * 1024 * 1024, format("{} MB buffer of text split into {} B", s, ss), ss);
+ gen_fill(s * 1024 * 1024 - ss, format("{} MB - {}B buffer of \'a\' split into {} B", s, ss, ss), ss);
+ gen_rand(s * 1024 * 1024 - ss, format("{} MB - {}B buffer of random split into {} B", s, ss, ss), ss);
+ gen_text(s * 1024 * 1024 - ss, format("{} MB - {}B buffer of random split into {} B", s, ss, ss), ss);
+ }
+ }
+
+ for (auto s : { 64*1024 + 5670, 16*1024 + 3421, 32*1024 - 321 }) {
+ gen_fill(s, format("{} bytes buffer of \'a\'", s));
+ gen_rand(s, format("{} bytes buffer of random", s));
+ gen_text(s, format("{} bytes buffer of text", s));
+ }
+
+ std::vector<std::tuple<sstring, std::function<rcv_buf(snd_buf)>>> transforms {
+ { "identity", [] (snd_buf snd) {
+ rcv_buf rcv;
+ rcv.size = snd.size;
+ rcv.bufs = std::move(snd.bufs);
+ return rcv;
+ } },
+ { "linearized", [&linearize] (snd_buf snd) {
+ rcv_buf rcv;
+ rcv.size = snd.size;
+ rcv.bufs = linearize(snd);
+ return rcv;
+ } },
+ { "split 1 B", [&] (snd_buf snd) {
+ rcv_buf rcv;
+ rcv.size = snd.size;
+ rcv.bufs = split_buffer(linearize(snd), 1);
+ return rcv;
+ } },
+ { "split 129 B", [&] (snd_buf snd) {
+ rcv_buf rcv;
+ rcv.size = snd.size;
+ rcv.bufs = split_buffer(linearize(snd), 129);
+ return rcv;
+ } },
+ { "split 4 kB", [&] (snd_buf snd) {
+ rcv_buf rcv;
+ rcv.size = snd.size;
+ rcv.bufs = split_buffer(linearize(snd), 4096);
+ return rcv;
+ } },
+ { "split 4 kB - 128", [&] (snd_buf snd) {
+ rcv_buf rcv;
+ rcv.size = snd.size;
+ rcv.bufs = split_buffer(linearize(snd), 4096 - 128);
+ return rcv;
+ } },
+ };
+
+ auto sanity_check = [&] (const auto& buffer) {
+ auto actual_size = seastar::visit(buffer.bufs,
+ [] (const temporary_buffer<char>& buf) {
+ return buf.size();
+ },
+ [] (const std::vector<temporary_buffer<char>>& bufs) {
+ return boost::accumulate(bufs, size_t(0), [] (size_t sz, const temporary_buffer<char>& buf) {
+ return sz + buf.size();
+ });
+ }
+ );
+ BOOST_CHECK_EQUAL(actual_size, buffer.size);
+ };
+
+ for (auto& in : inputs) {
+ BOOST_TEST_MESSAGE("Input: " << std::get<0>(in));
+ auto headroom = std::get<1>(in);
+ auto compressed = compressor->compress(headroom, clone(std::get<2>(in)));
+ sanity_check(compressed);
+
+ // Remove headroom
+ BOOST_CHECK_GE(compressed.size, headroom);
+ compressed.size -= headroom;
+ seastar::visit(compressed.bufs,
+ [&] (temporary_buffer<char>& buf) {
+ BOOST_CHECK_GE(buf.size(), headroom);
+ buf.trim_front(headroom);
+ },
+ [&] (std::vector<temporary_buffer<char>>& bufs) {
+ while (headroom) {
+ BOOST_CHECK(!bufs.empty());
+ auto to_remove = std::min(bufs.front().size(), headroom);
+ bufs.front().trim_front(to_remove);
+ if (bufs.front().empty() && bufs.size() > 1) {
+ bufs.erase(bufs.begin());
+ }
+ headroom -= to_remove;
+ }
+ }
+ );
+
+ auto in_l = linearize(std::get<2>(in));
+
+ for (auto& t : transforms) {
+ BOOST_TEST_MESSAGE(" Transform: " << std::get<0>(t));
+ auto received = std::get<1>(t)(clone(compressed));
+
+ auto decompressed = compressor->decompress(std::move(received));
+ sanity_check(decompressed);
+
+ BOOST_CHECK_EQUAL(decompressed.size, std::get<2>(in).size);
+
+ auto out_l = linearize(decompressed);
+
+ BOOST_CHECK_EQUAL(in_l.size(), out_l.size());
+ BOOST_CHECK(in_l == out_l);
+ }
+ }
+}
+
+SEASTAR_THREAD_TEST_CASE(test_lz4_compressor) {
+ test_compressor([] { return std::make_unique<rpc::lz4_compressor>(); });
+}
+
+SEASTAR_THREAD_TEST_CASE(test_lz4_fragmented_compressor) {
+ test_compressor([] { return std::make_unique<rpc::lz4_fragmented_compressor>(); });
+}
+
+// Test reproducing issue #671: If timeout is time_point::max(), translating
+// it to relative timeout in the sender and then back in the receiver, when
+// these calculations happen across a millisecond boundary, overflowed the
+// integer and mislead the receiver to think the requested timeout was
+// negative, and cause it drop its response, so the RPC call never completed.
+SEASTAR_TEST_CASE(test_max_absolute_timeout) {
+ // The typical failure of this test is a hang. So we use semaphore to
+ // stop the test either when it succeeds, or after a long enough hang.
+ auto success = make_lw_shared<bool>(false);
+ auto done = make_lw_shared<semaphore>(0);
+ auto abrt = make_lw_shared<abort_source>();
+ (void) seastar::sleep_abortable(std::chrono::seconds(3), *abrt).then([done, success] {
+ done->signal(1);
+ }).handle_exception([] (std::exception_ptr) {});
+ rpc::client_options co;
+ co.send_timeout_data = 1;
+ (void)rpc_test_env<>::do_with_thread(rpc_test_config(), co, [] (rpc_test_env<>& env, test_rpc_proto::client& c1) {
+ env.register_handler(1, [](int a, int b) {
+ return make_ready_future<int>(a+b);
+ }).get();
+ auto sum = env.proto().make_client<int (int, int)>(1);
+ // The bug only reproduces if the calculation done on the sender
+ // and receiver sides, happened across a millisecond boundary.
+ // We can't control when it happens, so we just need to loop many
+ // times, at least many milliseconds, to increase the probability
+ // that we catch the bug. Experimentally, if we loop for 200ms, we
+ // catch the bug in #671 virtually every time.
+ auto until = seastar::lowres_clock::now() + std::chrono::milliseconds(200);
+ while (seastar::lowres_clock::now() <= until) {
+ auto result = sum(c1, rpc::rpc_clock_type::time_point::max(), 2, 3).get0();
+ BOOST_REQUIRE_EQUAL(result, 2 + 3);
+ }
+ }).then([success, done, abrt] {
+ *success = true;
+ abrt->request_abort();
+ done->signal();
+ });
+ return done->wait().then([done, success] {
+ BOOST_REQUIRE(*success);
+ });
+}
+
+// Similar to the above test: Test that a relative timeout duration::max()
+// also works, and again doesn't cause the timeout wrapping around to the
+// past and causing dropped responses.
+SEASTAR_TEST_CASE(test_max_relative_timeout) {
+ // The typical failure of this test is a hang. So we use semaphore to
+ // stop the test either when it succeeds, or after a long enough hang.
+ auto success = make_lw_shared<bool>(false);
+ auto done = make_lw_shared<semaphore>(0);
+ auto abrt = make_lw_shared<abort_source>();
+ (void) seastar::sleep_abortable(std::chrono::seconds(3), *abrt).then([done, success] {
+ done->signal(1);
+ }).handle_exception([] (std::exception_ptr) {});
+ rpc::client_options co;
+ co.send_timeout_data = 1;
+ (void)rpc_test_env<>::do_with_thread(rpc_test_config(), co, [] (rpc_test_env<>& env, test_rpc_proto::client& c1) {
+ env.register_handler(1, [](int a, int b) {
+ return make_ready_future<int>(a+b);
+ }).get();
+ auto sum = env.proto().make_client<int (int, int)>(1);
+ // The following call used to always hang, when max()+now()
+ // overflowed and appeared to be a negative timeout.
+ auto result = sum(c1, rpc::rpc_clock_type::duration::max(), 2, 3).get0();
+ BOOST_REQUIRE_EQUAL(result, 2 + 3);
+ }).then([success, done, abrt] {
+ *success = true;
+ abrt->request_abort();
+ done->signal();
+ });
+ return done->wait().then([done, success] {
+ BOOST_REQUIRE(*success);
+ });
+}
+
+SEASTAR_TEST_CASE(test_rpc_tuple) {
+ return rpc_test_env<>::do_with_thread(rpc_test_config(), [] (rpc_test_env<>& env, test_rpc_proto::client& c1) {
+ env.register_handler(1, [] () {
+ return make_ready_future<rpc::tuple<int, long>>(rpc::tuple<int, long>(1, 0x7'0000'0000L));
+ }).get();
+ auto f1 = env.proto().make_client<rpc::tuple<int, long> ()>(1);
+ auto result = f1(c1).get0();
+ BOOST_REQUIRE_EQUAL(std::get<0>(result), 1);
+ BOOST_REQUIRE_EQUAL(std::get<1>(result), 0x7'0000'0000L);
+ });
+}
+
+SEASTAR_TEST_CASE(test_rpc_nonvariadic_client_variadic_server) {
+ return rpc_test_env<>::do_with_thread(rpc_test_config(), [] (rpc_test_env<>& env, test_rpc_proto::client& c1) {
+ // Server is variadic
+ env.register_handler(1, [] () {
+ return make_ready_future<rpc::tuple<int, long>>(rpc::tuple(1, 0x7'0000'0000L));
+ }).get();
+ // Client is non-variadic
+ auto f1 = env.proto().make_client<future<rpc::tuple<int, long>> ()>(1);
+ auto result = f1(c1).get0();
+ BOOST_REQUIRE_EQUAL(std::get<0>(result), 1);
+ BOOST_REQUIRE_EQUAL(std::get<1>(result), 0x7'0000'0000L);
+ });
+}
+
+SEASTAR_TEST_CASE(test_rpc_variadic_client_nonvariadic_server) {
+ return rpc_test_env<>::do_with_thread(rpc_test_config(), [] (rpc_test_env<>& env, test_rpc_proto::client& c1) {
+ // Server is nonvariadic
+ env.register_handler(1, [] () {
+ return make_ready_future<rpc::tuple<int, long>>(rpc::tuple<int, long>(1, 0x7'0000'0000L));
+ }).get();
+ // Client is variadic
+ auto f1 = env.proto().make_client<future<rpc::tuple<int, long>> ()>(1);
+ auto result = f1(c1).get0();
+ BOOST_REQUIRE_EQUAL(std::get<0>(result), 1);
+ BOOST_REQUIRE_EQUAL(std::get<1>(result), 0x7'0000'0000L);
+ });
+}
+
+SEASTAR_TEST_CASE(test_handler_registration) {
+ rpc_test_config cfg;
+ cfg.connect = false;
+ return rpc_test_env<>::do_with_thread(cfg, [] (rpc_test_env<>& env) {
+ auto& proto = env.proto();
+
+ // new proto must be empty
+ BOOST_REQUIRE(!proto.has_handlers());
+
+ // non-existing handler should not be found
+ BOOST_REQUIRE(!proto.has_handler(1));
+
+ // unregistered non-existing handler should return ready future
+ auto fut = proto.unregister_handler(1);
+ BOOST_REQUIRE(fut.available() && !fut.failed());
+ fut.get();
+
+ // existing handler should be found
+ auto handler = [] () { return make_ready_future<>(); };
+ proto.register_handler(1, handler);
+ BOOST_REQUIRE(proto.has_handler(1));
+
+ // cannot register handler if already registered
+ BOOST_REQUIRE_THROW(proto.register_handler(1, handler), std::runtime_error);
+
+ // unregistered handler should not be found
+ proto.unregister_handler(1).get();
+ BOOST_REQUIRE(!proto.has_handler(1));
+
+ // re-registering a handler should succeed
+ proto.register_handler(1, handler);
+ BOOST_REQUIRE(proto.has_handler(1));
+
+ // proto with handlers must not be empty
+ BOOST_REQUIRE(proto.has_handlers());
+ });
+}
+
+SEASTAR_TEST_CASE(test_unregister_handler) {
+ using namespace std::chrono_literals;
+ return rpc_test_env<>::do_with_thread(rpc_test_config(), [] (rpc_test_env<>& env, test_rpc_proto::client& c1) {
+ promise<> handler_called;
+ future<> f_handler_called = handler_called.get_future();
+ bool rpc_executed = false;
+ bool rpc_completed = false;
+
+ auto reset_state = [&f_handler_called, &rpc_executed, &rpc_completed] {
+ if (f_handler_called.available()) {
+ f_handler_called.get();
+ }
+ rpc_executed = false;
+ rpc_completed = false;
+ };
+
+ auto get_handler = [&handler_called, &rpc_executed, &rpc_completed] {
+ return [&handler_called, &rpc_executed, &rpc_completed] {
+ handler_called.set_value();
+ rpc_executed = true;
+ return sleep(1ms).then([&rpc_completed] {
+ rpc_completed = true;
+ });
+ };
+ };
+
+ // handler should not run if unregistered before being called
+ env.register_handler(1, get_handler()).get();
+ env.unregister_handler(1).get();
+ BOOST_REQUIRE(!f_handler_called.available());
+ BOOST_REQUIRE(!rpc_executed);
+ BOOST_REQUIRE(!rpc_completed);
+
+ // verify normal execution path
+ env.register_handler(1, get_handler()).get();
+ auto call = env.proto().make_client<void ()>(1);
+ call(c1).get();
+ BOOST_REQUIRE(f_handler_called.available());
+ BOOST_REQUIRE(rpc_executed);
+ BOOST_REQUIRE(rpc_completed);
+ reset_state();
+
+ // call should fail after handler is unregistered
+ env.unregister_handler(1).get();
+ try {
+ call(c1).get();
+ BOOST_REQUIRE(false);
+ } catch (rpc::unknown_verb_error&) {
+ // expected
+ } catch (...) {
+ std::cerr << "call failed in an unexpected way: " << std::current_exception() << std::endl;
+ BOOST_REQUIRE(false);
+ }
+ BOOST_REQUIRE(!f_handler_called.available());
+ BOOST_REQUIRE(!rpc_executed);
+ BOOST_REQUIRE(!rpc_completed);
+
+ // unregistered is allowed while call is in flight
+ auto delayed_unregister = [&env] {
+ return sleep(500us).then([&env] {
+ return env.unregister_handler(1);
+ });
+ };
+
+ env.register_handler(1, get_handler()).get();
+ try {
+ when_all_succeed(call(c1), delayed_unregister()).get();
+ reset_state();
+ } catch (rpc::unknown_verb_error&) {
+ // expected
+ } catch (...) {
+ std::cerr << "call failed in an unexpected way: " << std::current_exception() << std::endl;
+ BOOST_REQUIRE(false);
+ }
+ });
+}
+
+SEASTAR_TEST_CASE(test_loggers) {
+ static seastar::logger log("dummy");
+ log.set_level(log_level::debug);
+ return rpc_test_env<>::do_with_thread(rpc_test_config(), [] (rpc_test_env<>& env, test_rpc_proto::client& c1) {
+ socket_address dummy_addr;
+ auto& proto = env.proto();
+ auto& logger = proto.get_logger();
+ logger(dummy_addr, "Hello0");
+ logger(dummy_addr, log_level::debug, "Hello1");
+ proto.set_logger(&log);
+ logger(dummy_addr, "Hello2");
+ logger(dummy_addr, log_level::debug, "Hello3");
+ // We *want* to test the deprecated API, don't spam warnings about it.
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
+ proto.set_logger([] (const sstring& str) {
+ log.info("Test: {}", str);
+ });
+#pragma GCC diagnostic pop
+ logger(dummy_addr, "Hello4");
+ logger(dummy_addr, log_level::debug, "Hello5");
+ proto.set_logger(nullptr);
+ logger(dummy_addr, "Hello6");
+ logger(dummy_addr, log_level::debug, "Hello7");
+ });
+}
+
+static_assert(std::is_same_v<decltype(rpc::tuple(1U, 1L)), rpc::tuple<unsigned, long>>, "rpc::tuple deduction guid not working");
generated by cgit v1.2.3 (git 2.39.1) at 2024-09-12 18:03:03 +0000