Introduce ahead-of-time registration of (method,host) pairs

pull/1261/head
Craig Tiller 10 years ago committed by Craig Tiller
parent 2ab3998251
commit 08453374b2
  1. 253
      Makefile
  2. 10
      include/grpc/grpc.h
  3. 81
      src/core/surface/channel.c
  4. 1
      test/core/end2end/gen_build_json.py
  5. 220
      test/core/end2end/tests/registered_call.c
  6. 63
      tools/run_tests/tests.json

File diff suppressed because one or more lines are too long

@ -421,6 +421,16 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel,
const char *method, const char *host,
gpr_timespec deadline);
/* Pre-register a method/host pair on a channel. */
void *grpc_channel_register_call(grpc_channel *channel, const char *method,
const char *host);
/* Create a call given a handle returned from grpc_channel_register_call */
grpc_call *grpc_channel_create_registered_call(grpc_channel *channel,
grpc_completion_queue *completion_queue,
void *registered_call_handle,
gpr_timespec deadline);
/* Start a batch of operations defined in the array ops; when complete, post a
completion of type 'tag' to the completion queue bound to the call.
The order of ops specified in the batch has no significance.

@ -43,6 +43,12 @@
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
typedef struct registered_call {
grpc_mdelem *path;
grpc_mdelem *authority;
struct registered_call *next;
} registered_call;
struct grpc_channel {
int is_client;
gpr_refcount refs;
@ -51,6 +57,9 @@ struct grpc_channel {
grpc_mdstr *grpc_message_string;
grpc_mdstr *path_string;
grpc_mdstr *authority_string;
gpr_mu registered_call_mu;
registered_call *registered_calls;
};
#define CHANNEL_STACK_FROM_CHANNEL(c) ((grpc_channel_stack *)((c)+1))
@ -75,18 +84,17 @@ grpc_channel *grpc_channel_create_from_filters(
channel->authority_string = grpc_mdstr_from_string(mdctx, ":authority");
grpc_channel_stack_init(filters, num_filters, args, channel->metadata_context,
CHANNEL_STACK_FROM_CHANNEL(channel));
gpr_mu_init(&channel->registered_call_mu);
channel->registered_calls = NULL;
return channel;
}
static void do_nothing(void *ignored, grpc_op_error error) {}
grpc_call *grpc_channel_create_call(grpc_channel *channel,
grpc_completion_queue *cq,
const char *method, const char *host,
gpr_timespec absolute_deadline) {
static grpc_call *grpc_channel_create_call_internal(
grpc_channel *channel, grpc_completion_queue *cq,
grpc_mdelem *path_mdelem, grpc_mdelem *authority_mdelem, gpr_timespec deadline) {
grpc_call *call;
grpc_mdelem *path_mdelem;
grpc_mdelem *authority_mdelem;
grpc_call_op op;
if (!channel->is_client) {
@ -97,11 +105,6 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel,
call = grpc_call_create(channel, cq, NULL);
/* Add :path and :authority headers. */
/* TODO(klempner): Consider optimizing this by stashing mdelems for common
values of method and host. */
path_mdelem = grpc_mdelem_from_metadata_strings(
channel->metadata_context, grpc_mdstr_ref(channel->path_string),
grpc_mdstr_from_string(channel->metadata_context, method));
op.type = GRPC_SEND_METADATA;
op.dir = GRPC_CALL_DOWN;
op.flags = 0;
@ -110,18 +113,14 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel,
op.user_data = NULL;
grpc_call_execute_op(call, &op);
grpc_mdstr_ref(channel->authority_string);
authority_mdelem = grpc_mdelem_from_metadata_strings(
channel->metadata_context, channel->authority_string,
grpc_mdstr_from_string(channel->metadata_context, host));
op.data.metadata = authority_mdelem;
grpc_call_execute_op(call, &op);
if (0 != gpr_time_cmp(absolute_deadline, gpr_inf_future)) {
if (0 != gpr_time_cmp(deadline, gpr_inf_future)) {
op.type = GRPC_SEND_DEADLINE;
op.dir = GRPC_CALL_DOWN;
op.flags = 0;
op.data.deadline = absolute_deadline;
op.data.deadline = deadline;
op.done_cb = do_nothing;
op.user_data = NULL;
grpc_call_execute_op(call, &op);
@ -130,6 +129,22 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel,
return call;
}
grpc_call *grpc_channel_create_call(grpc_channel *channel,
grpc_completion_queue *cq,
const char *method, const char *host,
gpr_timespec deadline) {
return grpc_channel_create_call_internal(
channel,
cq,
grpc_mdelem_from_metadata_strings(
channel->metadata_context, grpc_mdstr_ref(channel->path_string),
grpc_mdstr_from_string(channel->metadata_context, method)),
grpc_mdelem_from_metadata_strings(
channel->metadata_context, grpc_mdstr_ref(channel->authority_string),
grpc_mdstr_from_string(channel->metadata_context, host)),
deadline);
}
grpc_call *grpc_channel_create_call_old(grpc_channel *channel,
const char *method, const char *host,
gpr_timespec absolute_deadline) {
@ -137,6 +152,30 @@ grpc_call *grpc_channel_create_call_old(grpc_channel *channel,
absolute_deadline);
}
void *grpc_channel_register_call(grpc_channel *channel, const char *method,
const char *host) {
registered_call *rc = gpr_malloc(sizeof(registered_call));
rc->path = grpc_mdelem_from_metadata_strings(
channel->metadata_context, grpc_mdstr_ref(channel->path_string),
grpc_mdstr_from_string(channel->metadata_context, method));
rc->authority = grpc_mdelem_from_metadata_strings(
channel->metadata_context, grpc_mdstr_ref(channel->authority_string),
grpc_mdstr_from_string(channel->metadata_context, host));
gpr_mu_lock(&channel->registered_call_mu);
rc->next = channel->registered_calls;
channel->registered_calls = rc;
gpr_mu_unlock(&channel->registered_call_mu);
return rc;
}
grpc_call *grpc_channel_create_registered_call(grpc_channel *channel,
grpc_completion_queue *completion_queue,
void *registered_call_handle,
gpr_timespec deadline) {
registered_call *rc = registered_call_handle;
return grpc_channel_create_call_internal(channel, completion_queue, grpc_mdelem_ref(rc->path), grpc_mdelem_ref(rc->authority), deadline);
}
void grpc_channel_internal_ref(grpc_channel *channel) {
gpr_ref(&channel->refs);
}
@ -148,7 +187,15 @@ static void destroy_channel(void *p, int ok) {
grpc_mdstr_unref(channel->grpc_message_string);
grpc_mdstr_unref(channel->path_string);
grpc_mdstr_unref(channel->authority_string);
while (channel->registered_calls) {
registered_call *rc = channel->registered_calls;
channel->registered_calls = rc->next;
grpc_mdelem_unref(rc->path);
grpc_mdelem_unref(rc->authority);
gpr_free(rc);
}
grpc_mdctx_unref(channel->metadata_context);
gpr_mu_destroy(&channel->registered_call_mu);
gpr_free(channel);
}

@ -69,6 +69,7 @@ END2END_TESTS = [
'request_with_payload',
'simple_delayed_request',
'simple_request',
'registered_call',
'thread_stress',
'writes_done_hangs_with_pending_read',

@ -0,0 +1,220 @@
/*
*
* 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 <unistd.h>
#include "src/core/support/string.h"
#include <grpc/byte_buffer.h>
#include <grpc/grpc.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"
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;
grpc_completion_type type;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
GPR_ASSERT(ev);
type = ev->type;
grpc_event_finish(ev);
} while (type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return;
grpc_server_shutdown(f->server);
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->server_cq);
drain_cq(f->server_cq);
grpc_completion_queue_destroy(f->server_cq);
grpc_completion_queue_shutdown(f->client_cq);
drain_cq(f->client_cq);
grpc_completion_queue_destroy(f->client_cq);
}
static void simple_request_body(grpc_end2end_test_fixture f, void *rc) {
grpc_call *c;
grpc_call *s;
gpr_timespec deadline = five_seconds_time();
cq_verifier *v_client = cq_verifier_create(f.client_cq);
cq_verifier *v_server = cq_verifier_create(f.server_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;
char *details = NULL;
size_t details_capacity = 0;
int was_cancelled = 2;
c = grpc_channel_create_registered_call(f.client, f.client_cq, rc, 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;
op->data.send_initial_metadata.count = 0;
op++;
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
op++;
op->op = GRPC_OP_RECV_INITIAL_METADATA;
op->data.recv_initial_metadata = &initial_metadata_recv;
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++;
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.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 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_UNIMPLEMENTED;
op->data.send_status_from_server.status_details = "xyz";
op++;
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
op->data.recv_close_on_server.cancelled = &was_cancelled;
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
cq_expect_completion(v_server, tag(102), GRPC_OP_OK);
cq_verify(v_server);
cq_expect_completion(v_client, tag(1), GRPC_OP_OK);
cq_verify(v_client);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
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:1234"));
GPR_ASSERT(was_cancelled == 0);
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(v_client);
cq_verifier_destroy(v_server);
}
static void test_invoke_simple_request(grpc_end2end_test_config config) {
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
void *rc = grpc_channel_register_call(f.client, "/foo", "foo.test.google.fr:1234");
simple_request_body(f, rc);
end_test(&f);
config.tear_down_data(&f);
}
static void test_invoke_10_simple_requests(grpc_end2end_test_config config) {
int i;
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
void *rc = grpc_channel_register_call(f.client, "/foo", "foo.test.google.fr:1234");
for (i = 0; i < 10; i++) {
simple_request_body(f, rc);
gpr_log(GPR_INFO, "Passed simple request %d", i);
}
end_test(&f);
config.tear_down_data(&f);
}
void grpc_end2end_tests(grpc_end2end_test_config config) {
test_invoke_simple_request(config);
test_invoke_10_simple_requests(config);
}

@ -918,6 +918,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fake_security_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1359,6 +1368,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -1800,6 +1818,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_fullstack_uds_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2241,6 +2268,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_simple_ssl_fullstack_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -2682,6 +2718,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_simple_ssl_with_oauth2_fullstack_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -3123,6 +3168,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",
@ -3564,6 +3618,15 @@
"posix"
]
},
{
"flaky": false,
"language": "c",
"name": "chttp2_socket_pair_one_byte_at_a_time_registered_call_test",
"platforms": [
"windows",
"posix"
]
},
{
"flaky": false,
"language": "c",

Loading…
Cancel
Save