feat grpc: retry on next channel

commit_hash:9f2c059fe510d157832d96c03adb160337abc83d

Author: kpavlov00

.mapping.json

@@ -2372,6 +2372,7 @@
   "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/fwd.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/fwd.hpp",
   "grpc/include/userver/ugrpc/client/impl/graceful_stream_finish.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/graceful_stream_finish.hpp",
+  "grpc/include/userver/ugrpc/client/impl/method_stubs.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/method_stubs.hpp",
   "grpc/include/userver/ugrpc/client/impl/middleware_hooks.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/middleware_hooks.hpp",
   "grpc/include/userver/ugrpc/client/impl/middleware_pipeline.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/middleware_pipeline.hpp",
   "grpc/include/userver/ugrpc/client/impl/perform_unary_call.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/perform_unary_call.hpp",
@@ -2381,8 +2382,6 @@
   "grpc/include/userver/ugrpc/client/impl/retry_policy.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/retry_policy.hpp",
   "grpc/include/userver/ugrpc/client/impl/rpc.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/rpc.hpp",
   "grpc/include/userver/ugrpc/client/impl/stream_read_future_impl_base.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/stream_read_future_impl_base.hpp",
-  "grpc/include/userver/ugrpc/client/impl/stub_handle.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/stub_handle.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/impl/stub_state.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/stub_state.hpp",
   "grpc/include/userver/ugrpc/client/impl/tracing.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/tracing.hpp",
   "grpc/include/userver/ugrpc/client/impl/unary_call.hpp":"taxi/uservices/userver/grpc/include/userver/ugrpc/client/impl/unary_call.hpp",
@@ -2520,12 +2519,11 @@
   "grpc/src/ugrpc/client/impl/compat/retry_policy.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/compat/retry_policy.cpp",
   "grpc/src/ugrpc/client/impl/compat/retry_policy.hpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/compat/retry_policy.hpp",
   "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/method_stubs.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/method_stubs.cpp",
   "grpc/src/ugrpc/client/impl/middleware_hooks.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/middleware_hooks.cpp",
   "grpc/src/ugrpc/client/impl/middleware_pipeline.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/middleware_pipeline.cpp",
   "grpc/src/ugrpc/client/impl/retry_backoff.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/retry_backoff.cpp",
   "grpc/src/ugrpc/client/impl/retry_policy.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/retry_policy.cpp",
-  "grpc/src/ugrpc/client/impl/stub_handle.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/stub_handle.cpp",
-  "grpc/src/ugrpc/client/impl/stub_pool.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/stub_pool.cpp",
   "grpc/src/ugrpc/client/impl/stub_state.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/stub_state.cpp",
   "grpc/src/ugrpc/client/impl/tracing.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/impl/tracing.cpp",
   "grpc/src/ugrpc/client/middlewares/baggage/middleware.cpp":"taxi/uservices/userver/grpc/src/ugrpc/client/middlewares/baggage/middleware.cpp",

grpc/functional_tests/basic_chaos/client.hpp

@@ -35,11 +35,11 @@ class GreeterClient final : public components::ComponentBase {
             context.FindComponent<ugrpc::client::SimpleClientComponent<Client>>("greeter-client-component").GetClient();
         const auto& data = ugrpc::client::impl::ClientDataAccessor::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);
+        UASSERT(stub_state->dedicated_stubs[0].stubs.size() == 3);
+        UASSERT(stub_state->dedicated_stubs[1].stubs.size() == 0);
+        UASSERT(stub_state->dedicated_stubs[2].stubs.size() == 2);
+        UASSERT(stub_state->dedicated_stubs[3].stubs.size() == 0);
+        UASSERT(stub_state->dedicated_stubs[4].stubs.size() == 0);
     }
 
     inline std::string SayHello(std::string name, bool is_small_timeout);

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

@@ -10,7 +10,7 @@
 #include <userver/ugrpc/client/call_options.hpp>
 #include <userver/ugrpc/client/fwd.hpp>
 #include <userver/ugrpc/client/generic_options.hpp>
-#include <userver/ugrpc/client/impl/stub_handle.hpp>
+#include <userver/ugrpc/client/impl/method_stubs.hpp>
 #include <userver/ugrpc/client/middlewares/fwd.hpp>
 #include <userver/ugrpc/impl/maybe_owned_string.hpp>
 #include <userver/ugrpc/impl/statistics.hpp>
