pull/36732/head
Craig Tiller
11 months ago
commit
1a66558de1
37 changed files with 5985 additions and 136 deletions
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,249 @@ |
||||
//
|
||||
// Copyright 2015 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// 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.
|
||||
//
|
||||
|
||||
#ifndef GRPC_SRC_CORE_CLIENT_CHANNEL_CLIENT_CHANNEL_H |
||||
#define GRPC_SRC_CORE_CLIENT_CHANNEL_CLIENT_CHANNEL_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "absl/status/status.h" |
||||
#include "absl/status/statusor.h" |
||||
#include "absl/strings/string_view.h" |
||||
|
||||
#include "src/core/client_channel/config_selector.h" |
||||
#include "src/core/client_channel/client_channel_factory.h" |
||||
#include "src/core/client_channel/subchannel.h" |
||||
#include "src/core/ext/filters/channel_idle/idle_filter_state.h" |
||||
#include "src/core/lib/gprpp/single_set_ptr.h" |
||||
#include "src/core/lib/promise/loop.h" |
||||
#include "src/core/lib/promise/observable.h" |
||||
#include "src/core/lib/surface/channel.h" |
||||
#include "src/core/lib/transport/call_filters.h" |
||||
#include "src/core/lib/transport/metadata.h" |
||||
#include "src/core/load_balancing/lb_policy.h" |
||||
#include "src/core/resolver/resolver.h" |
||||
#include "src/core/service_config/service_config.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
class ClientChannel : public Channel { |
||||
public: |
||||
static absl::StatusOr<OrphanablePtr<Channel>> Create( |
||||
std::string target, ChannelArgs channel_args, |
||||
grpc_channel_stack_type channel_stack_type); |
||||
|
||||
// Do not instantiate directly -- use Create() instead.
|
||||
ClientChannel(std::string target_uri, ChannelArgs args, |
||||
std::string uri_to_resolve, |
||||
RefCountedPtr<ServiceConfig> default_service_config, |
||||
ClientChannelFactory* client_channel_factory); |
||||
|
||||
~ClientChannel() override; |
||||
|
||||
void Orphan() override; |
||||
|
||||
grpc_call* CreateCall(grpc_call* parent_call, uint32_t propagation_mask, |
||||
grpc_completion_queue* cq, |
||||
grpc_pollset_set* /*pollset_set_alternative*/, |
||||
Slice path, absl::optional<Slice> authority, |
||||
Timestamp deadline, bool registered_method) override; |
||||
|
||||
grpc_event_engine::experimental::EventEngine* event_engine() |
||||
const override { |
||||
return event_engine_.get(); |
||||
} |
||||
|
||||
// FIXME: should we support lame channels somehow?
|
||||
bool IsLame() const override { return false; } |
||||
|
||||
bool SupportsConnectivityWatcher() const override { return true; } |
||||
|
||||
// Returns the current connectivity state. If try_to_connect is true,
|
||||
// triggers a connection attempt if not already connected.
|
||||
grpc_connectivity_state CheckConnectivityState(bool try_to_connect) override; |
||||
|
||||
void WatchConnectivityState( |
||||
grpc_connectivity_state last_observed_state, Timestamp deadline, |
||||
grpc_completion_queue* cq, void* tag) override; |
||||
|
||||
// Starts and stops a connectivity watch. The watcher will be initially
|
||||
// notified as soon as the state changes from initial_state and then on
|
||||
// every subsequent state change until either the watch is stopped or
|
||||
// it is notified that the state has changed to SHUTDOWN.
|
||||
//
|
||||
// This is intended to be used when starting watches from code inside of
|
||||
// C-core (e.g., for a nested control plane channel for things like xds).
|
||||
void AddConnectivityWatcher( |
||||
grpc_connectivity_state initial_state, |
||||
OrphanablePtr<AsyncConnectivityStateWatcherInterface> watcher) override; |
||||
void RemoveConnectivityWatcher( |
||||
AsyncConnectivityStateWatcherInterface* watcher) override; |
||||
|
||||
void GetInfo(const grpc_channel_info* channel_info) override; |
||||
|
||||
void ResetConnectionBackoff() override; |
||||
|
||||
void Ping(grpc_completion_queue* cq, void* tag) override; |
||||
|
||||
// Flag that this object gets stored in channel args as a raw pointer.
|
||||
struct RawPointerChannelArgTag {}; |
||||
static absl::string_view ChannelArgName() { |
||||
return "grpc.internal.client_channel"; |
||||
} |
||||
|
||||
private: |
||||
class ResolverResultHandler; |
||||
class ClientChannelControlHelper; |
||||
class SubchannelWrapper; |
||||
class LoadBalancedCallDestination; |
||||
|
||||
void CreateResolverLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(*work_serializer_); |
||||
void DestroyResolverAndLbPolicyLocked() |
||||
ABSL_EXCLUSIVE_LOCKS_REQUIRED(*work_serializer_); |
||||
|
||||
void TryToConnectLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(*work_serializer_); |
||||
|
||||
void OnResolverResultChangedLocked(Resolver::Result result) |
||||
ABSL_EXCLUSIVE_LOCKS_REQUIRED(*work_serializer_); |
||||
void OnResolverErrorLocked(absl::Status status) |
||||
ABSL_EXCLUSIVE_LOCKS_REQUIRED(*work_serializer_); |
||||
|
||||
absl::Status CreateOrUpdateLbPolicyLocked( |
||||
RefCountedPtr<LoadBalancingPolicy::Config> lb_policy_config, |
||||
const absl::optional<std::string>& health_check_service_name, |
||||
Resolver::Result result) ABSL_EXCLUSIVE_LOCKS_REQUIRED(*work_serializer_); |
||||
OrphanablePtr<LoadBalancingPolicy> CreateLbPolicyLocked( |
||||
const ChannelArgs& args) ABSL_EXCLUSIVE_LOCKS_REQUIRED(*work_serializer_); |
||||
|
||||
void UpdateServiceConfigInControlPlaneLocked( |
||||
RefCountedPtr<ServiceConfig> service_config, |
||||
RefCountedPtr<ConfigSelector> config_selector, std::string lb_policy_name) |
||||
ABSL_EXCLUSIVE_LOCKS_REQUIRED(*work_serializer_); |
||||
|
||||
void UpdateServiceConfigInDataPlaneLocked() |
||||
ABSL_EXCLUSIVE_LOCKS_REQUIRED(*work_serializer_); |
||||
|
||||
void UpdateStateLocked(grpc_connectivity_state state, |
||||
const absl::Status& status, const char* reason) |
||||
ABSL_EXCLUSIVE_LOCKS_REQUIRED(*work_serializer_); |
||||
|
||||
void UpdateStateAndPickerLocked( |
||||
grpc_connectivity_state state, const absl::Status& status, |
||||
const char* reason, |
||||
RefCountedPtr<LoadBalancingPolicy::SubchannelPicker> picker) |
||||
ABSL_EXCLUSIVE_LOCKS_REQUIRED(*work_serializer_); |
||||
|
||||
void StartIdleTimer(); |
||||
|
||||
CallInitiator CreateCall(ClientMetadataHandle client_initial_metadata, |
||||
Arena* arena); |
||||
|
||||
// Applies service config settings from config_selector to the call.
|
||||
// May modify call context and client_initial_metadata.
|
||||
absl::Status ApplyServiceConfigToCall( |
||||
ConfigSelector& config_selector, |
||||
ClientMetadataHandle& client_initial_metadata) const; |
||||
|
||||
// Does an LB pick for a call. Returns one of the following things:
|
||||
// - Continue{}, meaning to queue the pick
|
||||
// - a non-OK status, meaning to fail the call
|
||||
// - a connected subchannel, meaning that the pick is complete
|
||||
// When the pick is complete, pushes client_initial_metadata onto
|
||||
// call_initiator. Also adds the subchannel call tracker (if any) to
|
||||
// context.
|
||||
LoopCtl<absl::StatusOr<RefCountedPtr<ConnectedSubchannel>>> PickSubchannel( |
||||
LoadBalancingPolicy::SubchannelPicker& picker, |
||||
UnstartedCallHandler& unstarted_handler); |
||||
|
||||
//
|
||||
// Fields set at construction and never modified.
|
||||
//
|
||||
ChannelArgs channel_args_; |
||||
std::shared_ptr<grpc_event_engine::experimental::EventEngine> event_engine_; |
||||
std::string uri_to_resolve_; |
||||
const size_t service_config_parser_index_; |
||||
RefCountedPtr<ServiceConfig> default_service_config_; |
||||
ClientChannelFactory* client_channel_factory_; |
||||
std::string default_authority_; |
||||
channelz::ChannelNode* channelz_node_; |
||||
GlobalStatsPluginRegistry::StatsPluginGroup stats_plugin_group_; |
||||
grpc_pollset_set* interested_parties_; |
||||
|
||||
//
|
||||
// State for LB calls.
|
||||
//
|
||||
CallSizeEstimator lb_call_size_estimator_; |
||||
MemoryAllocator lb_call_allocator_; |
||||
|
||||
//
|
||||
// Idleness state.
|
||||
//
|
||||
const Duration idle_timeout_; |
||||
IdleFilterState idle_state_{false}; |
||||
SingleSetPtr<Activity, typename ActivityPtr::deleter_type> idle_activity_; |
||||
|
||||
//
|
||||
// Fields related to name resolution.
|
||||
//
|
||||
struct ResolverDataForCalls { |
||||
RefCountedPtr<ConfigSelector> config_selector; |
||||
RefCountedPtr<CallDestination> call_destination; |
||||
}; |
||||
Observable<absl::StatusOr<ResolverDataForCalls>> resolver_data_for_calls_; |
||||
|
||||
//
|
||||
// Fields related to LB picks.
|
||||
//
|
||||
Observable<RefCountedPtr<LoadBalancingPolicy::SubchannelPicker>> picker_; |
||||
|
||||
//
|
||||
// Fields used in the control plane. Guarded by work_serializer.
|
||||
//
|
||||
std::shared_ptr<WorkSerializer> work_serializer_; |
||||
ConnectivityStateTracker state_tracker_ ABSL_GUARDED_BY(*work_serializer_); |
||||
OrphanablePtr<Resolver> resolver_ ABSL_GUARDED_BY(*work_serializer_); |
||||
bool previous_resolution_contained_addresses_ |
||||
ABSL_GUARDED_BY(*work_serializer_) = false; |
||||
RefCountedPtr<ServiceConfig> saved_service_config_ |
||||
ABSL_GUARDED_BY(*work_serializer_); |
||||
RefCountedPtr<ConfigSelector> saved_config_selector_ |
||||
ABSL_GUARDED_BY(*work_serializer_); |
||||
OrphanablePtr<LoadBalancingPolicy> lb_policy_ |
||||
ABSL_GUARDED_BY(*work_serializer_); |
||||
RefCountedPtr<SubchannelPoolInterface> subchannel_pool_ |
||||
ABSL_GUARDED_BY(*work_serializer_); |
||||
// The number of SubchannelWrapper instances referencing a given Subchannel.
|
||||
std::map<Subchannel*, int> subchannel_refcount_map_ |
||||
ABSL_GUARDED_BY(*work_serializer_); |
||||
// The set of SubchannelWrappers that currently exist.
|
||||
// No need to hold a ref, since the set is updated in the control-plane
|
||||
// work_serializer when the SubchannelWrappers are created and destroyed.
|
||||
absl::flat_hash_set<SubchannelWrapper*> subchannel_wrappers_ |
||||
ABSL_GUARDED_BY(*work_serializer_); |
||||
int keepalive_time_ ABSL_GUARDED_BY(*work_serializer_) = -1; |
||||
absl::Status disconnect_error_ ABSL_GUARDED_BY(*work_serializer_); |
||||
|
||||
//
|
||||
// Fields accessed via GetChannelInfo().
|
||||
//
|
||||
Mutex info_mu_; |
||||
std::string info_lb_policy_name_ ABSL_GUARDED_BY(info_mu_); |
||||
std::string info_service_config_json_ ABSL_GUARDED_BY(info_mu_); |
||||
}; |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif // GRPC_SRC_CORE_CLIENT_CHANNEL_CLIENT_CHANNEL_H
|
||||
@ -0,0 +1,316 @@ |
||||
// Copyright 2022 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// 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.
|
||||
|
||||
// TODO(ctiller): Add a unit test suite for these filters once it's practical to
|
||||
// mock transport operations.
|
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/ext/filters/channel_idle/channel_idle_filter.h" |
||||
|
||||
#include <functional> |
||||
#include <utility> |
||||
|
||||
#include "absl/base/thread_annotations.h" |
||||
#include "absl/meta/type_traits.h" |
||||
#include "absl/random/random.h" |
||||
#include "absl/types/optional.h" |
||||
|
||||
#include <grpc/impl/channel_arg_names.h> |
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/core/lib/channel/channel_args.h" |
||||
#include "src/core/lib/channel/promise_based_filter.h" |
||||
#include "src/core/lib/config/core_configuration.h" |
||||
#include "src/core/lib/debug/trace.h" |
||||
#include "src/core/lib/experiments/experiments.h" |
||||
#include "src/core/lib/gprpp/debug_location.h" |
||||
#include "src/core/lib/gprpp/no_destruct.h" |
||||
#include "src/core/lib/gprpp/orphanable.h" |
||||
#include "src/core/lib/gprpp/per_cpu.h" |
||||
#include "src/core/lib/gprpp/status_helper.h" |
||||
#include "src/core/lib/gprpp/sync.h" |
||||
#include "src/core/lib/iomgr/closure.h" |
||||
#include "src/core/lib/iomgr/error.h" |
||||
#include "src/core/lib/iomgr/exec_ctx.h" |
||||
#include "src/core/lib/promise/exec_ctx_wakeup_scheduler.h" |
||||
#include "src/core/lib/promise/loop.h" |
||||
#include "src/core/lib/promise/poll.h" |
||||
#include "src/core/lib/promise/promise.h" |
||||
#include "src/core/lib/promise/sleep.h" |
||||
#include "src/core/lib/promise/try_seq.h" |
||||
#include "src/core/lib/surface/channel_stack_type.h" |
||||
#include "src/core/lib/transport/http2_errors.h" |
||||
#include "src/core/lib/transport/metadata_batch.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
const NoInterceptor ChannelIdleFilter::Call::OnClientInitialMetadata; |
||||
const NoInterceptor ChannelIdleFilter::Call::OnServerInitialMetadata; |
||||
const NoInterceptor ChannelIdleFilter::Call::OnServerTrailingMetadata; |
||||
const NoInterceptor ChannelIdleFilter::Call::OnClientToServerMessage; |
||||
const NoInterceptor ChannelIdleFilter::Call::OnServerToClientMessage; |
||||
|
||||
namespace { |
||||
|
||||
// TODO(roth): This can go back to being a constant when the experiment
|
||||
// is removed.
|
||||
Duration DefaultIdleTimeout() { |
||||
if (IsClientIdlenessEnabled()) return Duration::Minutes(30); |
||||
return Duration::Infinity(); |
||||
} |
||||
|
||||
// If these settings change, make sure that we are not sending a GOAWAY for
|
||||
// inproc transport, since a GOAWAY to inproc ends up destroying the transport.
|
||||
const auto kDefaultMaxConnectionAge = Duration::Infinity(); |
||||
const auto kDefaultMaxConnectionAgeGrace = Duration::Infinity(); |
||||
const auto kDefaultMaxConnectionIdle = Duration::Infinity(); |
||||
const auto kMaxConnectionAgeJitter = 0.1; |
||||
|
||||
TraceFlag grpc_trace_client_idle_filter(false, "client_idle_filter"); |
||||
} // namespace
|
||||
|
||||
#define GRPC_IDLE_FILTER_LOG(format, ...) \ |
||||
do { \
|
||||
if (GRPC_TRACE_FLAG_ENABLED(grpc_trace_client_idle_filter)) { \
|
||||
gpr_log(GPR_INFO, "(client idle filter) " format, ##__VA_ARGS__); \
|
||||
} \
|
||||
} while (0) |
||||
|
||||
Duration GetClientIdleTimeout(const ChannelArgs& args) { |
||||
return args.GetDurationFromIntMillis(GRPC_ARG_CLIENT_IDLE_TIMEOUT_MS) |
||||
.value_or(DefaultIdleTimeout()); |
||||
} |
||||
|
||||
struct MaxAgeFilter::Config { |
||||
Duration max_connection_age; |
||||
Duration max_connection_idle; |
||||
Duration max_connection_age_grace; |
||||
|
||||
bool enable() const { |
||||
return max_connection_age != Duration::Infinity() || |
||||
max_connection_idle != Duration::Infinity(); |
||||
} |
||||
|
||||
// A random jitter of +/-10% will be added to MAX_CONNECTION_AGE and
|
||||
// MAX_CONNECTION_IDLE to spread out reconnection storms.
|
||||
static Config FromChannelArgs(const ChannelArgs& args) { |
||||
const Duration args_max_age = |
||||
args.GetDurationFromIntMillis(GRPC_ARG_MAX_CONNECTION_AGE_MS) |
||||
.value_or(kDefaultMaxConnectionAge); |
||||
const Duration args_max_idle = |
||||
args.GetDurationFromIntMillis(GRPC_ARG_MAX_CONNECTION_IDLE_MS) |
||||
.value_or(kDefaultMaxConnectionIdle); |
||||
const Duration args_max_age_grace = |
||||
args.GetDurationFromIntMillis(GRPC_ARG_MAX_CONNECTION_AGE_GRACE_MS) |
||||
.value_or(kDefaultMaxConnectionAgeGrace); |
||||
// generate a random number between 1 - kMaxConnectionAgeJitter and
|
||||
// 1 + kMaxConnectionAgeJitter
|
||||
struct BitGen { |
||||
Mutex mu; |
||||
absl::BitGen bit_gen ABSL_GUARDED_BY(mu); |
||||
double MakeUniformDouble(double min, double max) { |
||||
MutexLock lock(&mu); |
||||
return absl::Uniform(bit_gen, min, max); |
||||
} |
||||
}; |
||||
static NoDestruct<PerCpu<BitGen>> bit_gen(PerCpuOptions().SetMaxShards(8)); |
||||
const double multiplier = bit_gen->this_cpu().MakeUniformDouble( |
||||
1.0 - kMaxConnectionAgeJitter, 1.0 + kMaxConnectionAgeJitter); |
||||
// GRPC_MILLIS_INF_FUTURE - 0.5 converts the value to float, so that result
|
||||
// will not be cast to int implicitly before the comparison.
|
||||
return Config{args_max_age * multiplier, args_max_idle * multiplier, |
||||
args_max_age_grace}; |
||||
} |
||||
}; |
||||
|
||||
absl::StatusOr<ClientIdleFilter> ClientIdleFilter::Create( |
||||
const ChannelArgs& args, ChannelFilter::Args filter_args) { |
||||
ClientIdleFilter filter(filter_args.channel_stack(), |
||||
GetClientIdleTimeout(args)); |
||||
return absl::StatusOr<ClientIdleFilter>(std::move(filter)); |
||||
} |
||||
|
||||
absl::StatusOr<MaxAgeFilter> MaxAgeFilter::Create( |
||||
const ChannelArgs& args, ChannelFilter::Args filter_args) { |
||||
MaxAgeFilter filter(filter_args.channel_stack(), |
||||
Config::FromChannelArgs(args)); |
||||
return absl::StatusOr<MaxAgeFilter>(std::move(filter)); |
||||
} |
||||
|
||||
void MaxAgeFilter::Shutdown() { |
||||
max_age_activity_.Reset(); |
||||
ChannelIdleFilter::Shutdown(); |
||||
} |
||||
|
||||
void MaxAgeFilter::PostInit() { |
||||
struct StartupClosure { |
||||
RefCountedPtr<grpc_channel_stack> channel_stack; |
||||
MaxAgeFilter* filter; |
||||
grpc_closure closure; |
||||
}; |
||||
auto run_startup = [](void* p, grpc_error_handle) { |
||||
auto* startup = static_cast<StartupClosure*>(p); |
||||
// Trigger idle timer
|
||||
startup->filter->IncreaseCallCount(); |
||||
startup->filter->DecreaseCallCount(); |
||||
grpc_transport_op* op = grpc_make_transport_op(nullptr); |
||||
op->start_connectivity_watch.reset( |
||||
new ConnectivityWatcher(startup->filter)); |
||||
op->start_connectivity_watch_state = GRPC_CHANNEL_IDLE; |
||||
grpc_channel_next_op( |
||||
grpc_channel_stack_element(startup->channel_stack.get(), 0), op); |
||||
delete startup; |
||||
}; |
||||
auto* startup = |
||||
new StartupClosure{this->channel_stack()->Ref(), this, grpc_closure{}}; |
||||
GRPC_CLOSURE_INIT(&startup->closure, run_startup, startup, nullptr); |
||||
ExecCtx::Run(DEBUG_LOCATION, &startup->closure, absl::OkStatus()); |
||||
|
||||
auto channel_stack = this->channel_stack()->Ref(); |
||||
|
||||
// Start the max age timer
|
||||
if (max_connection_age_ != Duration::Infinity()) { |
||||
max_age_activity_.Set(MakeActivity( |
||||
TrySeq( |
||||
// First sleep until the max connection age
|
||||
Sleep(Timestamp::Now() + max_connection_age_), |
||||
// Then send a goaway.
|
||||
[this] { |
||||
GRPC_CHANNEL_STACK_REF(this->channel_stack(), |
||||
"max_age send_goaway"); |
||||
// Jump out of the activity to send the goaway.
|
||||
auto fn = [](void* arg, grpc_error_handle) { |
||||
auto* channel_stack = static_cast<grpc_channel_stack*>(arg); |
||||
grpc_transport_op* op = grpc_make_transport_op(nullptr); |
||||
op->goaway_error = grpc_error_set_int( |
||||
GRPC_ERROR_CREATE("max_age"), |
||||
StatusIntProperty::kHttp2Error, GRPC_HTTP2_NO_ERROR); |
||||
grpc_channel_element* elem = |
||||
grpc_channel_stack_element(channel_stack, 0); |
||||
elem->filter->start_transport_op(elem, op); |
||||
GRPC_CHANNEL_STACK_UNREF(channel_stack, "max_age send_goaway"); |
||||
}; |
||||
ExecCtx::Run( |
||||
DEBUG_LOCATION, |
||||
GRPC_CLOSURE_CREATE(fn, this->channel_stack(), nullptr), |
||||
absl::OkStatus()); |
||||
return Immediate(absl::OkStatus()); |
||||
}, |
||||
// Sleep for the grace period
|
||||
[this] { |
||||
return Sleep(Timestamp::Now() + max_connection_age_grace_); |
||||
}), |
||||
ExecCtxWakeupScheduler(), |
||||
[channel_stack, this](absl::Status status) { |
||||
// OnDone -- close the connection if the promise completed
|
||||
// successfully.
|
||||
// (if it did not, it was cancelled)
|
||||
if (status.ok()) CloseChannel(); |
||||
}, |
||||
channel_stack->EventEngine())); |
||||
} |
||||
} |
||||
|
||||
bool ChannelIdleFilter::StartTransportOp(grpc_transport_op* op) { |
||||
// Catch the disconnect_with_error transport op.
|
||||
if (!op->disconnect_with_error.ok()) Shutdown(); |
||||
// Pass the op to the next filter.
|
||||
return false; |
||||
} |
||||
|
||||
void ChannelIdleFilter::Shutdown() { |
||||
// IncreaseCallCount() introduces a phony call and prevent the timer from
|
||||
// being reset by other threads.
|
||||
IncreaseCallCount(); |
||||
activity_.Reset(); |
||||
} |
||||
|
||||
void ChannelIdleFilter::IncreaseCallCount() { |
||||
idle_filter_state_->IncreaseCallCount(); |
||||
} |
||||
|
||||
void ChannelIdleFilter::DecreaseCallCount() { |
||||
if (idle_filter_state_->DecreaseCallCount()) { |
||||
// If there are no more calls in progress, start the idle timer.
|
||||
StartIdleTimer(); |
||||
} |
||||
} |
||||
|
||||
void ChannelIdleFilter::StartIdleTimer() { |
||||
GRPC_IDLE_FILTER_LOG("timer has started"); |
||||
auto idle_filter_state = idle_filter_state_; |
||||
// Hold a ref to the channel stack for the timer callback.
|
||||
auto channel_stack = channel_stack_->Ref(); |
||||
auto timeout = client_idle_timeout_; |
||||
auto promise = Loop([timeout, idle_filter_state]() { |
||||
return TrySeq(Sleep(Timestamp::Now() + timeout), |
||||
[idle_filter_state]() -> Poll<LoopCtl<absl::Status>> { |
||||
if (idle_filter_state->CheckTimer()) { |
||||
return Continue{}; |
||||
} else { |
||||
return absl::OkStatus(); |
||||
} |
||||
}); |
||||
}); |
||||
activity_.Set(MakeActivity( |
||||
std::move(promise), ExecCtxWakeupScheduler{}, |
||||
[channel_stack, this](absl::Status status) { |
||||
if (status.ok()) CloseChannel(); |
||||
}, |
||||
channel_stack->EventEngine())); |
||||
} |
||||
|
||||
void ChannelIdleFilter::CloseChannel() { |
||||
auto* op = grpc_make_transport_op(nullptr); |
||||
op->disconnect_with_error = grpc_error_set_int( |
||||
GRPC_ERROR_CREATE("enter idle"), |
||||
StatusIntProperty::ChannelConnectivityState, GRPC_CHANNEL_IDLE); |
||||
// Pass the transport op down to the channel stack.
|
||||
auto* elem = grpc_channel_stack_element(channel_stack_, 0); |
||||
elem->filter->start_transport_op(elem, op); |
||||
} |
||||
|
||||
const grpc_channel_filter ClientIdleFilter::kFilter = |
||||
MakePromiseBasedFilter<ClientIdleFilter, FilterEndpoint::kClient>( |
||||
"client_idle"); |
||||
const grpc_channel_filter MaxAgeFilter::kFilter = |
||||
MakePromiseBasedFilter<MaxAgeFilter, FilterEndpoint::kServer>("max_age"); |
||||
|
||||
void RegisterChannelIdleFilters(CoreConfiguration::Builder* builder) { |
||||
GPR_ASSERT(MaxAgeFilter::kFilter.init_call != nullptr); |
||||
if (!IsV3ChannelIdleFiltersEnabled()) return; |
||||
if (!IsCallV3Enabled()) { |
||||
builder->channel_init() |
||||
->RegisterFilter<ClientIdleFilter>(GRPC_CLIENT_CHANNEL) |
||||
.ExcludeFromMinimalStack() |
||||
.If([](const ChannelArgs& channel_args) { |
||||
return GetClientIdleTimeout(channel_args) != Duration::Infinity(); |
||||
}); |
||||
} |
||||
builder->channel_init() |
||||
->RegisterFilter<MaxAgeFilter>(GRPC_SERVER_CHANNEL) |
||||
.ExcludeFromMinimalStack() |
||||
.If([](const ChannelArgs& channel_args) { |
||||
return MaxAgeFilter::Config::FromChannelArgs(channel_args).enable(); |
||||
}); |
||||
} |
||||
|
||||
MaxAgeFilter::MaxAgeFilter(grpc_channel_stack* channel_stack, |
||||
const Config& max_age_config) |
||||
: ChannelIdleFilter(channel_stack, max_age_config.max_connection_idle), |
||||
max_connection_age_(max_age_config.max_connection_age), |
||||
max_connection_age_grace_(max_age_config.max_connection_age_grace) {} |
||||
|
||||
} // namespace grpc_core
|
||||
@ -0,0 +1,156 @@ |
||||
// Copyright 2022 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// 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.
|
||||
|
||||
#ifndef GRPC_SRC_CORE_EXT_FILTERS_CHANNEL_IDLE_CHANNEL_IDLE_FILTER_H |
||||
#define GRPC_SRC_CORE_EXT_FILTERS_CHANNEL_IDLE_CHANNEL_IDLE_FILTER_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <memory> |
||||
|
||||
#include "absl/status/status.h" |
||||
#include "absl/status/statusor.h" |
||||
|
||||
#include <grpc/impl/connectivity_state.h> |
||||
|
||||
#include "src/core/ext/filters/channel_idle/idle_filter_state.h" |
||||
#include "src/core/lib/channel/channel_args.h" |
||||
#include "src/core/lib/channel/channel_fwd.h" |
||||
#include "src/core/lib/channel/channel_stack.h" |
||||
#include "src/core/lib/channel/promise_based_filter.h" |
||||
#include "src/core/lib/gprpp/orphanable.h" |
||||
#include "src/core/lib/gprpp/ref_counted_ptr.h" |
||||
#include "src/core/lib/gprpp/single_set_ptr.h" |
||||
#include "src/core/lib/gprpp/time.h" |
||||
#include "src/core/lib/promise/activity.h" |
||||
#include "src/core/lib/promise/arena_promise.h" |
||||
#include "src/core/lib/transport/connectivity_state.h" |
||||
#include "src/core/lib/transport/transport.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
Duration GetClientIdleTimeout(const ChannelArgs& args); |
||||
|
||||
class ChannelIdleFilter : public ImplementChannelFilter<ChannelIdleFilter> { |
||||
public: |
||||
~ChannelIdleFilter() override = default; |
||||
|
||||
ChannelIdleFilter(const ChannelIdleFilter&) = delete; |
||||
ChannelIdleFilter& operator=(const ChannelIdleFilter&) = delete; |
||||
ChannelIdleFilter(ChannelIdleFilter&&) = default; |
||||
ChannelIdleFilter& operator=(ChannelIdleFilter&&) = default; |
||||
|
||||
class Call { |
||||
public: |
||||
explicit Call(ChannelIdleFilter* filter) : filter_(filter) { |
||||
filter_->IncreaseCallCount(); |
||||
} |
||||
~Call() { MaybeDecrement(); } |
||||
|
||||
static const NoInterceptor OnClientInitialMetadata; |
||||
static const NoInterceptor OnServerInitialMetadata; |
||||
static const NoInterceptor OnServerTrailingMetadata; |
||||
static const NoInterceptor OnClientToServerMessage; |
||||
static const NoInterceptor OnServerToClientMessage; |
||||
void OnFinalize(const grpc_call_final_info*) { MaybeDecrement(); } |
||||
|
||||
private: |
||||
void MaybeDecrement() { |
||||
auto* filter = std::exchange(filter_, nullptr); |
||||
if (filter != nullptr) filter->DecreaseCallCount(); |
||||
} |
||||
ChannelIdleFilter* filter_; |
||||
}; |
||||
|
||||
bool StartTransportOp(grpc_transport_op* op) override; |
||||
|
||||
protected: |
||||
using SingleSetActivityPtr = |
||||
SingleSetPtr<Activity, typename ActivityPtr::deleter_type>; |
||||
|
||||
ChannelIdleFilter(grpc_channel_stack* channel_stack, |
||||
Duration client_idle_timeout) |
||||
: channel_stack_(channel_stack), |
||||
client_idle_timeout_(client_idle_timeout) {} |
||||
|
||||
grpc_channel_stack* channel_stack() { return channel_stack_; }; |
||||
|
||||
virtual void Shutdown(); |
||||
void CloseChannel(); |
||||
|
||||
void IncreaseCallCount(); |
||||
void DecreaseCallCount(); |
||||
|
||||
private: |
||||
void StartIdleTimer(); |
||||
|
||||
// The channel stack to which we take refs for pending callbacks.
|
||||
grpc_channel_stack* channel_stack_; |
||||
Duration client_idle_timeout_; |
||||
std::shared_ptr<IdleFilterState> idle_filter_state_{ |
||||
std::make_shared<IdleFilterState>(false)}; |
||||
|
||||
SingleSetActivityPtr activity_; |
||||
}; |
||||
|
||||
class ClientIdleFilter final : public ChannelIdleFilter { |
||||
public: |
||||
static const grpc_channel_filter kFilter; |
||||
|
||||
static absl::StatusOr<ClientIdleFilter> Create( |
||||
const ChannelArgs& args, ChannelFilter::Args filter_args); |
||||
|
||||
private: |
||||
using ChannelIdleFilter::ChannelIdleFilter; |
||||
}; |
||||
|
||||
class MaxAgeFilter final : public ChannelIdleFilter { |
||||
public: |
||||
static const grpc_channel_filter kFilter; |
||||
struct Config; |
||||
|
||||
static absl::StatusOr<MaxAgeFilter> Create(const ChannelArgs& args, |
||||
ChannelFilter::Args filter_args); |
||||
|
||||
void PostInit() override; |
||||
|
||||
private: |
||||
class ConnectivityWatcher : public AsyncConnectivityStateWatcherInterface { |
||||
public: |
||||
explicit ConnectivityWatcher(MaxAgeFilter* filter) |
||||
: channel_stack_(filter->channel_stack()->Ref()), filter_(filter) {} |
||||
~ConnectivityWatcher() override = default; |
||||
|
||||
void OnConnectivityStateChange(grpc_connectivity_state new_state, |
||||
const absl::Status&) override { |
||||
if (new_state == GRPC_CHANNEL_SHUTDOWN) filter_->Shutdown(); |
||||
} |
||||
|
||||
private: |
||||
RefCountedPtr<grpc_channel_stack> channel_stack_; |
||||
MaxAgeFilter* filter_; |
||||
}; |
||||
|
||||
MaxAgeFilter(grpc_channel_stack* channel_stack, const Config& max_age_config); |
||||
|
||||
void Shutdown() override; |
||||
|
||||
SingleSetActivityPtr max_age_activity_; |
||||
Duration max_connection_age_; |
||||
Duration max_connection_age_grace_; |
||||
}; |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif // GRPC_SRC_CORE_EXT_FILTERS_CHANNEL_IDLE_CHANNEL_IDLE_FILTER_H
|
||||
Loading…
Reference in new issue