mirror of https://github.com/grpc/grpc.git
Reland: [promises] Compression filter conversion (#31686)
* Revert "Revert "[promises] Compression filter conversion (#31204)" (#31682)"
This reverts commit fa31b36cb1.
* fix?
* fix
pull/31688/head^2
Craig Tiller
2 years ago
committed by
GitHub
parent
cf666c4c20
commit
c545350633
45 changed files with 890 additions and 1430 deletions
@ -0,0 +1,315 @@ |
||||
// 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.
|
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/ext/filters/http/message_compress/compression_filter.h" |
||||
|
||||
#include <inttypes.h> |
||||
|
||||
#include <cstdint> |
||||
#include <functional> |
||||
#include <memory> |
||||
#include <utility> |
||||
|
||||
#include "absl/meta/type_traits.h" |
||||
#include "absl/status/status.h" |
||||
#include "absl/strings/str_cat.h" |
||||
#include "absl/strings/str_format.h" |
||||
#include "absl/types/optional.h" |
||||
|
||||
#include <grpc/compression.h> |
||||
#include <grpc/impl/codegen/compression_types.h> |
||||
#include <grpc/impl/codegen/grpc_types.h> |
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/core/ext/filters/message_size/message_size_filter.h" |
||||
#include "src/core/lib/channel/channel_args.h" |
||||
#include "src/core/lib/channel/channel_stack.h" |
||||
#include "src/core/lib/channel/context.h" |
||||
#include "src/core/lib/channel/promise_based_filter.h" |
||||
#include "src/core/lib/compression/compression_internal.h" |
||||
#include "src/core/lib/compression/message_compress.h" |
||||
#include "src/core/lib/debug/trace.h" |
||||
#include "src/core/lib/promise/context.h" |
||||
#include "src/core/lib/promise/for_each.h" |
||||
#include "src/core/lib/promise/latch.h" |
||||
#include "src/core/lib/promise/map_pipe.h" |
||||
#include "src/core/lib/promise/promise.h" |
||||
#include "src/core/lib/promise/seq.h" |
||||
#include "src/core/lib/promise/try_concurrently.h" |
||||
#include "src/core/lib/promise/try_seq.h" |
||||
#include "src/core/lib/resource_quota/arena.h" |
||||
#include "src/core/lib/slice/slice_buffer.h" |
||||
#include "src/core/lib/surface/call.h" |
||||
#include "src/core/lib/transport/metadata_batch.h" |
||||
#include "src/core/lib/transport/transport.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
const grpc_channel_filter ClientCompressionFilter::kFilter = |
||||
MakePromiseBasedFilter<ClientCompressionFilter, FilterEndpoint::kClient, |
||||
kFilterExaminesServerInitialMetadata | |
||||
kFilterExaminesInboundMessages | |
||||
kFilterExaminesOutboundMessages>("compression"); |
||||
const grpc_channel_filter ServerCompressionFilter::kFilter = |
||||
MakePromiseBasedFilter<ServerCompressionFilter, FilterEndpoint::kServer, |
||||
kFilterExaminesServerInitialMetadata | |
||||
kFilterExaminesInboundMessages | |
||||
kFilterExaminesOutboundMessages>("compression"); |
||||
|
||||
absl::StatusOr<ClientCompressionFilter> ClientCompressionFilter::Create( |
||||
const ChannelArgs& args, ChannelFilter::Args) { |
||||
return ClientCompressionFilter(args); |
||||
} |
||||
|
||||
absl::StatusOr<ServerCompressionFilter> ServerCompressionFilter::Create( |
||||
const ChannelArgs& args, ChannelFilter::Args) { |
||||
return ServerCompressionFilter(args); |
||||
} |
||||
|
||||
CompressionFilter::CompressionFilter(const ChannelArgs& args) |
||||
: max_recv_size_(GetMaxRecvSizeFromChannelArgs(args)), |
||||
message_size_service_config_parser_index_( |
||||
MessageSizeParser::ParserIndex()), |
||||
default_compression_algorithm_( |
||||
DefaultCompressionAlgorithmFromChannelArgs(args).value_or( |
||||
GRPC_COMPRESS_NONE)), |
||||
enabled_compression_algorithms_( |
||||
CompressionAlgorithmSet::FromChannelArgs(args)), |
||||
enable_compression_( |
||||
args.GetBool(GRPC_ARG_ENABLE_PER_MESSAGE_COMPRESSION).value_or(true)), |
||||
enable_decompression_( |
||||
args.GetBool(GRPC_ARG_ENABLE_PER_MESSAGE_DECOMPRESSION) |
||||
.value_or(true)) { |
||||
// Make sure the default is enabled.
|
||||
if (!enabled_compression_algorithms_.IsSet(default_compression_algorithm_)) { |
||||
const char* name; |
||||
if (!grpc_compression_algorithm_name(default_compression_algorithm_, |
||||
&name)) { |
||||
name = "<unknown>"; |
||||
} |
||||
gpr_log(GPR_ERROR, |
||||
"default compression algorithm %s not enabled: switching to none", |
||||
name); |
||||
default_compression_algorithm_ = GRPC_COMPRESS_NONE; |
||||
} |
||||
} |
||||
|
||||
MessageHandle CompressionFilter::CompressMessage( |
||||
MessageHandle message, grpc_compression_algorithm algorithm) const { |
||||
if (GRPC_TRACE_FLAG_ENABLED(grpc_compression_trace)) { |
||||
gpr_log(GPR_ERROR, "CompressMessage: len=%" PRIdPTR " alg=%d flags=%d", |
||||
message->payload()->Length(), algorithm, message->flags()); |
||||
} |
||||
// Check if we're allowed to compress this message
|
||||
// (apps might want to disable compression for certain messages to avoid
|
||||
// crime/beast like vulns).
|
||||
uint32_t& flags = message->mutable_flags(); |
||||
if (algorithm == GRPC_COMPRESS_NONE || !enable_compression_ || |
||||
(flags & (GRPC_WRITE_NO_COMPRESS | GRPC_WRITE_INTERNAL_COMPRESS))) { |
||||
return message; |
||||
} |
||||
// Try to compress the payload.
|
||||
SliceBuffer tmp; |
||||
SliceBuffer* payload = message->payload(); |
||||
bool did_compress = grpc_msg_compress(algorithm, payload->c_slice_buffer(), |
||||
tmp.c_slice_buffer()); |
||||
// If we achieved compression send it as compressed, otherwise send it as (to
|
||||
// avoid spending cycles on the receiver decompressing).
|
||||
if (did_compress) { |
||||
if (GRPC_TRACE_FLAG_ENABLED(grpc_compression_trace)) { |
||||
const char* algo_name; |
||||
const size_t before_size = payload->Length(); |
||||
const size_t after_size = tmp.Length(); |
||||
const float savings_ratio = 1.0f - static_cast<float>(after_size) / |
||||
static_cast<float>(before_size); |
||||
GPR_ASSERT(grpc_compression_algorithm_name(algorithm, &algo_name)); |
||||
gpr_log(GPR_INFO, |
||||
"Compressed[%s] %" PRIuPTR " bytes vs. %" PRIuPTR |
||||
" bytes (%.2f%% savings)", |
||||
algo_name, before_size, after_size, 100 * savings_ratio); |
||||
} |
||||
tmp.Swap(payload); |
||||
flags |= GRPC_WRITE_INTERNAL_COMPRESS; |
||||
} else { |
||||
if (GRPC_TRACE_FLAG_ENABLED(grpc_compression_trace)) { |
||||
const char* algo_name; |
||||
GPR_ASSERT(grpc_compression_algorithm_name(algorithm, &algo_name)); |
||||
gpr_log(GPR_INFO, |
||||
"Algorithm '%s' enabled but decided not to compress. Input size: " |
||||
"%" PRIuPTR, |
||||
algo_name, payload->Length()); |
||||
} |
||||
} |
||||
return message; |
||||
} |
||||
|
||||
absl::StatusOr<MessageHandle> CompressionFilter::DecompressMessage( |
||||
MessageHandle message, grpc_compression_algorithm algorithm, |
||||
absl::optional<uint32_t> max_recv_message_length) const { |
||||
if (GRPC_TRACE_FLAG_ENABLED(grpc_compression_trace)) { |
||||
gpr_log(GPR_ERROR, "DecompressMessage: len=%" PRIdPTR " max=%d alg=%d", |
||||
message->payload()->Length(), max_recv_message_length.value_or(-1), |
||||
algorithm); |
||||
} |
||||
// Check max message length.
|
||||
if (max_recv_message_length.has_value() && |
||||
message->payload()->Length() > |
||||
static_cast<size_t>(*max_recv_message_length)) { |
||||
return absl::ResourceExhaustedError(absl::StrFormat( |
||||
"Received message larger than max (%u vs. %d)", |
||||
message->payload()->Length(), *max_recv_message_length)); |
||||
} |
||||
// Check if decompression is enabled (if not, we can just pass the message
|
||||
// up).
|
||||
if (!enable_decompression_ || |
||||
(message->flags() & GRPC_WRITE_INTERNAL_COMPRESS) == 0) { |
||||
return std::move(message); |
||||
} |
||||
// Try to decompress the payload.
|
||||
SliceBuffer decompressed_slices; |
||||
if (grpc_msg_decompress(algorithm, message->payload()->c_slice_buffer(), |
||||
decompressed_slices.c_slice_buffer()) == 0) { |
||||
return absl::InternalError( |
||||
absl::StrCat("Unexpected error decompressing data for algorithm ", |
||||
CompressionAlgorithmAsString(algorithm))); |
||||
} |
||||
// Swap the decompressed slices into the message.
|
||||
message->payload()->Swap(&decompressed_slices); |
||||
message->mutable_flags() &= ~GRPC_WRITE_INTERNAL_COMPRESS; |
||||
message->mutable_flags() |= GRPC_WRITE_INTERNAL_TEST_ONLY_WAS_COMPRESSED; |
||||
return std::move(message); |
||||
} |
||||
|
||||
class CompressionFilter::DecompressLoop { |
||||
public: |
||||
explicit DecompressLoop(CompressionFilter* filter, CallArgs& call_args) |
||||
: filter_(filter), |
||||
mapper_(PipeMapper<MessageHandle>::Intercept( |
||||
*call_args.incoming_messages)) {} |
||||
|
||||
// Once we have a compression algorithm we can construct the decompression
|
||||
// loop.
|
||||
// Returns a promise that resolves to MessageHandle.
|
||||
auto TakeAndRun(grpc_compression_algorithm algorithm) { |
||||
// Configure max receive size.
|
||||
auto max_recv_message_length = filter_->max_recv_size_; |
||||
const MessageSizeParsedConfig* limits = |
||||
MessageSizeParsedConfig::GetFromCallContext( |
||||
GetContext<grpc_call_context_element>(), |
||||
filter_->message_size_service_config_parser_index_); |
||||
if (limits != nullptr && limits->max_recv_size().has_value() && |
||||
(!max_recv_message_length.has_value() || |
||||
*limits->max_recv_size() < *max_recv_message_length)) { |
||||
max_recv_message_length = *limits->max_recv_size(); |
||||
} |
||||
// Interject decompression into the message loop.
|
||||
return mapper_.TakeAndRun([algorithm, max_recv_message_length, |
||||
filter = filter_](MessageHandle message) { |
||||
return filter->DecompressMessage(std::move(message), algorithm, |
||||
max_recv_message_length); |
||||
}); |
||||
} |
||||
|
||||
private: |
||||
CompressionFilter* filter_; |
||||
PipeMapper<MessageHandle> mapper_; |
||||
}; |
||||
|
||||
class CompressionFilter::CompressLoop { |
||||
public: |
||||
explicit CompressLoop(CompressionFilter* filter, CallArgs& call_args) |
||||
: filter_(filter), |
||||
mapper_(PipeMapper<MessageHandle>::Intercept( |
||||
*call_args.outgoing_messages)) {} |
||||
|
||||
// Once we're ready to send initial metadata we can construct the compression
|
||||
// loop.
|
||||
// Returns a promise that resolves to MessageHandle.
|
||||
auto TakeAndRun(grpc_metadata_batch& outgoing_metadata) { |
||||
const auto algorithm = |
||||
outgoing_metadata.Take(GrpcInternalEncodingRequest()) |
||||
.value_or(filter_->default_compression_algorithm()); |
||||
// Convey supported compression algorithms.
|
||||
outgoing_metadata.Set(GrpcAcceptEncodingMetadata(), |
||||
filter_->enabled_compression_algorithms()); |
||||
if (algorithm != GRPC_COMPRESS_NONE) { |
||||
outgoing_metadata.Set(GrpcEncodingMetadata(), algorithm); |
||||
} |
||||
// Interject compression into the message loop.
|
||||
return mapper_.TakeAndRun([filter = filter_, algorithm](MessageHandle m) { |
||||
return filter->CompressMessage(std::move(m), algorithm); |
||||
}); |
||||
} |
||||
|
||||
private: |
||||
CompressionFilter* filter_; |
||||
PipeMapper<MessageHandle> mapper_; |
||||
}; |
||||
|
||||
ArenaPromise<ServerMetadataHandle> ClientCompressionFilter::MakeCallPromise( |
||||
CallArgs call_args, NextPromiseFactory next_promise_factory) { |
||||
auto compress_loop = CompressLoop(this, call_args) |
||||
.TakeAndRun(*call_args.client_initial_metadata); |
||||
DecompressLoop decompress_loop(this, call_args); |
||||
auto* server_initial_metadata = call_args.server_initial_metadata; |
||||
// Concurrently:
|
||||
// - call the next filter
|
||||
// - wait for initial metadata from the server and then commence decompression
|
||||
// - compress outgoing messages
|
||||
return TryConcurrently(next_promise_factory(std::move(call_args))) |
||||
.NecessaryPull(Seq(server_initial_metadata->Wait(), |
||||
[decompress_loop = std::move(decompress_loop)]( |
||||
ServerMetadata** server_initial_metadata) mutable |
||||
-> ArenaPromise<absl::Status> { |
||||
if (*server_initial_metadata == nullptr) { |
||||
return ImmediateOkStatus(); |
||||
} |
||||
return decompress_loop.TakeAndRun( |
||||
(*server_initial_metadata) |
||||
->get(GrpcEncodingMetadata()) |
||||
.value_or(GRPC_COMPRESS_NONE)); |
||||
})) |
||||
.Push(std::move(compress_loop)); |
||||
} |
||||
|
||||
ArenaPromise<ServerMetadataHandle> ServerCompressionFilter::MakeCallPromise( |
||||
CallArgs call_args, NextPromiseFactory next_promise_factory) { |
||||
CompressLoop compress_loop(this, call_args); |
||||
auto decompress_loop = DecompressLoop(this, call_args) |
||||
.TakeAndRun(call_args.client_initial_metadata |
||||
->get(GrpcEncodingMetadata()) |
||||
.value_or(GRPC_COMPRESS_NONE)); |
||||
auto* read_latch = GetContext<Arena>()->New<Latch<ServerMetadata*>>(); |
||||
auto* write_latch = |
||||
std::exchange(call_args.server_initial_metadata, read_latch); |
||||
// Concurrently:
|
||||
// - call the next filter
|
||||
// - decompress incoming messages
|
||||
// - wait for initial metadata to be sent, and then commence compression of
|
||||
// outgoing messages
|
||||
return TryConcurrently(next_promise_factory(std::move(call_args))) |
||||
.Pull(std::move(decompress_loop)) |
||||
.Push(Seq(read_latch->Wait(), |
||||
[write_latch, compress_loop = std::move(compress_loop)]( |
||||
ServerMetadata** md) mutable { |
||||
// Find the compression algorithm.
|
||||
auto loop = compress_loop.TakeAndRun(**md); |
||||
write_latch->Set(*md); |
||||
return loop; |
||||
})); |
||||
} |
||||
|
||||
} // namespace grpc_core
|
||||
@ -0,0 +1,133 @@ |
||||
//
|
||||
//
|
||||
// Copyright 2020 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_HTTP_MESSAGE_COMPRESS_COMPRESSION_FILTER_H |
||||
#define GRPC_CORE_EXT_FILTERS_HTTP_MESSAGE_COMPRESS_COMPRESSION_FILTER_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <stddef.h> |
||||
#include <stdint.h> |
||||
|
||||
#include "absl/status/statusor.h" |
||||
#include "absl/types/optional.h" |
||||
|
||||
#include <grpc/impl/codegen/compression_types.h> |
||||
|
||||
#include "src/core/lib/channel/channel_args.h" |
||||
#include "src/core/lib/channel/channel_fwd.h" |
||||
#include "src/core/lib/channel/promise_based_filter.h" |
||||
#include "src/core/lib/compression/compression_internal.h" |
||||
#include "src/core/lib/promise/arena_promise.h" |
||||
#include "src/core/lib/transport/transport.h" |
||||
|
||||
namespace grpc_core { |
||||
|
||||
/** Compression filter for messages.
|
||||
* |
||||
* See <grpc/compression.h> for the available compression settings. |
||||
* |
||||
* Compression settings may come from: |
||||
* - Channel configuration, as established at channel creation time. |
||||
* - The metadata accompanying the outgoing data to be compressed. This is |
||||
* taken as a request only. We may choose not to honor it. The metadata key |
||||
* is given by \a GRPC_COMPRESSION_REQUEST_ALGORITHM_MD_KEY. |
||||
* |
||||
* Compression can be disabled for concrete messages (for instance in order to |
||||
* prevent CRIME/BEAST type attacks) by having the GRPC_WRITE_NO_COMPRESS set in |
||||
* the MessageHandle flags. |
||||
* |
||||
* The attempted compression mechanism is added to the resulting initial |
||||
* metadata under the 'grpc-encoding' key. |
||||
* |
||||
* If compression is actually performed, the MessageHandle's flag is modified to |
||||
* incorporate GRPC_WRITE_INTERNAL_COMPRESS. Otherwise, and regardless of the |
||||
* aforementioned 'grpc-encoding' metadata value, data will pass through |
||||
* uncompressed. */ |
||||
|
||||
class CompressionFilter : public ChannelFilter { |
||||
protected: |
||||
explicit CompressionFilter(const ChannelArgs& args); |
||||
|
||||
class CompressLoop; |
||||
class DecompressLoop; |
||||
|
||||
grpc_compression_algorithm default_compression_algorithm() const { |
||||
return default_compression_algorithm_; |
||||
} |
||||
|
||||
CompressionAlgorithmSet enabled_compression_algorithms() const { |
||||
return enabled_compression_algorithms_; |
||||
} |
||||
|
||||
private: |
||||
// Compress one message synchronously.
|
||||
MessageHandle CompressMessage(MessageHandle message, |
||||
grpc_compression_algorithm algorithm) const; |
||||
// Decompress one message synchronously.
|
||||
absl::StatusOr<MessageHandle> DecompressMessage( |
||||
MessageHandle message, grpc_compression_algorithm algorithm, |
||||
absl::optional<uint32_t> max_recv_message_length) const; |
||||
|
||||
// Max receive message length, if set.
|
||||
absl::optional<uint32_t> max_recv_size_; |
||||
size_t message_size_service_config_parser_index_; |
||||
// The default, channel-level, compression algorithm.
|
||||
grpc_compression_algorithm default_compression_algorithm_; |
||||
// Enabled compression algorithms.
|
||||
CompressionAlgorithmSet enabled_compression_algorithms_; |
||||
// Is compression enabled?
|
||||
bool enable_compression_; |
||||
// Is decompression enabled?
|
||||
bool enable_decompression_; |
||||
}; |
||||
|
||||
class ClientCompressionFilter final : public CompressionFilter { |
||||
public: |
||||
static const grpc_channel_filter kFilter; |
||||
|
||||
static absl::StatusOr<ClientCompressionFilter> Create( |
||||
const ChannelArgs& args, ChannelFilter::Args filter_args); |
||||
|
||||
// Construct a promise for one call.
|
||||
ArenaPromise<ServerMetadataHandle> MakeCallPromise( |
||||
CallArgs call_args, NextPromiseFactory next_promise_factory) override; |
||||
|
||||
private: |
||||
using CompressionFilter::CompressionFilter; |
||||
}; |
||||
|
||||
class ServerCompressionFilter final : public CompressionFilter { |
||||
public: |
||||
static const grpc_channel_filter kFilter; |
||||
|
||||
static absl::StatusOr<ServerCompressionFilter> Create( |
||||
const ChannelArgs& args, ChannelFilter::Args filter_args); |
||||
|
||||
// Construct a promise for one call.
|
||||
ArenaPromise<ServerMetadataHandle> MakeCallPromise( |
||||
CallArgs call_args, NextPromiseFactory next_promise_factory) override; |
||||
|
||||
private: |
||||
using CompressionFilter::CompressionFilter; |
||||
}; |
||||
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif /* GRPC_CORE_EXT_FILTERS_HTTP_MESSAGE_COMPRESS_COMPRESSION_FILTER_H \ |
||||
*/ |
||||
@ -1,332 +0,0 @@ |
||||
/*
|
||||
* |
||||
* 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. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/ext/filters/http/message_compress/message_compress_filter.h" |
||||
|
||||
#include <inttypes.h> |
||||
#include <stdlib.h> |
||||
|
||||
#include <new> |
||||
#include <utility> |
||||
|
||||
#include "absl/meta/type_traits.h" |
||||
#include "absl/status/status.h" |
||||
#include "absl/types/optional.h" |
||||
|
||||
#include <grpc/compression.h> |
||||
#include <grpc/impl/codegen/compression_types.h> |
||||
#include <grpc/impl/codegen/grpc_types.h> |
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/core/lib/compression/compression_internal.h" |
||||
#include "src/core/lib/compression/message_compress.h" |
||||
#include "src/core/lib/debug/trace.h" |
||||
#include "src/core/lib/iomgr/call_combiner.h" |
||||
#include "src/core/lib/iomgr/closure.h" |
||||
#include "src/core/lib/iomgr/error.h" |
||||
#include "src/core/lib/slice/slice_buffer.h" |
||||
#include "src/core/lib/surface/call.h" |
||||
#include "src/core/lib/transport/metadata_batch.h" |
||||
#include "src/core/lib/transport/transport.h" |
||||
|
||||
namespace { |
||||
|
||||
class ChannelData { |
||||
public: |
||||
explicit ChannelData(grpc_channel_element_args* args) { |
||||
// Get the enabled and the default algorithms from channel args.
|
||||
enabled_compression_algorithms_ = |
||||
grpc_core::CompressionAlgorithmSet::FromChannelArgs(args->channel_args); |
||||
default_compression_algorithm_ = |
||||
grpc_core::DefaultCompressionAlgorithmFromChannelArgs( |
||||
args->channel_args) |
||||
.value_or(GRPC_COMPRESS_NONE); |
||||
// Make sure the default is enabled.
|
||||
if (!enabled_compression_algorithms_.IsSet( |
||||
default_compression_algorithm_)) { |
||||
const char* name; |
||||
if (!grpc_compression_algorithm_name(default_compression_algorithm_, |
||||
&name)) { |
||||
name = "<unknown>"; |
||||
} |
||||
gpr_log(GPR_ERROR, |
||||
"default compression algorithm %s not enabled: switching to none", |
||||
name); |
||||
default_compression_algorithm_ = GRPC_COMPRESS_NONE; |
||||
} |
||||
GPR_ASSERT(!args->is_last); |
||||
} |
||||
|
||||
grpc_compression_algorithm default_compression_algorithm() const { |
||||
return default_compression_algorithm_; |
||||
} |
||||
|
||||
grpc_core::CompressionAlgorithmSet enabled_compression_algorithms() const { |
||||
return enabled_compression_algorithms_; |
||||
} |
||||
|
||||
private: |
||||
/** The default, channel-level, compression algorithm */ |
||||
grpc_compression_algorithm default_compression_algorithm_; |
||||
/** Enabled compression algorithms */ |
||||
grpc_core::CompressionAlgorithmSet enabled_compression_algorithms_; |
||||
}; |
||||
|
||||
class CallData { |
||||
public: |
||||
CallData(grpc_call_element* elem, const grpc_call_element_args& args) |
||||
: call_combiner_(args.call_combiner) { |
||||
ChannelData* channeld = static_cast<ChannelData*>(elem->channel_data); |
||||
// The call's message compression algorithm is set to channel's default
|
||||
// setting. It can be overridden later by initial metadata.
|
||||
if (GPR_LIKELY(channeld->enabled_compression_algorithms().IsSet( |
||||
channeld->default_compression_algorithm()))) { |
||||
compression_algorithm_ = channeld->default_compression_algorithm(); |
||||
} |
||||
GRPC_CLOSURE_INIT(&forward_send_message_batch_in_call_combiner_, |
||||
ForwardSendMessageBatch, elem, grpc_schedule_on_exec_ctx); |
||||
} |
||||
|
||||
~CallData() {} |
||||
|
||||
void CompressStartTransportStreamOpBatch( |
||||
grpc_call_element* elem, grpc_transport_stream_op_batch* batch); |
||||
|
||||
private: |
||||
bool SkipMessageCompression(); |
||||
void FinishSendMessage(grpc_call_element* elem); |
||||
|
||||
void ProcessSendInitialMetadata(grpc_call_element* elem, |
||||
grpc_metadata_batch* initial_metadata); |
||||
|
||||
// Methods for processing a send_message batch
|
||||
static void FailSendMessageBatchInCallCombiner(void* calld_arg, |
||||
grpc_error_handle error); |
||||
static void ForwardSendMessageBatch(void* elem_arg, grpc_error_handle unused); |
||||
|
||||
grpc_core::CallCombiner* call_combiner_; |
||||
grpc_compression_algorithm compression_algorithm_ = GRPC_COMPRESS_NONE; |
||||
grpc_error_handle cancel_error_; |
||||
grpc_transport_stream_op_batch* send_message_batch_ = nullptr; |
||||
bool seen_initial_metadata_ = false; |
||||
grpc_closure forward_send_message_batch_in_call_combiner_; |
||||
}; |
||||
|
||||
// Returns true if we should skip message compression for the current message.
|
||||
bool CallData::SkipMessageCompression() { |
||||
// If the flags of this message indicate that it shouldn't be compressed, we
|
||||
// skip message compression.
|
||||
uint32_t flags = send_message_batch_->payload->send_message.flags; |
||||
if (flags & (GRPC_WRITE_NO_COMPRESS | GRPC_WRITE_INTERNAL_COMPRESS)) { |
||||
return true; |
||||
} |
||||
// If this call doesn't have any message compression algorithm set, skip
|
||||
// message compression.
|
||||
return compression_algorithm_ == GRPC_COMPRESS_NONE; |
||||
} |
||||
|
||||
void CallData::ProcessSendInitialMetadata( |
||||
grpc_call_element* elem, grpc_metadata_batch* initial_metadata) { |
||||
ChannelData* channeld = static_cast<ChannelData*>(elem->channel_data); |
||||
// Find the compression algorithm.
|
||||
compression_algorithm_ = |
||||
initial_metadata->Take(grpc_core::GrpcInternalEncodingRequest()) |
||||
.value_or(channeld->default_compression_algorithm()); |
||||
switch (compression_algorithm_) { |
||||
case GRPC_COMPRESS_NONE: |
||||
break; |
||||
case GRPC_COMPRESS_DEFLATE: |
||||
case GRPC_COMPRESS_GZIP: |
||||
initial_metadata->Set(grpc_core::GrpcEncodingMetadata(), |
||||
compression_algorithm_); |
||||
break; |
||||
case GRPC_COMPRESS_ALGORITHMS_COUNT: |
||||
abort(); |
||||
} |
||||
// Convey supported compression algorithms.
|
||||
initial_metadata->Set(grpc_core::GrpcAcceptEncodingMetadata(), |
||||
channeld->enabled_compression_algorithms()); |
||||
} |
||||
|
||||
void CallData::FinishSendMessage(grpc_call_element* elem) { |
||||
// Compress the data if appropriate.
|
||||
if (!SkipMessageCompression()) { |
||||
grpc_core::SliceBuffer tmp; |
||||
uint32_t& send_flags = send_message_batch_->payload->send_message.flags; |
||||
grpc_core::SliceBuffer* payload = |
||||
send_message_batch_->payload->send_message.send_message; |
||||
bool did_compress = |
||||
grpc_msg_compress(compression_algorithm_, payload->c_slice_buffer(), |
||||
tmp.c_slice_buffer()); |
||||
if (did_compress) { |
||||
if (GRPC_TRACE_FLAG_ENABLED(grpc_compression_trace)) { |
||||
const char* algo_name; |
||||
const size_t before_size = payload->Length(); |
||||
const size_t after_size = tmp.Length(); |
||||
const float savings_ratio = 1.0f - static_cast<float>(after_size) / |
||||
static_cast<float>(before_size); |
||||
GPR_ASSERT(grpc_compression_algorithm_name(compression_algorithm_, |
||||
&algo_name)); |
||||
gpr_log(GPR_INFO, |
||||
"Compressed[%s] %" PRIuPTR " bytes vs. %" PRIuPTR |
||||
" bytes (%.2f%% savings)", |
||||
algo_name, before_size, after_size, 100 * savings_ratio); |
||||
} |
||||
tmp.Swap(payload); |
||||
send_flags |= GRPC_WRITE_INTERNAL_COMPRESS; |
||||
} else { |
||||
if (GRPC_TRACE_FLAG_ENABLED(grpc_compression_trace)) { |
||||
const char* algo_name; |
||||
GPR_ASSERT(grpc_compression_algorithm_name(compression_algorithm_, |
||||
&algo_name)); |
||||
gpr_log( |
||||
GPR_INFO, |
||||
"Algorithm '%s' enabled but decided not to compress. Input size: " |
||||
"%" PRIuPTR, |
||||
algo_name, payload->Length()); |
||||
} |
||||
} |
||||
} |
||||
grpc_call_next_op(elem, std::exchange(send_message_batch_, nullptr)); |
||||
} |
||||
|
||||
void CallData::FailSendMessageBatchInCallCombiner(void* calld_arg, |
||||
grpc_error_handle error) { |
||||
CallData* calld = static_cast<CallData*>(calld_arg); |
||||
if (calld->send_message_batch_ != nullptr) { |
||||
grpc_transport_stream_op_batch_finish_with_failure( |
||||
calld->send_message_batch_, error, calld->call_combiner_); |
||||
calld->send_message_batch_ = nullptr; |
||||
} |
||||
} |
||||
|
||||
void CallData::ForwardSendMessageBatch(void* elem_arg, |
||||
grpc_error_handle /*unused*/) { |
||||
grpc_call_element* elem = static_cast<grpc_call_element*>(elem_arg); |
||||
CallData* calld = static_cast<CallData*>(elem->call_data); |
||||
calld->FinishSendMessage(elem); |
||||
} |
||||
|
||||
void CallData::CompressStartTransportStreamOpBatch( |
||||
grpc_call_element* elem, grpc_transport_stream_op_batch* batch) { |
||||
// Handle cancel_stream.
|
||||
if (batch->cancel_stream) { |
||||
cancel_error_ = batch->payload->cancel_stream.cancel_error; |
||||
if (send_message_batch_ != nullptr) { |
||||
if (!seen_initial_metadata_) { |
||||
GRPC_CALL_COMBINER_START( |
||||
call_combiner_, |
||||
GRPC_CLOSURE_CREATE(FailSendMessageBatchInCallCombiner, this, |
||||
grpc_schedule_on_exec_ctx), |
||||
cancel_error_, "failing send_message op"); |
||||
} |
||||
} |
||||
} else if (!cancel_error_.ok()) { |
||||
grpc_transport_stream_op_batch_finish_with_failure(batch, cancel_error_, |
||||
call_combiner_); |
||||
return; |
||||
} |
||||
// Handle send_initial_metadata.
|
||||
if (batch->send_initial_metadata) { |
||||
GPR_ASSERT(!seen_initial_metadata_); |
||||
ProcessSendInitialMetadata( |
||||
elem, batch->payload->send_initial_metadata.send_initial_metadata); |
||||
seen_initial_metadata_ = true; |
||||
// If we had previously received a batch containing a send_message op,
|
||||
// handle it now. Note that we need to re-enter the call combiner
|
||||
// for this, since we can't send two batches down while holding the
|
||||
// call combiner, since the connected_channel filter (at the bottom of
|
||||
// the call stack) will release the call combiner for each batch it sees.
|
||||
if (send_message_batch_ != nullptr) { |
||||
GRPC_CALL_COMBINER_START( |
||||
call_combiner_, &forward_send_message_batch_in_call_combiner_, |
||||
absl::OkStatus(), |
||||
"starting send_message after send_initial_metadata"); |
||||
} |
||||
} |
||||
// Handle send_message.
|
||||
if (batch->send_message) { |
||||
GPR_ASSERT(send_message_batch_ == nullptr); |
||||
send_message_batch_ = batch; |
||||
// If we have not yet seen send_initial_metadata, then we have to
|
||||
// wait. We save the batch and then drop the call combiner, which we'll
|
||||
// have to pick up again later when we get send_initial_metadata.
|
||||
if (!seen_initial_metadata_) { |
||||
GRPC_CALL_COMBINER_STOP( |
||||
call_combiner_, "send_message batch pending send_initial_metadata"); |
||||
return; |
||||
} |
||||
FinishSendMessage(elem); |
||||
} else { |
||||
// Pass control down the stack.
|
||||
grpc_call_next_op(elem, batch); |
||||
} |
||||
} |
||||
|
||||
void CompressStartTransportStreamOpBatch( |
||||
grpc_call_element* elem, grpc_transport_stream_op_batch* batch) { |
||||
CallData* calld = static_cast<CallData*>(elem->call_data); |
||||
calld->CompressStartTransportStreamOpBatch(elem, batch); |
||||
} |
||||
|
||||
/* Constructor for call_data */ |
||||
grpc_error_handle CompressInitCallElem(grpc_call_element* elem, |
||||
const grpc_call_element_args* args) { |
||||
new (elem->call_data) CallData(elem, *args); |
||||
return absl::OkStatus(); |
||||
} |
||||
|
||||
/* Destructor for call_data */ |
||||
void CompressDestroyCallElem(grpc_call_element* elem, |
||||
const grpc_call_final_info* /*final_info*/, |
||||
grpc_closure* /*ignored*/) { |
||||
CallData* calld = static_cast<CallData*>(elem->call_data); |
||||
calld->~CallData(); |
||||
} |
||||
|
||||
/* Constructor for ChannelData */ |
||||
grpc_error_handle CompressInitChannelElem(grpc_channel_element* elem, |
||||
grpc_channel_element_args* args) { |
||||
new (elem->channel_data) ChannelData(args); |
||||
return absl::OkStatus(); |
||||
} |
||||
|
||||
/* Destructor for channel data */ |
||||
void CompressDestroyChannelElem(grpc_channel_element* elem) { |
||||
ChannelData* channeld = static_cast<ChannelData*>(elem->channel_data); |
||||
channeld->~ChannelData(); |
||||
} |
||||
|
||||
} // namespace
|
||||
|
||||
const grpc_channel_filter grpc_message_compress_filter = { |
||||
CompressStartTransportStreamOpBatch, |
||||
nullptr, |
||||
grpc_channel_next_op, |
||||
sizeof(CallData), |
||||
CompressInitCallElem, |
||||
grpc_call_stack_ignore_set_pollset_or_pollset_set, |
||||
CompressDestroyCallElem, |
||||
sizeof(ChannelData), |
||||
CompressInitChannelElem, |
||||
grpc_channel_stack_no_post_init, |
||||
CompressDestroyChannelElem, |
||||
grpc_channel_next_get_info, |
||||
"message_compress"}; |
||||
@ -1,52 +0,0 @@ |
||||
/*
|
||||
* |
||||
* 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_CORE_EXT_FILTERS_HTTP_MESSAGE_COMPRESS_MESSAGE_COMPRESS_FILTER_H |
||||
#define GRPC_CORE_EXT_FILTERS_HTTP_MESSAGE_COMPRESS_MESSAGE_COMPRESS_FILTER_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/lib/channel/channel_fwd.h" |
||||
#include "src/core/lib/channel/channel_stack.h" |
||||
|
||||
/** Compression filter for outgoing data.
|
||||
* |
||||
* See <grpc/compression.h> for the available compression settings. |
||||
* |
||||
* Compression settings may come from: |
||||
* - Channel configuration, as established at channel creation time. |
||||
* - The metadata accompanying the outgoing data to be compressed. This is |
||||
* taken as a request only. We may choose not to honor it. The metadata key |
||||
* is given by \a GRPC_COMPRESSION_REQUEST_ALGORITHM_MD_KEY. |
||||
* |
||||
* Compression can be disabled for concrete messages (for instance in order to |
||||
* prevent CRIME/BEAST type attacks) by having the GRPC_WRITE_NO_COMPRESS set in |
||||
* the BEGIN_MESSAGE flags. |
||||
* |
||||
* The attempted compression mechanism is added to the resulting initial |
||||
* metadata under the'grpc-encoding' key. |
||||
* |
||||
* If compression is actually performed, BEGIN_MESSAGE's flag is modified to |
||||
* incorporate GRPC_WRITE_INTERNAL_COMPRESS. Otherwise, and regardless of the |
||||
* aforementioned 'grpc-encoding' metadata value, data will pass through |
||||
* uncompressed. */ |
||||
|
||||
extern const grpc_channel_filter grpc_message_compress_filter; |
||||
|
||||
#endif /* GRPC_CORE_EXT_FILTERS_HTTP_MESSAGE_COMPRESS_MESSAGE_COMPRESS_FILTER_H \ |
||||
*/ |
||||
@ -1,322 +0,0 @@ |
||||
//
|
||||
//
|
||||
// Copyright 2020 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/http/message_compress/message_decompress_filter.h" |
||||
|
||||
#include <string.h> |
||||
|
||||
#include <cstdint> |
||||
#include <new> |
||||
|
||||
#include "absl/status/status.h" |
||||
#include "absl/strings/str_cat.h" |
||||
#include "absl/strings/str_format.h" |
||||
#include "absl/types/optional.h" |
||||
|
||||
#include <grpc/impl/codegen/compression_types.h> |
||||
#include <grpc/status.h> |
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/core/ext/filters/message_size/message_size_filter.h" |
||||
#include "src/core/lib/channel/channel_args.h" |
||||
#include "src/core/lib/compression/message_compress.h" |
||||
#include "src/core/lib/gprpp/debug_location.h" |
||||
#include "src/core/lib/gprpp/status_helper.h" |
||||
#include "src/core/lib/iomgr/call_combiner.h" |
||||
#include "src/core/lib/iomgr/closure.h" |
||||
#include "src/core/lib/iomgr/error.h" |
||||
#include "src/core/lib/slice/slice_buffer.h" |
||||
#include "src/core/lib/transport/metadata_batch.h" |
||||
#include "src/core/lib/transport/transport.h" |
||||
|
||||
namespace grpc_core { |
||||
namespace { |
||||
|
||||
class ChannelData { |
||||
public: |
||||
explicit ChannelData(const grpc_channel_element_args* args) |
||||
: max_recv_size_(GetMaxRecvSizeFromChannelArgs( |
||||
ChannelArgs::FromC(args->channel_args))), |
||||
message_size_service_config_parser_index_( |
||||
MessageSizeParser::ParserIndex()) {} |
||||
|
||||
absl::optional<uint32_t> max_recv_size() const { return max_recv_size_; } |
||||
size_t message_size_service_config_parser_index() const { |
||||
return message_size_service_config_parser_index_; |
||||
} |
||||
|
||||
private: |
||||
absl::optional<uint32_t> max_recv_size_; |
||||
const size_t message_size_service_config_parser_index_; |
||||
}; |
||||
|
||||
class CallData { |
||||
public: |
||||
CallData(const grpc_call_element_args& args, const ChannelData* chand) |
||||
: call_combiner_(args.call_combiner), |
||||
max_recv_message_length_(chand->max_recv_size()) { |
||||
// Initialize state for recv_initial_metadata_ready callback
|
||||
GRPC_CLOSURE_INIT(&on_recv_initial_metadata_ready_, |
||||
OnRecvInitialMetadataReady, this, |
||||
grpc_schedule_on_exec_ctx); |
||||
// Initialize state for recv_message_ready callback
|
||||
GRPC_CLOSURE_INIT(&on_recv_message_ready_, OnRecvMessageReady, this, |
||||
grpc_schedule_on_exec_ctx); |
||||
// Initialize state for recv_trailing_metadata_ready callback
|
||||
GRPC_CLOSURE_INIT(&on_recv_trailing_metadata_ready_, |
||||
OnRecvTrailingMetadataReady, this, |
||||
grpc_schedule_on_exec_ctx); |
||||
const MessageSizeParsedConfig* limits = |
||||
MessageSizeParsedConfig::GetFromCallContext( |
||||
args.context, chand->message_size_service_config_parser_index()); |
||||
if (limits != nullptr && limits->max_recv_size().has_value() && |
||||
(!max_recv_message_length_.has_value() || |
||||
*limits->max_recv_size() < *max_recv_message_length_)) { |
||||
max_recv_message_length_ = *limits->max_recv_size(); |
||||
} |
||||
} |
||||
|
||||
void DecompressStartTransportStreamOpBatch( |
||||
grpc_call_element* elem, grpc_transport_stream_op_batch* batch); |
||||
|
||||
private: |
||||
static void OnRecvInitialMetadataReady(void* arg, grpc_error_handle error); |
||||
|
||||
// Methods for processing a receive message event
|
||||
void MaybeResumeOnRecvMessageReady(); |
||||
static void OnRecvMessageReady(void* arg, grpc_error_handle error); |
||||
void ContinueRecvMessageReadyCallback(grpc_error_handle error); |
||||
|
||||
// Methods for processing a recv_trailing_metadata event
|
||||
void MaybeResumeOnRecvTrailingMetadataReady(); |
||||
static void OnRecvTrailingMetadataReady(void* arg, grpc_error_handle error); |
||||
|
||||
CallCombiner* call_combiner_; |
||||
// Overall error for the call
|
||||
grpc_error_handle error_; |
||||
// Fields for handling recv_initial_metadata_ready callback
|
||||
grpc_closure on_recv_initial_metadata_ready_; |
||||
grpc_closure* original_recv_initial_metadata_ready_ = nullptr; |
||||
grpc_metadata_batch* recv_initial_metadata_ = nullptr; |
||||
// Fields for handling recv_message_ready callback
|
||||
bool seen_recv_message_ready_ = false; |
||||
absl::optional<uint32_t> max_recv_message_length_; |
||||
grpc_compression_algorithm algorithm_ = GRPC_COMPRESS_NONE; |
||||
absl::optional<SliceBuffer>* recv_message_ = nullptr; |
||||
uint32_t* recv_message_flags_ = nullptr; |
||||
grpc_closure on_recv_message_ready_; |
||||
grpc_closure* original_recv_message_ready_ = nullptr; |
||||
// Fields for handling recv_trailing_metadata_ready callback
|
||||
bool seen_recv_trailing_metadata_ready_ = false; |
||||
grpc_closure on_recv_trailing_metadata_ready_; |
||||
grpc_closure* original_recv_trailing_metadata_ready_ = nullptr; |
||||
grpc_error_handle on_recv_trailing_metadata_ready_error_; |
||||
}; |
||||
|
||||
void CallData::OnRecvInitialMetadataReady(void* arg, grpc_error_handle error) { |
||||
CallData* calld = static_cast<CallData*>(arg); |
||||
if (error.ok()) { |
||||
calld->algorithm_ = |
||||
calld->recv_initial_metadata_->get(GrpcEncodingMetadata()) |
||||
.value_or(GRPC_COMPRESS_NONE); |
||||
} |
||||
calld->MaybeResumeOnRecvMessageReady(); |
||||
calld->MaybeResumeOnRecvTrailingMetadataReady(); |
||||
grpc_closure* closure = calld->original_recv_initial_metadata_ready_; |
||||
calld->original_recv_initial_metadata_ready_ = nullptr; |
||||
Closure::Run(DEBUG_LOCATION, closure, error); |
||||
} |
||||
|
||||
void CallData::MaybeResumeOnRecvMessageReady() { |
||||
if (seen_recv_message_ready_) { |
||||
seen_recv_message_ready_ = false; |
||||
GRPC_CALL_COMBINER_START(call_combiner_, &on_recv_message_ready_, |
||||
absl::OkStatus(), |
||||
"continue recv_message_ready callback"); |
||||
} |
||||
} |
||||
|
||||
void CallData::OnRecvMessageReady(void* arg, grpc_error_handle error) { |
||||
CallData* calld = static_cast<CallData*>(arg); |
||||
if (error.ok()) { |
||||
if (calld->original_recv_initial_metadata_ready_ != nullptr) { |
||||
calld->seen_recv_message_ready_ = true; |
||||
GRPC_CALL_COMBINER_STOP(calld->call_combiner_, |
||||
"Deferring OnRecvMessageReady until after " |
||||
"OnRecvInitialMetadataReady"); |
||||
return; |
||||
} |
||||
if (calld->algorithm_ != GRPC_COMPRESS_NONE) { |
||||
// recv_message can be NULL if trailing metadata is received instead of
|
||||
// message, or it's possible that the message was not compressed.
|
||||
if (!calld->recv_message_->has_value() || |
||||
(*calld->recv_message_)->Length() == 0 || |
||||
((*calld->recv_message_flags_ & GRPC_WRITE_INTERNAL_COMPRESS) == 0)) { |
||||
return calld->ContinueRecvMessageReadyCallback(absl::OkStatus()); |
||||
} |
||||
if (calld->max_recv_message_length_.has_value() && |
||||
(*calld->recv_message_)->Length() > |
||||
static_cast<uint32_t>(*calld->max_recv_message_length_)) { |
||||
GPR_DEBUG_ASSERT(calld->error_.ok()); |
||||
calld->error_ = grpc_error_set_int( |
||||
GRPC_ERROR_CREATE( |
||||
absl::StrFormat("Received message larger than max (%u vs. %d)", |
||||
(*calld->recv_message_)->Length(), |
||||
*calld->max_recv_message_length_)), |
||||
StatusIntProperty::kRpcStatus, GRPC_STATUS_RESOURCE_EXHAUSTED); |
||||
return calld->ContinueRecvMessageReadyCallback(calld->error_); |
||||
} |
||||
SliceBuffer decompressed_slices; |
||||
if (grpc_msg_decompress(calld->algorithm_, |
||||
(*calld->recv_message_)->c_slice_buffer(), |
||||
decompressed_slices.c_slice_buffer()) == 0) { |
||||
GPR_DEBUG_ASSERT(calld->error_.ok()); |
||||
calld->error_ = GRPC_ERROR_CREATE(absl::StrCat( |
||||
"Unexpected error decompressing data for algorithm with " |
||||
"enum value ", |
||||
calld->algorithm_)); |
||||
} else { |
||||
*calld->recv_message_flags_ = |
||||
(*calld->recv_message_flags_ & (~GRPC_WRITE_INTERNAL_COMPRESS)) | |
||||
GRPC_WRITE_INTERNAL_TEST_ONLY_WAS_COMPRESSED; |
||||
(*calld->recv_message_)->Swap(&decompressed_slices); |
||||
} |
||||
return calld->ContinueRecvMessageReadyCallback(calld->error_); |
||||
} |
||||
} |
||||
calld->ContinueRecvMessageReadyCallback(error); |
||||
} |
||||
|
||||
void CallData::ContinueRecvMessageReadyCallback(grpc_error_handle error) { |
||||
MaybeResumeOnRecvTrailingMetadataReady(); |
||||
// The surface will clean up the receiving stream if there is an error.
|
||||
grpc_closure* closure = original_recv_message_ready_; |
||||
original_recv_message_ready_ = nullptr; |
||||
Closure::Run(DEBUG_LOCATION, closure, error); |
||||
} |
||||
|
||||
void CallData::MaybeResumeOnRecvTrailingMetadataReady() { |
||||
if (seen_recv_trailing_metadata_ready_) { |
||||
seen_recv_trailing_metadata_ready_ = false; |
||||
grpc_error_handle error = on_recv_trailing_metadata_ready_error_; |
||||
on_recv_trailing_metadata_ready_error_ = absl::OkStatus(); |
||||
GRPC_CALL_COMBINER_START(call_combiner_, &on_recv_trailing_metadata_ready_, |
||||
error, "Continuing OnRecvTrailingMetadataReady"); |
||||
} |
||||
} |
||||
|
||||
void CallData::OnRecvTrailingMetadataReady(void* arg, grpc_error_handle error) { |
||||
CallData* calld = static_cast<CallData*>(arg); |
||||
if (calld->original_recv_initial_metadata_ready_ != nullptr || |
||||
calld->original_recv_message_ready_ != nullptr) { |
||||
calld->seen_recv_trailing_metadata_ready_ = true; |
||||
calld->on_recv_trailing_metadata_ready_error_ = error; |
||||
GRPC_CALL_COMBINER_STOP( |
||||
calld->call_combiner_, |
||||
"Deferring OnRecvTrailingMetadataReady until after " |
||||
"OnRecvInitialMetadataReady and OnRecvMessageReady"); |
||||
return; |
||||
} |
||||
error = grpc_error_add_child(error, calld->error_); |
||||
calld->error_ = absl::OkStatus(); |
||||
grpc_closure* closure = calld->original_recv_trailing_metadata_ready_; |
||||
calld->original_recv_trailing_metadata_ready_ = nullptr; |
||||
Closure::Run(DEBUG_LOCATION, closure, error); |
||||
} |
||||
|
||||
void CallData::DecompressStartTransportStreamOpBatch( |
||||
grpc_call_element* elem, grpc_transport_stream_op_batch* batch) { |
||||
// Handle recv_initial_metadata.
|
||||
if (batch->recv_initial_metadata) { |
||||
recv_initial_metadata_ = |
||||
batch->payload->recv_initial_metadata.recv_initial_metadata; |
||||
original_recv_initial_metadata_ready_ = |
||||
batch->payload->recv_initial_metadata.recv_initial_metadata_ready; |
||||
batch->payload->recv_initial_metadata.recv_initial_metadata_ready = |
||||
&on_recv_initial_metadata_ready_; |
||||
} |
||||
// Handle recv_message
|
||||
if (batch->recv_message) { |
||||
recv_message_ = batch->payload->recv_message.recv_message; |
||||
recv_message_flags_ = batch->payload->recv_message.flags; |
||||
original_recv_message_ready_ = |
||||
batch->payload->recv_message.recv_message_ready; |
||||
batch->payload->recv_message.recv_message_ready = &on_recv_message_ready_; |
||||
} |
||||
// Handle recv_trailing_metadata
|
||||
if (batch->recv_trailing_metadata) { |
||||
original_recv_trailing_metadata_ready_ = |
||||
batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready; |
||||
batch->payload->recv_trailing_metadata.recv_trailing_metadata_ready = |
||||
&on_recv_trailing_metadata_ready_; |
||||
} |
||||
// Pass control down the stack.
|
||||
grpc_call_next_op(elem, batch); |
||||
} |
||||
|
||||
void DecompressStartTransportStreamOpBatch( |
||||
grpc_call_element* elem, grpc_transport_stream_op_batch* batch) { |
||||
CallData* calld = static_cast<CallData*>(elem->call_data); |
||||
calld->DecompressStartTransportStreamOpBatch(elem, batch); |
||||
} |
||||
|
||||
grpc_error_handle DecompressInitCallElem(grpc_call_element* elem, |
||||
const grpc_call_element_args* args) { |
||||
ChannelData* chand = static_cast<ChannelData*>(elem->channel_data); |
||||
new (elem->call_data) CallData(*args, chand); |
||||
return absl::OkStatus(); |
||||
} |
||||
|
||||
void DecompressDestroyCallElem(grpc_call_element* elem, |
||||
const grpc_call_final_info* /*final_info*/, |
||||
grpc_closure* /*ignored*/) { |
||||
CallData* calld = static_cast<CallData*>(elem->call_data); |
||||
calld->~CallData(); |
||||
} |
||||
|
||||
grpc_error_handle DecompressInitChannelElem(grpc_channel_element* elem, |
||||
grpc_channel_element_args* args) { |
||||
ChannelData* chand = static_cast<ChannelData*>(elem->channel_data); |
||||
new (chand) ChannelData(args); |
||||
return absl::OkStatus(); |
||||
} |
||||
|
||||
void DecompressDestroyChannelElem(grpc_channel_element* elem) { |
||||
ChannelData* chand = static_cast<ChannelData*>(elem->channel_data); |
||||
chand->~ChannelData(); |
||||
} |
||||
|
||||
} // namespace
|
||||
|
||||
const grpc_channel_filter MessageDecompressFilter = { |
||||
DecompressStartTransportStreamOpBatch, |
||||
nullptr, |
||||
grpc_channel_next_op, |
||||
sizeof(CallData), |
||||
DecompressInitCallElem, |
||||
grpc_call_stack_ignore_set_pollset_or_pollset_set, |
||||
DecompressDestroyCallElem, |
||||
sizeof(ChannelData), |
||||
DecompressInitChannelElem, |
||||
grpc_channel_stack_no_post_init, |
||||
DecompressDestroyChannelElem, |
||||
grpc_channel_next_get_info, |
||||
"message_decompress"}; |
||||
} // namespace grpc_core
|
||||
@ -1,32 +0,0 @@ |
||||
//
|
||||
//
|
||||
// Copyright 2020 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_HTTP_MESSAGE_COMPRESS_MESSAGE_DECOMPRESS_FILTER_H |
||||
#define GRPC_CORE_EXT_FILTERS_HTTP_MESSAGE_COMPRESS_MESSAGE_DECOMPRESS_FILTER_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/lib/channel/channel_fwd.h" |
||||
#include "src/core/lib/channel/channel_stack.h" |
||||
|
||||
namespace grpc_core { |
||||
extern const grpc_channel_filter MessageDecompressFilter; |
||||
} // namespace grpc_core
|
||||
|
||||
#endif /* GRPC_CORE_EXT_FILTERS_HTTP_MESSAGE_COMPRESS_MESSAGE_DECOMPRESS_FILTER_H \ |
||||
*/ |
||||
@ -1,103 +0,0 @@ |
||||
// Copyright 2021 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 "test/core/compression/args_utils.h" |
||||
|
||||
#include <string.h> |
||||
|
||||
#include "absl/types/optional.h" |
||||
|
||||
#include <grpc/compression.h> |
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/core/lib/channel/channel_args.h" |
||||
#include "src/core/lib/compression/compression_internal.h" |
||||
#include "src/core/lib/gpr/useful.h" |
||||
|
||||
const grpc_channel_args* |
||||
grpc_channel_args_set_channel_default_compression_algorithm( |
||||
const grpc_channel_args* a, grpc_compression_algorithm algorithm) { |
||||
GPR_ASSERT(algorithm < GRPC_COMPRESS_ALGORITHMS_COUNT); |
||||
grpc_arg tmp; |
||||
tmp.type = GRPC_ARG_INTEGER; |
||||
tmp.key = const_cast<char*>(GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM); |
||||
tmp.value.integer = algorithm; |
||||
return grpc_channel_args_copy_and_add(a, &tmp, 1); |
||||
} |
||||
|
||||
/** Returns 1 if the argument for compression algorithm's enabled states bitset
|
||||
* was found in \a a, returning the arg's value in \a states. Otherwise, returns |
||||
* 0. */ |
||||
static int find_compression_algorithm_states_bitset(const grpc_channel_args* a, |
||||
int** states_arg) { |
||||
if (a != nullptr) { |
||||
size_t i; |
||||
for (i = 0; i < a->num_args; ++i) { |
||||
if (a->args[i].type == GRPC_ARG_INTEGER && |
||||
!strcmp(GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET, |
||||
a->args[i].key)) { |
||||
*states_arg = &a->args[i].value.integer; |
||||
**states_arg = |
||||
(**states_arg & ((1 << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1)) | |
||||
0x1; /* forcefully enable support for no compression */ |
||||
return 1; |
||||
} |
||||
} |
||||
} |
||||
return 0; /* GPR_FALSE */ |
||||
} |
||||
|
||||
const grpc_channel_args* grpc_channel_args_compression_algorithm_set_state( |
||||
const grpc_channel_args** a, grpc_compression_algorithm algorithm, |
||||
int state) { |
||||
int* states_arg = nullptr; |
||||
const grpc_channel_args* result = *a; |
||||
const int states_arg_found = |
||||
find_compression_algorithm_states_bitset(*a, &states_arg); |
||||
|
||||
if (grpc_core::DefaultCompressionAlgorithmFromChannelArgs(*a) == algorithm && |
||||
state == 0) { |
||||
const char* algo_name = nullptr; |
||||
GPR_ASSERT(grpc_compression_algorithm_name(algorithm, &algo_name) != 0); |
||||
gpr_log(GPR_ERROR, |
||||
"Tried to disable default compression algorithm '%s'. The " |
||||
"operation has been ignored.", |
||||
algo_name); |
||||
} else if (states_arg_found) { |
||||
if (state != 0) { |
||||
grpc_core::SetBit(reinterpret_cast<unsigned*>(states_arg), algorithm); |
||||
} else if (algorithm != GRPC_COMPRESS_NONE) { |
||||
grpc_core::ClearBit(reinterpret_cast<unsigned*>(states_arg), algorithm); |
||||
} |
||||
} else { |
||||
/* create a new arg */ |
||||
grpc_arg tmp; |
||||
tmp.type = GRPC_ARG_INTEGER; |
||||
tmp.key = |
||||
const_cast<char*>(GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET); |
||||
/* all enabled by default */ |
||||
tmp.value.integer = (1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1; |
||||
if (state != 0) { |
||||
grpc_core::SetBit(reinterpret_cast<unsigned*>(&tmp.value.integer), |
||||
algorithm); |
||||
} else if (algorithm != GRPC_COMPRESS_NONE) { |
||||
grpc_core::ClearBit(reinterpret_cast<unsigned*>(&tmp.value.integer), |
||||
algorithm); |
||||
} |
||||
result = grpc_channel_args_copy_and_add(*a, &tmp, 1); |
||||
grpc_channel_args_destroy(*a); |
||||
*a = result; |
||||
} |
||||
return result; |
||||
} |
||||
@ -1,34 +0,0 @@ |
||||
// Copyright 2021 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_TEST_CORE_COMPRESSION_ARGS_UTILS_H_H |
||||
#define GRPC_TEST_CORE_COMPRESSION_ARGS_UTILS_H_H |
||||
|
||||
#include <grpc/compression.h> |
||||
#include <grpc/grpc.h> |
||||
|
||||
// TODO(ctiller): when we do the channel args migration, just delete this.
|
||||
const grpc_channel_args* |
||||
grpc_channel_args_set_channel_default_compression_algorithm( |
||||
const grpc_channel_args* a, grpc_compression_algorithm algorithm); |
||||
|
||||
const grpc_channel_args* grpc_channel_args_compression_algorithm_set_state( |
||||
const grpc_channel_args** a, grpc_compression_algorithm algorithm, |
||||
int state); |
||||
|
||||
const grpc_channel_args* |
||||
grpc_channel_args_set_channel_default_compression_algorithm( |
||||
const grpc_channel_args* a, grpc_compression_algorithm algorithm); |
||||
|
||||
#endif |
||||
Loading…
Reference in new issue