feat grpc: grpc client subscribe on config update

commit_hash:67bc5c92974883cf587f2c449bbf4727e19560b1

Author: kpavlov00

.mapping.json

@@ -1927,6 +1927,8 @@
   "grpc/include/userver/ugrpc/client/impl/codegen_declarations.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/codegen_declarations.hpp",
   "grpc/include/userver/ugrpc/client/impl/codegen_definitions.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/codegen_definitions.hpp",
   "grpc/include/userver/ugrpc/client/impl/completion_queue_pool.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/completion_queue_pool.hpp",
+  "grpc/include/userver/ugrpc/client/impl/stub_any.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/stub_any.hpp",
+  "grpc/include/userver/ugrpc/client/impl/stub_pool.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/stub_pool.hpp",
   "grpc/include/userver/ugrpc/client/middlewares/baggage/component.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/middlewares/baggage/component.hpp",
   "grpc/include/userver/ugrpc/client/middlewares/base.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/middlewares/base.hpp",
   "grpc/include/userver/ugrpc/client/middlewares/deadline_propagation/component.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/middlewares/deadline_propagation/component.hpp",
@@ -2023,6 +2025,7 @@
   "grpc/src/ugrpc/client/impl/client_factory_config.hpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/client_factory_config.hpp",
   "grpc/src/ugrpc/client/impl/client_qos.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/client_qos.cpp",
   "grpc/src/ugrpc/client/impl/completion_queue_pool.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/completion_queue_pool.cpp",
+  "grpc/src/ugrpc/client/impl/stub_pool.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/stub_pool.cpp",
   "grpc/src/ugrpc/client/middlewares/baggage/component.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/middlewares/baggage/component.cpp",
   "grpc/src/ugrpc/client/middlewares/baggage/middleware.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/middlewares/baggage/middleware.cpp",
   "grpc/src/ugrpc/client/middlewares/baggage/middleware.hpp":"taxi/uservices/userver/grpc/src/ugrpc/client/middlewares/baggage/middleware.hpp",

grpc/functional_tests/basic_chaos/client.hpp

@@ -31,11 +31,13 @@ class GreeterClient final : public components::ComponentBase {
         // Tests dedicated-channel-count from SimpleClientComponent
         auto& client =
             context.FindComponent<ugrpc::client::SimpleClientComponent<Client>>("greeter-client-component").GetClient();
-        UASSERT(ugrpc::client::impl::GetClientData(client).GetDedicatedChannelCount(0) == 3);
-        UASSERT(ugrpc::client::impl::GetClientData(client).GetDedicatedChannelCount(1) == 0);
-        UASSERT(ugrpc::client::impl::GetClientData(client).GetDedicatedChannelCount(2) == 2);
-        UASSERT(ugrpc::client::impl::GetClientData(client).GetDedicatedChannelCount(3) == 0);
-        UASSERT(ugrpc::client::impl::GetClientData(client).GetDedicatedChannelCount(4) == 0);
+        const auto& data = ugrpc::client::impl::GetClientData(client);
+        const auto stub_state = data.GetStubState();
+        UASSERT(stub_state->dedicated_stubs[0].Size() == 3);
+        UASSERT(stub_state->dedicated_stubs[1].Size() == 0);
+        UASSERT(stub_state->dedicated_stubs[2].Size() == 2);
+        UASSERT(stub_state->dedicated_stubs[3].Size() == 0);
+        UASSERT(stub_state->dedicated_stubs[4].Size() == 0);
     }
 
     inline std::string SayHello(std::string name, bool is_small_timeout);

grpc/include/userver/ugrpc/client/impl/async_methods.hpp

