mirror of https://github.com/grpc/grpc.git
Craig Tiller
10 years ago
commit
ad339108f1
48 changed files with 5855 additions and 1118 deletions
@ -0,0 +1,325 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#include <assert.h> |
||||
#include <string.h> |
||||
|
||||
#include <grpc/compression.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/slice_buffer.h> |
||||
|
||||
#include "src/core/channel/compress_filter.h" |
||||
#include "src/core/channel/channel_args.h" |
||||
#include "src/core/compression/message_compress.h" |
||||
|
||||
typedef struct call_data { |
||||
gpr_slice_buffer slices; /**< Buffers up input slices to be compressed */ |
||||
grpc_linked_mdelem compression_algorithm_storage; |
||||
int remaining_slice_bytes; /**< Input data to be read, as per BEGIN_MESSAGE */ |
||||
int written_initial_metadata; /**< Already processed initial md? */ |
||||
/** Compression algorithm we'll try to use. It may be given by incoming
|
||||
* metadata, or by the channel's default compression settings. */ |
||||
grpc_compression_algorithm compression_algorithm; |
||||
/** If true, contents of \a compression_algorithm are authoritative */ |
||||
int has_compression_algorithm; |
||||
} call_data; |
||||
|
||||
typedef struct channel_data { |
||||
/** Metadata key for the incoming (requested) compression algorithm */ |
||||
grpc_mdstr *mdstr_request_compression_algorithm_key; |
||||
/** Metadata key for the outgoing (used) compression algorithm */ |
||||
grpc_mdstr *mdstr_outgoing_compression_algorithm_key; |
||||
/** Precomputed metadata elements for all available compression algorithms */ |
||||
grpc_mdelem *mdelem_compression_algorithms[GRPC_COMPRESS_ALGORITHMS_COUNT]; |
||||
/** The default, channel-level, compression algorithm */ |
||||
grpc_compression_algorithm default_compression_algorithm; |
||||
} channel_data; |
||||
|
||||
/** Compress \a slices in place using \a algorithm. Returns 1 if compression did
|
||||
* actually happen, 0 otherwise (for example if the compressed output size was |
||||
* larger than the raw input). |
||||
* |
||||
* Returns 1 if the data was actually compress and 0 otherwise. */ |
||||
static int compress_send_sb(grpc_compression_algorithm algorithm, |
||||
gpr_slice_buffer *slices) { |
||||
int did_compress; |
||||
gpr_slice_buffer tmp; |
||||
gpr_slice_buffer_init(&tmp); |
||||
did_compress = grpc_msg_compress(algorithm, slices, &tmp); |
||||
if (did_compress) { |
||||
gpr_slice_buffer_swap(slices, &tmp); |
||||
} |
||||
gpr_slice_buffer_destroy(&tmp); |
||||
return did_compress; |
||||
} |
||||
|
||||
/** For each \a md element from the incoming metadata, filter out the entry for
|
||||
* "grpc-encoding", using its value to populate the call data's |
||||
* compression_algorithm field. */ |
||||
static grpc_mdelem* compression_md_filter(void *user_data, grpc_mdelem *md) { |
||||
grpc_call_element *elem = user_data; |
||||
call_data *calld = elem->call_data; |
||||
channel_data *channeld = elem->channel_data; |
||||
|
||||
if (md->key == channeld->mdstr_request_compression_algorithm_key) { |
||||
const char *md_c_str = grpc_mdstr_as_c_string(md->value); |
||||
if (!grpc_compression_algorithm_parse(md_c_str, |
||||
&calld->compression_algorithm)) { |
||||
gpr_log(GPR_ERROR, "Invalid compression algorithm: '%s'. Ignoring.", |
||||
md_c_str); |
||||
calld->compression_algorithm = GRPC_COMPRESS_NONE; |
||||
} |
||||
calld->has_compression_algorithm = 1; |
||||
return NULL; |
||||
} |
||||
|
||||
return md; |
||||
} |
||||
|
||||
static int skip_compression(channel_data *channeld, call_data *calld) { |
||||
if (calld->has_compression_algorithm) { |
||||
if (calld->compression_algorithm == GRPC_COMPRESS_NONE) { |
||||
return 1; |
||||
} |
||||
return 0; /* we have an actual call-specific algorithm */ |
||||
} |
||||
/* no per-call compression override */ |
||||
return channeld->default_compression_algorithm == GRPC_COMPRESS_NONE; |
||||
} |
||||
|
||||
/** Assembles a new grpc_stream_op_buffer with the compressed slices, modifying
|
||||
* the associated GRPC_OP_BEGIN_MESSAGE accordingly (new compressed length, |
||||
* flags indicating compression is in effect) and replaces \a send_ops with it. |
||||
* */ |
||||
static void finish_compressed_sopb(grpc_stream_op_buffer *send_ops, |
||||
grpc_call_element *elem) { |
||||
size_t i; |
||||
call_data *calld = elem->call_data; |
||||
int new_slices_added = 0; /* GPR_FALSE */ |
||||
grpc_metadata_batch metadata; |
||||
grpc_stream_op_buffer new_send_ops; |
||||
grpc_sopb_init(&new_send_ops); |
||||
|
||||
for (i = 0; i < send_ops->nops; i++) { |
||||
grpc_stream_op *sop = &send_ops->ops[i]; |
||||
switch (sop->type) { |
||||
case GRPC_OP_BEGIN_MESSAGE: |
||||
grpc_sopb_add_begin_message( |
||||
&new_send_ops, calld->slices.length, |
||||
sop->data.begin_message.flags | GRPC_WRITE_INTERNAL_COMPRESS); |
||||
break; |
||||
case GRPC_OP_SLICE: |
||||
/* Once we reach the slices section of the original buffer, simply add
|
||||
* all the new (compressed) slices. We obviously want to do this only |
||||
* once, hence the "new_slices_added" guard. */ |
||||
if (!new_slices_added) { |
||||
size_t j; |
||||
for (j = 0; j < calld->slices.count; ++j) { |
||||
grpc_sopb_add_slice(&new_send_ops, |
||||
gpr_slice_ref(calld->slices.slices[j])); |
||||
} |
||||
new_slices_added = 1; /* GPR_TRUE */ |
||||
} |
||||
break; |
||||
case GRPC_OP_METADATA: |
||||
/* move the metadata to the new buffer. */ |
||||
grpc_metadata_batch_move(&metadata, &sop->data.metadata); |
||||
grpc_sopb_add_metadata(&new_send_ops, metadata); |
||||
break; |
||||
case GRPC_NO_OP: |
||||
break; |
||||
} |
||||
} |
||||
grpc_sopb_swap(send_ops, &new_send_ops); |
||||
grpc_sopb_destroy(&new_send_ops); |
||||
} |
||||
|
||||
/** Filter's "main" function, called for any incoming grpc_transport_stream_op
|
||||
* instance that holds a non-zero number of send operations, accesible to this |
||||
* function in \a send_ops. */ |
||||
static void process_send_ops(grpc_call_element *elem, |
||||
grpc_stream_op_buffer *send_ops) { |
||||
call_data *calld = elem->call_data; |
||||
channel_data *channeld = elem->channel_data; |
||||
size_t i; |
||||
int did_compress = 0; |
||||
|
||||
for (i = 0; i < send_ops->nops; ++i) { |
||||
grpc_stream_op *sop = &send_ops->ops[i]; |
||||
switch (sop->type) { |
||||
case GRPC_OP_BEGIN_MESSAGE: |
||||
/* buffer up slices until we've processed all the expected ones (as
|
||||
* given by GRPC_OP_BEGIN_MESSAGE) */ |
||||
calld->remaining_slice_bytes = sop->data.begin_message.length; |
||||
if (sop->data.begin_message.flags & GRPC_WRITE_NO_COMPRESS) { |
||||
calld->has_compression_algorithm = 1; /* GPR_TRUE */ |
||||
calld->compression_algorithm = GRPC_COMPRESS_NONE; |
||||
} |
||||
break; |
||||
case GRPC_OP_METADATA: |
||||
if (!calld->written_initial_metadata) { |
||||
/* Parse incoming request for compression. If any, it'll be available
|
||||
* at calld->compression_algorithm */ |
||||
grpc_metadata_batch_filter(&(sop->data.metadata), |
||||
compression_md_filter, elem); |
||||
if (!calld->has_compression_algorithm) { |
||||
/* If no algorithm was found in the metadata and we aren't
|
||||
* exceptionally skipping compression, fall back to the channel |
||||
* default */ |
||||
calld->compression_algorithm = |
||||
channeld->default_compression_algorithm; |
||||
calld->has_compression_algorithm = 1; /* GPR_TRUE */ |
||||
} |
||||
grpc_metadata_batch_add_head( |
||||
&(sop->data.metadata), &calld->compression_algorithm_storage, |
||||
grpc_mdelem_ref(channeld->mdelem_compression_algorithms |
||||
[calld->compression_algorithm])); |
||||
calld->written_initial_metadata = 1; /* GPR_TRUE */ |
||||
} |
||||
break; |
||||
case GRPC_OP_SLICE: |
||||
if (skip_compression(channeld, calld)) continue; |
||||
GPR_ASSERT(calld->remaining_slice_bytes > 0); |
||||
/* Increase input ref count, gpr_slice_buffer_add takes ownership. */ |
||||
gpr_slice_buffer_add(&calld->slices, gpr_slice_ref(sop->data.slice)); |
||||
calld->remaining_slice_bytes -= GPR_SLICE_LENGTH(sop->data.slice); |
||||
if (calld->remaining_slice_bytes == 0) { |
||||
did_compress = |
||||
compress_send_sb(calld->compression_algorithm, &calld->slices); |
||||
} |
||||
break; |
||||
case GRPC_NO_OP: |
||||
break; |
||||
} |
||||
} |
||||
|
||||
/* Modify the send_ops stream_op_buffer depending on whether compression was
|
||||
* carried out */ |
||||
if (did_compress) { |
||||
finish_compressed_sopb(send_ops, elem); |
||||
} |
||||
} |
||||
|
||||
/* Called either:
|
||||
- in response to an API call (or similar) from above, to send something |
||||
- a network event (or similar) from below, to receive something |
||||
op contains type and call direction information, in addition to the data |
||||
that is being sent or received. */ |
||||
static void compress_start_transport_stream_op(grpc_call_element *elem, |
||||
grpc_transport_stream_op *op) { |
||||
if (op->send_ops && op->send_ops->nops > 0) { |
||||
process_send_ops(elem, op->send_ops); |
||||
} |
||||
|
||||
/* pass control down the stack */ |
||||
grpc_call_next_op(elem, op); |
||||
} |
||||
|
||||
/* Constructor for call_data */ |
||||
static void init_call_elem(grpc_call_element *elem, |
||||
const void *server_transport_data, |
||||
grpc_transport_stream_op *initial_op) { |
||||
/* grab pointers to our data from the call element */ |
||||
call_data *calld = elem->call_data; |
||||
|
||||
/* initialize members */ |
||||
gpr_slice_buffer_init(&calld->slices); |
||||
calld->has_compression_algorithm = 0; |
||||
calld->written_initial_metadata = 0; /* GPR_FALSE */ |
||||
|
||||
if (initial_op) { |
||||
if (initial_op->send_ops && initial_op->send_ops->nops > 0) { |
||||
process_send_ops(elem, initial_op->send_ops); |
||||
} |
||||
} |
||||
} |
||||
|
||||
/* Destructor for call_data */ |
||||
static void destroy_call_elem(grpc_call_element *elem) { |
||||
/* grab pointers to our data from the call element */ |
||||
call_data *calld = elem->call_data; |
||||
gpr_slice_buffer_destroy(&calld->slices); |
||||
} |
||||
|
||||
/* Constructor for channel_data */ |
||||
static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, |
||||
const grpc_channel_args *args, grpc_mdctx *mdctx, |
||||
int is_first, int is_last) { |
||||
channel_data *channeld = elem->channel_data; |
||||
grpc_compression_algorithm algo_idx; |
||||
|
||||
channeld->default_compression_algorithm = |
||||
grpc_channel_args_get_compression_algorithm(args); |
||||
|
||||
channeld->mdstr_request_compression_algorithm_key = |
||||
grpc_mdstr_from_string(mdctx, GRPC_COMPRESS_REQUEST_ALGORITHM_KEY); |
||||
|
||||
channeld->mdstr_outgoing_compression_algorithm_key = |
||||
grpc_mdstr_from_string(mdctx, "grpc-encoding"); |
||||
|
||||
for (algo_idx = 0; algo_idx < GRPC_COMPRESS_ALGORITHMS_COUNT; ++algo_idx) { |
||||
char *algorith_name; |
||||
GPR_ASSERT(grpc_compression_algorithm_name(algo_idx, &algorith_name) != 0); |
||||
channeld->mdelem_compression_algorithms[algo_idx] = |
||||
grpc_mdelem_from_metadata_strings( |
||||
mdctx, |
||||
grpc_mdstr_ref(channeld->mdstr_outgoing_compression_algorithm_key), |
||||
grpc_mdstr_from_string(mdctx, algorith_name)); |
||||
} |
||||
|
||||
GPR_ASSERT(!is_last); |
||||
} |
||||
|
||||
/* Destructor for channel data */ |
||||
static void destroy_channel_elem(grpc_channel_element *elem) { |
||||
channel_data *channeld = elem->channel_data; |
||||
grpc_compression_algorithm algo_idx; |
||||
|
||||
grpc_mdstr_unref(channeld->mdstr_request_compression_algorithm_key); |
||||
grpc_mdstr_unref(channeld->mdstr_outgoing_compression_algorithm_key); |
||||
for (algo_idx = 0; algo_idx < GRPC_COMPRESS_ALGORITHMS_COUNT; |
||||
++algo_idx) { |
||||
grpc_mdelem_unref(channeld->mdelem_compression_algorithms[algo_idx]); |
||||
} |
||||
} |
||||
|
||||
const grpc_channel_filter grpc_compress_filter = { |
||||
compress_start_transport_stream_op, |
||||
grpc_channel_next_op, |
||||
sizeof(call_data), |
||||
init_call_elem, |
||||
destroy_call_elem, |
||||
sizeof(channel_data), |
||||
init_channel_elem, |
||||
destroy_channel_elem, |
||||
"compress"}; |
||||
@ -0,0 +1,65 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_INTERNAL_CORE_CHANNEL_COMPRESS_FILTER_H |
||||
#define GRPC_INTERNAL_CORE_CHANNEL_COMPRESS_FILTER_H |
||||
|
||||
#include "src/core/channel/channel_stack.h" |
||||
|
||||
#define GRPC_COMPRESS_REQUEST_ALGORITHM_KEY "internal:grpc-encoding-request" |
||||
|
||||
/** 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_COMPRESS_REQUEST_ALGORITHM_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_compress_filter; |
||||
|
||||
#endif /* GRPC_INTERNAL_CORE_CHANNEL_COMPRESS_FILTER_H */ |
||||
@ -0,0 +1,135 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#include "test/core/end2end/end2end_tests.h" |
||||
|
||||
#include <string.h> |
||||
|
||||
#include "src/core/channel/channel_args.h" |
||||
#include "src/core/channel/client_channel.h" |
||||
#include "src/core/channel/connected_channel.h" |
||||
#include "src/core/channel/http_server_filter.h" |
||||
#include "src/core/surface/channel.h" |
||||
#include "src/core/surface/server.h" |
||||
#include "src/core/transport/chttp2_transport.h" |
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/host_port.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/sync.h> |
||||
#include <grpc/support/thd.h> |
||||
#include <grpc/support/useful.h> |
||||
#include "test/core/util/port.h" |
||||
#include "test/core/util/test_config.h" |
||||
|
||||
typedef struct fullstack_compression_fixture_data { |
||||
char *localaddr; |
||||
grpc_channel_args* client_args_compression; |
||||
grpc_channel_args* server_args_compression; |
||||
} fullstack_compression_fixture_data; |
||||
|
||||
static grpc_end2end_test_fixture chttp2_create_fixture_fullstack_compression( |
||||
grpc_channel_args *client_args, grpc_channel_args *server_args) { |
||||
grpc_end2end_test_fixture f; |
||||
int port = grpc_pick_unused_port_or_die(); |
||||
fullstack_compression_fixture_data *ffd = |
||||
gpr_malloc(sizeof(fullstack_compression_fixture_data)); |
||||
memset(ffd, 0, sizeof(fullstack_compression_fixture_data)); |
||||
|
||||
gpr_join_host_port(&ffd->localaddr, "localhost", port); |
||||
|
||||
memset(&f, 0, sizeof(f)); |
||||
f.fixture_data = ffd; |
||||
f.cq = grpc_completion_queue_create(); |
||||
|
||||
return f; |
||||
} |
||||
|
||||
void chttp2_init_client_fullstack_compression(grpc_end2end_test_fixture *f, |
||||
grpc_channel_args *client_args) { |
||||
fullstack_compression_fixture_data *ffd = f->fixture_data; |
||||
if (ffd->client_args_compression != NULL) { |
||||
grpc_channel_args_destroy(ffd->client_args_compression); |
||||
} |
||||
ffd->client_args_compression = grpc_channel_args_set_compression_algorithm( |
||||
client_args, GRPC_COMPRESS_GZIP); |
||||
f->client = grpc_channel_create(ffd->localaddr, ffd->client_args_compression); |
||||
} |
||||
|
||||
void chttp2_init_server_fullstack_compression(grpc_end2end_test_fixture *f, |
||||
grpc_channel_args *server_args) { |
||||
fullstack_compression_fixture_data *ffd = f->fixture_data; |
||||
if (ffd->server_args_compression != NULL) { |
||||
grpc_channel_args_destroy(ffd->server_args_compression); |
||||
} |
||||
ffd->server_args_compression = grpc_channel_args_set_compression_algorithm( |
||||
server_args, GRPC_COMPRESS_GZIP); |
||||
if (f->server) { |
||||
grpc_server_destroy(f->server); |
||||
} |
||||
f->server = grpc_server_create(ffd->server_args_compression); |
||||
grpc_server_register_completion_queue(f->server, f->cq); |
||||
GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr)); |
||||
grpc_server_start(f->server); |
||||
} |
||||
|
||||
void chttp2_tear_down_fullstack_compression(grpc_end2end_test_fixture *f) { |
||||
fullstack_compression_fixture_data *ffd = f->fixture_data; |
||||
grpc_channel_args_destroy(ffd->client_args_compression); |
||||
grpc_channel_args_destroy(ffd->server_args_compression); |
||||
gpr_free(ffd->localaddr); |
||||
gpr_free(ffd); |
||||
} |
||||
|
||||
/* All test configurations */ |
||||
static grpc_end2end_test_config configs[] = { |
||||
{"chttp2/fullstack_compression", FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION, |
||||
chttp2_create_fixture_fullstack_compression, |
||||
chttp2_init_client_fullstack_compression, |
||||
chttp2_init_server_fullstack_compression, |
||||
chttp2_tear_down_fullstack_compression}, |
||||
}; |
||||
|
||||
int main(int argc, char **argv) { |
||||
size_t i; |
||||
|
||||
grpc_test_init(argc, argv); |
||||
grpc_init(); |
||||
|
||||
for (i = 0; i < sizeof(configs) / sizeof(*configs); i++) { |
||||
grpc_end2end_tests(configs[i]); |
||||
} |
||||
|
||||
grpc_shutdown(); |
||||
|
||||
return 0; |
||||
} |
||||
@ -0,0 +1,315 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#include "test/core/end2end/end2end_tests.h" |
||||
|
||||
#include <stdio.h> |
||||
#include <string.h> |
||||
|
||||
#include <grpc/byte_buffer.h> |
||||
#include <grpc/byte_buffer_reader.h> |
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/time.h> |
||||
#include <grpc/support/useful.h> |
||||
|
||||
#include "test/core/end2end/cq_verifier.h" |
||||
#include "src/core/channel/channel_args.h" |
||||
#include "src/core/channel/compress_filter.h" |
||||
|
||||
enum { TIMEOUT = 200000 }; |
||||
|
||||
static void *tag(gpr_intptr t) { return (void *)t; } |
||||
|
||||
static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config, |
||||
const char *test_name, |
||||
grpc_channel_args *client_args, |
||||
grpc_channel_args *server_args) { |
||||
grpc_end2end_test_fixture f; |
||||
gpr_log(GPR_INFO, "%s/%s", test_name, config.name); |
||||
f = config.create_fixture(client_args, server_args); |
||||
config.init_client(&f, client_args); |
||||
config.init_server(&f, server_args); |
||||
return f; |
||||
} |
||||
|
||||
static gpr_timespec n_seconds_time(int n) { |
||||
return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(n); |
||||
} |
||||
|
||||
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); } |
||||
|
||||
static void drain_cq(grpc_completion_queue *cq) { |
||||
grpc_event ev; |
||||
do { |
||||
ev = grpc_completion_queue_next(cq, five_seconds_time()); |
||||
} while (ev.type != GRPC_QUEUE_SHUTDOWN); |
||||
} |
||||
|
||||
static void shutdown_server(grpc_end2end_test_fixture *f) { |
||||
if (!f->server) return; |
||||
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); |
||||
GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(1000), |
||||
GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5)) |
||||
.type == GRPC_OP_COMPLETE); |
||||
grpc_server_destroy(f->server); |
||||
f->server = NULL; |
||||
} |
||||
|
||||
static void shutdown_client(grpc_end2end_test_fixture *f) { |
||||
if (!f->client) return; |
||||
grpc_channel_destroy(f->client); |
||||
f->client = NULL; |
||||
} |
||||
|
||||
static void end_test(grpc_end2end_test_fixture *f) { |
||||
shutdown_server(f); |
||||
shutdown_client(f); |
||||
|
||||
grpc_completion_queue_shutdown(f->cq); |
||||
drain_cq(f->cq); |
||||
grpc_completion_queue_destroy(f->cq); |
||||
} |
||||
|
||||
static void request_with_payload_template( |
||||
grpc_end2end_test_config config, const char *test_name, |
||||
gpr_uint32 send_flags_bitmask, |
||||
grpc_compression_algorithm requested_compression_algorithm, |
||||
grpc_compression_algorithm expected_compression_algorithm, |
||||
grpc_metadata *client_metadata) { |
||||
grpc_call *c; |
||||
grpc_call *s; |
||||
gpr_slice request_payload_slice; |
||||
grpc_byte_buffer *request_payload; |
||||
gpr_timespec deadline = five_seconds_time(); |
||||
grpc_channel_args *client_args; |
||||
grpc_channel_args *server_args; |
||||
grpc_end2end_test_fixture f; |
||||
grpc_op ops[6]; |
||||
grpc_op *op; |
||||
grpc_metadata_array initial_metadata_recv; |
||||
grpc_metadata_array trailing_metadata_recv; |
||||
grpc_metadata_array request_metadata_recv; |
||||
grpc_byte_buffer *request_payload_recv = NULL; |
||||
grpc_call_details call_details; |
||||
grpc_status_code status; |
||||
char *details = NULL; |
||||
size_t details_capacity = 0; |
||||
int was_cancelled = 2; |
||||
cq_verifier *cqv; |
||||
char str[1024]; |
||||
|
||||
memset(str, 'x', 1023); str[1023] = '\0'; |
||||
request_payload_slice = gpr_slice_from_copied_string(str); |
||||
request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); |
||||
|
||||
client_args = grpc_channel_args_set_compression_algorithm( |
||||
NULL, requested_compression_algorithm); |
||||
server_args = grpc_channel_args_set_compression_algorithm( |
||||
NULL, requested_compression_algorithm); |
||||
|
||||
f = begin_test(config, test_name, client_args, server_args); |
||||
cqv = cq_verifier_create(f.cq); |
||||
|
||||
c = grpc_channel_create_call(f.client, f.cq, "/foo", |
||||
"foo.test.google.fr", deadline); |
||||
GPR_ASSERT(c); |
||||
|
||||
grpc_metadata_array_init(&initial_metadata_recv); |
||||
grpc_metadata_array_init(&trailing_metadata_recv); |
||||
grpc_metadata_array_init(&request_metadata_recv); |
||||
grpc_call_details_init(&call_details); |
||||
|
||||
op = ops; |
||||
op->op = GRPC_OP_SEND_INITIAL_METADATA; |
||||
if (client_metadata != NULL) { |
||||
op->data.send_initial_metadata.count = 1; |
||||
op->data.send_initial_metadata.metadata = client_metadata; |
||||
} else { |
||||
op->data.send_initial_metadata.count = 0; |
||||
} |
||||
op->flags = 0; |
||||
op++; |
||||
op->op = GRPC_OP_SEND_MESSAGE; |
||||
op->data.send_message = request_payload; |
||||
op->flags = send_flags_bitmask; |
||||
op++; |
||||
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; |
||||
op->flags = 0; |
||||
op++; |
||||
op->op = GRPC_OP_RECV_INITIAL_METADATA; |
||||
op->data.recv_initial_metadata = &initial_metadata_recv; |
||||
op->flags = 0; |
||||
op++; |
||||
op->op = GRPC_OP_RECV_STATUS_ON_CLIENT; |
||||
op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv; |
||||
op->data.recv_status_on_client.status = &status; |
||||
op->data.recv_status_on_client.status_details = &details; |
||||
op->data.recv_status_on_client.status_details_capacity = &details_capacity; |
||||
op->flags = 0; |
||||
op++; |
||||
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1))); |
||||
|
||||
GPR_ASSERT(GRPC_CALL_OK == |
||||
grpc_server_request_call(f.server, &s, &call_details, |
||||
&request_metadata_recv, f.cq, |
||||
f.cq, tag(101))); |
||||
cq_expect_completion(cqv, tag(101), 1); |
||||
cq_verify(cqv); |
||||
|
||||
op = ops; |
||||
op->op = GRPC_OP_SEND_INITIAL_METADATA; |
||||
op->data.send_initial_metadata.count = 0; |
||||
op->flags = 0; |
||||
op++; |
||||
op->op = GRPC_OP_RECV_MESSAGE; |
||||
op->data.recv_message = &request_payload_recv; |
||||
op->flags = 0; |
||||
op++; |
||||
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102))); |
||||
|
||||
cq_expect_completion(cqv, tag(102), 1); |
||||
cq_verify(cqv); |
||||
|
||||
op = ops; |
||||
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; |
||||
op->data.recv_close_on_server.cancelled = &was_cancelled; |
||||
op->flags = 0; |
||||
op++; |
||||
op->op = GRPC_OP_SEND_STATUS_FROM_SERVER; |
||||
op->data.send_status_from_server.trailing_metadata_count = 0; |
||||
op->data.send_status_from_server.status = GRPC_STATUS_OK; |
||||
op->data.send_status_from_server.status_details = "xyz"; |
||||
op->flags = 0; |
||||
op++; |
||||
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(103))); |
||||
|
||||
cq_expect_completion(cqv, tag(103), 1); |
||||
cq_expect_completion(cqv, tag(1), 1); |
||||
cq_verify(cqv); |
||||
|
||||
GPR_ASSERT(status == GRPC_STATUS_OK); |
||||
GPR_ASSERT(0 == strcmp(details, "xyz")); |
||||
GPR_ASSERT(0 == strcmp(call_details.method, "/foo")); |
||||
GPR_ASSERT(0 == strcmp(call_details.host, "foo.test.google.fr")); |
||||
GPR_ASSERT(was_cancelled == 0); |
||||
|
||||
GPR_ASSERT(request_payload_recv->type == GRPC_BB_RAW); |
||||
GPR_ASSERT(request_payload_recv->data.raw.compression == |
||||
expected_compression_algorithm); |
||||
|
||||
GPR_ASSERT(byte_buffer_eq_string(request_payload_recv, str)); |
||||
|
||||
gpr_free(details); |
||||
grpc_metadata_array_destroy(&initial_metadata_recv); |
||||
grpc_metadata_array_destroy(&trailing_metadata_recv); |
||||
grpc_metadata_array_destroy(&request_metadata_recv); |
||||
grpc_call_details_destroy(&call_details); |
||||
|
||||
grpc_call_destroy(c); |
||||
grpc_call_destroy(s); |
||||
|
||||
cq_verifier_destroy(cqv); |
||||
|
||||
gpr_slice_unref(request_payload_slice); |
||||
grpc_byte_buffer_destroy(request_payload); |
||||
grpc_byte_buffer_destroy(request_payload_recv); |
||||
|
||||
grpc_channel_args_destroy(client_args); |
||||
grpc_channel_args_destroy(server_args); |
||||
|
||||
end_test(&f); |
||||
config.tear_down_data(&f); |
||||
} |
||||
|
||||
static void test_invoke_request_with_exceptionally_uncompressed_payload( |
||||
grpc_end2end_test_config config) { |
||||
request_with_payload_template( |
||||
config, "test_invoke_request_with_exceptionally_uncompressed_payload", |
||||
GRPC_WRITE_NO_COMPRESS, GRPC_COMPRESS_GZIP, GRPC_COMPRESS_NONE, |
||||
NULL); |
||||
} |
||||
|
||||
static void test_invoke_request_with_uncompressed_payload( |
||||
grpc_end2end_test_config config) { |
||||
request_with_payload_template( |
||||
config, "test_invoke_request_with_uncompressed_payload", 0, |
||||
GRPC_COMPRESS_NONE, GRPC_COMPRESS_NONE, NULL); |
||||
} |
||||
|
||||
static void test_invoke_request_with_compressed_payload( |
||||
grpc_end2end_test_config config) { |
||||
request_with_payload_template( |
||||
config, "test_invoke_request_with_compressed_payload", 0, |
||||
GRPC_COMPRESS_GZIP, GRPC_COMPRESS_GZIP, NULL); |
||||
} |
||||
|
||||
static void test_invoke_request_with_compressed_payload_md_override( |
||||
grpc_end2end_test_config config) { |
||||
grpc_metadata gzip_compression_override; |
||||
grpc_metadata none_compression_override; |
||||
|
||||
gzip_compression_override.key = GRPC_COMPRESS_REQUEST_ALGORITHM_KEY; |
||||
gzip_compression_override.value = "gzip"; |
||||
gzip_compression_override.value_length = 4; |
||||
memset(&gzip_compression_override.internal_data, 0, |
||||
sizeof(gzip_compression_override.internal_data)); |
||||
|
||||
none_compression_override.key = GRPC_COMPRESS_REQUEST_ALGORITHM_KEY; |
||||
none_compression_override.value = "none"; |
||||
none_compression_override.value_length = 4; |
||||
memset(&none_compression_override.internal_data, 0, |
||||
sizeof(none_compression_override.internal_data)); |
||||
|
||||
/* Channel default NONE, call override to GZIP */ |
||||
request_with_payload_template( |
||||
config, "test_invoke_request_with_compressed_payload_md_override_1", 0, |
||||
GRPC_COMPRESS_NONE, GRPC_COMPRESS_GZIP, &gzip_compression_override); |
||||
|
||||
/* Channel default DEFLATE, call override to GZIP */ |
||||
request_with_payload_template( |
||||
config, "test_invoke_request_with_compressed_payload_md_override_2", 0, |
||||
GRPC_COMPRESS_DEFLATE, GRPC_COMPRESS_GZIP, &gzip_compression_override); |
||||
|
||||
/* Channel default DEFLATE, call override to NONE */ |
||||
request_with_payload_template( |
||||
config, "test_invoke_request_with_compressed_payload_md_override_3", 0, |
||||
GRPC_COMPRESS_DEFLATE, GRPC_COMPRESS_NONE, &none_compression_override); |
||||
} |
||||
|
||||
void grpc_end2end_tests(grpc_end2end_test_config config) { |
||||
test_invoke_request_with_exceptionally_uncompressed_payload(config); |
||||
test_invoke_request_with_uncompressed_payload(config); |
||||
test_invoke_request_with_compressed_payload(config); |
||||
test_invoke_request_with_compressed_payload_md_override(config); |
||||
} |
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
File diff suppressed because one or more lines are too long
Loading…
Reference in new issue