@@ -32,7 +32,7 @@ struct CallParams {
     std::string_view service_name;
     std::string_view method_name;
     CallOptions call_options;
-    StubHandle stub;
+    MethodStubs method_stubs;
     const Middlewares& middlewares;
     ugrpc::impl::MethodStatistics& statistics;
     const testsuite::GrpcControl& testsuite_grpc;

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

@@ -12,8 +12,8 @@
 #include <userver/tracing/in_place_span.hpp>
 
 #include <userver/ugrpc/client/fwd.hpp>
+#include <userver/ugrpc/client/impl/method_stubs.hpp>
 #include <userver/ugrpc/client/impl/middleware_pipeline.hpp>
-#include <userver/ugrpc/client/impl/stub_handle.hpp>
 #include <userver/ugrpc/impl/async_method_invocation.hpp>
 #include <userver/ugrpc/impl/maybe_owned_string.hpp>
 #include <userver/ugrpc/impl/statistics_scope.hpp>
@@ -44,7 +44,7 @@ class CallState {
     CallState(CallState&&) noexcept = delete;
     CallState& operator=(CallState&&) noexcept = delete;
 
-    ugrpc::impl::StubAny& GetStub() noexcept;
+    ugrpc::impl::StubAny& GetStub() const noexcept;
 
     void SetClientContext(std::unique_ptr<grpc::ClientContext> client_context) noexcept;
 
@@ -83,7 +83,7 @@ class CallState {
     void ResetSpan() noexcept;
 
 private:
-    StubHandle stub_;
+    MethodStubs method_stubs_;
 
     std::unique_ptr<grpc::ClientContext> client_context_;
 

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

@@ -109,7 +109,7 @@ class ClientData final {
 
 private:
     template <typename Service>
-    static StubPool MakeStubs(
+    static StubArray MakeStubs(
         std::size_t size,
         const ChannelFactory& channel_factory,
         std::string_view target,
@@ -121,11 +121,11 @@ class ClientData final {
         auto stubs = utils::GenerateFixedArray(channels.size(), [&channels](std::size_t index) {
             return ugrpc::impl::AsyncService<Service>::NewStub(channels[index]);
         });
-        return StubPool{std::move(channels), std::move(stubs)};
+        return StubArray{std::move(channels), std::move(stubs)};
     }
 
     template <typename Service>
-    static utils::FixedArray<StubPool> MakeDedicatedStubs(
+    static utils::FixedArray<StubArray> MakeDedicatedStubs(
         const ugrpc::impl::StaticServiceMetadata& metadata,
         const DedicatedMethodsConfig& dedicated_methods_config,
         const ChannelFactory& channel_factory,
@@ -184,7 +184,7 @@ class ClientData final {
                       target,
                       channel_args
                   )
-                : utils::FixedArray<StubPool>{};
+                : utils::FixedArray<StubArray>{};
 
         stub_state_.Assign({client_qos, std::move(stubs), std::move(dedicated_stubs)});
     }

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

@@ -0,0 +1,36 @@
+#pragma once
+
+#include <optional>
+
+#include <userver/rcu/rcu.hpp>
+
+#include <userver/ugrpc/impl/stub_any.hpp>
+
+USERVER_NAMESPACE_BEGIN
+
+namespace ugrpc::client::impl {
+
+struct StubState;
+struct StubArray;
+
+class MethodStubs final {
+public:
+    MethodStubs(rcu::ReadablePtr<StubState>&& stub_state, const StubArray& stubs);
+
+    MethodStubs(MethodStubs&&) noexcept = default;
+    MethodStubs& operator=(MethodStubs&&) = delete;
+
+    MethodStubs(const MethodStubs&) = delete;
+    MethodStubs& operator=(const MethodStubs&) = delete;
+
+    ugrpc::impl::StubAny& GetStub() const;
+
+private:
+    const rcu::ReadablePtr<StubState> stub_state_;
+    const StubArray& stubs_;
+    mutable std::optional<std::size_t> last_;
+};
+
+}  // namespace ugrpc::client::impl
+
+USERVER_NAMESPACE_END

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

@@ -1,25 +1,32 @@
 #pragma once
 
+#include <grpcpp/channel.h>
+
 #include <userver/utils/fixed_array.hpp>
 
 #include <userver/ugrpc/client/client_qos.hpp>
-#include <userver/ugrpc/client/impl/stub_pool.hpp>
+#include <userver/ugrpc/impl/stub_any.hpp>
 
 USERVER_NAMESPACE_BEGIN
 
 namespace ugrpc::client::impl {
 
+struct StubArray {
+    utils::FixedArray<std::shared_ptr<grpc::Channel>> channels;
+    mutable utils::FixedArray<ugrpc::impl::StubAny> stubs;
+};
+
 struct StubState {
     ClientQos client_qos;
 
-    StubPool stubs;
+    StubArray stubs;
     // method_id -> stub_pool
-    utils::FixedArray<StubPool> dedicated_stubs;
+    utils::FixedArray<StubArray> dedicated_stubs;
 };
 
-ugrpc::impl::StubAny& NextStub(const StubState& stub_state, std::size_t method_id);
+const StubArray& GetMethodStubs(const StubState& stub_state, std::size_t method_id);
 
-ugrpc::impl::StubAny& NextGenericStub(const StubState& stub_state);
+const StubArray& GetGenericMethodStubs(const StubState& stub_state);
 
 }  // namespace ugrpc::client::impl
 

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

@@ -52,7 +52,7 @@ namespace {
     engine::TaskProcessor& blocking_task_processor
 ) {
     const auto stub_state = client_data.GetStubState();
-    const auto& channels = stub_state->stubs.GetChannels();
+    const auto& channels = stub_state->stubs.channels;
 
     auto& queue = client_data.NextQueue();