@@ -74,6 +74,8 @@ class RpcData final {
     RpcData& operator=(RpcData&&) noexcept = delete;
     ~RpcData();
 
+    ClientData::StubHandle& GetStub() noexcept;
+
     const grpc::ClientContext& GetContext() const noexcept;
 
     grpc::ClientContext& GetContext() noexcept;
@@ -154,6 +156,7 @@ class RpcData final {
     };
 
 private:
+    ClientData::StubHandle stub_;
     std::unique_ptr<grpc::ClientContext> context_;
     std::string client_name_;
     ugrpc::impl::MaybeOwnedString call_name_;

grpc/include/userver/ugrpc/client/impl/call_params.hpp

@@ -23,6 +23,7 @@ struct CallParams {
     grpc::CompletionQueue& queue;
     dynamic_config::Snapshot config;
     ugrpc::impl::MaybeOwnedString call_name;
+    ClientData::StubHandle stub;
     std::unique_ptr<grpc::ClientContext> context;
     ugrpc::impl::MethodStatistics& statistics;
     const Middlewares& mws;

grpc/include/userver/ugrpc/client/impl/client_data.hpp

@@ -9,13 +9,16 @@
 
 #include <userver/dynamic_config/source.hpp>
 #include <userver/engine/task/task_processor_fwd.hpp>
+#include <userver/rcu/rcu.hpp>
 #include <userver/testsuite/grpc_control.hpp>
 #include <userver/utils/fixed_array.hpp>
 
 #include <userver/ugrpc/client/client_factory_settings.hpp>
 #include <userver/ugrpc/client/client_settings.hpp>
 #include <userver/ugrpc/client/fwd.hpp>
 #include <userver/ugrpc/client/impl/channel_factory.hpp>
+#include <userver/ugrpc/client/impl/stub_any.hpp>
+#include <userver/ugrpc/client/impl/stub_pool.hpp>
 #include <userver/ugrpc/client/middlewares/fwd.hpp>
 #include <userver/ugrpc/impl/static_service_metadata.hpp>
 #include <userver/ugrpc/impl/statistics.hpp>
@@ -36,7 +39,7 @@ struct ClientDependencies final {
     Middlewares mws;
     ugrpc::impl::CompletionQueuePoolBase& completion_queues;
     ugrpc::impl::StatisticsStorage& statistics_storage;
-    const dynamic_config::Source config_source;
+    dynamic_config::Source config_source;
     testsuite::GrpcControl& testsuite_grpc;
     const dynamic_config::Key<ClientQos>* qos{nullptr};
     const ClientFactorySettings& client_factory_settings;
@@ -51,8 +54,31 @@ struct GenericClientTag final {
 /// The internal state of generated gRPC clients
 class ClientData final {
 public:
-    template <typename Service>
-    using Stub = typename Service::Stub;
+    struct StubState {
+        StubPool stubs;
+        // method_id -> stub_pool
+        utils::FixedArray<StubPool> dedicated_stubs;
+    };
+
+    class StubHandle {
+    public:
+        StubHandle(rcu::ReadablePtr<StubState>&& state, StubAny& stub) : state_{std::move(state)}, stub_{stub} {}
+
+        StubHandle(StubHandle&&) noexcept = default;
+        StubHandle& operator=(StubHandle&&) = delete;
+
+        StubHandle(const StubHandle&) = delete;
+        StubHandle& operator=(const StubHandle&) = delete;
+
+        template <typename Stub>
+        Stub& Get() {
+            return StubCast<Stub>(stub_);
+        }
+
+    private:
+        rcu::ReadablePtr<StubState> state_;
+        StubAny& stub_;
+    };
 
     ClientData() = delete;
 
@@ -62,36 +88,42 @@ class ClientData final {
           metadata_(metadata),
           service_statistics_(&GetServiceStatistics()),
           channel_factory_(CreateChannelFactory(dependencies_)),
-          channels_(CreateChannels(channel_factory_, dependencies_.client_factory_settings.channel_count)),
-          stubs_(MakeStubs<Service>(channels_)),
-          dedicated_stubs_(
-              MakeDedicatedStubs<Service>(channel_factory_, *metadata_, dependencies_.dedicated_methods_config)
-          ) {}
+          stub_state_(std::make_unique<rcu::Variable<StubState>>()) {
+        if (dependencies_.qos) {
+            SubscribeOnConfigUpdate<Service>(*dependencies_.qos);
+        } else {
+            ConstructStubState<typename Service::Stub>();
+        }
+    }
 
     template <typename Service>
     ClientData(ClientDependencies&& dependencies, GenericClientTag, std::in_place_type_t<Service>)
         : dependencies_(std::move(dependencies)),
           channel_factory_(CreateChannelFactory(dependencies_)),
-          channels_(CreateChannels(channel_factory_, dependencies_.client_factory_settings.channel_count)),
-          stubs_(MakeStubs<Service>(channels_)) {}
+          stub_state_(std::make_unique<rcu::Variable<StubState>>()) {
+        ConstructStubState<typename Service::Stub>();
+    }
+
+    ~ClientData();
 
     ClientData(ClientData&&) noexcept = default;
     ClientData& operator=(ClientData&&) = delete;
 
     ClientData(const ClientData&) = delete;
     ClientData& operator=(const ClientData&) = delete;
 
-    template <typename Service>
-    Stub<Service>& NextStubFromMethodId(std::size_t method_id) const {
-        if (!dedicated_stubs_[method_id].empty()) {
-            return *static_cast<Stub<Service>*>(NextStubPtr(dedicated_stubs_[method_id]).get());
-        }
-        return NextGenericStub<Service>();
+    StubHandle NextStubFromMethodId(std::size_t method_id) const {
+        auto stub_state = stub_state_->Read();
+        auto& dedicated_stubs = stub_state->dedicated_stubs[method_id];
+        auto& stubs = dedicated_stubs.Size() ? dedicated_stubs : stub_state->stubs;
+        auto& stub = stubs.NextStub();
+        return StubHandle{std::move(stub_state), stub};
     }
 
-    template <typename Service>
-    Stub<Service>& NextGenericStub() const {
-        return *static_cast<Stub<Service>*>(NextStubPtr(stubs_).get());
+    StubHandle NextStub() const {
+        auto stub_state = stub_state_->Read();
+        auto& stub = stub_state->stubs.NextStub();
+        return StubHandle{std::move(stub_state), stub};
     }
 
     grpc::CompletionQueue& NextQueue() const;
@@ -102,8 +134,6 @@ class ClientData final {
 
     ugrpc::impl::MethodStatistics& GetGenericStatistics(std::string_view call_name) const;
 
-    const utils::FixedArray<std::shared_ptr<grpc::Channel>>& GetChannels() { return channels_; }
-
     std::string_view GetClientName() const { return dependencies_.client_name; }
 
     const Middlewares& GetMiddlewares() const { return dependencies_.mws; }
@@ -114,37 +144,12 @@ class ClientData final {
 
     const dynamic_config::Key<ClientQos>* GetClientQos() const;
 
-    std::size_t GetDedicatedChannelCount(std::size_t method_id) const;
+    rcu::ReadablePtr<StubState> GetStubState() const { return stub_state_->Read(); }
 
 private:
     static ChannelFactory CreateChannelFactory(const ClientDependencies& dependencies);
 
-    static utils::FixedArray<std::shared_ptr<grpc::Channel>>
-    CreateChannels(const ChannelFactory& channel_factory, std::size_t channel_count);
-
-    using StubDeleterType = void (*)(void*);
-    using StubPtr = std::unique_ptr<void, StubDeleterType>;
-
-    using StubPool = utils::FixedArray<StubPtr>;
-
-    template <typename Service>
-    static void StubDeleter(void* ptr) noexcept {
-        delete static_cast<Stub<Service>*>(ptr);
-    }
-
-    template <typename Service>
-    static StubPtr MakeStub(const std::shared_ptr<grpc::Channel>& channel) {
-        return StubPtr(Service::NewStub(channel).release(), &StubDeleter<Service>);
-    }
-
-    template <typename Service>
-    static StubPool MakeStubs(const utils::FixedArray<std::shared_ptr<grpc::Channel>>& channels) {
-        return utils::GenerateFixedArray(channels.size(), [&](std::size_t index) {
-            return MakeStub<Service>(channels[index]);
-        });
-    }
-
-    template <typename Service>
+    template <typename Stub>
     static utils::FixedArray<StubPool> MakeDedicatedStubs(
         const ChannelFactory& channel_factory,
         const ugrpc::impl::StaticServiceMetadata& metadata,
@@ -153,28 +158,45 @@ class ClientData final {
         return utils::GenerateFixedArray(GetMethodsCount(metadata), [&](std::size_t method_id) {
             const auto method_channel_count =
                 GetMethodChannelCount(dedicated_methods_config, GetMethodName(metadata, method_id));
-            return utils::GenerateFixedArray(method_channel_count, [&](std::size_t) {
-                const auto channel = channel_factory.CreateChannel();
-                return MakeStub<Service>(channel);
-            });
+            return StubPool::Create<Stub>(method_channel_count, channel_factory);
         });
     }
 
-    const StubPtr& NextStubPtr(const StubPool& stubs) const;
-
     ugrpc::impl::ServiceStatistics& GetServiceStatistics();
 
+    template <typename Service>
+    void SubscribeOnConfigUpdate(const dynamic_config::Key<ClientQos>& qos) {
+        config_subscription_ = dependencies_.config_source.UpdateAndListen(
+            this, dependencies_.client_name, &ClientData::OnConfigUpdate<Service>, qos
+        );
+    }
+
+    template <typename Service>
+    void OnConfigUpdate(const dynamic_config::Snapshot& /*config*/) {
+        ConstructStubState<typename Service::Stub>();
+    }
+
+    template <typename Stub>
+    void ConstructStubState() {
+        auto stubs = StubPool::Create<Stub>(dependencies_.client_factory_settings.channel_count, channel_factory_);
+
+        auto dedicated_stubs =
+            metadata_.has_value()
+                ? MakeDedicatedStubs<Stub>(channel_factory_, *metadata_, dependencies_.dedicated_methods_config)
+                : utils::FixedArray<StubPool>{};
+
+        stub_state_->Assign({std::move(stubs), std::move(dedicated_stubs)});
+    }
+
     ClientDependencies dependencies_;
     std::optional<ugrpc::impl::StaticServiceMetadata> metadata_{std::nullopt};
     ugrpc::impl::ServiceStatistics* service_statistics_{nullptr};
 
     ChannelFactory channel_factory_;
+    std::unique_ptr<rcu::Variable<StubState>> stub_state_;
 
-    utils::FixedArray<std::shared_ptr<grpc::Channel>> channels_;
-
-    StubPool stubs_;
-    // method_id -> stub_pool
-    utils::FixedArray<StubPool> dedicated_stubs_;
+    // These fields must be the last ones
+    concurrent::AsyncEventSubscriberScope config_subscription_;
 };
 
 template <typename Client>

grpc/include/userver/ugrpc/client/impl/stub_any.hpp

@@ -0,0 +1,29 @@
+#pragma once
+
+#include <grpcpp/channel.h>
+#include <grpcpp/generic/generic_stub.h>
+
+#include <userver/utils/any_movable.hpp>
+
+USERVER_NAMESPACE_BEGIN
+
+namespace ugrpc::client::impl {
+
+using StubAny = utils::AnyMovable;
+
+template <typename Stub>
+StubAny MakeStub(const std::shared_ptr<grpc::Channel>& channel) {
+    if constexpr (std::is_same_v<grpc::GenericStub, Stub>) {
+        return StubAny{std::in_place_type<grpc::GenericStub>, std::shared_ptr<grpc::Channel>{channel}};
+    }
+    return StubAny{std::in_place_type<Stub>, channel};
+}
+
+template <typename Stub>
+Stub& StubCast(StubAny& stub) {
+    return utils::AnyCast<Stub&>(stub);
+}
+
+}  // namespace ugrpc::client::impl
+
+USERVER_NAMESPACE_END

grpc/include/userver/ugrpc/client/impl/stub_pool.hpp

@@ -0,0 +1,48 @@
+#pragma once
+
+#include <grpcpp/channel.h>
+
+#include <userver/utils/fixed_array.hpp>
+
+#include <userver/ugrpc/client/impl/channel_factory.hpp>
+#include <userver/ugrpc/client/impl/stub_any.hpp>
+
+USERVER_NAMESPACE_BEGIN
+
+namespace ugrpc::client::impl {
+
+class StubPool final {
+public:
+    StubPool() = default;
+
+    std::size_t Size() const { return stubs_.size(); }
+
+    StubAny& NextStub() const;
+
+    const utils::FixedArray<std::shared_ptr<grpc::Channel>>& GetChannels() const { return channels_; }
+
+    const utils::FixedArray<StubAny>& GetStubs() const { return stubs_; }
+
+    template <typename Stub>
+    static StubPool Create(std::size_t size, const ChannelFactory& channel_factory) {
+        auto channels = utils::GenerateFixedArray(size, [&channel_factory](std::size_t) {
+            return channel_factory.CreateChannel();
+        });
+        auto stubs = utils::GenerateFixedArray(channels.size(), [&channels](std::size_t index) {
+            return MakeStub<Stub>(channels[index]);
+        });
+        return StubPool{std::move(channels), std::move(stubs)};
+    }
+
+private:
+    StubPool(utils::FixedArray<std::shared_ptr<grpc::Channel>>&& channels, utils::FixedArray<StubAny>&& stubs)
+        : channels_{std::move(channels)}, stubs_{std::move(stubs)} {}
+
+    utils::FixedArray<std::shared_ptr<grpc::Channel>> channels_;
+
+    mutable utils::FixedArray<StubAny> stubs_;
+};
+
+}  // namespace ugrpc::client::impl
+
+USERVER_NAMESPACE_END