|
|
|
|
@ -45,18 +45,30 @@ static void send_message_on_complete(void* arg, grpc_error* error); |
|
|
|
|
static void on_send_message_next_done(void* arg, grpc_error* error); |
|
|
|
|
|
|
|
|
|
namespace { |
|
|
|
|
enum initial_metadata_state { |
|
|
|
|
// Initial metadata not yet seen.
|
|
|
|
|
INITIAL_METADATA_UNSEEN = 0, |
|
|
|
|
// Initial metadata seen; compression algorithm set.
|
|
|
|
|
HAS_COMPRESSION_ALGORITHM, |
|
|
|
|
// Initial metadata seen; no compression algorithm set.
|
|
|
|
|
NO_COMPRESSION_ALGORITHM, |
|
|
|
|
|
|
|
|
|
struct channel_data { |
|
|
|
|
/** The default, channel-level, compression algorithm */ |
|
|
|
|
grpc_compression_algorithm default_compression_algorithm; |
|
|
|
|
/** Bitset of enabled compression algorithms */ |
|
|
|
|
uint32_t enabled_compression_algorithms_bitset; |
|
|
|
|
/** Bitset of enabled message compression algorithms */ |
|
|
|
|
uint32_t enabled_message_compression_algorithms_bitset; |
|
|
|
|
/** Bitset of enabled stream compression algorithms */ |
|
|
|
|
uint32_t enabled_stream_compression_algorithms_bitset; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
struct call_data { |
|
|
|
|
call_data(grpc_call_element* elem, const grpc_call_element_args& args) |
|
|
|
|
: call_combiner(args.call_combiner) { |
|
|
|
|
channel_data* channeld = static_cast<channel_data*>(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(GPR_BITGET(channeld->enabled_compression_algorithms_bitset, |
|
|
|
|
channeld->default_compression_algorithm))) { |
|
|
|
|
message_compression_algorithm = |
|
|
|
|
grpc_compression_algorithm_to_message_compression_algorithm( |
|
|
|
|
channeld->default_compression_algorithm); |
|
|
|
|
} |
|
|
|
|
GRPC_CLOSURE_INIT(&start_send_message_batch_in_call_combiner, |
|
|
|
|
start_send_message_batch, elem, |
|
|
|
|
grpc_schedule_on_exec_ctx); |
|
|
|
|
@ -73,15 +85,13 @@ struct call_data { |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_core::CallCombiner* call_combiner; |
|
|
|
|
grpc_linked_mdelem compression_algorithm_storage; |
|
|
|
|
grpc_linked_mdelem message_compression_algorithm_storage; |
|
|
|
|
grpc_linked_mdelem stream_compression_algorithm_storage; |
|
|
|
|
grpc_linked_mdelem accept_encoding_storage; |
|
|
|
|
grpc_linked_mdelem accept_stream_encoding_storage; |
|
|
|
|
/** Compression algorithm we'll try to use. It may be given by incoming
|
|
|
|
|
* metadata, or by the channel's default compression settings. */ |
|
|
|
|
grpc_message_compression_algorithm message_compression_algorithm = |
|
|
|
|
GRPC_MESSAGE_COMPRESS_NONE; |
|
|
|
|
initial_metadata_state send_initial_metadata_state = INITIAL_METADATA_UNSEEN; |
|
|
|
|
bool seen_initial_metadata = false; |
|
|
|
|
grpc_error* cancel_error = GRPC_ERROR_NONE; |
|
|
|
|
grpc_closure start_send_message_batch_in_call_combiner; |
|
|
|
|
grpc_transport_stream_op_batch* send_message_batch = nullptr; |
|
|
|
|
@ -93,130 +103,104 @@ struct call_data { |
|
|
|
|
grpc_closure on_send_message_next_done; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
struct channel_data { |
|
|
|
|
/** The default, channel-level, compression algorithm */ |
|
|
|
|
grpc_compression_algorithm default_compression_algorithm; |
|
|
|
|
/** Bitset of enabled compression algorithms */ |
|
|
|
|
uint32_t enabled_algorithms_bitset; |
|
|
|
|
/** Supported compression algorithms */ |
|
|
|
|
uint32_t supported_message_compression_algorithms; |
|
|
|
|
/** Supported stream compression algorithms */ |
|
|
|
|
uint32_t supported_stream_compression_algorithms; |
|
|
|
|
}; |
|
|
|
|
} // namespace
|
|
|
|
|
|
|
|
|
|
static bool skip_compression(grpc_call_element* elem, uint32_t flags, |
|
|
|
|
bool has_compression_algorithm) { |
|
|
|
|
// Returns true if we should skip message compression for the current message.
|
|
|
|
|
static bool skip_message_compression(grpc_call_element* elem) { |
|
|
|
|
call_data* calld = static_cast<call_data*>(elem->call_data); |
|
|
|
|
channel_data* channeld = static_cast<channel_data*>(elem->channel_data); |
|
|
|
|
|
|
|
|
|
// If the flags of this message indicate that it shouldn't be compressed, we
|
|
|
|
|
// skip message compression.
|
|
|
|
|
uint32_t flags = |
|
|
|
|
calld->send_message_batch->payload->send_message.send_message->flags(); |
|
|
|
|
if (flags & (GRPC_WRITE_NO_COMPRESS | GRPC_WRITE_INTERNAL_COMPRESS)) { |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
if (has_compression_algorithm) { |
|
|
|
|
if (calld->message_compression_algorithm == GRPC_MESSAGE_COMPRESS_NONE) { |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
return false; /* we have an actual call-specific algorithm */ |
|
|
|
|
// If this call doesn't have any message compression algorithm set, skip
|
|
|
|
|
// message compression.
|
|
|
|
|
return calld->message_compression_algorithm == GRPC_MESSAGE_COMPRESS_NONE; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Determines the compression algorithm from the initial metadata and the
|
|
|
|
|
// channel's default setting.
|
|
|
|
|
static grpc_compression_algorithm find_compression_algorithm( |
|
|
|
|
grpc_metadata_batch* initial_metadata, channel_data* channeld) { |
|
|
|
|
if (initial_metadata->idx.named.grpc_internal_encoding_request == nullptr) { |
|
|
|
|
return channeld->default_compression_algorithm; |
|
|
|
|
} |
|
|
|
|
/* no per-call compression override */ |
|
|
|
|
return channeld->default_compression_algorithm == GRPC_COMPRESS_NONE; |
|
|
|
|
grpc_compression_algorithm compression_algorithm; |
|
|
|
|
// Parse the compression algorithm from the initial metadata.
|
|
|
|
|
grpc_mdelem md = |
|
|
|
|
initial_metadata->idx.named.grpc_internal_encoding_request->md; |
|
|
|
|
GPR_ASSERT(grpc_compression_algorithm_parse(GRPC_MDVALUE(md), |
|
|
|
|
&compression_algorithm)); |
|
|
|
|
// Remove this metadata since it's an internal one (i.e., it won't be
|
|
|
|
|
// transmitted out).
|
|
|
|
|
grpc_metadata_batch_remove( |
|
|
|
|
initial_metadata, |
|
|
|
|
initial_metadata->idx.named.grpc_internal_encoding_request); |
|
|
|
|
// Check if that algorithm is enabled. Note that GRPC_COMPRESS_NONE is always
|
|
|
|
|
// enabled.
|
|
|
|
|
// TODO(juanlishen): Maybe use channel default or abort() if the algorithm
|
|
|
|
|
// from the initial metadata is disabled.
|
|
|
|
|
if (GPR_LIKELY(GPR_BITGET(channeld->enabled_compression_algorithms_bitset, |
|
|
|
|
compression_algorithm))) { |
|
|
|
|
return compression_algorithm; |
|
|
|
|
} |
|
|
|
|
const char* algorithm_name; |
|
|
|
|
GPR_ASSERT( |
|
|
|
|
grpc_compression_algorithm_name(compression_algorithm, &algorithm_name)); |
|
|
|
|
gpr_log(GPR_ERROR, |
|
|
|
|
"Invalid compression algorithm from initial metadata: '%s' " |
|
|
|
|
"(previously disabled). " |
|
|
|
|
"Will not compress.", |
|
|
|
|
algorithm_name); |
|
|
|
|
return GRPC_COMPRESS_NONE; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** Filter initial metadata */ |
|
|
|
|
static grpc_error* process_send_initial_metadata( |
|
|
|
|
grpc_call_element* elem, grpc_metadata_batch* initial_metadata, |
|
|
|
|
bool* has_compression_algorithm) GRPC_MUST_USE_RESULT; |
|
|
|
|
grpc_call_element* elem, |
|
|
|
|
grpc_metadata_batch* initial_metadata) GRPC_MUST_USE_RESULT; |
|
|
|
|
static grpc_error* process_send_initial_metadata( |
|
|
|
|
grpc_call_element* elem, grpc_metadata_batch* initial_metadata, |
|
|
|
|
bool* has_compression_algorithm) { |
|
|
|
|
grpc_call_element* elem, grpc_metadata_batch* initial_metadata) { |
|
|
|
|
call_data* calld = static_cast<call_data*>(elem->call_data); |
|
|
|
|
channel_data* channeld = static_cast<channel_data*>(elem->channel_data); |
|
|
|
|
*has_compression_algorithm = false; |
|
|
|
|
grpc_compression_algorithm compression_algorithm; |
|
|
|
|
// Find the compression algorithm.
|
|
|
|
|
grpc_compression_algorithm compression_algorithm = |
|
|
|
|
find_compression_algorithm(initial_metadata, channeld); |
|
|
|
|
// Note that at most one of the following algorithms can be set.
|
|
|
|
|
calld->message_compression_algorithm = |
|
|
|
|
grpc_compression_algorithm_to_message_compression_algorithm( |
|
|
|
|
compression_algorithm); |
|
|
|
|
grpc_stream_compression_algorithm stream_compression_algorithm = |
|
|
|
|
GRPC_STREAM_COMPRESS_NONE; |
|
|
|
|
if (initial_metadata->idx.named.grpc_internal_encoding_request != nullptr) { |
|
|
|
|
grpc_mdelem md = |
|
|
|
|
initial_metadata->idx.named.grpc_internal_encoding_request->md; |
|
|
|
|
if (GPR_UNLIKELY(!grpc_compression_algorithm_parse( |
|
|
|
|
GRPC_MDVALUE(md), &compression_algorithm))) { |
|
|
|
|
char* val = grpc_slice_to_c_string(GRPC_MDVALUE(md)); |
|
|
|
|
gpr_log(GPR_ERROR, |
|
|
|
|
"Invalid compression algorithm: '%s' (unknown). Ignoring.", val); |
|
|
|
|
gpr_free(val); |
|
|
|
|
calld->message_compression_algorithm = GRPC_MESSAGE_COMPRESS_NONE; |
|
|
|
|
stream_compression_algorithm = GRPC_STREAM_COMPRESS_NONE; |
|
|
|
|
} |
|
|
|
|
if (GPR_UNLIKELY(!GPR_BITGET(channeld->enabled_algorithms_bitset, |
|
|
|
|
compression_algorithm))) { |
|
|
|
|
char* val = grpc_slice_to_c_string(GRPC_MDVALUE(md)); |
|
|
|
|
gpr_log(GPR_ERROR, |
|
|
|
|
"Invalid compression algorithm: '%s' (previously disabled). " |
|
|
|
|
"Ignoring.", |
|
|
|
|
val); |
|
|
|
|
gpr_free(val); |
|
|
|
|
calld->message_compression_algorithm = GRPC_MESSAGE_COMPRESS_NONE; |
|
|
|
|
stream_compression_algorithm = GRPC_STREAM_COMPRESS_NONE; |
|
|
|
|
} |
|
|
|
|
*has_compression_algorithm = true; |
|
|
|
|
grpc_metadata_batch_remove( |
|
|
|
|
initial_metadata, |
|
|
|
|
initial_metadata->idx.named.grpc_internal_encoding_request); |
|
|
|
|
calld->message_compression_algorithm = |
|
|
|
|
grpc_compression_algorithm_to_message_compression_algorithm( |
|
|
|
|
compression_algorithm); |
|
|
|
|
stream_compression_algorithm = |
|
|
|
|
grpc_compression_algorithm_to_stream_compression_algorithm( |
|
|
|
|
compression_algorithm); |
|
|
|
|
} else { |
|
|
|
|
/* If no algorithm was found in the metadata and we aren't
|
|
|
|
|
* exceptionally skipping compression, fall back to the channel |
|
|
|
|
* default */ |
|
|
|
|
if (channeld->default_compression_algorithm != GRPC_COMPRESS_NONE) { |
|
|
|
|
calld->message_compression_algorithm = |
|
|
|
|
grpc_compression_algorithm_to_message_compression_algorithm( |
|
|
|
|
channeld->default_compression_algorithm); |
|
|
|
|
stream_compression_algorithm = |
|
|
|
|
grpc_compression_algorithm_to_stream_compression_algorithm( |
|
|
|
|
channeld->default_compression_algorithm); |
|
|
|
|
} |
|
|
|
|
*has_compression_algorithm = true; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_compression_algorithm_to_stream_compression_algorithm( |
|
|
|
|
compression_algorithm); |
|
|
|
|
// Hint compression algorithm.
|
|
|
|
|
grpc_error* error = GRPC_ERROR_NONE; |
|
|
|
|
/* hint compression algorithm */ |
|
|
|
|
if (stream_compression_algorithm != GRPC_STREAM_COMPRESS_NONE) { |
|
|
|
|
if (calld->message_compression_algorithm != GRPC_MESSAGE_COMPRESS_NONE) { |
|
|
|
|
error = grpc_metadata_batch_add_tail( |
|
|
|
|
initial_metadata, &calld->stream_compression_algorithm_storage, |
|
|
|
|
grpc_stream_compression_encoding_mdelem(stream_compression_algorithm)); |
|
|
|
|
} else if (calld->message_compression_algorithm != |
|
|
|
|
GRPC_MESSAGE_COMPRESS_NONE) { |
|
|
|
|
error = grpc_metadata_batch_add_tail( |
|
|
|
|
initial_metadata, &calld->compression_algorithm_storage, |
|
|
|
|
initial_metadata, &calld->message_compression_algorithm_storage, |
|
|
|
|
grpc_message_compression_encoding_mdelem( |
|
|
|
|
calld->message_compression_algorithm)); |
|
|
|
|
} else if (stream_compression_algorithm != GRPC_STREAM_COMPRESS_NONE) { |
|
|
|
|
error = grpc_metadata_batch_add_tail( |
|
|
|
|
initial_metadata, &calld->stream_compression_algorithm_storage, |
|
|
|
|
grpc_stream_compression_encoding_mdelem(stream_compression_algorithm)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if (error != GRPC_ERROR_NONE) return error; |
|
|
|
|
|
|
|
|
|
/* convey supported compression algorithms */ |
|
|
|
|
// Convey supported compression algorithms.
|
|
|
|
|
error = grpc_metadata_batch_add_tail( |
|
|
|
|
initial_metadata, &calld->accept_encoding_storage, |
|
|
|
|
GRPC_MDELEM_ACCEPT_ENCODING_FOR_ALGORITHMS( |
|
|
|
|
channeld->supported_message_compression_algorithms)); |
|
|
|
|
|
|
|
|
|
channeld->enabled_message_compression_algorithms_bitset)); |
|
|
|
|
if (error != GRPC_ERROR_NONE) return error; |
|
|
|
|
|
|
|
|
|
/* Do not overwrite accept-encoding header if it already presents (e.g. added
|
|
|
|
|
* by some proxy). */ |
|
|
|
|
// Do not overwrite accept-encoding header if it already presents (e.g., added
|
|
|
|
|
// by some proxy).
|
|
|
|
|
if (!initial_metadata->idx.named.accept_encoding) { |
|
|
|
|
error = grpc_metadata_batch_add_tail( |
|
|
|
|
initial_metadata, &calld->accept_stream_encoding_storage, |
|
|
|
|
GRPC_MDELEM_ACCEPT_STREAM_ENCODING_FOR_ALGORITHMS( |
|
|
|
|
channeld->supported_stream_compression_algorithms)); |
|
|
|
|
channeld->enabled_stream_compression_algorithms_bitset)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
return error; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -358,12 +342,7 @@ static void on_send_message_next_done(void* arg, grpc_error* error) { |
|
|
|
|
|
|
|
|
|
static void start_send_message_batch(void* arg, grpc_error* unused) { |
|
|
|
|
grpc_call_element* elem = static_cast<grpc_call_element*>(arg); |
|
|
|
|
call_data* calld = static_cast<call_data*>(elem->call_data); |
|
|
|
|
if (skip_compression( |
|
|
|
|
elem, |
|
|
|
|
calld->send_message_batch->payload->send_message.send_message |
|
|
|
|
->flags(), |
|
|
|
|
calld->send_initial_metadata_state == HAS_COMPRESSION_ALGORITHM)) { |
|
|
|
|
if (skip_message_compression(elem)) { |
|
|
|
|
send_message_batch_continue(elem); |
|
|
|
|
} else { |
|
|
|
|
continue_reading_send_message(elem); |
|
|
|
|
@ -380,7 +359,7 @@ static void compress_start_transport_stream_op_batch( |
|
|
|
|
calld->cancel_error = |
|
|
|
|
GRPC_ERROR_REF(batch->payload->cancel_stream.cancel_error); |
|
|
|
|
if (calld->send_message_batch != nullptr) { |
|
|
|
|
if (calld->send_initial_metadata_state == INITIAL_METADATA_UNSEEN) { |
|
|
|
|
if (!calld->seen_initial_metadata) { |
|
|
|
|
GRPC_CALL_COMBINER_START( |
|
|
|
|
calld->call_combiner, |
|
|
|
|
GRPC_CLOSURE_CREATE(fail_send_message_batch_in_call_combiner, calld, |
|
|
|
|
@ -398,19 +377,15 @@ static void compress_start_transport_stream_op_batch( |
|
|
|
|
} |
|
|
|
|
// Handle send_initial_metadata.
|
|
|
|
|
if (batch->send_initial_metadata) { |
|
|
|
|
GPR_ASSERT(calld->send_initial_metadata_state == INITIAL_METADATA_UNSEEN); |
|
|
|
|
bool has_compression_algorithm; |
|
|
|
|
GPR_ASSERT(!calld->seen_initial_metadata); |
|
|
|
|
grpc_error* error = process_send_initial_metadata( |
|
|
|
|
elem, batch->payload->send_initial_metadata.send_initial_metadata, |
|
|
|
|
&has_compression_algorithm); |
|
|
|
|
elem, batch->payload->send_initial_metadata.send_initial_metadata); |
|
|
|
|
if (error != GRPC_ERROR_NONE) { |
|
|
|
|
grpc_transport_stream_op_batch_finish_with_failure(batch, error, |
|
|
|
|
calld->call_combiner); |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
calld->send_initial_metadata_state = has_compression_algorithm |
|
|
|
|
? HAS_COMPRESSION_ALGORITHM |
|
|
|
|
: NO_COMPRESSION_ALGORITHM; |
|
|
|
|
calld->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
|
|
|
|
|
@ -431,7 +406,7 @@ static void compress_start_transport_stream_op_batch( |
|
|
|
|
// wait. We save the batch in calld and then drop the call
|
|
|
|
|
// combiner, which we'll have to pick up again later when we get
|
|
|
|
|
// send_initial_metadata.
|
|
|
|
|
if (calld->send_initial_metadata_state == INITIAL_METADATA_UNSEEN) { |
|
|
|
|
if (!calld->seen_initial_metadata) { |
|
|
|
|
GRPC_CALL_COMBINER_STOP( |
|
|
|
|
calld->call_combiner, |
|
|
|
|
"send_message batch pending send_initial_metadata"); |
|
|
|
|
@ -463,34 +438,29 @@ static void destroy_call_elem(grpc_call_element* elem, |
|
|
|
|
static grpc_error* init_channel_elem(grpc_channel_element* elem, |
|
|
|
|
grpc_channel_element_args* args) { |
|
|
|
|
channel_data* channeld = static_cast<channel_data*>(elem->channel_data); |
|
|
|
|
|
|
|
|
|
channeld->enabled_algorithms_bitset = |
|
|
|
|
// Get the enabled and the default algorithms from channel args.
|
|
|
|
|
channeld->enabled_compression_algorithms_bitset = |
|
|
|
|
grpc_channel_args_compression_algorithm_get_states(args->channel_args); |
|
|
|
|
channeld->default_compression_algorithm = |
|
|
|
|
grpc_channel_args_get_compression_algorithm(args->channel_args); |
|
|
|
|
|
|
|
|
|
/* Make sure the default isn't disabled. */ |
|
|
|
|
if (!GPR_BITGET(channeld->enabled_algorithms_bitset, |
|
|
|
|
grpc_channel_args_get_channel_default_compression_algorithm( |
|
|
|
|
args->channel_args); |
|
|
|
|
// Make sure the default is enabled.
|
|
|
|
|
if (!GPR_BITGET(channeld->enabled_compression_algorithms_bitset, |
|
|
|
|
channeld->default_compression_algorithm)) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"compression algorithm %d not enabled: switching to none", |
|
|
|
|
channeld->default_compression_algorithm); |
|
|
|
|
const char* name; |
|
|
|
|
GPR_ASSERT(grpc_compression_algorithm_name( |
|
|
|
|
channeld->default_compression_algorithm, &name) == 1); |
|
|
|
|
gpr_log(GPR_ERROR, |
|
|
|
|
"default compression algorithm %s not enabled: switching to none", |
|
|
|
|
name); |
|
|
|
|
channeld->default_compression_algorithm = GRPC_COMPRESS_NONE; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
uint32_t supported_compression_algorithms = |
|
|
|
|
(((1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1) & |
|
|
|
|
channeld->enabled_algorithms_bitset) | |
|
|
|
|
1u; |
|
|
|
|
|
|
|
|
|
channeld->supported_message_compression_algorithms = |
|
|
|
|
channeld->enabled_message_compression_algorithms_bitset = |
|
|
|
|
grpc_compression_bitset_to_message_bitset( |
|
|
|
|
supported_compression_algorithms); |
|
|
|
|
|
|
|
|
|
channeld->supported_stream_compression_algorithms = |
|
|
|
|
channeld->enabled_compression_algorithms_bitset); |
|
|
|
|
channeld->enabled_stream_compression_algorithms_bitset = |
|
|
|
|
grpc_compression_bitset_to_stream_bitset( |
|
|
|
|
supported_compression_algorithms); |
|
|
|
|
|
|
|
|
|
channeld->enabled_compression_algorithms_bitset); |
|
|
|
|
GPR_ASSERT(!args->is_last); |
|
|
|
|
return GRPC_ERROR_NONE; |
|
|
|
|
} |
|
|
|
|
|
Juanli Shen
6 years ago