Sree Kuchibhotla
8 years ago
commit
0e90957458
110 changed files with 9555 additions and 3678 deletions
@ -0,0 +1,36 @@ |
|||||||
|
/*
|
||||||
|
* |
||||||
|
* Copyright 2017 gRPC authors. |
||||||
|
* |
||||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||||
|
* you may not use this file except in compliance with the License. |
||||||
|
* You may obtain a copy of the License at |
||||||
|
* |
||||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
* |
||||||
|
* Unless required by applicable law or agreed to in writing, software |
||||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||||
|
* See the License for the specific language governing permissions and |
||||||
|
* limitations under the License. |
||||||
|
* |
||||||
|
*/ |
||||||
|
|
||||||
|
#ifndef GRPC_IMPL_CODEGEN_SYNC_CUSTOM_H |
||||||
|
#define GRPC_IMPL_CODEGEN_SYNC_CUSTOM_H |
||||||
|
|
||||||
|
#include <grpc/impl/codegen/sync_generic.h> |
||||||
|
|
||||||
|
/* Users defining GPR_CUSTOM_SYNC need to define the following macros. */ |
||||||
|
|
||||||
|
#ifdef GPR_CUSTOM_SYNC |
||||||
|
|
||||||
|
typedef GPR_CUSTOM_MU_TYPE gpr_mu; |
||||||
|
typedef GPR_CUSTOM_CV_TYPE gpr_cv; |
||||||
|
typedef GPR_CUSTOM_ONCE_TYPE gpr_once; |
||||||
|
|
||||||
|
#define GPR_ONCE_INIT GPR_CUSTOM_ONCE_INIT |
||||||
|
|
||||||
|
#endif |
||||||
|
|
||||||
|
#endif /* GRPC_IMPL_CODEGEN_SYNC_CUSTOM_H */ |
||||||
@ -0,0 +1,180 @@ |
|||||||
|
/*
|
||||||
|
* |
||||||
|
* Copyright 2017 gRPC authors. |
||||||
|
* |
||||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||||
|
* you may not use this file except in compliance with the License. |
||||||
|
* You may obtain a copy of the License at |
||||||
|
* |
||||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
* |
||||||
|
* Unless required by applicable law or agreed to in writing, software |
||||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||||
|
* See the License for the specific language governing permissions and |
||||||
|
* limitations under the License. |
||||||
|
* |
||||||
|
*/ |
||||||
|
|
||||||
|
#include "src/core/lib/iomgr/call_combiner.h" |
||||||
|
|
||||||
|
#include <grpc/support/log.h> |
||||||
|
|
||||||
|
grpc_tracer_flag grpc_call_combiner_trace = |
||||||
|
GRPC_TRACER_INITIALIZER(false, "call_combiner"); |
||||||
|
|
||||||
|
static grpc_error* decode_cancel_state_error(gpr_atm cancel_state) { |
||||||
|
if (cancel_state & 1) { |
||||||
|
return (grpc_error*)(cancel_state & ~(gpr_atm)1); |
||||||
|
} |
||||||
|
return GRPC_ERROR_NONE; |
||||||
|
} |
||||||
|
|
||||||
|
static gpr_atm encode_cancel_state_error(grpc_error* error) { |
||||||
|
return (gpr_atm)1 | (gpr_atm)error; |
||||||
|
} |
||||||
|
|
||||||
|
void grpc_call_combiner_init(grpc_call_combiner* call_combiner) { |
||||||
|
gpr_mpscq_init(&call_combiner->queue); |
||||||
|
} |
||||||
|
|
||||||
|
void grpc_call_combiner_destroy(grpc_call_combiner* call_combiner) { |
||||||
|
gpr_mpscq_destroy(&call_combiner->queue); |
||||||
|
GRPC_ERROR_UNREF(decode_cancel_state_error(call_combiner->cancel_state)); |
||||||
|
} |
||||||
|
|
||||||
|
#ifndef NDEBUG |
||||||
|
#define DEBUG_ARGS , const char *file, int line |
||||||
|
#define DEBUG_FMT_STR "%s:%d: " |
||||||
|
#define DEBUG_FMT_ARGS , file, line |
||||||
|
#else |
||||||
|
#define DEBUG_ARGS |
||||||
|
#define DEBUG_FMT_STR |
||||||
|
#define DEBUG_FMT_ARGS |
||||||
|
#endif |
||||||
|
|
||||||
|
void grpc_call_combiner_start(grpc_exec_ctx* exec_ctx, |
||||||
|
grpc_call_combiner* call_combiner, |
||||||
|
grpc_closure* closure, |
||||||
|
grpc_error* error DEBUG_ARGS, |
||||||
|
const char* reason) { |
||||||
|
if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { |
||||||
|
gpr_log(GPR_DEBUG, |
||||||
|
"==> grpc_call_combiner_start() [%p] closure=%p [" DEBUG_FMT_STR |
||||||
|
"%s] error=%s", |
||||||
|
call_combiner, closure DEBUG_FMT_ARGS, reason, |
||||||
|
grpc_error_string(error)); |
||||||
|
} |
||||||
|
size_t prev_size = |
||||||
|
(size_t)gpr_atm_full_fetch_add(&call_combiner->size, (gpr_atm)1); |
||||||
|
if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { |
||||||
|
gpr_log(GPR_DEBUG, " size: %" PRIdPTR " -> %" PRIdPTR, prev_size, |
||||||
|
prev_size + 1); |
||||||
|
} |
||||||
|
if (prev_size == 0) { |
||||||
|
if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { |
||||||
|
gpr_log(GPR_DEBUG, " EXECUTING IMMEDIATELY"); |
||||||
|
} |
||||||
|
// Queue was empty, so execute this closure immediately.
|
||||||
|
GRPC_CLOSURE_SCHED(exec_ctx, closure, error); |
||||||
|
} else { |
||||||
|
if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { |
||||||
|
gpr_log(GPR_INFO, " QUEUING"); |
||||||
|
} |
||||||
|
// Queue was not empty, so add closure to queue.
|
||||||
|
closure->error_data.error = error; |
||||||
|
gpr_mpscq_push(&call_combiner->queue, (gpr_mpscq_node*)closure); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
void grpc_call_combiner_stop(grpc_exec_ctx* exec_ctx, |
||||||
|
grpc_call_combiner* call_combiner DEBUG_ARGS, |
||||||
|
const char* reason) { |
||||||
|
if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { |
||||||
|
gpr_log(GPR_DEBUG, |
||||||
|
"==> grpc_call_combiner_stop() [%p] [" DEBUG_FMT_STR "%s]", |
||||||
|
call_combiner DEBUG_FMT_ARGS, reason); |
||||||
|
} |
||||||
|
size_t prev_size = |
||||||
|
(size_t)gpr_atm_full_fetch_add(&call_combiner->size, (gpr_atm)-1); |
||||||
|
if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { |
||||||
|
gpr_log(GPR_DEBUG, " size: %" PRIdPTR " -> %" PRIdPTR, prev_size, |
||||||
|
prev_size - 1); |
||||||
|
} |
||||||
|
GPR_ASSERT(prev_size >= 1); |
||||||
|
if (prev_size > 1) { |
||||||
|
while (true) { |
||||||
|
if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { |
||||||
|
gpr_log(GPR_DEBUG, " checking queue"); |
||||||
|
} |
||||||
|
bool empty; |
||||||
|
grpc_closure* closure = (grpc_closure*)gpr_mpscq_pop_and_check_end( |
||||||
|
&call_combiner->queue, &empty); |
||||||
|
if (closure == NULL) { |
||||||
|
// This can happen either due to a race condition within the mpscq
|
||||||
|
// code or because of a race with grpc_call_combiner_start().
|
||||||
|
if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { |
||||||
|
gpr_log(GPR_DEBUG, " queue returned no result; checking again"); |
||||||
|
} |
||||||
|
continue; |
||||||
|
} |
||||||
|
if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { |
||||||
|
gpr_log(GPR_DEBUG, " EXECUTING FROM QUEUE: closure=%p error=%s", |
||||||
|
closure, grpc_error_string(closure->error_data.error)); |
||||||
|
} |
||||||
|
GRPC_CLOSURE_SCHED(exec_ctx, closure, closure->error_data.error); |
||||||
|
break; |
||||||
|
} |
||||||
|
} else if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { |
||||||
|
gpr_log(GPR_DEBUG, " queue empty"); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
void grpc_call_combiner_set_notify_on_cancel(grpc_exec_ctx* exec_ctx, |
||||||
|
grpc_call_combiner* call_combiner, |
||||||
|
grpc_closure* closure) { |
||||||
|
while (true) { |
||||||
|
// Decode original state.
|
||||||
|
gpr_atm original_state = gpr_atm_acq_load(&call_combiner->cancel_state); |
||||||
|
grpc_error* original_error = decode_cancel_state_error(original_state); |
||||||
|
// If error is set, invoke the cancellation closure immediately.
|
||||||
|
// Otherwise, store the new closure.
|
||||||
|
if (original_error != GRPC_ERROR_NONE) { |
||||||
|
GRPC_CLOSURE_SCHED(exec_ctx, closure, GRPC_ERROR_REF(original_error)); |
||||||
|
break; |
||||||
|
} else { |
||||||
|
if (gpr_atm_full_cas(&call_combiner->cancel_state, original_state, |
||||||
|
(gpr_atm)closure)) { |
||||||
|
break; |
||||||
|
} |
||||||
|
} |
||||||
|
// cas failed, try again.
|
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
void grpc_call_combiner_cancel(grpc_exec_ctx* exec_ctx, |
||||||
|
grpc_call_combiner* call_combiner, |
||||||
|
grpc_error* error) { |
||||||
|
while (true) { |
||||||
|
gpr_atm original_state = gpr_atm_acq_load(&call_combiner->cancel_state); |
||||||
|
grpc_error* original_error = decode_cancel_state_error(original_state); |
||||||
|
if (original_error != GRPC_ERROR_NONE) { |
||||||
|
GRPC_ERROR_UNREF(error); |
||||||
|
break; |
||||||
|
} |
||||||
|
if (gpr_atm_full_cas(&call_combiner->cancel_state, original_state, |
||||||
|
encode_cancel_state_error(error))) { |
||||||
|
if (original_state != 0) { |
||||||
|
grpc_closure* notify_on_cancel = (grpc_closure*)original_state; |
||||||
|
if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { |
||||||
|
gpr_log(GPR_DEBUG, |
||||||
|
"call_combiner=%p: scheduling notify_on_cancel callback=%p", |
||||||
|
call_combiner, notify_on_cancel); |
||||||
|
} |
||||||
|
GRPC_CLOSURE_SCHED(exec_ctx, notify_on_cancel, GRPC_ERROR_REF(error)); |
||||||
|
} |
||||||
|
break; |
||||||
|
} |
||||||
|
// cas failed, try again.
|
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,104 @@ |
|||||||
|
/*
|
||||||
|
* |
||||||
|
* Copyright 2017 gRPC authors. |
||||||
|
* |
||||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||||
|
* you may not use this file except in compliance with the License. |
||||||
|
* You may obtain a copy of the License at |
||||||
|
* |
||||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
* |
||||||
|
* Unless required by applicable law or agreed to in writing, software |
||||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||||
|
* See the License for the specific language governing permissions and |
||||||
|
* limitations under the License. |
||||||
|
* |
||||||
|
*/ |
||||||
|
|
||||||
|
#ifndef GRPC_CORE_LIB_IOMGR_CALL_COMBINER_H |
||||||
|
#define GRPC_CORE_LIB_IOMGR_CALL_COMBINER_H |
||||||
|
|
||||||
|
#include <stddef.h> |
||||||
|
|
||||||
|
#include <grpc/support/atm.h> |
||||||
|
|
||||||
|
#include "src/core/lib/iomgr/closure.h" |
||||||
|
#include "src/core/lib/iomgr/exec_ctx.h" |
||||||
|
#include "src/core/lib/support/mpscq.h" |
||||||
|
|
||||||
|
// A simple, lock-free mechanism for serializing activity related to a
|
||||||
|
// single call. This is similar to a combiner but is more lightweight.
|
||||||
|
//
|
||||||
|
// It requires the callback (or, in the common case where the callback
|
||||||
|
// actually kicks off a chain of callbacks, the last callback in that
|
||||||
|
// chain) to explicitly indicate (by calling GRPC_CALL_COMBINER_STOP())
|
||||||
|
// when it is done with the action that was kicked off by the original
|
||||||
|
// callback.
|
||||||
|
|
||||||
|
extern grpc_tracer_flag grpc_call_combiner_trace; |
||||||
|
|
||||||
|
typedef struct { |
||||||
|
gpr_atm size; // size_t, num closures in queue or currently executing
|
||||||
|
gpr_mpscq queue; |
||||||
|
// Either 0 (if not cancelled and no cancellation closure set),
|
||||||
|
// a grpc_closure* (if the lowest bit is 0),
|
||||||
|
// or a grpc_error* (if the lowest bit is 1).
|
||||||
|
gpr_atm cancel_state; |
||||||
|
} grpc_call_combiner; |
||||||
|
|
||||||
|
// Assumes memory was initialized to zero.
|
||||||
|
void grpc_call_combiner_init(grpc_call_combiner* call_combiner); |
||||||
|
|
||||||
|
void grpc_call_combiner_destroy(grpc_call_combiner* call_combiner); |
||||||
|
|
||||||
|
#ifndef NDEBUG |
||||||
|
#define GRPC_CALL_COMBINER_START(exec_ctx, call_combiner, closure, error, \ |
||||||
|
reason) \
|
||||||
|
grpc_call_combiner_start((exec_ctx), (call_combiner), (closure), (error), \
|
||||||
|
__FILE__, __LINE__, (reason)) |
||||||
|
#define GRPC_CALL_COMBINER_STOP(exec_ctx, call_combiner, reason) \ |
||||||
|
grpc_call_combiner_stop((exec_ctx), (call_combiner), __FILE__, __LINE__, \
|
||||||
|
(reason)) |
||||||
|
/// Starts processing \a closure on \a call_combiner.
|
||||||
|
void grpc_call_combiner_start(grpc_exec_ctx* exec_ctx, |
||||||
|
grpc_call_combiner* call_combiner, |
||||||
|
grpc_closure* closure, grpc_error* error, |
||||||
|
const char* file, int line, const char* reason); |
||||||
|
/// Yields the call combiner to the next closure in the queue, if any.
|
||||||
|
void grpc_call_combiner_stop(grpc_exec_ctx* exec_ctx, |
||||||
|
grpc_call_combiner* call_combiner, |
||||||
|
const char* file, int line, const char* reason); |
||||||
|
#else |
||||||
|
#define GRPC_CALL_COMBINER_START(exec_ctx, call_combiner, closure, error, \ |
||||||
|
reason) \
|
||||||
|
grpc_call_combiner_start((exec_ctx), (call_combiner), (closure), (error), \
|
||||||
|
(reason)) |
||||||
|
#define GRPC_CALL_COMBINER_STOP(exec_ctx, call_combiner, reason) \ |
||||||
|
grpc_call_combiner_stop((exec_ctx), (call_combiner), (reason)) |
||||||
|
/// Starts processing \a closure on \a call_combiner.
|
||||||
|
void grpc_call_combiner_start(grpc_exec_ctx* exec_ctx, |
||||||
|
grpc_call_combiner* call_combiner, |
||||||
|
grpc_closure* closure, grpc_error* error, |
||||||
|
const char* reason); |
||||||
|
/// Yields the call combiner to the next closure in the queue, if any.
|
||||||
|
void grpc_call_combiner_stop(grpc_exec_ctx* exec_ctx, |
||||||
|
grpc_call_combiner* call_combiner, |
||||||
|
const char* reason); |
||||||
|
#endif |
||||||
|
|
||||||
|
/// Tells \a call_combiner to invoke \a closure when
|
||||||
|
/// grpc_call_combiner_cancel() is called. If grpc_call_combiner_cancel()
|
||||||
|
/// was previously called, \a closure will be invoked immediately.
|
||||||
|
/// If \a closure is NULL, then no closure will be invoked on
|
||||||
|
/// cancellation; this effectively unregisters the previously set closure.
|
||||||
|
void grpc_call_combiner_set_notify_on_cancel(grpc_exec_ctx* exec_ctx, |
||||||
|
grpc_call_combiner* call_combiner, |
||||||
|
grpc_closure* closure); |
||||||
|
|
||||||
|
/// Indicates that the call has been cancelled.
|
||||||
|
void grpc_call_combiner_cancel(grpc_exec_ctx* exec_ctx, |
||||||
|
grpc_call_combiner* call_combiner, |
||||||
|
grpc_error* error); |
||||||
|
|
||||||
|
#endif /* GRPC_CORE_LIB_IOMGR_CALL_COMBINER_H */ |
||||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,652 @@ |
|||||||
|
/*
|
||||||
|
* |
||||||
|
* 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 "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/compression.h> |
||||||
|
#include <grpc/support/alloc.h> |
||||||
|
#include <grpc/support/log.h> |
||||||
|
#include <grpc/support/string_util.h> |
||||||
|
#include <grpc/support/time.h> |
||||||
|
#include <grpc/support/useful.h> |
||||||
|
|
||||||
|
#include "src/core/lib/channel/channel_args.h" |
||||||
|
#include "src/core/lib/surface/call.h" |
||||||
|
#include "src/core/lib/surface/call_test_only.h" |
||||||
|
#include "src/core/lib/transport/static_metadata.h" |
||||||
|
#include "test/core/end2end/cq_verifier.h" |
||||||
|
|
||||||
|
static void *tag(intptr_t 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, "Running test: %s/%s", test_name, config.name); |
||||||
|
f = config.create_fixture(client_args, server_args); |
||||||
|
config.init_server(&f, server_args); |
||||||
|
config.init_client(&f, client_args); |
||||||
|
return f; |
||||||
|
} |
||||||
|
|
||||||
|
static gpr_timespec n_seconds_from_now(int n) { |
||||||
|
return grpc_timeout_seconds_to_deadline(n); |
||||||
|
} |
||||||
|
|
||||||
|
static gpr_timespec five_seconds_from_now(void) { |
||||||
|
return n_seconds_from_now(5); |
||||||
|
} |
||||||
|
|
||||||
|
static void drain_cq(grpc_completion_queue *cq) { |
||||||
|
grpc_event ev; |
||||||
|
do { |
||||||
|
ev = grpc_completion_queue_next(cq, five_seconds_from_now(), NULL); |
||||||
|
} 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->shutdown_cq, tag(1000)); |
||||||
|
GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), |
||||||
|
grpc_timeout_seconds_to_deadline(5), |
||||||
|
NULL) |
||||||
|
.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); |
||||||
|
grpc_completion_queue_destroy(f->shutdown_cq); |
||||||
|
} |
||||||
|
|
||||||
|
static void request_for_disabled_algorithm( |
||||||
|
grpc_end2end_test_config config, const char *test_name, |
||||||
|
uint32_t send_flags_bitmask, |
||||||
|
grpc_stream_compression_algorithm algorithm_to_disable, |
||||||
|
grpc_stream_compression_algorithm requested_client_compression_algorithm, |
||||||
|
grpc_status_code expected_error, grpc_metadata *client_metadata) { |
||||||
|
grpc_call *c; |
||||||
|
grpc_call *s; |
||||||
|
grpc_slice request_payload_slice; |
||||||
|
grpc_byte_buffer *request_payload; |
||||||
|
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; |
||||||
|
grpc_call_error error; |
||||||
|
grpc_slice details; |
||||||
|
int was_cancelled = 2; |
||||||
|
cq_verifier *cqv; |
||||||
|
char str[1024]; |
||||||
|
|
||||||
|
memset(str, 'x', 1023); |
||||||
|
str[1023] = '\0'; |
||||||
|
request_payload_slice = grpc_slice_from_copied_string(str); |
||||||
|
request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); |
||||||
|
|
||||||
|
client_args = grpc_channel_args_set_stream_compression_algorithm( |
||||||
|
NULL, requested_client_compression_algorithm); |
||||||
|
server_args = grpc_channel_args_set_stream_compression_algorithm( |
||||||
|
NULL, GRPC_STREAM_COMPRESS_NONE); |
||||||
|
{ |
||||||
|
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; |
||||||
|
server_args = grpc_channel_args_stream_compression_algorithm_set_state( |
||||||
|
&exec_ctx, &server_args, algorithm_to_disable, false); |
||||||
|
grpc_exec_ctx_finish(&exec_ctx); |
||||||
|
} |
||||||
|
|
||||||
|
f = begin_test(config, test_name, client_args, server_args); |
||||||
|
cqv = cq_verifier_create(f.cq); |
||||||
|
|
||||||
|
gpr_timespec deadline = five_seconds_from_now(); |
||||||
|
c = grpc_channel_create_call( |
||||||
|
f.client, NULL, GRPC_PROPAGATE_DEFAULTS, f.cq, |
||||||
|
grpc_slice_from_static_string("/foo"), |
||||||
|
get_host_override_slice("foo.test.google.fr:1234", config), deadline, |
||||||
|
NULL); |
||||||
|
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); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
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->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_SEND_MESSAGE; |
||||||
|
op->data.send_message.send_message = request_payload; |
||||||
|
op->flags = send_flags_bitmask; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_RECV_INITIAL_METADATA; |
||||||
|
op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
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->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
error = |
||||||
|
grpc_server_request_call(f.server, &s, &call_details, |
||||||
|
&request_metadata_recv, f.cq, f.cq, tag(101)); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(101), true); |
||||||
|
cq_verify(cqv); |
||||||
|
|
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_INITIAL_METADATA; |
||||||
|
op->data.send_initial_metadata.count = 0; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_RECV_MESSAGE; |
||||||
|
op->data.recv_message.recv_message = &request_payload_recv; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(102), false); |
||||||
|
|
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; |
||||||
|
op->data.recv_close_on_server.cancelled = &was_cancelled; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(103), true); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(1), true); |
||||||
|
cq_verify(cqv); |
||||||
|
|
||||||
|
/* call was cancelled (closed) ... */ |
||||||
|
GPR_ASSERT(was_cancelled != 0); |
||||||
|
/* with a certain error */ |
||||||
|
GPR_ASSERT(status == expected_error); |
||||||
|
|
||||||
|
char *algo_name = NULL; |
||||||
|
GPR_ASSERT( |
||||||
|
grpc_stream_compression_algorithm_name(algorithm_to_disable, &algo_name)); |
||||||
|
char *expected_details = NULL; |
||||||
|
gpr_asprintf(&expected_details, |
||||||
|
"Stream compression algorithm '%s' is disabled.", algo_name); |
||||||
|
/* and we expect a specific reason for it */ |
||||||
|
GPR_ASSERT(0 == grpc_slice_str_cmp(details, expected_details)); |
||||||
|
gpr_free(expected_details); |
||||||
|
GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo")); |
||||||
|
validate_host_override_string("foo.test.google.fr:1234", call_details.host, |
||||||
|
config); |
||||||
|
|
||||||
|
grpc_slice_unref(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_unref(c); |
||||||
|
grpc_call_unref(s); |
||||||
|
|
||||||
|
cq_verifier_destroy(cqv); |
||||||
|
|
||||||
|
grpc_slice_unref(request_payload_slice); |
||||||
|
grpc_byte_buffer_destroy(request_payload); |
||||||
|
grpc_byte_buffer_destroy(request_payload_recv); |
||||||
|
|
||||||
|
{ |
||||||
|
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; |
||||||
|
grpc_channel_args_destroy(&exec_ctx, client_args); |
||||||
|
grpc_channel_args_destroy(&exec_ctx, server_args); |
||||||
|
grpc_exec_ctx_finish(&exec_ctx); |
||||||
|
} |
||||||
|
|
||||||
|
end_test(&f); |
||||||
|
config.tear_down_data(&f); |
||||||
|
} |
||||||
|
|
||||||
|
static void request_with_payload_template( |
||||||
|
grpc_end2end_test_config config, const char *test_name, |
||||||
|
uint32_t client_send_flags_bitmask, |
||||||
|
grpc_stream_compression_algorithm |
||||||
|
default_client_channel_compression_algorithm, |
||||||
|
grpc_stream_compression_algorithm |
||||||
|
default_server_channel_compression_algorithm, |
||||||
|
grpc_stream_compression_algorithm expected_client_compression_algorithm, |
||||||
|
grpc_stream_compression_algorithm expected_server_compression_algorithm, |
||||||
|
grpc_metadata *client_init_metadata, bool set_server_level, |
||||||
|
grpc_stream_compression_level server_compression_level, |
||||||
|
bool send_message_before_initial_metadata, |
||||||
|
bool set_default_server_message_compression_algorithm, |
||||||
|
grpc_compression_algorithm default_server_message_compression_algorithm) { |
||||||
|
grpc_call *c; |
||||||
|
grpc_call *s; |
||||||
|
grpc_slice request_payload_slice; |
||||||
|
grpc_byte_buffer *request_payload = NULL; |
||||||
|
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_byte_buffer *response_payload; |
||||||
|
grpc_byte_buffer *response_payload_recv; |
||||||
|
grpc_call_details call_details; |
||||||
|
grpc_status_code status; |
||||||
|
grpc_call_error error; |
||||||
|
grpc_slice details; |
||||||
|
int was_cancelled = 2; |
||||||
|
cq_verifier *cqv; |
||||||
|
char request_str[1024]; |
||||||
|
char response_str[1024]; |
||||||
|
|
||||||
|
memset(request_str, 'x', 1023); |
||||||
|
request_str[1023] = '\0'; |
||||||
|
|
||||||
|
memset(response_str, 'y', 1023); |
||||||
|
response_str[1023] = '\0'; |
||||||
|
|
||||||
|
request_payload_slice = grpc_slice_from_copied_string(request_str); |
||||||
|
grpc_slice response_payload_slice = |
||||||
|
grpc_slice_from_copied_string(response_str); |
||||||
|
|
||||||
|
client_args = grpc_channel_args_set_stream_compression_algorithm( |
||||||
|
NULL, default_client_channel_compression_algorithm); |
||||||
|
if (set_default_server_message_compression_algorithm) { |
||||||
|
server_args = grpc_channel_args_set_compression_algorithm( |
||||||
|
NULL, default_server_message_compression_algorithm); |
||||||
|
} else { |
||||||
|
server_args = grpc_channel_args_set_stream_compression_algorithm( |
||||||
|
NULL, default_server_channel_compression_algorithm); |
||||||
|
} |
||||||
|
|
||||||
|
f = begin_test(config, test_name, client_args, server_args); |
||||||
|
cqv = cq_verifier_create(f.cq); |
||||||
|
|
||||||
|
gpr_timespec deadline = five_seconds_from_now(); |
||||||
|
c = grpc_channel_create_call( |
||||||
|
f.client, NULL, GRPC_PROPAGATE_DEFAULTS, f.cq, |
||||||
|
grpc_slice_from_static_string("/foo"), |
||||||
|
get_host_override_slice("foo.test.google.fr:1234", config), deadline, |
||||||
|
NULL); |
||||||
|
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); |
||||||
|
|
||||||
|
if (send_message_before_initial_metadata) { |
||||||
|
request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); |
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_MESSAGE; |
||||||
|
op->data.send_message.send_message = request_payload; |
||||||
|
op->flags = client_send_flags_bitmask; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(2), true); |
||||||
|
} |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_INITIAL_METADATA; |
||||||
|
if (client_init_metadata != NULL) { |
||||||
|
op->data.send_initial_metadata.count = 1; |
||||||
|
op->data.send_initial_metadata.metadata = client_init_metadata; |
||||||
|
} else { |
||||||
|
op->data.send_initial_metadata.count = 0; |
||||||
|
} |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_RECV_INITIAL_METADATA; |
||||||
|
op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
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->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
error = |
||||||
|
grpc_server_request_call(f.server, &s, &call_details, |
||||||
|
&request_metadata_recv, f.cq, f.cq, tag(100)); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(100), true); |
||||||
|
cq_verify(cqv); |
||||||
|
|
||||||
|
GPR_ASSERT(GPR_BITCOUNT(grpc_call_test_only_get_encodings_accepted_by_peer( |
||||||
|
s)) == GRPC_COMPRESS_ALGORITHMS_COUNT); |
||||||
|
GPR_ASSERT(GPR_BITGET(grpc_call_test_only_get_encodings_accepted_by_peer(s), |
||||||
|
GRPC_COMPRESS_NONE) != 0); |
||||||
|
GPR_ASSERT(GPR_BITGET(grpc_call_test_only_get_encodings_accepted_by_peer(s), |
||||||
|
GRPC_COMPRESS_DEFLATE) != 0); |
||||||
|
GPR_ASSERT(GPR_BITGET(grpc_call_test_only_get_encodings_accepted_by_peer(s), |
||||||
|
GRPC_COMPRESS_GZIP) != 0); |
||||||
|
GPR_ASSERT( |
||||||
|
GPR_BITCOUNT(grpc_call_test_only_get_stream_encodings_accepted_by_peer( |
||||||
|
s)) == GRPC_STREAM_COMPRESS_ALGORITHMS_COUNT); |
||||||
|
GPR_ASSERT( |
||||||
|
GPR_BITGET(grpc_call_test_only_get_stream_encodings_accepted_by_peer(s), |
||||||
|
GRPC_STREAM_COMPRESS_NONE) != 0); |
||||||
|
GPR_ASSERT( |
||||||
|
GPR_BITGET(grpc_call_test_only_get_stream_encodings_accepted_by_peer(s), |
||||||
|
GRPC_STREAM_COMPRESS_GZIP) != 0); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_INITIAL_METADATA; |
||||||
|
op->data.send_initial_metadata.count = 0; |
||||||
|
if (set_server_level) { |
||||||
|
op->data.send_initial_metadata.maybe_stream_compression_level.is_set = true; |
||||||
|
op->data.send_initial_metadata.maybe_stream_compression_level.level = |
||||||
|
server_compression_level; |
||||||
|
} |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; |
||||||
|
op->data.recv_close_on_server.cancelled = &was_cancelled; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(101), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
for (int i = 0; i < 2; i++) { |
||||||
|
response_payload = grpc_raw_byte_buffer_create(&response_payload_slice, 1); |
||||||
|
|
||||||
|
if (i > 0 || !send_message_before_initial_metadata) { |
||||||
|
request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); |
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_MESSAGE; |
||||||
|
op->data.send_message.send_message = request_payload; |
||||||
|
op->flags = client_send_flags_bitmask; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(2), 1); |
||||||
|
} |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_RECV_MESSAGE; |
||||||
|
op->data.recv_message.recv_message = &request_payload_recv; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(102), 1); |
||||||
|
cq_verify(cqv); |
||||||
|
|
||||||
|
GPR_ASSERT(request_payload_recv->type == GRPC_BB_RAW); |
||||||
|
GPR_ASSERT(byte_buffer_eq_string(request_payload_recv, request_str)); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_MESSAGE; |
||||||
|
op->data.send_message.send_message = response_payload; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_RECV_MESSAGE; |
||||||
|
op->data.recv_message.recv_message = &response_payload_recv; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(3), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(103), 1); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(3), 1); |
||||||
|
cq_verify(cqv); |
||||||
|
|
||||||
|
GPR_ASSERT(response_payload_recv->type == GRPC_BB_RAW); |
||||||
|
GPR_ASSERT(byte_buffer_eq_string(response_payload_recv, response_str)); |
||||||
|
|
||||||
|
grpc_byte_buffer_destroy(request_payload); |
||||||
|
grpc_byte_buffer_destroy(response_payload); |
||||||
|
grpc_byte_buffer_destroy(request_payload_recv); |
||||||
|
grpc_byte_buffer_destroy(response_payload_recv); |
||||||
|
} |
||||||
|
|
||||||
|
grpc_slice_unref(request_payload_slice); |
||||||
|
grpc_slice_unref(response_payload_slice); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(4), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
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; |
||||||
|
grpc_slice status_details = grpc_slice_from_static_string("xyz"); |
||||||
|
op->data.send_status_from_server.status_details = &status_details; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(104), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(1), 1); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(4), 1); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(101), 1); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(104), 1); |
||||||
|
cq_verify(cqv); |
||||||
|
|
||||||
|
GPR_ASSERT(status == GRPC_STATUS_OK); |
||||||
|
GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz")); |
||||||
|
GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo")); |
||||||
|
validate_host_override_string("foo.test.google.fr:1234", call_details.host, |
||||||
|
config); |
||||||
|
GPR_ASSERT(was_cancelled == 0); |
||||||
|
|
||||||
|
grpc_slice_unref(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_unref(c); |
||||||
|
grpc_call_unref(s); |
||||||
|
|
||||||
|
cq_verifier_destroy(cqv); |
||||||
|
|
||||||
|
{ |
||||||
|
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; |
||||||
|
grpc_channel_args_destroy(&exec_ctx, client_args); |
||||||
|
grpc_channel_args_destroy(&exec_ctx, server_args); |
||||||
|
grpc_exec_ctx_finish(&exec_ctx); |
||||||
|
} |
||||||
|
|
||||||
|
end_test(&f); |
||||||
|
config.tear_down_data(&f); |
||||||
|
} |
||||||
|
|
||||||
|
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_STREAM_COMPRESS_GZIP, GRPC_STREAM_COMPRESS_GZIP, |
||||||
|
GRPC_STREAM_COMPRESS_GZIP, GRPC_STREAM_COMPRESS_GZIP, NULL, |
||||||
|
false, /* ignored */ |
||||||
|
GRPC_STREAM_COMPRESS_LEVEL_NONE, false, false, GRPC_COMPRESS_NONE); |
||||||
|
} |
||||||
|
|
||||||
|
static void test_invoke_request_with_send_message_before_initial_metadata( |
||||||
|
grpc_end2end_test_config config) { |
||||||
|
request_with_payload_template( |
||||||
|
config, "test_invoke_request_with_send_message_before_initial_metadata", |
||||||
|
0, GRPC_STREAM_COMPRESS_GZIP, GRPC_STREAM_COMPRESS_GZIP, |
||||||
|
GRPC_STREAM_COMPRESS_GZIP, GRPC_STREAM_COMPRESS_GZIP, NULL, |
||||||
|
false, /* ignored */ |
||||||
|
GRPC_STREAM_COMPRESS_LEVEL_NONE, true, false, GRPC_COMPRESS_NONE); |
||||||
|
} |
||||||
|
|
||||||
|
static void test_invoke_request_with_server_level( |
||||||
|
grpc_end2end_test_config config) { |
||||||
|
request_with_payload_template( |
||||||
|
config, "test_invoke_request_with_server_level", 0, |
||||||
|
GRPC_STREAM_COMPRESS_NONE, GRPC_STREAM_COMPRESS_NONE, |
||||||
|
GRPC_STREAM_COMPRESS_NONE, GRPC_STREAM_COMPRESS_GZIP, |
||||||
|
/* ignored */ NULL, true, GRPC_STREAM_COMPRESS_LEVEL_HIGH, false, false, |
||||||
|
GRPC_COMPRESS_NONE); |
||||||
|
} |
||||||
|
|
||||||
|
static void test_invoke_request_with_compressed_payload_md_override( |
||||||
|
grpc_end2end_test_config config) { |
||||||
|
grpc_metadata gzip_compression_override; |
||||||
|
grpc_metadata identity_compression_override; |
||||||
|
|
||||||
|
gzip_compression_override.key = |
||||||
|
GRPC_MDSTR_GRPC_INTERNAL_STREAM_ENCODING_REQUEST; |
||||||
|
gzip_compression_override.value = grpc_slice_from_static_string("gzip"); |
||||||
|
memset(&gzip_compression_override.internal_data, 0, |
||||||
|
sizeof(gzip_compression_override.internal_data)); |
||||||
|
|
||||||
|
identity_compression_override.key = |
||||||
|
GRPC_MDSTR_GRPC_INTERNAL_STREAM_ENCODING_REQUEST; |
||||||
|
identity_compression_override.value = |
||||||
|
grpc_slice_from_static_string("identity"); |
||||||
|
memset(&identity_compression_override.internal_data, 0, |
||||||
|
sizeof(identity_compression_override.internal_data)); |
||||||
|
|
||||||
|
/* Channel default NONE (aka IDENTITY), call override to stream GZIP */ |
||||||
|
request_with_payload_template( |
||||||
|
config, "test_invoke_request_with_compressed_payload_md_override_1", 0, |
||||||
|
GRPC_STREAM_COMPRESS_NONE, GRPC_STREAM_COMPRESS_NONE, |
||||||
|
GRPC_STREAM_COMPRESS_GZIP, GRPC_STREAM_COMPRESS_NONE, |
||||||
|
&gzip_compression_override, false, |
||||||
|
/*ignored*/ GRPC_STREAM_COMPRESS_LEVEL_NONE, false, false, |
||||||
|
GRPC_COMPRESS_NONE); |
||||||
|
|
||||||
|
/* Channel default stream GZIP, call override to NONE (aka IDENTITY) */ |
||||||
|
request_with_payload_template( |
||||||
|
config, "test_invoke_request_with_compressed_payload_md_override_3", 0, |
||||||
|
GRPC_STREAM_COMPRESS_GZIP, GRPC_STREAM_COMPRESS_NONE, |
||||||
|
GRPC_STREAM_COMPRESS_NONE, GRPC_STREAM_COMPRESS_NONE, |
||||||
|
&identity_compression_override, false, |
||||||
|
/*ignored*/ GRPC_STREAM_COMPRESS_LEVEL_NONE, false, false, |
||||||
|
GRPC_COMPRESS_NONE); |
||||||
|
} |
||||||
|
|
||||||
|
static void test_invoke_request_with_disabled_algorithm( |
||||||
|
grpc_end2end_test_config config) { |
||||||
|
request_for_disabled_algorithm( |
||||||
|
config, "test_invoke_request_with_disabled_algorithm", 0, |
||||||
|
GRPC_STREAM_COMPRESS_GZIP, GRPC_STREAM_COMPRESS_GZIP, |
||||||
|
GRPC_STATUS_UNIMPLEMENTED, NULL); |
||||||
|
} |
||||||
|
|
||||||
|
static void test_stream_compression_override_message_compression( |
||||||
|
grpc_end2end_test_config config) { |
||||||
|
grpc_stream_compression_level level = GRPC_STREAM_COMPRESS_LEVEL_MED; |
||||||
|
request_with_payload_template( |
||||||
|
config, "test_stream_compression_override_message_compression", 0, |
||||||
|
GRPC_STREAM_COMPRESS_NONE, GRPC_STREAM_COMPRESS_NONE, |
||||||
|
GRPC_STREAM_COMPRESS_NONE, |
||||||
|
grpc_stream_compression_algorithm_for_level( |
||||||
|
level, (1u << GRPC_STREAM_COMPRESS_ALGORITHMS_COUNT) - 1), |
||||||
|
/* ignored */ NULL, true, level, false, true, GRPC_COMPRESS_GZIP); |
||||||
|
} |
||||||
|
|
||||||
|
void stream_compression_compressed_payload(grpc_end2end_test_config config) { |
||||||
|
test_invoke_request_with_compressed_payload(config); |
||||||
|
test_invoke_request_with_send_message_before_initial_metadata(config); |
||||||
|
test_invoke_request_with_server_level(config); |
||||||
|
test_invoke_request_with_compressed_payload_md_override(config); |
||||||
|
test_invoke_request_with_disabled_algorithm(config); |
||||||
|
test_stream_compression_override_message_compression(config); |
||||||
|
} |
||||||
|
|
||||||
|
void stream_compression_compressed_payload_pre_init(void) {} |
||||||
@ -0,0 +1,305 @@ |
|||||||
|
/*
|
||||||
|
* |
||||||
|
* 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 "test/core/end2end/end2end_tests.h" |
||||||
|
|
||||||
|
#include <stdio.h> |
||||||
|
#include <string.h> |
||||||
|
|
||||||
|
#include <grpc/byte_buffer.h> |
||||||
|
#include <grpc/compression.h> |
||||||
|
#include <grpc/compression.h> |
||||||
|
#include <grpc/support/alloc.h> |
||||||
|
#include <grpc/support/log.h> |
||||||
|
#include <grpc/support/time.h> |
||||||
|
#include <grpc/support/useful.h> |
||||||
|
#include "src/core/lib/channel/channel_args.h" |
||||||
|
#include "src/core/lib/surface/call.h" |
||||||
|
#include "test/core/end2end/cq_verifier.h" |
||||||
|
|
||||||
|
static void *tag(intptr_t 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, "Running test: %s/%s", test_name, config.name); |
||||||
|
f = config.create_fixture(client_args, server_args); |
||||||
|
config.init_server(&f, server_args); |
||||||
|
config.init_client(&f, client_args); |
||||||
|
return f; |
||||||
|
} |
||||||
|
|
||||||
|
static gpr_timespec n_seconds_from_now(int n) { |
||||||
|
return grpc_timeout_seconds_to_deadline(n); |
||||||
|
} |
||||||
|
|
||||||
|
static gpr_timespec five_seconds_from_now(void) { |
||||||
|
return n_seconds_from_now(5); |
||||||
|
} |
||||||
|
|
||||||
|
static void drain_cq(grpc_completion_queue *cq) { |
||||||
|
grpc_event ev; |
||||||
|
do { |
||||||
|
ev = grpc_completion_queue_next(cq, five_seconds_from_now(), NULL); |
||||||
|
} 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->shutdown_cq, tag(1000)); |
||||||
|
GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), |
||||||
|
grpc_timeout_seconds_to_deadline(5), |
||||||
|
NULL) |
||||||
|
.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); |
||||||
|
grpc_completion_queue_destroy(f->shutdown_cq); |
||||||
|
} |
||||||
|
|
||||||
|
/* Creates and returns a grpc_slice containing random alphanumeric characters.
|
||||||
|
*/ |
||||||
|
static grpc_slice generate_random_slice() { |
||||||
|
size_t i; |
||||||
|
static const char chars[] = "abcdefghijklmnopqrstuvwxyz1234567890"; |
||||||
|
char *output; |
||||||
|
const size_t output_size = 1024 * 1024; |
||||||
|
output = gpr_malloc(output_size); |
||||||
|
for (i = 0; i < output_size - 1; ++i) { |
||||||
|
output[i] = chars[rand() % (int)(sizeof(chars) - 1)]; |
||||||
|
} |
||||||
|
output[output_size - 1] = '\0'; |
||||||
|
grpc_slice out = grpc_slice_from_copied_string(output); |
||||||
|
gpr_free(output); |
||||||
|
return out; |
||||||
|
} |
||||||
|
|
||||||
|
static void request_response_with_payload(grpc_end2end_test_config config, |
||||||
|
grpc_end2end_test_fixture f) { |
||||||
|
/* Create large request and response bodies. These are big enough to require
|
||||||
|
* multiple round trips to deliver to the peer, and their exact contents of |
||||||
|
* will be verified on completion. */ |
||||||
|
grpc_slice request_payload_slice = generate_random_slice(); |
||||||
|
grpc_slice response_payload_slice = generate_random_slice(); |
||||||
|
|
||||||
|
grpc_call *c; |
||||||
|
grpc_call *s; |
||||||
|
grpc_byte_buffer *request_payload = |
||||||
|
grpc_raw_byte_buffer_create(&request_payload_slice, 1); |
||||||
|
grpc_byte_buffer *response_payload = |
||||||
|
grpc_raw_byte_buffer_create(&response_payload_slice, 1); |
||||||
|
cq_verifier *cqv = cq_verifier_create(f.cq); |
||||||
|
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_byte_buffer *response_payload_recv = NULL; |
||||||
|
grpc_call_details call_details; |
||||||
|
grpc_status_code status; |
||||||
|
grpc_call_error error; |
||||||
|
grpc_slice details; |
||||||
|
int was_cancelled = 2; |
||||||
|
|
||||||
|
gpr_timespec deadline = n_seconds_from_now(60); |
||||||
|
c = grpc_channel_create_call( |
||||||
|
f.client, NULL, GRPC_PROPAGATE_DEFAULTS, f.cq, |
||||||
|
grpc_slice_from_static_string("/foo"), |
||||||
|
get_host_override_slice("foo.test.google.fr:1234", config), deadline, |
||||||
|
NULL); |
||||||
|
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); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_INITIAL_METADATA; |
||||||
|
op->data.send_initial_metadata.count = 0; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_SEND_MESSAGE; |
||||||
|
op->data.send_message.send_message = request_payload; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_RECV_INITIAL_METADATA; |
||||||
|
op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_RECV_MESSAGE; |
||||||
|
op->data.recv_message.recv_message = &response_payload_recv; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
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->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
error = |
||||||
|
grpc_server_request_call(f.server, &s, &call_details, |
||||||
|
&request_metadata_recv, f.cq, f.cq, tag(101)); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(101), 1); |
||||||
|
cq_verify(cqv); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_INITIAL_METADATA; |
||||||
|
op->data.send_initial_metadata.count = 0; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_RECV_MESSAGE; |
||||||
|
op->data.recv_message.recv_message = &request_payload_recv; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(102), 1); |
||||||
|
cq_verify(cqv); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; |
||||||
|
op->data.recv_close_on_server.cancelled = &was_cancelled; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_SEND_MESSAGE; |
||||||
|
op->data.send_message.send_message = response_payload; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
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; |
||||||
|
grpc_slice status_details = grpc_slice_from_static_string("xyz"); |
||||||
|
op->data.send_status_from_server.status_details = &status_details; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
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 == grpc_slice_str_cmp(details, "xyz")); |
||||||
|
GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo")); |
||||||
|
validate_host_override_string("foo.test.google.fr:1234", call_details.host, |
||||||
|
config); |
||||||
|
GPR_ASSERT(was_cancelled == 0); |
||||||
|
GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice)); |
||||||
|
GPR_ASSERT( |
||||||
|
byte_buffer_eq_slice(response_payload_recv, response_payload_slice)); |
||||||
|
|
||||||
|
grpc_slice_unref(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_unref(c); |
||||||
|
grpc_call_unref(s); |
||||||
|
|
||||||
|
cq_verifier_destroy(cqv); |
||||||
|
|
||||||
|
grpc_byte_buffer_destroy(request_payload); |
||||||
|
grpc_byte_buffer_destroy(response_payload); |
||||||
|
grpc_byte_buffer_destroy(request_payload_recv); |
||||||
|
grpc_byte_buffer_destroy(response_payload_recv); |
||||||
|
} |
||||||
|
|
||||||
|
/* Client sends a request with payload, server reads then returns a response
|
||||||
|
payload and status. */ |
||||||
|
static void test_invoke_request_response_with_payload( |
||||||
|
grpc_end2end_test_config config) { |
||||||
|
grpc_channel_args *client_args = |
||||||
|
grpc_channel_args_set_stream_compression_algorithm( |
||||||
|
NULL, GRPC_STREAM_COMPRESS_GZIP); |
||||||
|
grpc_channel_args *server_args = |
||||||
|
grpc_channel_args_set_stream_compression_algorithm( |
||||||
|
NULL, GRPC_STREAM_COMPRESS_GZIP); |
||||||
|
grpc_end2end_test_fixture f = |
||||||
|
begin_test(config, "test_invoke_request_response_with_payload", |
||||||
|
client_args, server_args); |
||||||
|
request_response_with_payload(config, f); |
||||||
|
end_test(&f); |
||||||
|
config.tear_down_data(&f); |
||||||
|
{ |
||||||
|
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; |
||||||
|
grpc_channel_args_destroy(&exec_ctx, client_args); |
||||||
|
grpc_channel_args_destroy(&exec_ctx, server_args); |
||||||
|
grpc_exec_ctx_finish(&exec_ctx); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
static void test_invoke_10_request_response_with_payload( |
||||||
|
grpc_end2end_test_config config) { |
||||||
|
int i; |
||||||
|
grpc_end2end_test_fixture f = begin_test( |
||||||
|
config, "test_invoke_10_request_response_with_payload", NULL, NULL); |
||||||
|
for (i = 0; i < 10; i++) { |
||||||
|
request_response_with_payload(config, f); |
||||||
|
} |
||||||
|
end_test(&f); |
||||||
|
config.tear_down_data(&f); |
||||||
|
} |
||||||
|
|
||||||
|
void stream_compression_payload(grpc_end2end_test_config config) { |
||||||
|
test_invoke_request_response_with_payload(config); |
||||||
|
test_invoke_10_request_response_with_payload(config); |
||||||
|
} |
||||||
|
|
||||||
|
void stream_compression_payload_pre_init(void) {} |
||||||
@ -0,0 +1,291 @@ |
|||||||
|
/*
|
||||||
|
* |
||||||
|
* 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 "test/core/end2end/end2end_tests.h" |
||||||
|
|
||||||
|
#include <stdio.h> |
||||||
|
#include <string.h> |
||||||
|
|
||||||
|
#include <grpc/byte_buffer.h> |
||||||
|
#include <grpc/compression.h> |
||||||
|
#include <grpc/support/alloc.h> |
||||||
|
#include <grpc/support/log.h> |
||||||
|
#include <grpc/support/time.h> |
||||||
|
#include <grpc/support/useful.h> |
||||||
|
#include "src/core/lib/channel/channel_args.h" |
||||||
|
#include "src/core/lib/surface/call.h" |
||||||
|
#include "test/core/end2end/cq_verifier.h" |
||||||
|
|
||||||
|
static void *tag(intptr_t 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, "Running test: %s/%s", test_name, config.name); |
||||||
|
f = config.create_fixture(client_args, server_args); |
||||||
|
config.init_server(&f, server_args); |
||||||
|
config.init_client(&f, client_args); |
||||||
|
return f; |
||||||
|
} |
||||||
|
|
||||||
|
static gpr_timespec n_seconds_from_now(int n) { |
||||||
|
return grpc_timeout_seconds_to_deadline(n); |
||||||
|
} |
||||||
|
|
||||||
|
static gpr_timespec five_seconds_from_now(void) { |
||||||
|
return n_seconds_from_now(5); |
||||||
|
} |
||||||
|
|
||||||
|
static void drain_cq(grpc_completion_queue *cq) { |
||||||
|
grpc_event ev; |
||||||
|
do { |
||||||
|
ev = grpc_completion_queue_next(cq, five_seconds_from_now(), NULL); |
||||||
|
} 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->shutdown_cq, tag(1000)); |
||||||
|
GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000), |
||||||
|
grpc_timeout_seconds_to_deadline(5), |
||||||
|
NULL) |
||||||
|
.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); |
||||||
|
grpc_completion_queue_destroy(f->shutdown_cq); |
||||||
|
} |
||||||
|
|
||||||
|
/* Client pings and server pongs. Repeat messages rounds before finishing. */ |
||||||
|
static void test_pingpong_streaming(grpc_end2end_test_config config, |
||||||
|
int messages) { |
||||||
|
grpc_channel_args *client_args = |
||||||
|
grpc_channel_args_set_stream_compression_algorithm( |
||||||
|
NULL, GRPC_STREAM_COMPRESS_GZIP); |
||||||
|
grpc_channel_args *server_args = |
||||||
|
grpc_channel_args_set_stream_compression_algorithm( |
||||||
|
NULL, GRPC_STREAM_COMPRESS_GZIP); |
||||||
|
grpc_end2end_test_fixture f = |
||||||
|
begin_test(config, "test_pingpong_streaming", client_args, server_args); |
||||||
|
grpc_call *c; |
||||||
|
grpc_call *s; |
||||||
|
cq_verifier *cqv = cq_verifier_create(f.cq); |
||||||
|
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_call_details call_details; |
||||||
|
grpc_status_code status; |
||||||
|
grpc_call_error error; |
||||||
|
grpc_slice details; |
||||||
|
int was_cancelled = 2; |
||||||
|
grpc_byte_buffer *request_payload; |
||||||
|
grpc_byte_buffer *request_payload_recv; |
||||||
|
grpc_byte_buffer *response_payload; |
||||||
|
grpc_byte_buffer *response_payload_recv; |
||||||
|
int i; |
||||||
|
grpc_slice request_payload_slice = |
||||||
|
grpc_slice_from_copied_string("hello world"); |
||||||
|
grpc_slice response_payload_slice = |
||||||
|
grpc_slice_from_copied_string("hello you"); |
||||||
|
|
||||||
|
gpr_timespec deadline = five_seconds_from_now(); |
||||||
|
c = grpc_channel_create_call( |
||||||
|
f.client, NULL, GRPC_PROPAGATE_DEFAULTS, f.cq, |
||||||
|
grpc_slice_from_static_string("/foo"), |
||||||
|
get_host_override_slice("foo.test.google.fr:1234", config), deadline, |
||||||
|
NULL); |
||||||
|
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); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_INITIAL_METADATA; |
||||||
|
op->data.send_initial_metadata.count = 0; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_RECV_INITIAL_METADATA; |
||||||
|
op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
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->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(1), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
error = |
||||||
|
grpc_server_request_call(f.server, &s, &call_details, |
||||||
|
&request_metadata_recv, f.cq, f.cq, tag(100)); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(100), 1); |
||||||
|
cq_verify(cqv); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_INITIAL_METADATA; |
||||||
|
op->data.send_initial_metadata.count = 0; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER; |
||||||
|
op->data.recv_close_on_server.cancelled = &was_cancelled; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(101), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
for (i = 0; i < messages; i++) { |
||||||
|
request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1); |
||||||
|
response_payload = grpc_raw_byte_buffer_create(&response_payload_slice, 1); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_MESSAGE; |
||||||
|
op->data.send_message.send_message = request_payload; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
op->op = GRPC_OP_RECV_MESSAGE; |
||||||
|
op->data.recv_message.recv_message = &response_payload_recv; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_RECV_MESSAGE; |
||||||
|
op->data.recv_message.recv_message = &request_payload_recv; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(102), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(102), 1); |
||||||
|
cq_verify(cqv); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_MESSAGE; |
||||||
|
op->data.send_message.send_message = response_payload; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(103), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(103), 1); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(2), 1); |
||||||
|
cq_verify(cqv); |
||||||
|
|
||||||
|
grpc_byte_buffer_destroy(request_payload); |
||||||
|
grpc_byte_buffer_destroy(response_payload); |
||||||
|
grpc_byte_buffer_destroy(request_payload_recv); |
||||||
|
grpc_byte_buffer_destroy(response_payload_recv); |
||||||
|
} |
||||||
|
|
||||||
|
grpc_slice_unref(request_payload_slice); |
||||||
|
grpc_slice_unref(response_payload_slice); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(3), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
memset(ops, 0, sizeof(ops)); |
||||||
|
op = ops; |
||||||
|
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_UNIMPLEMENTED; |
||||||
|
grpc_slice status_details = grpc_slice_from_static_string("xyz"); |
||||||
|
op->data.send_status_from_server.status_details = &status_details; |
||||||
|
op->flags = 0; |
||||||
|
op->reserved = NULL; |
||||||
|
op++; |
||||||
|
error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(104), NULL); |
||||||
|
GPR_ASSERT(GRPC_CALL_OK == error); |
||||||
|
|
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(1), 1); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(3), 1); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(101), 1); |
||||||
|
CQ_EXPECT_COMPLETION(cqv, tag(104), 1); |
||||||
|
cq_verify(cqv); |
||||||
|
|
||||||
|
grpc_call_unref(c); |
||||||
|
grpc_call_unref(s); |
||||||
|
|
||||||
|
cq_verifier_destroy(cqv); |
||||||
|
|
||||||
|
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_slice_unref(details); |
||||||
|
|
||||||
|
end_test(&f); |
||||||
|
config.tear_down_data(&f); |
||||||
|
{ |
||||||
|
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; |
||||||
|
grpc_channel_args_destroy(&exec_ctx, client_args); |
||||||
|
grpc_channel_args_destroy(&exec_ctx, server_args); |
||||||
|
grpc_exec_ctx_finish(&exec_ctx); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
void stream_compression_ping_pong_streaming(grpc_end2end_test_config config) { |
||||||
|
int i; |
||||||
|
|
||||||
|
for (i = 1; i < 10; i++) { |
||||||
|
test_pingpong_streaming(config, i); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
void stream_compression_ping_pong_streaming_pre_init(void) {} |
||||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue