mirror of https://github.com/grpc/grpc.git
Convert max_age filter to promises (#28904)
* start * stuff * build * Automated change: Fix sanity tests * fix * finish him * fix * better * fix * review feedback * Update channel_idle_filter.cc * Automated change: Fix sanity tests Co-authored-by: ctiller <ctiller@users.noreply.github.com>pull/28957/head
Craig Tiller
3 years ago
committed by
GitHub
parent
7992f162f3
commit
ae4b45d75a
24 changed files with 461 additions and 892 deletions
@ -0,0 +1,382 @@ |
||||
// 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 <limits.h> |
||||
#include <stdlib.h> |
||||
|
||||
#include <atomic> |
||||
|
||||
#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_stack_builder.h" |
||||
#include "src/core/lib/channel/promise_based_filter.h" |
||||
#include "src/core/lib/config/core_configuration.h" |
||||
#include "src/core/lib/gprpp/capture.h" |
||||
#include "src/core/lib/iomgr/timer.h" |
||||
#include "src/core/lib/promise/exec_ctx_wakeup_scheduler.h" |
||||
#include "src/core/lib/promise/loop.h" |
||||
#include "src/core/lib/promise/sleep.h" |
||||
#include "src/core/lib/promise/try_seq.h" |
||||
#include "src/core/lib/transport/http2_errors.h" |
||||
|
||||
// TODO(juanlishen): The idle filter was disabled in client channel by default
|
||||
// due to b/143502997. Now the bug is fixed enable the filter by default.
|
||||
#define DEFAULT_IDLE_TIMEOUT_MS INT_MAX |
||||
// The user input idle timeout smaller than this would be capped to it.
|
||||
#define MIN_IDLE_TIMEOUT_MS (1 /*second*/ * 1000) |
||||
|
||||
// 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.
|
||||
#define DEFAULT_MAX_CONNECTION_AGE_MS INT_MAX |
||||
#define DEFAULT_MAX_CONNECTION_AGE_GRACE_MS INT_MAX |
||||
#define DEFAULT_MAX_CONNECTION_IDLE_MS INT_MAX |
||||
#define MAX_CONNECTION_AGE_JITTER 0.1 |
||||
|
||||
#define MAX_CONNECTION_AGE_INTEGER_OPTIONS \ |
||||
{ DEFAULT_MAX_CONNECTION_AGE_MS, 1, INT_MAX } |
||||
#define MAX_CONNECTION_IDLE_INTEGER_OPTIONS \ |
||||
{ DEFAULT_MAX_CONNECTION_IDLE_MS, 1, INT_MAX } |
||||
|
||||
namespace grpc_core { |
||||
|
||||
TraceFlag grpc_trace_client_idle_filter(false, "client_idle_filter"); |
||||
|
||||
#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) |
||||
|
||||
namespace { |
||||
|
||||
grpc_millis GetClientIdleTimeout(const grpc_channel_args* args) { |
||||
return std::max( |
||||
grpc_channel_arg_get_integer( |
||||
grpc_channel_args_find(args, GRPC_ARG_CLIENT_IDLE_TIMEOUT_MS), |
||||
{DEFAULT_IDLE_TIMEOUT_MS, 0, INT_MAX}), |
||||
MIN_IDLE_TIMEOUT_MS); |
||||
} |
||||
|
||||
struct MaxAgeConfig { |
||||
grpc_millis max_connection_age; |
||||
grpc_millis max_connection_idle; |
||||
grpc_millis max_connection_age_grace; |
||||
|
||||
bool enable() const { |
||||
return max_connection_age != GRPC_MILLIS_INF_FUTURE || |
||||
max_connection_idle != GRPC_MILLIS_INF_FUTURE; |
||||
} |
||||
}; |
||||
|
||||
/* A random jitter of +/-10% will be added to MAX_CONNECTION_AGE to spread out
|
||||
connection storms. Note that the MAX_CONNECTION_AGE option without jitter |
||||
would not create connection storms by itself, but if there happened to be a |
||||
connection storm it could cause it to repeat at a fixed period. */ |
||||
MaxAgeConfig GetMaxAgeConfig(const grpc_channel_args* args) { |
||||
const int args_max_age = grpc_channel_arg_get_integer( |
||||
grpc_channel_args_find(args, GRPC_ARG_MAX_CONNECTION_AGE_MS), |
||||
MAX_CONNECTION_AGE_INTEGER_OPTIONS); |
||||
const int args_max_idle = grpc_channel_arg_get_integer( |
||||
grpc_channel_args_find(args, GRPC_ARG_MAX_CONNECTION_IDLE_MS), |
||||
MAX_CONNECTION_IDLE_INTEGER_OPTIONS); |
||||
const int args_max_age_grace = grpc_channel_arg_get_integer( |
||||
grpc_channel_args_find(args, GRPC_ARG_MAX_CONNECTION_AGE_GRACE_MS), |
||||
{DEFAULT_MAX_CONNECTION_AGE_GRACE_MS, 0, INT_MAX}); |
||||
/* generate a random number between 1 - MAX_CONNECTION_AGE_JITTER and
|
||||
1 + MAX_CONNECTION_AGE_JITTER */ |
||||
const double multiplier = |
||||
rand() * MAX_CONNECTION_AGE_JITTER * 2.0 / RAND_MAX + 1.0 - |
||||
MAX_CONNECTION_AGE_JITTER; |
||||
const double max_age = multiplier * args_max_age; |
||||
/* 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 MaxAgeConfig{ |
||||
args_max_age == INT_MAX || |
||||
max_age > (static_cast<double>(GRPC_MILLIS_INF_FUTURE)) - 0.5 |
||||
? GRPC_MILLIS_INF_FUTURE |
||||
: static_cast<grpc_millis>(args_max_age), |
||||
args_max_idle == INT_MAX ? GRPC_MILLIS_INF_FUTURE : args_max_idle, |
||||
args_max_age_grace}; |
||||
} |
||||
|
||||
class ChannelIdleFilter : public ChannelFilter { |
||||
public: |
||||
~ChannelIdleFilter() override = default; |
||||
|
||||
ChannelIdleFilter(const ChannelIdleFilter&) = delete; |
||||
ChannelIdleFilter& operator=(const ChannelIdleFilter&) = delete; |
||||
ChannelIdleFilter(ChannelIdleFilter&&) = default; |
||||
ChannelIdleFilter& operator=(ChannelIdleFilter&&) = default; |
||||
|
||||
// Construct a promise for one call.
|
||||
ArenaPromise<TrailingMetadata> MakeCallPromise( |
||||
ClientInitialMetadata initial_metadata, |
||||
NextPromiseFactory next_promise_factory) override; |
||||
|
||||
bool StartTransportOp(grpc_transport_op* op) override; |
||||
|
||||
protected: |
||||
ChannelIdleFilter(grpc_channel_stack* channel_stack, |
||||
grpc_millis 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(); |
||||
|
||||
struct CallCountDecreaser { |
||||
void operator()(ChannelIdleFilter* filter) const { |
||||
filter->DecreaseCallCount(); |
||||
} |
||||
}; |
||||
|
||||
// The channel stack to which we take refs for pending callbacks.
|
||||
grpc_channel_stack* channel_stack_; |
||||
grpc_millis client_idle_timeout_; |
||||
std::shared_ptr<IdleFilterState> idle_filter_state_{ |
||||
std::make_shared<IdleFilterState>(false)}; |
||||
|
||||
ActivityPtr activity_; |
||||
}; |
||||
|
||||
class ClientIdleFilter final : public ChannelIdleFilter { |
||||
public: |
||||
static absl::StatusOr<ClientIdleFilter> Create( |
||||
const grpc_channel_args* args, ChannelFilter::Args filter_args); |
||||
|
||||
private: |
||||
using ChannelIdleFilter::ChannelIdleFilter; |
||||
}; |
||||
|
||||
class MaxAgeFilter final : public ChannelIdleFilter { |
||||
public: |
||||
static absl::StatusOr<MaxAgeFilter> Create(const grpc_channel_args* args, |
||||
ChannelFilter::Args filter_args); |
||||
|
||||
void Start(); |
||||
|
||||
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 MaxAgeConfig& 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) {} |
||||
|
||||
void Shutdown() override; |
||||
|
||||
ActivityPtr max_age_activity_; |
||||
grpc_millis max_connection_age_; |
||||
grpc_millis max_connection_age_grace_; |
||||
}; |
||||
|
||||
absl::StatusOr<ClientIdleFilter> ClientIdleFilter::Create( |
||||
const grpc_channel_args* 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 grpc_channel_args* args, ChannelFilter::Args filter_args) { |
||||
const auto config = GetMaxAgeConfig(args); |
||||
MaxAgeFilter filter(filter_args.channel_stack(), config); |
||||
return absl::StatusOr<MaxAgeFilter>(std::move(filter)); |
||||
} |
||||
|
||||
void MaxAgeFilter::Shutdown() { |
||||
max_age_activity_.reset(); |
||||
ChannelIdleFilter::Shutdown(); |
||||
} |
||||
|
||||
void MaxAgeFilter::Start() { |
||||
// Trigger idle timer immediately
|
||||
IncreaseCallCount(); |
||||
DecreaseCallCount(); |
||||
|
||||
auto channel_stack = this->channel_stack()->Ref(); |
||||
|
||||
// Start the max age timer
|
||||
if (max_connection_age_ != GRPC_MILLIS_INF_FUTURE) { |
||||
max_age_activity_ = MakeActivity( |
||||
TrySeq( |
||||
// First sleep until the max connection age
|
||||
Sleep(ExecCtx::Get()->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_FROM_STATIC_STRING("max_age"), |
||||
GRPC_ERROR_INT_HTTP2_ERROR, 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), |
||||
GRPC_ERROR_NONE); |
||||
return Immediate(absl::OkStatus()); |
||||
}, |
||||
// Sleep for the grace period
|
||||
[this] { |
||||
return Sleep(ExecCtx::Get()->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(); |
||||
}); |
||||
} |
||||
} |
||||
|
||||
// Construct a promise for one call.
|
||||
ArenaPromise<TrailingMetadata> ChannelIdleFilter::MakeCallPromise( |
||||
ClientInitialMetadata initial_metadata, |
||||
NextPromiseFactory next_promise_factory) { |
||||
using Decrementer = std::unique_ptr<ChannelIdleFilter, CallCountDecreaser>; |
||||
IncreaseCallCount(); |
||||
return ArenaPromise<TrailingMetadata>(Capture( |
||||
[](Decrementer*, ArenaPromise<TrailingMetadata>* next) |
||||
-> Poll<TrailingMetadata> { return (*next)(); }, |
||||
Decrementer(this), next_promise_factory(std::move(initial_metadata)))); |
||||
} |
||||
|
||||
bool ChannelIdleFilter::StartTransportOp(grpc_transport_op* op) { |
||||
// Catch the disconnect_with_error transport op.
|
||||
if (op->disconnect_with_error != GRPC_ERROR_NONE) 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(ExecCtx::Get()->Now() + timeout), |
||||
[idle_filter_state]() -> Poll<LoopCtl<absl::Status>> { |
||||
if (idle_filter_state->CheckTimer()) { |
||||
return Continue{}; |
||||
} else { |
||||
return absl::OkStatus(); |
||||
} |
||||
}); |
||||
}); |
||||
activity_ = MakeActivity(std::move(promise), ExecCtxWakeupScheduler{}, |
||||
[channel_stack, this](absl::Status status) { |
||||
if (status.ok()) CloseChannel(); |
||||
}); |
||||
} |
||||
|
||||
void ChannelIdleFilter::CloseChannel() { |
||||
auto* op = grpc_make_transport_op(nullptr); |
||||
op->disconnect_with_error = grpc_error_set_int( |
||||
GRPC_ERROR_CREATE_FROM_STATIC_STRING("enter idle"), |
||||
GRPC_ERROR_INT_CHANNEL_CONNECTIVITY_STATE, 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 grpc_client_idle_filter = |
||||
MakePromiseBasedFilter<ClientIdleFilter, FilterEndpoint::kClient>( |
||||
"client_idle"); |
||||
const grpc_channel_filter grpc_max_age_filter = |
||||
MakePromiseBasedFilter<MaxAgeFilter, FilterEndpoint::kServer>("max_age"); |
||||
|
||||
} // namespace
|
||||
|
||||
void RegisterChannelIdleFilters(CoreConfiguration::Builder* builder) { |
||||
builder->channel_init()->RegisterStage( |
||||
GRPC_CLIENT_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, |
||||
[](ChannelStackBuilder* builder) { |
||||
const grpc_channel_args* channel_args = builder->channel_args(); |
||||
if (!grpc_channel_args_want_minimal_stack(channel_args) && |
||||
GetClientIdleTimeout(channel_args) != INT_MAX) { |
||||
builder->PrependFilter(&grpc_client_idle_filter, nullptr); |
||||
} |
||||
return true; |
||||
}); |
||||
builder->channel_init()->RegisterStage( |
||||
GRPC_SERVER_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, |
||||
[](ChannelStackBuilder* builder) { |
||||
const grpc_channel_args* channel_args = builder->channel_args(); |
||||
if (GetMaxAgeConfig(channel_args).enable()) { |
||||
builder->PrependFilter( |
||||
&grpc_max_age_filter, |
||||
[](grpc_channel_stack*, grpc_channel_element* elem) { |
||||
static_cast<MaxAgeFilter*>(elem->channel_data)->Start(); |
||||
}); |
||||
} |
||||
return true; |
||||
}); |
||||
} |
||||
} // namespace grpc_core
|
||||
@ -1,199 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2019 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. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <limits.h> |
||||
|
||||
#include <atomic> |
||||
|
||||
#include "src/core/ext/filters/client_idle/idle_filter_state.h" |
||||
#include "src/core/lib/channel/channel_args.h" |
||||
#include "src/core/lib/channel/channel_stack_builder.h" |
||||
#include "src/core/lib/channel/promise_based_filter.h" |
||||
#include "src/core/lib/config/core_configuration.h" |
||||
#include "src/core/lib/gprpp/capture.h" |
||||
#include "src/core/lib/iomgr/timer.h" |
||||
#include "src/core/lib/promise/exec_ctx_wakeup_scheduler.h" |
||||
#include "src/core/lib/promise/loop.h" |
||||
#include "src/core/lib/promise/sleep.h" |
||||
#include "src/core/lib/promise/try_seq.h" |
||||
#include "src/core/lib/transport/http2_errors.h" |
||||
|
||||
// TODO(juanlishen): The idle filter is disabled in client channel by default
|
||||
// due to b/143502997. Try to fix the bug and enable the filter by default.
|
||||
#define DEFAULT_IDLE_TIMEOUT_MS INT_MAX |
||||
// The user input idle timeout smaller than this would be capped to it.
|
||||
#define MIN_IDLE_TIMEOUT_MS (1 /*second*/ * 1000) |
||||
|
||||
namespace grpc_core { |
||||
|
||||
TraceFlag grpc_trace_client_idle_filter(false, "client_idle_filter"); |
||||
|
||||
#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) |
||||
|
||||
namespace { |
||||
|
||||
grpc_millis GetClientIdleTimeout(const grpc_channel_args* args) { |
||||
return std::max( |
||||
grpc_channel_arg_get_integer( |
||||
grpc_channel_args_find(args, GRPC_ARG_CLIENT_IDLE_TIMEOUT_MS), |
||||
{DEFAULT_IDLE_TIMEOUT_MS, 0, INT_MAX}), |
||||
MIN_IDLE_TIMEOUT_MS); |
||||
} |
||||
|
||||
class ClientIdleFilter : public ChannelFilter { |
||||
public: |
||||
static absl::StatusOr<ClientIdleFilter> Create( |
||||
const grpc_channel_args* args, ChannelFilter::Args filter_args); |
||||
~ClientIdleFilter() override = default; |
||||
|
||||
ClientIdleFilter(const ClientIdleFilter&) = delete; |
||||
ClientIdleFilter& operator=(const ClientIdleFilter&) = delete; |
||||
ClientIdleFilter(ClientIdleFilter&&) = default; |
||||
ClientIdleFilter& operator=(ClientIdleFilter&&) = default; |
||||
|
||||
// Construct a promise for one call.
|
||||
ArenaPromise<TrailingMetadata> MakeCallPromise( |
||||
ClientInitialMetadata initial_metadata, |
||||
NextPromiseFactory next_promise_factory) override; |
||||
|
||||
bool StartTransportOp(grpc_transport_op* op) override; |
||||
|
||||
private: |
||||
ClientIdleFilter(grpc_channel_stack* channel_stack, |
||||
grpc_millis client_idle_timeout) |
||||
: channel_stack_(channel_stack), |
||||
client_idle_timeout_(client_idle_timeout) {} |
||||
|
||||
void StartIdleTimer(); |
||||
|
||||
void IncreaseCallCount(); |
||||
void DecreaseCallCount(); |
||||
|
||||
struct CallCountDecreaser { |
||||
void operator()(ClientIdleFilter* filter) const { |
||||
filter->DecreaseCallCount(); |
||||
} |
||||
}; |
||||
|
||||
// The channel stack to which we take refs for pending callbacks.
|
||||
grpc_channel_stack* channel_stack_; |
||||
grpc_millis client_idle_timeout_; |
||||
std::shared_ptr<IdleFilterState> idle_filter_state_{ |
||||
std::make_shared<IdleFilterState>(false)}; |
||||
|
||||
ActivityPtr activity_; |
||||
}; |
||||
|
||||
absl::StatusOr<ClientIdleFilter> ClientIdleFilter::Create( |
||||
const grpc_channel_args* args, ChannelFilter::Args filter_args) { |
||||
ClientIdleFilter filter(filter_args.channel_stack(), |
||||
GetClientIdleTimeout(args)); |
||||
return absl::StatusOr<ClientIdleFilter>(std::move(filter)); |
||||
} |
||||
|
||||
// Construct a promise for one call.
|
||||
ArenaPromise<TrailingMetadata> ClientIdleFilter::MakeCallPromise( |
||||
ClientInitialMetadata initial_metadata, |
||||
NextPromiseFactory next_promise_factory) { |
||||
using Decrementer = std::unique_ptr<ClientIdleFilter, CallCountDecreaser>; |
||||
IncreaseCallCount(); |
||||
return ArenaPromise<TrailingMetadata>(Capture( |
||||
[](Decrementer*, ArenaPromise<TrailingMetadata>* next) |
||||
-> Poll<TrailingMetadata> { return (*next)(); }, |
||||
Decrementer(this), next_promise_factory(std::move(initial_metadata)))); |
||||
} |
||||
|
||||
bool ClientIdleFilter::StartTransportOp(grpc_transport_op* op) { |
||||
// Catch the disconnect_with_error transport op.
|
||||
if (op->disconnect_with_error != GRPC_ERROR_NONE) { |
||||
// IncreaseCallCount() introduces a phony call and prevent the timer from
|
||||
// being reset by other threads.
|
||||
IncreaseCallCount(); |
||||
activity_.reset(); |
||||
} |
||||
// Pass the op to the next filter.
|
||||
return false; |
||||
} |
||||
|
||||
void ClientIdleFilter::IncreaseCallCount() { |
||||
idle_filter_state_->IncreaseCallCount(); |
||||
} |
||||
|
||||
void ClientIdleFilter::DecreaseCallCount() { |
||||
if (idle_filter_state_->DecreaseCallCount()) { |
||||
// If there are no more calls in progress, start the idle timer.
|
||||
StartIdleTimer(); |
||||
} |
||||
} |
||||
|
||||
void ClientIdleFilter::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(ExecCtx::Get()->Now() + timeout), |
||||
[idle_filter_state]() -> Poll<LoopCtl<absl::Status>> { |
||||
if (idle_filter_state->CheckTimer()) { |
||||
return Continue{}; |
||||
} else { |
||||
return absl::OkStatus(); |
||||
} |
||||
}); |
||||
}); |
||||
activity_ = MakeActivity( |
||||
std::move(promise), ExecCtxWakeupScheduler{}, |
||||
[channel_stack](absl::Status status) { |
||||
if (!status.ok()) return; |
||||
auto* op = grpc_make_transport_op(nullptr); |
||||
op->disconnect_with_error = grpc_error_set_int( |
||||
GRPC_ERROR_CREATE_FROM_STATIC_STRING("enter idle"), |
||||
GRPC_ERROR_INT_CHANNEL_CONNECTIVITY_STATE, GRPC_CHANNEL_IDLE); |
||||
// Pass the transport op down to the channel stack.
|
||||
auto* elem = grpc_channel_stack_element(channel_stack.get(), 0); |
||||
elem->filter->start_transport_op(elem, op); |
||||
}); |
||||
} |
||||
|
||||
const grpc_channel_filter grpc_client_idle_filter = |
||||
MakePromiseBasedFilter<ClientIdleFilter, FilterEndpoint::kClient>( |
||||
"client_idle"); |
||||
|
||||
} // namespace
|
||||
|
||||
void RegisterClientIdleFilter(CoreConfiguration::Builder* builder) { |
||||
builder->channel_init()->RegisterStage( |
||||
GRPC_CLIENT_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, |
||||
[](ChannelStackBuilder* builder) { |
||||
const grpc_channel_args* channel_args = builder->channel_args(); |
||||
if (!grpc_channel_args_want_minimal_stack(channel_args) && |
||||
GetClientIdleTimeout(channel_args) != INT_MAX) { |
||||
builder->PrependFilter(&grpc_client_idle_filter, nullptr); |
||||
} |
||||
return true; |
||||
}); |
||||
} |
||||
} // namespace grpc_core
|
||||
@ -1,558 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2017 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. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/ext/filters/max_age/max_age_filter.h" |
||||
|
||||
#include <limits.h> |
||||
#include <string.h> |
||||
|
||||
#include "src/core/lib/channel/channel_args.h" |
||||
#include "src/core/lib/channel/channel_stack_builder.h" |
||||
#include "src/core/lib/config/core_configuration.h" |
||||
#include "src/core/lib/iomgr/timer.h" |
||||
#include "src/core/lib/transport/http2_errors.h" |
||||
|
||||
/* 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. |
||||
*/ |
||||
#define DEFAULT_MAX_CONNECTION_AGE_MS INT_MAX |
||||
#define DEFAULT_MAX_CONNECTION_AGE_GRACE_MS INT_MAX |
||||
#define DEFAULT_MAX_CONNECTION_IDLE_MS INT_MAX |
||||
#define MAX_CONNECTION_AGE_JITTER 0.1 |
||||
|
||||
#define MAX_CONNECTION_AGE_INTEGER_OPTIONS \ |
||||
{ DEFAULT_MAX_CONNECTION_AGE_MS, 1, INT_MAX } |
||||
#define MAX_CONNECTION_IDLE_INTEGER_OPTIONS \ |
||||
{ DEFAULT_MAX_CONNECTION_IDLE_MS, 1, INT_MAX } |
||||
|
||||
/* States for idle_state in channel_data */ |
||||
#define MAX_IDLE_STATE_INIT ((gpr_atm)0) |
||||
#define MAX_IDLE_STATE_SEEN_EXIT_IDLE ((gpr_atm)1) |
||||
#define MAX_IDLE_STATE_SEEN_ENTER_IDLE ((gpr_atm)2) |
||||
#define MAX_IDLE_STATE_TIMER_SET ((gpr_atm)3) |
||||
|
||||
namespace { |
||||
struct channel_data { |
||||
/* The channel stack to which we take refs for pending callbacks. */ |
||||
grpc_channel_stack* channel_stack; |
||||
/* Guards access to max_age_timer, max_age_timer_pending, max_age_grace_timer
|
||||
and max_age_grace_timer_pending */ |
||||
grpc_core::Mutex max_age_timer_mu; |
||||
/* True if the max_age timer callback is currently pending */ |
||||
bool max_age_timer_pending ABSL_GUARDED_BY(max_age_timer_mu) = false; |
||||
/* True if the max_age_grace timer callback is currently pending */ |
||||
bool max_age_grace_timer_pending ABSL_GUARDED_BY(max_age_timer_mu) = false; |
||||
/* The timer for checking if the channel has reached its max age */ |
||||
grpc_timer max_age_timer ABSL_GUARDED_BY(max_age_timer_mu); |
||||
/* The timer for checking if the max-aged channel has uesed up the grace
|
||||
period */ |
||||
grpc_timer max_age_grace_timer ABSL_GUARDED_BY(max_age_timer_mu); |
||||
/* The timer for checking if the channel's idle duration reaches
|
||||
max_connection_idle */ |
||||
grpc_timer max_idle_timer; |
||||
/* Allowed max time a channel may have no outstanding rpcs */ |
||||
grpc_millis max_connection_idle; |
||||
/* Allowed max time a channel may exist */ |
||||
grpc_millis max_connection_age; |
||||
/* Allowed grace period after the channel reaches its max age */ |
||||
grpc_millis max_connection_age_grace; |
||||
/* Closure to run when the channel's idle duration reaches max_connection_idle
|
||||
and should be closed gracefully */ |
||||
grpc_closure max_idle_timer_cb; |
||||
/* Closure to run when the channel reaches its max age and should be closed
|
||||
gracefully */ |
||||
grpc_closure close_max_age_channel; |
||||
/* Closure to run the channel uses up its max age grace time and should be
|
||||
closed forcibly */ |
||||
grpc_closure force_close_max_age_channel; |
||||
/* Closure to run when the init fo channel stack is done and the max_idle
|
||||
timer should be started */ |
||||
grpc_closure start_max_idle_timer_after_init; |
||||
/* Closure to run when the init fo channel stack is done and the max_age timer
|
||||
should be started */ |
||||
grpc_closure start_max_age_timer_after_init; |
||||
/* Closure to run when the goaway op is finished and the max_age_timer */ |
||||
grpc_closure start_max_age_grace_timer_after_goaway_op; |
||||
/* Number of active calls */ |
||||
gpr_atm call_count; |
||||
/* TODO(zyc): C++lize this state machine */ |
||||
/* 'idle_state' holds the states of max_idle_timer and channel idleness.
|
||||
It can contain one of the following values: |
||||
+--------------------------------+----------------+---------+ |
||||
| idle_state | max_idle_timer | channel | |
||||
+--------------------------------+----------------+---------+ |
||||
| MAX_IDLE_STATE_INIT | unset | busy | |
||||
| MAX_IDLE_STATE_TIMER_SET | set, valid | idle | |
||||
| MAX_IDLE_STATE_SEEN_EXIT_IDLE | set, invalid | busy | |
||||
| MAX_IDLE_STATE_SEEN_ENTER_IDLE | set, invalid | idle | |
||||
+--------------------------------+----------------+---------+ |
||||
|
||||
MAX_IDLE_STATE_INIT: The initial and final state of 'idle_state'. The |
||||
channel has 1 or 1+ active calls, and the timer is not set. Note that |
||||
we may put a virtual call to hold this state at channel initialization or |
||||
shutdown, so that the channel won't enter other states. |
||||
|
||||
MAX_IDLE_STATE_TIMER_SET: The state after the timer is set and no calls |
||||
have arrived after the timer is set. The channel must have 0 active call in |
||||
this state. If the timer is fired in this state, we will close the channel |
||||
due to idleness. |
||||
|
||||
MAX_IDLE_STATE_SEEN_EXIT_IDLE: The state after the timer is set and at |
||||
least one call has arrived after the timer is set. The channel must have 1 |
||||
or 1+ active calls in this state. If the timer is fired in this state, we |
||||
won't reschudle it. |
||||
|
||||
MAX_IDLE_STATE_SEEN_ENTER_IDLE: The state after the timer is set and the at |
||||
least one call has arrived after the timer is set, BUT the channel |
||||
currently has 0 active calls. If the timer is fired in this state, we will |
||||
reschudle it. |
||||
|
||||
max_idle_timer will not be cancelled (unless the channel is shutting down). |
||||
If the timer callback is called when the max_idle_timer is valid (i.e. |
||||
idle_state is MAX_IDLE_STATE_TIMER_SET), the channel will be closed due to |
||||
idleness, otherwise the channel won't be changed. |
||||
|
||||
State transitions: |
||||
MAX_IDLE_STATE_INIT <-------3------ MAX_IDLE_STATE_SEEN_EXIT_IDLE |
||||
^ | ^ ^ | |
||||
| | | | | |
||||
1 2 +-----------4------------+ 6 7 |
||||
| | | | | |
||||
| v | | v |
||||
MAX_IDLE_STATE_TIMER_SET <----5------ MAX_IDLE_STATE_SEEN_ENTER_IDLE |
||||
|
||||
For 1, 3, 5 : See max_idle_timer_cb() function |
||||
For 2, 7 : See decrease_call_count() function |
||||
For 4, 6 : See increase_call_count() function */ |
||||
gpr_atm idle_state; |
||||
/* Time when the channel finished its last outstanding call, in grpc_millis */ |
||||
gpr_atm last_enter_idle_time_millis; |
||||
}; |
||||
} // namespace
|
||||
|
||||
/* Increase the nubmer of active calls. Before the increasement, if there are no
|
||||
calls, the max_idle_timer should be cancelled. */ |
||||
static void increase_call_count(channel_data* chand) { |
||||
/* Exit idle */ |
||||
if (gpr_atm_full_fetch_add(&chand->call_count, 1) == 0) { |
||||
while (true) { |
||||
gpr_atm idle_state = gpr_atm_acq_load(&chand->idle_state); |
||||
switch (idle_state) { |
||||
case MAX_IDLE_STATE_TIMER_SET: |
||||
/* max_idle_timer_cb may have already set idle_state to
|
||||
MAX_IDLE_STATE_INIT, in this case, we don't need to set it to |
||||
MAX_IDLE_STATE_SEEN_EXIT_IDLE */ |
||||
gpr_atm_rel_cas(&chand->idle_state, MAX_IDLE_STATE_TIMER_SET, |
||||
MAX_IDLE_STATE_SEEN_EXIT_IDLE); |
||||
return; |
||||
case MAX_IDLE_STATE_SEEN_ENTER_IDLE: |
||||
gpr_atm_rel_store(&chand->idle_state, MAX_IDLE_STATE_SEEN_EXIT_IDLE); |
||||
return; |
||||
default: |
||||
/* try again */ |
||||
break; |
||||
} |
||||
} |
||||
} |
||||
} |
||||
|
||||
/* Decrease the nubmer of active calls. After the decrement, if there are no
|
||||
calls, the max_idle_timer should be started. */ |
||||
static void decrease_call_count(channel_data* chand) { |
||||
/* Enter idle */ |
||||
if (gpr_atm_full_fetch_add(&chand->call_count, -1) == 1) { |
||||
gpr_atm_no_barrier_store(&chand->last_enter_idle_time_millis, |
||||
(gpr_atm)grpc_core::ExecCtx::Get()->Now()); |
||||
while (true) { |
||||
gpr_atm idle_state = gpr_atm_acq_load(&chand->idle_state); |
||||
switch (idle_state) { |
||||
case MAX_IDLE_STATE_INIT: |
||||
GRPC_CHANNEL_STACK_REF(chand->channel_stack, |
||||
"max_age max_idle_timer"); |
||||
grpc_timer_init( |
||||
&chand->max_idle_timer, |
||||
grpc_core::ExecCtx::Get()->Now() + chand->max_connection_idle, |
||||
&chand->max_idle_timer_cb); |
||||
gpr_atm_rel_store(&chand->idle_state, MAX_IDLE_STATE_TIMER_SET); |
||||
return; |
||||
case MAX_IDLE_STATE_SEEN_EXIT_IDLE: |
||||
if (gpr_atm_rel_cas(&chand->idle_state, MAX_IDLE_STATE_SEEN_EXIT_IDLE, |
||||
MAX_IDLE_STATE_SEEN_ENTER_IDLE)) { |
||||
return; |
||||
} |
||||
break; |
||||
default: |
||||
/* try again */ |
||||
break; |
||||
} |
||||
} |
||||
} |
||||
} |
||||
|
||||
static void start_max_idle_timer_after_init(void* arg, |
||||
grpc_error_handle /*error*/) { |
||||
channel_data* chand = static_cast<channel_data*>(arg); |
||||
/* Decrease call_count. If there are no active calls at this time,
|
||||
max_idle_timer will start here. If the number of active calls is not 0, |
||||
max_idle_timer will start after all the active calls end. */ |
||||
decrease_call_count(chand); |
||||
GRPC_CHANNEL_STACK_UNREF(chand->channel_stack, |
||||
"max_age start_max_idle_timer_after_init"); |
||||
} |
||||
|
||||
namespace grpc_core { |
||||
|
||||
class ConnectivityWatcher : public AsyncConnectivityStateWatcherInterface { |
||||
public: |
||||
explicit ConnectivityWatcher(channel_data* chand) : chand_(chand) { |
||||
GRPC_CHANNEL_STACK_REF(chand_->channel_stack, "max_age conn_watch"); |
||||
} |
||||
|
||||
~ConnectivityWatcher() override { |
||||
GRPC_CHANNEL_STACK_UNREF(chand_->channel_stack, "max_age conn_watch"); |
||||
} |
||||
|
||||
private: |
||||
void OnConnectivityStateChange(grpc_connectivity_state new_state, |
||||
const absl::Status& /* status */) override { |
||||
if (new_state != GRPC_CHANNEL_SHUTDOWN) return; |
||||
{ |
||||
MutexLock lock(&chand_->max_age_timer_mu); |
||||
if (chand_->max_age_timer_pending) { |
||||
grpc_timer_cancel(&chand_->max_age_timer); |
||||
chand_->max_age_timer_pending = false; |
||||
} |
||||
if (chand_->max_age_grace_timer_pending) { |
||||
grpc_timer_cancel(&chand_->max_age_grace_timer); |
||||
chand_->max_age_grace_timer_pending = false; |
||||
} |
||||
} |
||||
/* If there are no active calls, this increasement will cancel
|
||||
max_idle_timer, and prevent max_idle_timer from being started in the |
||||
future. */ |
||||
increase_call_count(chand_); |
||||
if (gpr_atm_acq_load(&chand_->idle_state) == |
||||
MAX_IDLE_STATE_SEEN_EXIT_IDLE) { |
||||
grpc_timer_cancel(&chand_->max_idle_timer); |
||||
} |
||||
} |
||||
|
||||
channel_data* chand_; |
||||
}; |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
static void start_max_age_timer_after_init(void* arg, |
||||
grpc_error_handle /*error*/) { |
||||
channel_data* chand = static_cast<channel_data*>(arg); |
||||
{ |
||||
grpc_core::MutexLock lock(&chand->max_age_timer_mu); |
||||
chand->max_age_timer_pending = true; |
||||
GRPC_CHANNEL_STACK_REF(chand->channel_stack, "max_age max_age_timer"); |
||||
grpc_timer_init( |
||||
&chand->max_age_timer, |
||||
grpc_core::ExecCtx::Get()->Now() + chand->max_connection_age, |
||||
&chand->close_max_age_channel); |
||||
} |
||||
grpc_transport_op* op = grpc_make_transport_op(nullptr); |
||||
op->start_connectivity_watch.reset(new grpc_core::ConnectivityWatcher(chand)); |
||||
op->start_connectivity_watch_state = GRPC_CHANNEL_IDLE; |
||||
grpc_channel_next_op(grpc_channel_stack_element(chand->channel_stack, 0), op); |
||||
GRPC_CHANNEL_STACK_UNREF(chand->channel_stack, |
||||
"max_age start_max_age_timer_after_init"); |
||||
} |
||||
|
||||
static void start_max_age_grace_timer_after_goaway_op( |
||||
void* arg, grpc_error_handle /*error*/) { |
||||
channel_data* chand = static_cast<channel_data*>(arg); |
||||
{ |
||||
grpc_core::MutexLock lock(&chand->max_age_timer_mu); |
||||
chand->max_age_grace_timer_pending = true; |
||||
GRPC_CHANNEL_STACK_REF(chand->channel_stack, "max_age max_age_grace_timer"); |
||||
grpc_timer_init(&chand->max_age_grace_timer, |
||||
chand->max_connection_age_grace == GRPC_MILLIS_INF_FUTURE |
||||
? GRPC_MILLIS_INF_FUTURE |
||||
: grpc_core::ExecCtx::Get()->Now() + |
||||
chand->max_connection_age_grace, |
||||
&chand->force_close_max_age_channel); |
||||
} |
||||
GRPC_CHANNEL_STACK_UNREF(chand->channel_stack, |
||||
"max_age start_max_age_grace_timer_after_goaway_op"); |
||||
} |
||||
|
||||
static void close_max_idle_channel(channel_data* chand) { |
||||
/* Prevent the max idle timer from being set again */ |
||||
gpr_atm_no_barrier_fetch_add(&chand->call_count, 1); |
||||
grpc_transport_op* op = grpc_make_transport_op(nullptr); |
||||
op->goaway_error = |
||||
grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING("max_idle"), |
||||
GRPC_ERROR_INT_HTTP2_ERROR, GRPC_HTTP2_NO_ERROR); |
||||
grpc_channel_element* elem = |
||||
grpc_channel_stack_element(chand->channel_stack, 0); |
||||
elem->filter->start_transport_op(elem, op); |
||||
} |
||||
|
||||
static void max_idle_timer_cb(void* arg, grpc_error_handle error) { |
||||
channel_data* chand = static_cast<channel_data*>(arg); |
||||
if (error == GRPC_ERROR_NONE) { |
||||
bool try_again = true; |
||||
while (try_again) { |
||||
gpr_atm idle_state = gpr_atm_acq_load(&chand->idle_state); |
||||
switch (idle_state) { |
||||
case MAX_IDLE_STATE_TIMER_SET: |
||||
close_max_idle_channel(chand); |
||||
/* This MAX_IDLE_STATE_INIT is a final state, we don't have to check
|
||||
* if idle_state has been changed */ |
||||
gpr_atm_rel_store(&chand->idle_state, MAX_IDLE_STATE_INIT); |
||||
try_again = false; |
||||
break; |
||||
case MAX_IDLE_STATE_SEEN_EXIT_IDLE: |
||||
if (gpr_atm_rel_cas(&chand->idle_state, MAX_IDLE_STATE_SEEN_EXIT_IDLE, |
||||
MAX_IDLE_STATE_INIT)) { |
||||
try_again = false; |
||||
} |
||||
break; |
||||
case MAX_IDLE_STATE_SEEN_ENTER_IDLE: |
||||
GRPC_CHANNEL_STACK_REF(chand->channel_stack, |
||||
"max_age max_idle_timer"); |
||||
grpc_timer_init(&chand->max_idle_timer, |
||||
static_cast<grpc_millis>(gpr_atm_no_barrier_load( |
||||
&chand->last_enter_idle_time_millis)) + |
||||
chand->max_connection_idle, |
||||
&chand->max_idle_timer_cb); |
||||
/* idle_state may have already been set to
|
||||
MAX_IDLE_STATE_SEEN_EXIT_IDLE by increase_call_count(), in this |
||||
case, we don't need to set it to MAX_IDLE_STATE_TIMER_SET */ |
||||
gpr_atm_rel_cas(&chand->idle_state, MAX_IDLE_STATE_SEEN_ENTER_IDLE, |
||||
MAX_IDLE_STATE_TIMER_SET); |
||||
try_again = false; |
||||
break; |
||||
default: |
||||
/* try again */ |
||||
break; |
||||
} |
||||
} |
||||
} |
||||
GRPC_CHANNEL_STACK_UNREF(chand->channel_stack, "max_age max_idle_timer"); |
||||
} |
||||
|
||||
static void close_max_age_channel(void* arg, grpc_error_handle error) { |
||||
channel_data* chand = static_cast<channel_data*>(arg); |
||||
{ |
||||
grpc_core::MutexLock lock(&chand->max_age_timer_mu); |
||||
chand->max_age_timer_pending = false; |
||||
} |
||||
if (error == GRPC_ERROR_NONE) { |
||||
GRPC_CHANNEL_STACK_REF(chand->channel_stack, |
||||
"max_age start_max_age_grace_timer_after_goaway_op"); |
||||
grpc_transport_op* op = grpc_make_transport_op( |
||||
&chand->start_max_age_grace_timer_after_goaway_op); |
||||
op->goaway_error = |
||||
grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING("max_age"), |
||||
GRPC_ERROR_INT_HTTP2_ERROR, GRPC_HTTP2_NO_ERROR); |
||||
grpc_channel_element* elem = |
||||
grpc_channel_stack_element(chand->channel_stack, 0); |
||||
elem->filter->start_transport_op(elem, op); |
||||
} else if (error != GRPC_ERROR_CANCELLED) { |
||||
GRPC_LOG_IF_ERROR("close_max_age_channel", error); |
||||
} |
||||
GRPC_CHANNEL_STACK_UNREF(chand->channel_stack, "max_age max_age_timer"); |
||||
} |
||||
|
||||
static void force_close_max_age_channel(void* arg, grpc_error_handle error) { |
||||
channel_data* chand = static_cast<channel_data*>(arg); |
||||
{ |
||||
grpc_core::MutexLock lock(&chand->max_age_timer_mu); |
||||
chand->max_age_grace_timer_pending = false; |
||||
} |
||||
if (error == GRPC_ERROR_NONE) { |
||||
grpc_transport_op* op = grpc_make_transport_op(nullptr); |
||||
op->disconnect_with_error = |
||||
GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel reaches max age"); |
||||
grpc_channel_element* elem = |
||||
grpc_channel_stack_element(chand->channel_stack, 0); |
||||
elem->filter->start_transport_op(elem, op); |
||||
} else if (error != GRPC_ERROR_CANCELLED) { |
||||
GRPC_LOG_IF_ERROR("force_close_max_age_channel", error); |
||||
} |
||||
GRPC_CHANNEL_STACK_UNREF(chand->channel_stack, "max_age max_age_grace_timer"); |
||||
} |
||||
|
||||
/* A random jitter of +/-10% will be added to MAX_CONNECTION_AGE to spread out
|
||||
connection storms. Note that the MAX_CONNECTION_AGE option without jitter |
||||
would not create connection storms by itself, but if there happened to be a |
||||
connection storm it could cause it to repeat at a fixed period. */ |
||||
static grpc_millis |
||||
add_random_max_connection_age_jitter_and_convert_to_grpc_millis(int value) { |
||||
/* generate a random number between 1 - MAX_CONNECTION_AGE_JITTER and
|
||||
1 + MAX_CONNECTION_AGE_JITTER */ |
||||
double multiplier = rand() * MAX_CONNECTION_AGE_JITTER * 2.0 / RAND_MAX + |
||||
1.0 - MAX_CONNECTION_AGE_JITTER; |
||||
double result = multiplier * value; |
||||
/* INT_MAX - 0.5 converts the value to float, so that result will not be
|
||||
cast to int implicitly before the comparison. */ |
||||
return result > (static_cast<double>(GRPC_MILLIS_INF_FUTURE)) - 0.5 |
||||
? GRPC_MILLIS_INF_FUTURE |
||||
: static_cast<grpc_millis>(result); |
||||
} |
||||
|
||||
/* Constructor for call_data. */ |
||||
static grpc_error_handle max_age_init_call_elem( |
||||
grpc_call_element* elem, const grpc_call_element_args* /*args*/) { |
||||
channel_data* chand = static_cast<channel_data*>(elem->channel_data); |
||||
increase_call_count(chand); |
||||
return GRPC_ERROR_NONE; |
||||
} |
||||
|
||||
/* Destructor for call_data. */ |
||||
static void max_age_destroy_call_elem( |
||||
grpc_call_element* elem, const grpc_call_final_info* /*final_info*/, |
||||
grpc_closure* /*ignored*/) { |
||||
channel_data* chand = static_cast<channel_data*>(elem->channel_data); |
||||
decrease_call_count(chand); |
||||
} |
||||
|
||||
/* Constructor for channel_data. */ |
||||
static grpc_error_handle max_age_init_channel_elem( |
||||
grpc_channel_element* elem, grpc_channel_element_args* args) { |
||||
channel_data* chand = static_cast<channel_data*>(elem->channel_data); |
||||
new (chand) channel_data(); |
||||
chand->channel_stack = args->channel_stack; |
||||
chand->max_connection_age = |
||||
add_random_max_connection_age_jitter_and_convert_to_grpc_millis( |
||||
DEFAULT_MAX_CONNECTION_AGE_MS); |
||||
chand->max_connection_age_grace = |
||||
DEFAULT_MAX_CONNECTION_AGE_GRACE_MS == INT_MAX |
||||
? GRPC_MILLIS_INF_FUTURE |
||||
: DEFAULT_MAX_CONNECTION_AGE_GRACE_MS; |
||||
chand->max_connection_idle = DEFAULT_MAX_CONNECTION_IDLE_MS == INT_MAX |
||||
? GRPC_MILLIS_INF_FUTURE |
||||
: DEFAULT_MAX_CONNECTION_IDLE_MS; |
||||
chand->idle_state = MAX_IDLE_STATE_INIT; |
||||
gpr_atm_no_barrier_store(&chand->last_enter_idle_time_millis, GPR_ATM_MIN); |
||||
for (size_t i = 0; i < args->channel_args->num_args; ++i) { |
||||
if (0 == strcmp(args->channel_args->args[i].key, |
||||
GRPC_ARG_MAX_CONNECTION_AGE_MS)) { |
||||
const int value = grpc_channel_arg_get_integer( |
||||
&args->channel_args->args[i], MAX_CONNECTION_AGE_INTEGER_OPTIONS); |
||||
chand->max_connection_age = |
||||
add_random_max_connection_age_jitter_and_convert_to_grpc_millis( |
||||
value); |
||||
} else if (0 == strcmp(args->channel_args->args[i].key, |
||||
GRPC_ARG_MAX_CONNECTION_AGE_GRACE_MS)) { |
||||
const int value = grpc_channel_arg_get_integer( |
||||
&args->channel_args->args[i], |
||||
{DEFAULT_MAX_CONNECTION_AGE_GRACE_MS, 0, INT_MAX}); |
||||
chand->max_connection_age_grace = |
||||
value == INT_MAX ? GRPC_MILLIS_INF_FUTURE : value; |
||||
} else if (0 == strcmp(args->channel_args->args[i].key, |
||||
GRPC_ARG_MAX_CONNECTION_IDLE_MS)) { |
||||
const int value = grpc_channel_arg_get_integer( |
||||
&args->channel_args->args[i], MAX_CONNECTION_IDLE_INTEGER_OPTIONS); |
||||
chand->max_connection_idle = |
||||
value == INT_MAX ? GRPC_MILLIS_INF_FUTURE : value; |
||||
} |
||||
} |
||||
GRPC_CLOSURE_INIT(&chand->max_idle_timer_cb, max_idle_timer_cb, chand, |
||||
grpc_schedule_on_exec_ctx); |
||||
GRPC_CLOSURE_INIT(&chand->close_max_age_channel, close_max_age_channel, chand, |
||||
grpc_schedule_on_exec_ctx); |
||||
GRPC_CLOSURE_INIT(&chand->force_close_max_age_channel, |
||||
force_close_max_age_channel, chand, |
||||
grpc_schedule_on_exec_ctx); |
||||
GRPC_CLOSURE_INIT(&chand->start_max_idle_timer_after_init, |
||||
start_max_idle_timer_after_init, chand, |
||||
grpc_schedule_on_exec_ctx); |
||||
GRPC_CLOSURE_INIT(&chand->start_max_age_timer_after_init, |
||||
start_max_age_timer_after_init, chand, |
||||
grpc_schedule_on_exec_ctx); |
||||
GRPC_CLOSURE_INIT(&chand->start_max_age_grace_timer_after_goaway_op, |
||||
start_max_age_grace_timer_after_goaway_op, chand, |
||||
grpc_schedule_on_exec_ctx); |
||||
|
||||
if (chand->max_connection_age != GRPC_MILLIS_INF_FUTURE) { |
||||
/* When the channel reaches its max age, we send down an op with
|
||||
goaway_error set. However, we can't send down any ops until after the |
||||
channel stack is fully initialized. If we start the timer here, we have |
||||
no guarantee that the timer won't pop before channel stack initialization |
||||
is finished. To avoid that problem, we create a closure to start the |
||||
timer, and we schedule that closure to be run after call stack |
||||
initialization is done. */ |
||||
GRPC_CHANNEL_STACK_REF(chand->channel_stack, |
||||
"max_age start_max_age_timer_after_init"); |
||||
grpc_core::ExecCtx::Run(DEBUG_LOCATION, |
||||
&chand->start_max_age_timer_after_init, |
||||
GRPC_ERROR_NONE); |
||||
} |
||||
|
||||
/* Initialize the number of calls as 1, so that the max_idle_timer will not
|
||||
start until start_max_idle_timer_after_init is invoked. */ |
||||
gpr_atm_rel_store(&chand->call_count, 1); |
||||
if (chand->max_connection_idle != GRPC_MILLIS_INF_FUTURE) { |
||||
GRPC_CHANNEL_STACK_REF(chand->channel_stack, |
||||
"max_age start_max_idle_timer_after_init"); |
||||
grpc_core::ExecCtx::Run(DEBUG_LOCATION, |
||||
&chand->start_max_idle_timer_after_init, |
||||
GRPC_ERROR_NONE); |
||||
} |
||||
return GRPC_ERROR_NONE; |
||||
} |
||||
|
||||
/* Destructor for channel_data. */ |
||||
static void max_age_destroy_channel_elem(grpc_channel_element* elem) { |
||||
channel_data* chand = static_cast<channel_data*>(elem->channel_data); |
||||
chand->~channel_data(); |
||||
} |
||||
|
||||
const grpc_channel_filter grpc_max_age_filter = { |
||||
grpc_call_next_op, |
||||
nullptr, |
||||
grpc_channel_next_op, |
||||
0, /* sizeof_call_data */ |
||||
max_age_init_call_elem, |
||||
grpc_call_stack_ignore_set_pollset_or_pollset_set, |
||||
max_age_destroy_call_elem, |
||||
sizeof(channel_data), |
||||
max_age_init_channel_elem, |
||||
max_age_destroy_channel_elem, |
||||
grpc_channel_next_get_info, |
||||
"max_age"}; |
||||
|
||||
namespace grpc_core { |
||||
void RegisterMaxAgeFilter(CoreConfiguration::Builder* builder) { |
||||
builder->channel_init()->RegisterStage( |
||||
GRPC_SERVER_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, |
||||
[](ChannelStackBuilder* builder) { |
||||
const grpc_channel_args* channel_args = builder->channel_args(); |
||||
bool enable = grpc_channel_arg_get_integer( |
||||
grpc_channel_args_find( |
||||
channel_args, GRPC_ARG_MAX_CONNECTION_AGE_MS), |
||||
MAX_CONNECTION_AGE_INTEGER_OPTIONS) != INT_MAX || |
||||
grpc_channel_arg_get_integer( |
||||
grpc_channel_args_find( |
||||
channel_args, GRPC_ARG_MAX_CONNECTION_IDLE_MS), |
||||
MAX_CONNECTION_IDLE_INTEGER_OPTIONS) != INT_MAX; |
||||
if (enable) { |
||||
builder->PrependFilter(&grpc_max_age_filter, nullptr); |
||||
} |
||||
return true; |
||||
}); |
||||
} |
||||
} // namespace grpc_core
|
||||
@ -1,26 +0,0 @@ |
||||
//
|
||||
// Copyright 2017 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_CORE_EXT_FILTERS_MAX_AGE_MAX_AGE_FILTER_H |
||||
#define GRPC_CORE_EXT_FILTERS_MAX_AGE_MAX_AGE_FILTER_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/lib/channel/channel_stack.h" |
||||
|
||||
extern const grpc_channel_filter grpc_max_age_filter; |
||||
|
||||
#endif /* GRPC_CORE_EXT_FILTERS_MAX_AGE_MAX_AGE_FILTER_H */ |
||||
Loading…
Reference in new issue