diff options
Diffstat (limited to 'src/core')
43 files changed, 1194 insertions, 1239 deletions
diff --git a/src/core/census/aggregation.h b/src/core/census/aggregation.h new file mode 100644 index 0000000000..e9bc6ada96 --- /dev/null +++ b/src/core/census/aggregation.h @@ -0,0 +1,66 @@ +/* + * + * 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 <stddef.h> + +#ifndef GRPC_INTERNAL_CORE_CENSUS_AGGREGATION_H +#define GRPC_INTERNAL_CORE_CENSUS_AGGREGATION_H + +/** Structure used to describe an aggregation type. */ +struct census_aggregation_ops { + /* Create a new aggregation. The pointer returned can be used in future calls + to clone(), free(), record(), data() and reset(). */ + void *(*create)(const void *create_arg); + /* Make a copy of an aggregation created by create() */ + void *(*clone)(const void *aggregation); + /* Destroy an aggregation created by create() */ + void (*free)(void *aggregation); + /* Record a new value against aggregation. */ + void (*record)(void *aggregation, double value); + /* Return current aggregation data. The caller must cast this object into + the correct type for the aggregation result. The object returned can be + freed by using free_data(). */ + void *(*data)(const void *aggregation); + /* free data returned by data() */ + void (*free_data)(void *data); + /* Reset an aggregation to default (zero) values. */ + void (*reset)(void *aggregation); + /* Merge 'from' aggregation into 'to'. Both aggregations must be compatible */ + void (*merge)(void *to, const void *from); + /* Fill buffer with printable string version of aggregation contents. For + debugging only. Returns the number of bytes added to buffer (a value == n + implies the buffer was of insufficient size). */ + size_t (*print)(const void *aggregation, char *buffer, size_t n); +}; + +#endif /* GRPC_INTERNAL_CORE_CENSUS_AGGREGATION_H */ diff --git a/src/core/census/context.c b/src/core/census/context.c index df238ec98c..cab58b653c 100644 --- a/src/core/census/context.c +++ b/src/core/census/context.c @@ -44,16 +44,3 @@ size_t census_context_serialize(const census_context *context, char *buffer, /* TODO(aveitch): implement serialization */ return 0; } - -int census_context_deserialize(const char *buffer, census_context **context) { - int ret = 0; - if (buffer != NULL) { - /* TODO(aveitch): implement deserialization. */ - ret = 1; - } - *context = gpr_malloc(sizeof(census_context)); - memset(*context, 0, sizeof(census_context)); - return ret; -} - -void census_context_destroy(census_context *context) { gpr_free(context); } diff --git a/src/core/census/grpc_context.c b/src/core/census/grpc_context.c index 11f1eb3d5d..429f3ec9db 100644 --- a/src/core/census/grpc_context.c +++ b/src/core/census/grpc_context.c @@ -35,24 +35,11 @@ #include <grpc/grpc.h> #include "src/core/surface/call.h" -static void grpc_census_context_destroy(void *context) { - census_context_destroy((census_context *)context); -} - void grpc_census_call_set_context(grpc_call *call, census_context *context) { if (census_enabled() == CENSUS_FEATURE_NONE) { return; } - if (context == NULL) { - if (grpc_call_is_client(call)) { - census_context *context_ptr; - census_context_deserialize(NULL, &context_ptr); - grpc_call_context_set(call, GRPC_CONTEXT_TRACING, context_ptr, - grpc_census_context_destroy); - } else { - /* TODO(aveitch): server side context code to be implemented. */ - } - } else { + if (context != NULL) { grpc_call_context_set(call, GRPC_CONTEXT_TRACING, context, NULL); } } diff --git a/src/core/census/grpc_filter.c b/src/core/census/grpc_filter.c index fbedb35661..8b6ba1d472 100644 --- a/src/core/census/grpc_filter.c +++ b/src/core/census/grpc_filter.c @@ -36,12 +36,11 @@ #include <stdio.h> #include <string.h> -#include "include/grpc/census.h" -#include "src/core/census/rpc_stat_id.h" #include "src/core/channel/channel_stack.h" #include "src/core/channel/noop_filter.h" #include "src/core/statistics/census_interface.h" #include "src/core/statistics/census_rpc_stats.h" +#include <grpc/census.h> #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <grpc/support/slice.h> @@ -173,25 +172,15 @@ static void destroy_channel_elem(grpc_channel_element* elem) { } const grpc_channel_filter grpc_client_census_filter = { - client_start_transport_op, - grpc_channel_next_op, - sizeof(call_data), - client_init_call_elem, - client_destroy_call_elem, - sizeof(channel_data), - init_channel_elem, - destroy_channel_elem, - grpc_call_next_get_peer, - "census-client"}; + client_start_transport_op, grpc_channel_next_op, + sizeof(call_data), client_init_call_elem, + client_destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, + grpc_call_next_get_peer, "census-client"}; const grpc_channel_filter grpc_server_census_filter = { - server_start_transport_op, - grpc_channel_next_op, - sizeof(call_data), - server_init_call_elem, - server_destroy_call_elem, - sizeof(channel_data), - init_channel_elem, - destroy_channel_elem, - grpc_call_next_get_peer, - "census-server"}; + server_start_transport_op, grpc_channel_next_op, + sizeof(call_data), server_init_call_elem, + server_destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, + grpc_call_next_get_peer, "census-server"}; diff --git a/src/core/census/grpc_filter.h b/src/core/census/grpc_filter.h index 1453c05d28..b3de3adc94 100644 --- a/src/core/census/grpc_filter.h +++ b/src/core/census/grpc_filter.h @@ -31,8 +31,8 @@ * */ -#ifndef GRPC_INTERNAL_CORE_CHANNEL_CENSUS_FILTER_H -#define GRPC_INTERNAL_CORE_CHANNEL_CENSUS_FILTER_H +#ifndef GRPC_INTERNAL_CORE_CENSUS_GRPC_FILTER_H +#define GRPC_INTERNAL_CORE_CENSUS_GRPC_FILTER_H #include "src/core/channel/channel_stack.h" @@ -41,4 +41,4 @@ extern const grpc_channel_filter grpc_client_census_filter; extern const grpc_channel_filter grpc_server_census_filter; -#endif /* GRPC_INTERNAL_CORE_CHANNEL_CENSUS_FILTER_H */ +#endif /* GRPC_INTERNAL_CORE_CENSUS_GRPC_FILTER_H */ diff --git a/src/core/census/operation.c b/src/core/census/operation.c new file mode 100644 index 0000000000..118eb0a47a --- /dev/null +++ b/src/core/census/operation.c @@ -0,0 +1,63 @@ +/* + * 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 <grpc/census.h> + +/* TODO(aveitch): These are all placeholder implementations. */ + +census_timestamp census_start_rpc_op_timestamp(void) { + census_timestamp ct; + /* TODO(aveitch): assumes gpr_timespec implementation of census_timestamp. */ + ct.ts = gpr_now(GPR_CLOCK_MONOTONIC); + return ct; +} + +census_context *census_start_client_rpc_op( + const census_context *context, gpr_int64 rpc_name_id, + const census_rpc_name_info *rpc_name_info, const char *peer, int trace_mask, + const census_timestamp *start_time) { + return NULL; +} + +census_context *census_start_server_rpc_op( + const char *buffer, gpr_int64 rpc_name_id, + const census_rpc_name_info *rpc_name_info, const char *peer, int trace_mask, + census_timestamp *start_time) { + return NULL; +} + +census_context *census_start_op(census_context *context, const char *family, + const char *name, int trace_mask) { + return NULL; +} + +void census_end_op(census_context *context, int status) {} diff --git a/src/core/census/rpc_stat_id.h b/src/core/census/rpc_metric_id.h index fc0aa6f43f..1d8e8806f5 100644 --- a/src/core/census/rpc_stat_id.h +++ b/src/core/census/rpc_metric_id.h @@ -31,16 +31,21 @@ * */ -#ifndef CENSUS_RPC_STAT_ID_H -#define CENSUS_RPC_STAT_ID_H +#ifndef CENSUS_RPC_METRIC_ID_H +#define CENSUS_RPC_METRIC_ID_H -/* Stats ID's used for RPC measurements. */ -#define CENSUS_INVALID_STAT_ID 0 /* ID 0 is always invalid */ -#define CENSUS_RPC_CLIENT_REQUESTS 1 /* Count of client requests sent. */ -#define CENSUS_RPC_SERVER_REQUESTS 2 /* Count of server requests sent. */ -#define CENSUS_RPC_CLIENT_ERRORS 3 /* Client error counts. */ -#define CENSUS_RPC_SERVER_ERRORS 4 /* Server error counts. */ -#define CENSUS_RPC_CLIENT_LATENCY 5 /* Client side request latency. */ -#define CENSUS_RPC_SERVER_LATENCY 6 /* Server side request latency. */ +/* Metric ID's used for RPC measurements. */ +/* Count of client requests sent. */ +#define CENSUS_METRIC_RPC_CLIENT_REQUESTS ((gpr_uint32)0) +/* Count of server requests sent. */ +#define CENSUS_METRIC_RPC_SERVER_REQUESTS ((gpr_uint32)1) +/* Client error counts. */ +#define CENSUS_METRIC_RPC_CLIENT_ERRORS ((gpr_uint32)2) +/* Server error counts. */ +#define CENSUS_METRIC_RPC_SERVER_ERRORS ((gpr_uint32)3) +/* Client side request latency. */ +#define CENSUS_METRIC_RPC_CLIENT_LATENCY ((gpr_uint32)4) +/* Server side request latency. */ +#define CENSUS_METRIC_RPC_SERVER_LATENCY ((gpr_uint32)5) -#endif /* CENSUS_RPC_STAT_ID_H */ +#endif /* CENSUS_RPC_METRIC_ID_H */ diff --git a/src/core/census/record_stat.c b/src/core/census/tracing.c index 3dd918618b..ae38773c0a 100644 --- a/src/core/census/record_stat.c +++ b/src/core/census/tracing.c @@ -32,7 +32,14 @@ */ #include <grpc/census.h> -#include "src/core/census/rpc_stat_id.h" -void census_record_stat(census_context *context, census_stat *stats, - size_t nstats) {} +/* TODO(aveitch): These are all placeholder implementations. */ + +int census_trace_mask(const census_context *context) { + return CENSUS_TRACE_MASK_NONE; +} + +void census_set_trace_mask(int trace_mask) {} + +void census_trace_print(census_context *context, gpr_uint32 type, + const char *buffer, size_t n) {} diff --git a/src/core/channel/http_client_filter.c b/src/core/channel/http_client_filter.c index 2b61d33c29..ec832a0367 100644 --- a/src/core/channel/http_client_filter.c +++ b/src/core/channel/http_client_filter.c @@ -85,16 +85,14 @@ static grpc_mdelem *client_filter(void *user_data, grpc_mdelem *md) { static void hc_on_recv(void *user_data, int success) { grpc_call_element *elem = user_data; call_data *calld = elem->call_data; - if (success) { - size_t i; - size_t nops = calld->recv_ops->nops; - grpc_stream_op *ops = calld->recv_ops->ops; - for (i = 0; i < nops; i++) { - grpc_stream_op *op = &ops[i]; - if (op->type != GRPC_OP_METADATA) continue; - calld->got_initial_metadata = 1; - grpc_metadata_batch_filter(&op->data.metadata, client_filter, elem); - } + size_t i; + size_t nops = calld->recv_ops->nops; + grpc_stream_op *ops = calld->recv_ops->ops; + for (i = 0; i < nops; i++) { + grpc_stream_op *op = &ops[i]; + if (op->type != GRPC_OP_METADATA) continue; + calld->got_initial_metadata = 1; + grpc_metadata_batch_filter(&op->data.metadata, client_filter, elem); } calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success); } diff --git a/src/core/httpcli/httpcli.c b/src/core/httpcli/httpcli.c index 9012070e8e..1e38479eb1 100644 --- a/src/core/httpcli/httpcli.c +++ b/src/core/httpcli/httpcli.c @@ -61,6 +61,10 @@ typedef struct { grpc_httpcli_context *context; grpc_pollset *pollset; grpc_iomgr_object iomgr_obj; + gpr_slice_buffer incoming; + gpr_slice_buffer outgoing; + grpc_iomgr_closure on_read; + grpc_iomgr_closure done_write; } internal_request; static grpc_httpcli_get_override g_get_override = NULL; @@ -99,73 +103,70 @@ static void finish(internal_request *req, int success) { gpr_slice_unref(req->request_text); gpr_free(req->host); grpc_iomgr_unregister_object(&req->iomgr_obj); + gpr_slice_buffer_destroy(&req->incoming); + gpr_slice_buffer_destroy(&req->outgoing); gpr_free(req); } -static void on_read(void *user_data, gpr_slice *slices, size_t nslices, - grpc_endpoint_cb_status status) { +static void on_read(void *user_data, int success); + +static void do_read(internal_request *req) { + switch (grpc_endpoint_read(req->ep, &req->incoming, &req->on_read)) { + case GRPC_ENDPOINT_DONE: + on_read(req, 1); + break; + case GRPC_ENDPOINT_PENDING: + break; + case GRPC_ENDPOINT_ERROR: + on_read(req, 0); + break; + } +} + +static void on_read(void *user_data, int success) { internal_request *req = user_data; size_t i; - for (i = 0; i < nslices; i++) { - if (GPR_SLICE_LENGTH(slices[i])) { + for (i = 0; i < req->incoming.count; i++) { + if (GPR_SLICE_LENGTH(req->incoming.slices[i])) { req->have_read_byte = 1; - if (!grpc_httpcli_parser_parse(&req->parser, slices[i])) { + if (!grpc_httpcli_parser_parse(&req->parser, req->incoming.slices[i])) { finish(req, 0); - goto done; + return; } } } - switch (status) { - case GRPC_ENDPOINT_CB_OK: - grpc_endpoint_notify_on_read(req->ep, on_read, req); - break; - case GRPC_ENDPOINT_CB_EOF: - case GRPC_ENDPOINT_CB_ERROR: - case GRPC_ENDPOINT_CB_SHUTDOWN: - if (!req->have_read_byte) { - next_address(req); - } else { - finish(req, grpc_httpcli_parser_eof(&req->parser)); - } - break; - } - -done: - for (i = 0; i < nslices; i++) { - gpr_slice_unref(slices[i]); + if (success) { + do_read(req); + } else if (!req->have_read_byte) { + next_address(req); + } else { + finish(req, grpc_httpcli_parser_eof(&req->parser)); } } -static void on_written(internal_request *req) { - grpc_endpoint_notify_on_read(req->ep, on_read, req); -} +static void on_written(internal_request *req) { do_read(req); } -static void done_write(void *arg, grpc_endpoint_cb_status status) { +static void done_write(void *arg, int success) { internal_request *req = arg; - switch (status) { - case GRPC_ENDPOINT_CB_OK: - on_written(req); - break; - case GRPC_ENDPOINT_CB_EOF: - case GRPC_ENDPOINT_CB_SHUTDOWN: - case GRPC_ENDPOINT_CB_ERROR: - next_address(req); - break; + if (success) { + on_written(req); + } else { + next_address(req); } } static void start_write(internal_request *req) { gpr_slice_ref(req->request_text); - switch ( - grpc_endpoint_write(req->ep, &req->request_text, 1, done_write, req)) { - case GRPC_ENDPOINT_WRITE_DONE: + gpr_slice_buffer_add(&req->outgoing, req->request_text); + switch (grpc_endpoint_write(req->ep, &req->outgoing, &req->done_write)) { + case GRPC_ENDPOINT_DONE: on_written(req); break; - case GRPC_ENDPOINT_WRITE_PENDING: + case GRPC_ENDPOINT_PENDING: break; - case GRPC_ENDPOINT_WRITE_ERROR: + case GRPC_ENDPOINT_ERROR: finish(req, 0); break; } @@ -237,6 +238,10 @@ void grpc_httpcli_get(grpc_httpcli_context *context, grpc_pollset *pollset, request->handshaker ? request->handshaker : &grpc_httpcli_plaintext; req->context = context; req->pollset = pollset; + grpc_iomgr_closure_init(&req->on_read, on_read, req); + grpc_iomgr_closure_init(&req->done_write, done_write, req); + gpr_slice_buffer_init(&req->incoming); + gpr_slice_buffer_init(&req->outgoing); gpr_asprintf(&name, "HTTP:GET:%s:%s", request->host, request->path); grpc_iomgr_register_object(&req->iomgr_obj, name); gpr_free(name); @@ -270,7 +275,11 @@ void grpc_httpcli_post(grpc_httpcli_context *context, grpc_pollset *pollset, request->handshaker ? request->handshaker : &grpc_httpcli_plaintext; req->context = context; req->pollset = pollset; - gpr_asprintf(&name, "HTTP:GET:%s:%s", request->host, request->path); + grpc_iomgr_closure_init(&req->on_read, on_read, req); + grpc_iomgr_closure_init(&req->done_write, done_write, req); + gpr_slice_buffer_init(&req->incoming); + gpr_slice_buffer_init(&req->outgoing); + gpr_asprintf(&name, "HTTP:POST:%s:%s", request->host, request->path); grpc_iomgr_register_object(&req->iomgr_obj, name); gpr_free(name); req->host = gpr_strdup(request->host); diff --git a/src/core/iomgr/endpoint.c b/src/core/iomgr/endpoint.c index 8ee14bce9b..a7878e31dd 100644 --- a/src/core/iomgr/endpoint.c +++ b/src/core/iomgr/endpoint.c @@ -33,17 +33,16 @@ #include "src/core/iomgr/endpoint.h" -void grpc_endpoint_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb, - void *user_data) { - ep->vtable->notify_on_read(ep, cb, user_data); +grpc_endpoint_op_status grpc_endpoint_read(grpc_endpoint *ep, + gpr_slice_buffer *slices, + grpc_iomgr_closure *cb) { + return ep->vtable->read(ep, slices, cb); } -grpc_endpoint_write_status grpc_endpoint_write(grpc_endpoint *ep, - gpr_slice *slices, - size_t nslices, - grpc_endpoint_write_cb cb, - void *user_data) { - return ep->vtable->write(ep, slices, nslices, cb, user_data); +grpc_endpoint_op_status grpc_endpoint_write(grpc_endpoint *ep, + gpr_slice_buffer *slices, + grpc_iomgr_closure *cb) { + return ep->vtable->write(ep, slices, cb); } void grpc_endpoint_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) { diff --git a/src/core/iomgr/endpoint.h b/src/core/iomgr/endpoint.h index ea92a500e8..d14d52d561 100644 --- a/src/core/iomgr/endpoint.h +++ b/src/core/iomgr/endpoint.h @@ -37,6 +37,7 @@ #include "src/core/iomgr/pollset.h" #include "src/core/iomgr/pollset_set.h" #include <grpc/support/slice.h> +#include <grpc/support/slice_buffer.h> #include <grpc/support/time.h> /* An endpoint caps a streaming channel between two communicating processes. @@ -45,31 +46,17 @@ typedef struct grpc_endpoint grpc_endpoint; typedef struct grpc_endpoint_vtable grpc_endpoint_vtable; -typedef enum grpc_endpoint_cb_status { - GRPC_ENDPOINT_CB_OK = 0, /* Call completed successfully */ - GRPC_ENDPOINT_CB_EOF, /* Call completed successfully, end of file reached */ - GRPC_ENDPOINT_CB_SHUTDOWN, /* Call interrupted by shutdown */ - GRPC_ENDPOINT_CB_ERROR /* Call interrupted by socket error */ -} grpc_endpoint_cb_status; - -typedef enum grpc_endpoint_write_status { - GRPC_ENDPOINT_WRITE_DONE, /* completed immediately, cb won't be called */ - GRPC_ENDPOINT_WRITE_PENDING, /* cb will be called when completed */ - GRPC_ENDPOINT_WRITE_ERROR /* write errored out, cb won't be called */ -} grpc_endpoint_write_status; - -typedef void (*grpc_endpoint_read_cb)(void *user_data, gpr_slice *slices, - size_t nslices, - grpc_endpoint_cb_status error); -typedef void (*grpc_endpoint_write_cb)(void *user_data, - grpc_endpoint_cb_status error); +typedef enum grpc_endpoint_op_status { + GRPC_ENDPOINT_DONE, /* completed immediately, cb won't be called */ + GRPC_ENDPOINT_PENDING, /* cb will be called when completed */ + GRPC_ENDPOINT_ERROR /* write errored out, cb won't be called */ +} grpc_endpoint_op_status; struct grpc_endpoint_vtable { - void (*notify_on_read)(grpc_endpoint *ep, grpc_endpoint_read_cb cb, - void *user_data); - grpc_endpoint_write_status (*write)(grpc_endpoint *ep, gpr_slice *slices, - size_t nslices, grpc_endpoint_write_cb cb, - void *user_data); + grpc_endpoint_op_status (*read)(grpc_endpoint *ep, gpr_slice_buffer *slices, + grpc_iomgr_closure *cb); + grpc_endpoint_op_status (*write)(grpc_endpoint *ep, gpr_slice_buffer *slices, + grpc_iomgr_closure *cb); void (*add_to_pollset)(grpc_endpoint *ep, grpc_pollset *pollset); void (*add_to_pollset_set)(grpc_endpoint *ep, grpc_pollset_set *pollset); void (*shutdown)(grpc_endpoint *ep); @@ -77,26 +64,32 @@ struct grpc_endpoint_vtable { char *(*get_peer)(grpc_endpoint *ep); }; -/* When data is available on the connection, calls the callback with slices. */ -void grpc_endpoint_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb, - void *user_data); +/* When data is available on the connection, calls the callback with slices. + Callback success indicates that the endpoint can accept more reads, failure + indicates the endpoint is closed. + Valid slices may be placed into \a slices even on callback success == 0. */ +grpc_endpoint_op_status grpc_endpoint_read( + grpc_endpoint *ep, gpr_slice_buffer *slices, + grpc_iomgr_closure *cb) GRPC_MUST_USE_RESULT; char *grpc_endpoint_get_peer(grpc_endpoint *ep); /* Write slices out to the socket. If the connection is ready for more data after the end of the call, it - returns GRPC_ENDPOINT_WRITE_DONE. - Otherwise it returns GRPC_ENDPOINT_WRITE_PENDING and calls cb when the - connection is ready for more data. */ -grpc_endpoint_write_status grpc_endpoint_write(grpc_endpoint *ep, - gpr_slice *slices, - size_t nslices, - grpc_endpoint_write_cb cb, - void *user_data); + returns GRPC_ENDPOINT_DONE. + Otherwise it returns GRPC_ENDPOINT_PENDING and calls cb when the + connection is ready for more data. + \a slices may be mutated at will by the endpoint until cb is called. + No guarantee is made to the content of slices after a write EXCEPT that + it is a valid slice buffer. + */ +grpc_endpoint_op_status grpc_endpoint_write( + grpc_endpoint *ep, gpr_slice_buffer *slices, + grpc_iomgr_closure *cb) GRPC_MUST_USE_RESULT; /* Causes any pending read/write callbacks to run immediately with - GRPC_ENDPOINT_CB_SHUTDOWN status */ + success==0 */ void grpc_endpoint_shutdown(grpc_endpoint *ep); void grpc_endpoint_destroy(grpc_endpoint *ep); diff --git a/src/core/iomgr/iocp_windows.c b/src/core/iomgr/iocp_windows.c index 09a457dd9a..006f8b2abf 100644 --- a/src/core/iomgr/iocp_windows.c +++ b/src/core/iomgr/iocp_windows.c @@ -52,7 +52,6 @@ static OVERLAPPED g_iocp_custom_overlap; static gpr_event g_shutdown_iocp; static gpr_event g_iocp_done; -static gpr_atm g_orphans = 0; static gpr_atm g_custom_events = 0; static HANDLE g_iocp; @@ -92,22 +91,13 @@ static void do_iocp_work() { gpr_log(GPR_ERROR, "Unknown IOCP operation"); abort(); } - GPR_ASSERT(info->outstanding); - if (socket->orphan) { - info->outstanding = 0; - if (!socket->read_info.outstanding && !socket->write_info.outstanding) { - grpc_winsocket_destroy(socket); - gpr_atm_full_fetch_add(&g_orphans, -1); - } - return; - } success = WSAGetOverlappedResult(socket->socket, &info->overlapped, &bytes, FALSE, &flags); info->bytes_transfered = bytes; info->wsa_error = success ? 0 : WSAGetLastError(); GPR_ASSERT(overlapped == &info->overlapped); - gpr_mu_lock(&socket->state_mu); GPR_ASSERT(!info->has_pending_iocp); + gpr_mu_lock(&socket->state_mu); if (info->cb) { f = info->cb; opaque = info->opaque; @@ -120,9 +110,8 @@ static void do_iocp_work() { } static void iocp_loop(void *p) { - while (gpr_atm_acq_load(&g_orphans) || gpr_atm_acq_load(&g_custom_events) || + while (gpr_atm_acq_load(&g_custom_events) || !gpr_event_get(&g_shutdown_iocp)) { - grpc_maybe_call_delayed_callbacks(NULL, 1); do_iocp_work(); } @@ -175,12 +164,6 @@ void grpc_iocp_add_socket(grpc_winsocket *socket) { GPR_ASSERT(ret == g_iocp); } -void grpc_iocp_socket_orphan(grpc_winsocket *socket) { - GPR_ASSERT(!socket->orphan); - gpr_atm_full_fetch_add(&g_orphans, 1); - socket->orphan = 1; -} - /* Calling notify_on_read or write means either of two things: -) The IOCP already completed in the background, and we need to call the callback now. diff --git a/src/core/iomgr/iocp_windows.h b/src/core/iomgr/iocp_windows.h index ee3847a229..7d2dc45176 100644 --- a/src/core/iomgr/iocp_windows.h +++ b/src/core/iomgr/iocp_windows.h @@ -42,7 +42,6 @@ void grpc_iocp_init(void); void grpc_iocp_kick(void); void grpc_iocp_shutdown(void); void grpc_iocp_add_socket(grpc_winsocket *); -void grpc_iocp_socket_orphan(grpc_winsocket *); void grpc_socket_notify_on_write(grpc_winsocket *, void (*cb)(void *, int success), void *opaque); diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c index fdc9adf4af..1dd03992ae 100644 --- a/src/core/iomgr/iomgr.c +++ b/src/core/iomgr/iomgr.c @@ -108,8 +108,14 @@ static size_t count_objects(void) { return n; } -void grpc_iomgr_shutdown(void) { +static void dump_objects(const char *kind) { grpc_iomgr_object *obj; + for (obj = g_root_object.next; obj != &g_root_object; obj = obj->next) { + gpr_log(GPR_DEBUG, "%s OBJECT: %s %p", kind, obj->name, obj); + } +} + +void grpc_iomgr_shutdown(void) { grpc_iomgr_closure *closure; gpr_timespec shutdown_deadline = gpr_time_add( gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_seconds(10, GPR_TIMESPAN)); @@ -151,12 +157,14 @@ void grpc_iomgr_shutdown(void) { } if (g_root_object.next != &g_root_object) { int timeout = 0; - gpr_timespec short_deadline = gpr_time_add( + while (g_cbs_head == NULL) { + gpr_timespec short_deadline = gpr_time_add( gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(100, GPR_TIMESPAN)); - while (gpr_cv_wait(&g_rcv, &g_mu, short_deadline) && g_cbs_head == NULL) { - if (gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), shutdown_deadline) > 0) { - timeout = 1; - break; + if (gpr_cv_wait(&g_rcv, &g_mu, short_deadline) && g_cbs_head == NULL) { + if (gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), shutdown_deadline) > 0) { + timeout = 1; + break; + } } } if (timeout) { @@ -164,9 +172,7 @@ void grpc_iomgr_shutdown(void) { "Failed to free %d iomgr objects before shutdown deadline: " "memory leaks are likely", count_objects()); - for (obj = g_root_object.next; obj != &g_root_object; obj = obj->next) { - gpr_log(GPR_DEBUG, "LEAKED OBJECT: %s %p", obj->name, obj); - } + dump_objects("LEAKED"); break; } } @@ -188,7 +194,7 @@ void grpc_iomgr_register_object(grpc_iomgr_object *obj, const char *name) { obj->name = gpr_strdup(name); gpr_mu_lock(&g_mu); obj->next = &g_root_object; - obj->prev = obj->next->prev; + obj->prev = g_root_object.prev; obj->next->prev = obj->prev->next = obj; gpr_mu_unlock(&g_mu); } diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c index fe66ebed77..8f62ce2954 100644 --- a/src/core/iomgr/pollset_multipoller_with_epoll.c +++ b/src/core/iomgr/pollset_multipoller_with_epoll.c @@ -99,7 +99,6 @@ static void perform_delayed_add(void *arg, int iomgr_status) { if (da->pollset->shutting_down) { /* We don't care about this pollset anymore. */ if (da->pollset->in_flight_cbs == 0 && !da->pollset->called_shutdown) { - GPR_ASSERT(!grpc_pollset_has_workers(da->pollset)); da->pollset->called_shutdown = 1; do_shutdown_cb = 1; } diff --git a/src/core/iomgr/socket_windows.c b/src/core/iomgr/socket_windows.c index 7d8421376b..557ca82226 100644 --- a/src/core/iomgr/socket_windows.c +++ b/src/core/iomgr/socket_windows.c @@ -35,8 +35,12 @@ #ifdef GPR_WINSOCK_SOCKET +#include <winsock2.h> +#include <mswsock.h> + #include <grpc/support/alloc.h> #include <grpc/support/log.h> +#include <grpc/support/log_win32.h> #include <grpc/support/string_util.h> #include "src/core/iomgr/iocp_windows.h" @@ -62,46 +66,30 @@ grpc_winsocket *grpc_winsocket_create(SOCKET socket, const char *name) { operations to abort them. We need to do that this way because of the various callsites of that function, which happens to be in various mutex hold states, and that'd be unsafe to call them directly. */ -int grpc_winsocket_shutdown(grpc_winsocket *winsocket) { - int callbacks_set = 0; - SOCKET socket; - gpr_mu_lock(&winsocket->state_mu); - socket = winsocket->socket; - if (winsocket->read_info.cb) { - callbacks_set++; - grpc_iomgr_closure_init(&winsocket->shutdown_closure, - winsocket->read_info.cb, - winsocket->read_info.opaque); - grpc_iomgr_add_delayed_callback(&winsocket->shutdown_closure, 0); - } - if (winsocket->write_info.cb) { - callbacks_set++; - grpc_iomgr_closure_init(&winsocket->shutdown_closure, - winsocket->write_info.cb, - winsocket->write_info.opaque); - grpc_iomgr_add_delayed_callback(&winsocket->shutdown_closure, 0); - } - gpr_mu_unlock(&winsocket->state_mu); - closesocket(socket); - return callbacks_set; -} +void grpc_winsocket_shutdown(grpc_winsocket *winsocket) { + /* Grab the function pointer for DisconnectEx for that specific socket. + It may change depending on the interface. */ + int status; + GUID guid = WSAID_DISCONNECTEX; + LPFN_DISCONNECTEX DisconnectEx; + DWORD ioctl_num_bytes; -/* Abandons a socket. Either we're going to queue it up for garbage collecting - from the IO Completion Port thread, or destroy it immediately. Note that this - mechanisms assumes that we're either always waiting for an operation, or we - explicitly know that we don't. If there is a future case where we can have - an "idle" socket which is neither trying to read or write, we'd start leaking - both memory and sockets. */ -void grpc_winsocket_orphan(grpc_winsocket *winsocket) { - grpc_iomgr_unregister_object(&winsocket->iomgr_object); - if (winsocket->read_info.outstanding || winsocket->write_info.outstanding) { - grpc_iocp_socket_orphan(winsocket); + status = WSAIoctl(winsocket->socket, SIO_GET_EXTENSION_FUNCTION_POINTER, + &guid, sizeof(guid), &DisconnectEx, sizeof(DisconnectEx), + &ioctl_num_bytes, NULL, NULL); + + if (status == 0) { + DisconnectEx(winsocket->socket, NULL, 0, 0); } else { - grpc_winsocket_destroy(winsocket); + char *utf8_message = gpr_format_message(WSAGetLastError()); + gpr_log(GPR_ERROR, "Unable to retrieve DisconnectEx pointer : %s", utf8_message); + gpr_free(utf8_message); } + closesocket(winsocket->socket); } void grpc_winsocket_destroy(grpc_winsocket *winsocket) { + grpc_iomgr_unregister_object(&winsocket->iomgr_object); gpr_mu_destroy(&winsocket->state_mu); gpr_free(winsocket); } diff --git a/src/core/iomgr/socket_windows.h b/src/core/iomgr/socket_windows.h index ecf2530173..498921e0fd 100644 --- a/src/core/iomgr/socket_windows.h +++ b/src/core/iomgr/socket_windows.h @@ -68,8 +68,6 @@ typedef struct grpc_winsocket_callback_info { /* The results of the overlapped operation. */ DWORD bytes_transfered; int wsa_error; - /* A boolean indicating that we started an operation. */ - int outstanding; } grpc_winsocket_callback_info; /* This is a wrapper to a Windows socket. A socket can have one outstanding @@ -92,10 +90,6 @@ typedef struct grpc_winsocket { /* You can't add the same socket twice to the same IO Completion Port. This prevents that. */ int added_to_iocp; - /* A boolean to indicate that the caller has abandoned that socket, but - there is a pending operation that the IO Completion Port will have to - wait for. The socket will be collected at that time. */ - int orphan; grpc_iomgr_closure shutdown_closure; @@ -108,14 +102,10 @@ typedef struct grpc_winsocket { grpc_winsocket *grpc_winsocket_create(SOCKET socket, const char *name); /* Initiate an asynchronous shutdown of the socket. Will call off any pending - operation to cancel them. Returns the number of callbacks that got setup. */ -int grpc_winsocket_shutdown(grpc_winsocket *socket); + operation to cancel them. */ +void grpc_winsocket_shutdown(grpc_winsocket *socket); -/* Abandon a socket. */ -void grpc_winsocket_orphan(grpc_winsocket *socket); - -/* Destroy a socket. Should only be called by the IO Completion Port thread, - or by grpc_winsocket_orphan if there's no pending operation. */ +/* Destroy a socket. Should only be called if there's no pending operation. */ void grpc_winsocket_destroy(grpc_winsocket *socket); #endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKET_WINDOWS_H */ diff --git a/src/core/iomgr/tcp_client_windows.c b/src/core/iomgr/tcp_client_windows.c index 79a58fe2af..05198dbff4 100644 --- a/src/core/iomgr/tcp_client_windows.c +++ b/src/core/iomgr/tcp_client_windows.c @@ -60,13 +60,13 @@ typedef struct { grpc_alarm alarm; char *addr_name; int refs; - int aborted; } async_connect; -static void async_connect_cleanup(async_connect *ac) { +static void async_connect_unlock_and_cleanup(async_connect *ac) { int done = (--ac->refs == 0); gpr_mu_unlock(&ac->mu); if (done) { + if (ac->socket != NULL) grpc_winsocket_destroy(ac->socket); gpr_mu_destroy(&ac->mu); gpr_free(ac->addr_name); gpr_free(ac); @@ -77,10 +77,11 @@ static void on_alarm(void *acp, int occured) { async_connect *ac = acp; gpr_mu_lock(&ac->mu); /* If the alarm didn't occur, it got cancelled. */ + gpr_log(GPR_DEBUG, "on_alarm: %p", ac->socket); if (ac->socket != NULL && occured) { grpc_winsocket_shutdown(ac->socket); } - async_connect_cleanup(ac); + async_connect_unlock_and_cleanup(ac); } static void on_connect(void *acp, int from_iocp) { @@ -90,51 +91,33 @@ static void on_connect(void *acp, int from_iocp) { grpc_winsocket_callback_info *info = &ac->socket->write_info; void (*cb)(void *arg, grpc_endpoint *tcp) = ac->cb; void *cb_arg = ac->cb_arg; - int aborted; - + grpc_alarm_cancel(&ac->alarm); gpr_mu_lock(&ac->mu); - aborted = ac->aborted; + + gpr_log(GPR_DEBUG, "on_connect: %p", ac->socket); if (from_iocp) { DWORD transfered_bytes = 0; DWORD flags; BOOL wsa_success = WSAGetOverlappedResult(sock, &info->overlapped, &transfered_bytes, FALSE, &flags); - info->outstanding = 0; GPR_ASSERT(transfered_bytes == 0); if (!wsa_success) { char *utf8_message = gpr_format_message(WSAGetLastError()); gpr_log(GPR_ERROR, "on_connect error: %s", utf8_message); gpr_free(utf8_message); - } else if (!aborted) { + } else { ep = grpc_tcp_create(ac->socket, ac->addr_name); + ac->socket = NULL; } - } else { - gpr_log(GPR_ERROR, "on_connect is shutting down"); - /* If the connection timeouts, we will still get a notification from - the IOCP whatever happens. So we're just going to flag that connection - as being in the process of being aborted, and wait for the IOCP. We - can't just orphan the socket now, because the IOCP might already have - gotten a successful connection, which is our worst-case scenario. - We need to call our callback now to respect the deadline. */ - ac->aborted = 1; - gpr_mu_unlock(&ac->mu); - cb(cb_arg, NULL); - return; } - ac->socket->write_info.outstanding = 0; - - /* If we don't have an endpoint, it means the connection failed, - so it doesn't matter if it aborted or failed. We need to orphan - that socket. */ - if (!ep || aborted) grpc_winsocket_orphan(ac->socket); - async_connect_cleanup(ac); + async_connect_unlock_and_cleanup(ac); /* If the connection was aborted, the callback was already called when the deadline was met. */ - if (!aborted) cb(cb_arg, ep); + cb(cb_arg, ep); } /* Tries to issue one async connection, then schedules both an IOCP @@ -196,7 +179,6 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp), socket = grpc_winsocket_create(sock, "client"); info = &socket->write_info; - info->outstanding = 1; success = ConnectEx(sock, addr, addr_len, NULL, 0, NULL, &info->overlapped); /* It wouldn't be unusual to get a success immediately. But we'll still get @@ -216,11 +198,9 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp), gpr_mu_init(&ac->mu); ac->refs = 2; ac->addr_name = grpc_sockaddr_to_uri(addr); - ac->aborted = 0; grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, gpr_now(GPR_CLOCK_MONOTONIC)); - socket->write_info.outstanding = 1; grpc_socket_notify_on_write(socket, on_connect, ac); return; @@ -228,8 +208,8 @@ failure: utf8_message = gpr_format_message(WSAGetLastError()); gpr_log(GPR_ERROR, message, utf8_message); gpr_free(utf8_message); - if (socket) { - grpc_winsocket_orphan(socket); + if (socket != NULL) { + grpc_winsocket_destroy(socket); } else if (sock != INVALID_SOCKET) { closesocket(sock); } diff --git a/src/core/iomgr/tcp_posix.c b/src/core/iomgr/tcp_posix.c index 360e6ebd8c..0db7cd9f0e 100644 --- a/src/core/iomgr/tcp_posix.c +++ b/src/core/iomgr/tcp_posix.c @@ -61,209 +61,8 @@ #define SENDMSG_FLAGS 0 #endif -/* Holds a slice array and associated state. */ -typedef struct grpc_tcp_slice_state { - gpr_slice *slices; /* Array of slices */ - size_t nslices; /* Size of slices array. */ - ssize_t first_slice; /* First valid slice in array */ - ssize_t last_slice; /* Last valid slice in array */ - gpr_slice working_slice; /* pointer to original final slice */ - int working_slice_valid; /* True if there is a working slice */ - int memory_owned; /* True if slices array is owned */ -} grpc_tcp_slice_state; - int grpc_tcp_trace = 0; -static void slice_state_init(grpc_tcp_slice_state *state, gpr_slice *slices, - size_t nslices, size_t valid_slices) { - state->slices = slices; - state->nslices = nslices; - if (valid_slices == 0) { - state->first_slice = -1; - } else { - state->first_slice = 0; - } - state->last_slice = valid_slices - 1; - state->working_slice_valid = 0; - state->memory_owned = 0; -} - -/* Returns true if there is still available data */ -static int slice_state_has_available(grpc_tcp_slice_state *state) { - return state->first_slice != -1 && state->last_slice >= state->first_slice; -} - -static ssize_t slice_state_slices_allocated(grpc_tcp_slice_state *state) { - if (state->first_slice == -1) { - return 0; - } else { - return state->last_slice - state->first_slice + 1; - } -} - -static void slice_state_realloc(grpc_tcp_slice_state *state, size_t new_size) { - /* TODO(klempner): use realloc instead when first_slice is 0 */ - /* TODO(klempner): Avoid a realloc in cases where it is unnecessary */ - gpr_slice *slices = state->slices; - size_t original_size = slice_state_slices_allocated(state); - size_t i; - gpr_slice *new_slices = gpr_malloc(sizeof(gpr_slice) * new_size); - - for (i = 0; i < original_size; ++i) { - new_slices[i] = slices[i + state->first_slice]; - } - - state->slices = new_slices; - state->last_slice = original_size - 1; - if (original_size > 0) { - state->first_slice = 0; - } else { - state->first_slice = -1; - } - state->nslices = new_size; - - if (state->memory_owned) { - gpr_free(slices); - } - state->memory_owned = 1; -} - -static void slice_state_remove_prefix(grpc_tcp_slice_state *state, - size_t prefix_bytes) { - gpr_slice *current_slice = &state->slices[state->first_slice]; - size_t current_slice_size; - - while (slice_state_has_available(state)) { - current_slice_size = GPR_SLICE_LENGTH(*current_slice); - if (current_slice_size > prefix_bytes) { - /* TODO(klempner): Get rid of the extra refcount created here by adding a - native "trim the first N bytes" operation to splice */ - /* TODO(klempner): This really shouldn't be modifying the current slice - unless we own the slices array. */ - gpr_slice tail; - tail = gpr_slice_split_tail(current_slice, prefix_bytes); - gpr_slice_unref(*current_slice); - *current_slice = tail; - return; - } else { - gpr_slice_unref(*current_slice); - ++state->first_slice; - ++current_slice; - prefix_bytes -= current_slice_size; - } - } -} - -static void slice_state_destroy(grpc_tcp_slice_state *state) { - while (slice_state_has_available(state)) { - gpr_slice_unref(state->slices[state->first_slice]); - ++state->first_slice; - } - - if (state->memory_owned) { - gpr_free(state->slices); - state->memory_owned = 0; - } -} - -void slice_state_transfer_ownership(grpc_tcp_slice_state *state, - gpr_slice **slices, size_t *nslices) { - *slices = state->slices + state->first_slice; - *nslices = state->last_slice - state->first_slice + 1; - - state->first_slice = -1; - state->last_slice = -1; -} - -/* Fills iov with the first min(iov_size, available) slices, returns number - filled */ -static size_t slice_state_to_iovec(grpc_tcp_slice_state *state, - struct iovec *iov, size_t iov_size) { - size_t nslices = state->last_slice - state->first_slice + 1; - gpr_slice *slices = state->slices + state->first_slice; - size_t i; - if (nslices < iov_size) { - iov_size = nslices; - } - - for (i = 0; i < iov_size; ++i) { - iov[i].iov_base = GPR_SLICE_START_PTR(slices[i]); - iov[i].iov_len = GPR_SLICE_LENGTH(slices[i]); - } - return iov_size; -} - -/* Makes n blocks available at the end of state, writes them into iov, and - returns the number of bytes allocated */ -static size_t slice_state_append_blocks_into_iovec(grpc_tcp_slice_state *state, - struct iovec *iov, size_t n, - size_t slice_size) { - size_t target_size; - size_t i; - size_t allocated_bytes; - ssize_t allocated_slices = slice_state_slices_allocated(state); - - if (n - state->working_slice_valid >= state->nslices - state->last_slice) { - /* Need to grow the slice array */ - target_size = state->nslices; - do { - target_size = target_size * 2; - } while (target_size < allocated_slices + n - state->working_slice_valid); - /* TODO(klempner): If this ever needs to support both prefix removal and - append, we should be smarter about the growth logic here */ - slice_state_realloc(state, target_size); - } - - i = 0; - allocated_bytes = 0; - - if (state->working_slice_valid) { - iov[0].iov_base = GPR_SLICE_END_PTR(state->slices[state->last_slice]); - iov[0].iov_len = GPR_SLICE_LENGTH(state->working_slice) - - GPR_SLICE_LENGTH(state->slices[state->last_slice]); - allocated_bytes += iov[0].iov_len; - ++i; - state->slices[state->last_slice] = state->working_slice; - state->working_slice_valid = 0; - } - - for (; i < n; ++i) { - ++state->last_slice; - state->slices[state->last_slice] = gpr_slice_malloc(slice_size); - iov[i].iov_base = GPR_SLICE_START_PTR(state->slices[state->last_slice]); - iov[i].iov_len = slice_size; - allocated_bytes += slice_size; - } - if (state->first_slice == -1) { - state->first_slice = 0; - } - return allocated_bytes; -} - -/* Remove the last n bytes from state */ -/* TODO(klempner): Consider having this defer actual deletion until later */ -static void slice_state_remove_last(grpc_tcp_slice_state *state, size_t bytes) { - while (bytes > 0 && slice_state_has_available(state)) { - if (GPR_SLICE_LENGTH(state->slices[state->last_slice]) > bytes) { - state->working_slice = state->slices[state->last_slice]; - state->working_slice_valid = 1; - /* TODO(klempner): Combine these into a single operation that doesn't need - to refcount */ - gpr_slice_unref(gpr_slice_split_tail( - &state->slices[state->last_slice], - GPR_SLICE_LENGTH(state->slices[state->last_slice]) - bytes)); - bytes = 0; - } else { - bytes -= GPR_SLICE_LENGTH(state->slices[state->last_slice]); - gpr_slice_unref(state->slices[state->last_slice]); - --state->last_slice; - if (state->last_slice == -1) { - state->first_slice = -1; - } - } - } -} - typedef struct { grpc_endpoint base; grpc_fd *em_fd; @@ -273,80 +72,111 @@ typedef struct { size_t slice_size; gpr_refcount refcount; - grpc_endpoint_read_cb read_cb; - void *read_user_data; - grpc_endpoint_write_cb write_cb; - void *write_user_data; + gpr_slice_buffer *incoming_buffer; + gpr_slice_buffer *outgoing_buffer; + /** slice within outgoing_buffer to write next */ + size_t outgoing_slice_idx; + /** byte within outgoing_buffer->slices[outgoing_slice_idx] to write next */ + size_t outgoing_byte_idx; - grpc_tcp_slice_state write_state; + grpc_iomgr_closure *read_cb; + grpc_iomgr_closure *write_cb; grpc_iomgr_closure read_closure; grpc_iomgr_closure write_closure; - grpc_iomgr_closure handle_read_closure; - char *peer_string; } grpc_tcp; -static void grpc_tcp_handle_read(void *arg /* grpc_tcp */, int success); -static void grpc_tcp_handle_write(void *arg /* grpc_tcp */, int success); +static void tcp_handle_read(void *arg /* grpc_tcp */, int success); +static void tcp_handle_write(void *arg /* grpc_tcp */, int success); -static void grpc_tcp_shutdown(grpc_endpoint *ep) { +static void tcp_shutdown(grpc_endpoint *ep) { grpc_tcp *tcp = (grpc_tcp *)ep; grpc_fd_shutdown(tcp->em_fd); } -static void grpc_tcp_unref(grpc_tcp *tcp) { - int refcount_zero = gpr_unref(&tcp->refcount); - if (refcount_zero) { - grpc_fd_orphan(tcp->em_fd, NULL, "tcp_unref_orphan"); - gpr_free(tcp->peer_string); - gpr_free(tcp); +static void tcp_free(grpc_tcp *tcp) { + grpc_fd_orphan(tcp->em_fd, NULL, "tcp_unref_orphan"); + gpr_free(tcp->peer_string); + gpr_free(tcp); +} + +/*#define GRPC_TCP_REFCOUNT_DEBUG*/ +#ifdef GRPC_TCP_REFCOUNT_DEBUG +#define TCP_UNREF(tcp, reason) tcp_unref((tcp), (reason), __FILE__, __LINE__) +#define TCP_REF(tcp, reason) tcp_ref((tcp), (reason), __FILE__, __LINE__) +static void tcp_unref(grpc_tcp *tcp, const char *reason, const char *file, + int line) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP unref %p : %s %d -> %d", tcp, + reason, tcp->refcount.count, tcp->refcount.count - 1); + if (gpr_unref(&tcp->refcount)) { + tcp_free(tcp); } } -static void grpc_tcp_destroy(grpc_endpoint *ep) { +static void tcp_ref(grpc_tcp *tcp, const char *reason, const char *file, + int line) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP ref %p : %s %d -> %d", tcp, + reason, tcp->refcount.count, tcp->refcount.count + 1); + gpr_ref(&tcp->refcount); +} +#else +#define TCP_UNREF(tcp, reason) tcp_unref((tcp)) +#define TCP_REF(tcp, reason) tcp_ref((tcp)) +static void tcp_unref(grpc_tcp *tcp) { + if (gpr_unref(&tcp->refcount)) { + tcp_free(tcp); + } +} + +static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); } +#endif + +static void tcp_destroy(grpc_endpoint *ep) { grpc_tcp *tcp = (grpc_tcp *)ep; - grpc_tcp_unref(tcp); + TCP_UNREF(tcp, "destroy"); } -static void call_read_cb(grpc_tcp *tcp, gpr_slice *slices, size_t nslices, - grpc_endpoint_cb_status status) { - grpc_endpoint_read_cb cb = tcp->read_cb; +static void call_read_cb(grpc_tcp *tcp, int success) { + grpc_iomgr_closure *cb = tcp->read_cb; if (grpc_tcp_trace) { size_t i; - gpr_log(GPR_DEBUG, "read: status=%d", status); - for (i = 0; i < nslices; i++) { - char *dump = gpr_dump_slice(slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII); + gpr_log(GPR_DEBUG, "read: success=%d", success); + for (i = 0; i < tcp->incoming_buffer->count; i++) { + char *dump = gpr_dump_slice(tcp->incoming_buffer->slices[i], + GPR_DUMP_HEX | GPR_DUMP_ASCII); gpr_log(GPR_DEBUG, "READ %p: %s", tcp, dump); gpr_free(dump); } } tcp->read_cb = NULL; - cb(tcp->read_user_data, slices, nslices, status); + tcp->incoming_buffer = NULL; + cb->cb(cb->cb_arg, success); } -#define INLINE_SLICE_BUFFER_SIZE 8 #define MAX_READ_IOVEC 4 -static void grpc_tcp_continue_read(grpc_tcp *tcp) { - gpr_slice static_read_slices[INLINE_SLICE_BUFFER_SIZE]; +static void tcp_continue_read(grpc_tcp *tcp) { struct msghdr msg; struct iovec iov[MAX_READ_IOVEC]; ssize_t read_bytes; - ssize_t allocated_bytes; - struct grpc_tcp_slice_state read_state; - gpr_slice *final_slices; - size_t final_nslices; + size_t i; GPR_ASSERT(!tcp->finished_edge); + GPR_ASSERT(tcp->iov_size <= MAX_READ_IOVEC); + GPR_ASSERT(tcp->incoming_buffer->count <= MAX_READ_IOVEC); GRPC_TIMER_BEGIN(GRPC_PTAG_HANDLE_READ, 0); - slice_state_init(&read_state, static_read_slices, INLINE_SLICE_BUFFER_SIZE, - 0); - allocated_bytes = slice_state_append_blocks_into_iovec( - &read_state, iov, tcp->iov_size, tcp->slice_size); + while (tcp->incoming_buffer->count < (size_t)tcp->iov_size) { + gpr_slice_buffer_add_indexed(tcp->incoming_buffer, + gpr_slice_malloc(tcp->slice_size)); + } + for (i = 0; i < tcp->incoming_buffer->count; i++) { + iov[i].iov_base = GPR_SLICE_START_PTR(tcp->incoming_buffer->slices[i]); + iov[i].iov_len = GPR_SLICE_LENGTH(tcp->incoming_buffer->slices[i]); + } msg.msg_name = NULL; msg.msg_namelen = 0; @@ -362,106 +192,105 @@ static void grpc_tcp_continue_read(grpc_tcp *tcp) { } while (read_bytes < 0 && errno == EINTR); GRPC_TIMER_END(GRPC_PTAG_RECVMSG, 0); - if (read_bytes < allocated_bytes) { - /* TODO(klempner): Consider a second read first, in hopes of getting a - * quick EAGAIN and saving a bunch of allocations. */ - slice_state_remove_last(&read_state, read_bytes < 0 - ? allocated_bytes - : allocated_bytes - read_bytes); - } - if (read_bytes < 0) { - /* NB: After calling the user_cb a parallel call of the read handler may + /* NB: After calling call_read_cb a parallel call of the read handler may * be running. */ if (errno == EAGAIN) { if (tcp->iov_size > 1) { tcp->iov_size /= 2; } - if (slice_state_has_available(&read_state)) { - /* TODO(klempner): We should probably do the call into the application - without all this junk on the stack */ - /* FIXME(klempner): Refcount properly */ - slice_state_transfer_ownership(&read_state, &final_slices, - &final_nslices); - tcp->finished_edge = 1; - call_read_cb(tcp, final_slices, final_nslices, GRPC_ENDPOINT_CB_OK); - slice_state_destroy(&read_state); - grpc_tcp_unref(tcp); - } else { - /* We've consumed the edge, request a new one */ - slice_state_destroy(&read_state); - grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure); - } + /* We've consumed the edge, request a new one */ + grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure); } else { /* TODO(klempner): Log interesting errors */ - call_read_cb(tcp, NULL, 0, GRPC_ENDPOINT_CB_ERROR); - slice_state_destroy(&read_state); - grpc_tcp_unref(tcp); + gpr_slice_buffer_reset_and_unref(tcp->incoming_buffer); + call_read_cb(tcp, 0); + TCP_UNREF(tcp, "read"); } } else if (read_bytes == 0) { /* 0 read size ==> end of stream */ - if (slice_state_has_available(&read_state)) { - /* there were bytes already read: pass them up to the application */ - slice_state_transfer_ownership(&read_state, &final_slices, - &final_nslices); - call_read_cb(tcp, final_slices, final_nslices, GRPC_ENDPOINT_CB_EOF); - } else { - call_read_cb(tcp, NULL, 0, GRPC_ENDPOINT_CB_EOF); - } - slice_state_destroy(&read_state); - grpc_tcp_unref(tcp); + gpr_slice_buffer_reset_and_unref(tcp->incoming_buffer); + call_read_cb(tcp, 0); + TCP_UNREF(tcp, "read"); } else { - if (tcp->iov_size < MAX_READ_IOVEC) { + GPR_ASSERT((size_t)read_bytes <= tcp->incoming_buffer->length); + if ((size_t)read_bytes < tcp->incoming_buffer->length) { + gpr_slice_buffer_trim_end(tcp->incoming_buffer, + tcp->incoming_buffer->length - read_bytes); + } else if (tcp->iov_size < MAX_READ_IOVEC) { ++tcp->iov_size; } - GPR_ASSERT(slice_state_has_available(&read_state)); - slice_state_transfer_ownership(&read_state, &final_slices, &final_nslices); - call_read_cb(tcp, final_slices, final_nslices, GRPC_ENDPOINT_CB_OK); - slice_state_destroy(&read_state); - grpc_tcp_unref(tcp); + GPR_ASSERT((size_t)read_bytes == tcp->incoming_buffer->length); + call_read_cb(tcp, 1); + TCP_UNREF(tcp, "read"); } GRPC_TIMER_END(GRPC_PTAG_HANDLE_READ, 0); } -static void grpc_tcp_handle_read(void *arg /* grpc_tcp */, int success) { +static void tcp_handle_read(void *arg /* grpc_tcp */, int success) { grpc_tcp *tcp = (grpc_tcp *)arg; GPR_ASSERT(!tcp->finished_edge); if (!success) { - call_read_cb(tcp, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN); - grpc_tcp_unref(tcp); + gpr_slice_buffer_reset_and_unref(tcp->incoming_buffer); + call_read_cb(tcp, 0); + TCP_UNREF(tcp, "read"); } else { - grpc_tcp_continue_read(tcp); + tcp_continue_read(tcp); } } -static void grpc_tcp_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb, - void *user_data) { +static grpc_endpoint_op_status tcp_read(grpc_endpoint *ep, + gpr_slice_buffer *incoming_buffer, + grpc_iomgr_closure *cb) { grpc_tcp *tcp = (grpc_tcp *)ep; GPR_ASSERT(tcp->read_cb == NULL); tcp->read_cb = cb; - tcp->read_user_data = user_data; - gpr_ref(&tcp->refcount); + tcp->incoming_buffer = incoming_buffer; + gpr_slice_buffer_reset_and_unref(incoming_buffer); + TCP_REF(tcp, "read"); if (tcp->finished_edge) { tcp->finished_edge = 0; grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure); } else { - tcp->handle_read_closure.cb_arg = tcp; - grpc_iomgr_add_delayed_callback(&tcp->handle_read_closure, 1); + grpc_iomgr_add_delayed_callback(&tcp->read_closure, 1); } + /* TODO(ctiller): immediate return */ + return GRPC_ENDPOINT_PENDING; } #define MAX_WRITE_IOVEC 16 -static grpc_endpoint_write_status grpc_tcp_flush(grpc_tcp *tcp) { +static grpc_endpoint_op_status tcp_flush(grpc_tcp *tcp) { struct msghdr msg; struct iovec iov[MAX_WRITE_IOVEC]; int iov_size; ssize_t sent_length; - grpc_tcp_slice_state *state = &tcp->write_state; + ssize_t sending_length; + ssize_t trailing; + ssize_t unwind_slice_idx; + ssize_t unwind_byte_idx; for (;;) { - iov_size = slice_state_to_iovec(state, iov, MAX_WRITE_IOVEC); + sending_length = 0; + unwind_slice_idx = tcp->outgoing_slice_idx; + unwind_byte_idx = tcp->outgoing_byte_idx; + for (iov_size = 0; tcp->outgoing_slice_idx != tcp->outgoing_buffer->count && + iov_size != MAX_WRITE_IOVEC; + iov_size++) { + iov[iov_size].iov_base = + GPR_SLICE_START_PTR( + tcp->outgoing_buffer->slices[tcp->outgoing_slice_idx]) + + tcp->outgoing_byte_idx; + iov[iov_size].iov_len = + GPR_SLICE_LENGTH( + tcp->outgoing_buffer->slices[tcp->outgoing_slice_idx]) - + tcp->outgoing_byte_idx; + sending_length += iov[iov_size].iov_len; + tcp->outgoing_slice_idx++; + tcp->outgoing_byte_idx = 0; + } + GPR_ASSERT(iov_size > 0); msg.msg_name = NULL; msg.msg_namelen = 0; @@ -480,70 +309,75 @@ static grpc_endpoint_write_status grpc_tcp_flush(grpc_tcp *tcp) { if (sent_length < 0) { if (errno == EAGAIN) { - return GRPC_ENDPOINT_WRITE_PENDING; + tcp->outgoing_slice_idx = unwind_slice_idx; + tcp->outgoing_byte_idx = unwind_byte_idx; + return GRPC_ENDPOINT_PENDING; } else { /* TODO(klempner): Log some of these */ - slice_state_destroy(state); - return GRPC_ENDPOINT_WRITE_ERROR; + return GRPC_ENDPOINT_ERROR; } } - /* TODO(klempner): Probably better to batch this after we finish flushing */ - slice_state_remove_prefix(state, sent_length); + GPR_ASSERT(tcp->outgoing_byte_idx == 0); + trailing = sending_length - sent_length; + while (trailing > 0) { + ssize_t slice_length; + + tcp->outgoing_slice_idx--; + slice_length = GPR_SLICE_LENGTH( + tcp->outgoing_buffer->slices[tcp->outgoing_slice_idx]); + if (slice_length > trailing) { + tcp->outgoing_byte_idx = slice_length - trailing; + break; + } else { + trailing -= slice_length; + } + } - if (!slice_state_has_available(state)) { - return GRPC_ENDPOINT_WRITE_DONE; + if (tcp->outgoing_slice_idx == tcp->outgoing_buffer->count) { + return GRPC_ENDPOINT_DONE; } }; } -static void grpc_tcp_handle_write(void *arg /* grpc_tcp */, int success) { +static void tcp_handle_write(void *arg /* grpc_tcp */, int success) { grpc_tcp *tcp = (grpc_tcp *)arg; - grpc_endpoint_write_status write_status; - grpc_endpoint_cb_status cb_status; - grpc_endpoint_write_cb cb; + grpc_endpoint_op_status status; + grpc_iomgr_closure *cb; if (!success) { - slice_state_destroy(&tcp->write_state); cb = tcp->write_cb; tcp->write_cb = NULL; - cb(tcp->write_user_data, GRPC_ENDPOINT_CB_SHUTDOWN); - grpc_tcp_unref(tcp); + cb->cb(cb->cb_arg, 0); + TCP_UNREF(tcp, "write"); return; } GRPC_TIMER_BEGIN(GRPC_PTAG_TCP_CB_WRITE, 0); - write_status = grpc_tcp_flush(tcp); - if (write_status == GRPC_ENDPOINT_WRITE_PENDING) { + status = tcp_flush(tcp); + if (status == GRPC_ENDPOINT_PENDING) { grpc_fd_notify_on_write(tcp->em_fd, &tcp->write_closure); } else { - slice_state_destroy(&tcp->write_state); - if (write_status == GRPC_ENDPOINT_WRITE_DONE) { - cb_status = GRPC_ENDPOINT_CB_OK; - } else { - cb_status = GRPC_ENDPOINT_CB_ERROR; - } cb = tcp->write_cb; tcp->write_cb = NULL; - cb(tcp->write_user_data, cb_status); - grpc_tcp_unref(tcp); + cb->cb(cb->cb_arg, status == GRPC_ENDPOINT_DONE); + TCP_UNREF(tcp, "write"); } GRPC_TIMER_END(GRPC_PTAG_TCP_CB_WRITE, 0); } -static grpc_endpoint_write_status grpc_tcp_write(grpc_endpoint *ep, - gpr_slice *slices, - size_t nslices, - grpc_endpoint_write_cb cb, - void *user_data) { +static grpc_endpoint_op_status tcp_write(grpc_endpoint *ep, + gpr_slice_buffer *buf, + grpc_iomgr_closure *cb) { grpc_tcp *tcp = (grpc_tcp *)ep; - grpc_endpoint_write_status status; + grpc_endpoint_op_status status; if (grpc_tcp_trace) { size_t i; - for (i = 0; i < nslices; i++) { - char *data = gpr_dump_slice(slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII); + for (i = 0; i < buf->count; i++) { + char *data = + gpr_dump_slice(buf->slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII); gpr_log(GPR_DEBUG, "WRITE %p: %s", tcp, data); gpr_free(data); } @@ -551,15 +385,19 @@ static grpc_endpoint_write_status grpc_tcp_write(grpc_endpoint *ep, GRPC_TIMER_BEGIN(GRPC_PTAG_TCP_WRITE, 0); GPR_ASSERT(tcp->write_cb == NULL); - slice_state_init(&tcp->write_state, slices, nslices, nslices); - status = grpc_tcp_flush(tcp); - if (status == GRPC_ENDPOINT_WRITE_PENDING) { - /* TODO(klempner): Consider inlining rather than malloc for small nslices */ - slice_state_realloc(&tcp->write_state, nslices); - gpr_ref(&tcp->refcount); + if (buf->length == 0) { + GRPC_TIMER_END(GRPC_PTAG_TCP_WRITE, 0); + return GRPC_ENDPOINT_DONE; + } + tcp->outgoing_buffer = buf; + tcp->outgoing_slice_idx = 0; + tcp->outgoing_byte_idx = 0; + + status = tcp_flush(tcp); + if (status == GRPC_ENDPOINT_PENDING) { + TCP_REF(tcp, "write"); tcp->write_cb = cb; - tcp->write_user_data = user_data; grpc_fd_notify_on_write(tcp->em_fd, &tcp->write_closure); } @@ -567,27 +405,25 @@ static grpc_endpoint_write_status grpc_tcp_write(grpc_endpoint *ep, return status; } -static void grpc_tcp_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) { +static void tcp_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) { grpc_tcp *tcp = (grpc_tcp *)ep; grpc_pollset_add_fd(pollset, tcp->em_fd); } -static void grpc_tcp_add_to_pollset_set(grpc_endpoint *ep, - grpc_pollset_set *pollset_set) { +static void tcp_add_to_pollset_set(grpc_endpoint *ep, + grpc_pollset_set *pollset_set) { grpc_tcp *tcp = (grpc_tcp *)ep; grpc_pollset_set_add_fd(pollset_set, tcp->em_fd); } -static char *grpc_tcp_get_peer(grpc_endpoint *ep) { +static char *tcp_get_peer(grpc_endpoint *ep) { grpc_tcp *tcp = (grpc_tcp *)ep; return gpr_strdup(tcp->peer_string); } static const grpc_endpoint_vtable vtable = { - grpc_tcp_notify_on_read, grpc_tcp_write, - grpc_tcp_add_to_pollset, grpc_tcp_add_to_pollset_set, - grpc_tcp_shutdown, grpc_tcp_destroy, - grpc_tcp_get_peer}; + tcp_read, tcp_write, tcp_add_to_pollset, tcp_add_to_pollset_set, + tcp_shutdown, tcp_destroy, tcp_get_peer}; grpc_endpoint *grpc_tcp_create(grpc_fd *em_fd, size_t slice_size, const char *peer_string) { @@ -597,21 +433,18 @@ grpc_endpoint *grpc_tcp_create(grpc_fd *em_fd, size_t slice_size, tcp->fd = em_fd->fd; tcp->read_cb = NULL; tcp->write_cb = NULL; - tcp->read_user_data = NULL; - tcp->write_user_data = NULL; + tcp->incoming_buffer = NULL; tcp->slice_size = slice_size; tcp->iov_size = 1; tcp->finished_edge = 1; - slice_state_init(&tcp->write_state, NULL, 0, 0); /* paired with unref in grpc_tcp_destroy */ gpr_ref_init(&tcp->refcount, 1); tcp->em_fd = em_fd; - tcp->read_closure.cb = grpc_tcp_handle_read; + tcp->read_closure.cb = tcp_handle_read; tcp->read_closure.cb_arg = tcp; - tcp->write_closure.cb = grpc_tcp_handle_write; + tcp->write_closure.cb = tcp_handle_write; tcp->write_closure.cb_arg = tcp; - tcp->handle_read_closure.cb = grpc_tcp_handle_read; return &tcp->base; } diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c index d0478d3604..b513d854aa 100644 --- a/src/core/iomgr/tcp_server_windows.c +++ b/src/core/iomgr/tcp_server_windows.c @@ -75,18 +75,18 @@ struct grpc_tcp_server { void *cb_arg; gpr_mu mu; - gpr_cv cv; /* active port count: how many ports are actually still listening */ int active_ports; - /* number of iomgr callbacks that have been explicitly scheduled during - * shutdown */ - int iomgr_callbacks_pending; /* all listening ports */ server_port *ports; size_t nports; size_t port_capacity; + + /* shutdown callback */ + void(*shutdown_complete)(void *); + void *shutdown_complete_arg; }; /* Public function. Allocates the proper data structures to hold a @@ -94,48 +94,61 @@ struct grpc_tcp_server { grpc_tcp_server *grpc_tcp_server_create(void) { grpc_tcp_server *s = gpr_malloc(sizeof(grpc_tcp_server)); gpr_mu_init(&s->mu); - gpr_cv_init(&s->cv); s->active_ports = 0; - s->iomgr_callbacks_pending = 0; s->cb = NULL; s->cb_arg = NULL; s->ports = gpr_malloc(sizeof(server_port) * INIT_PORT_CAP); s->nports = 0; s->port_capacity = INIT_PORT_CAP; + s->shutdown_complete = NULL; return s; } +static void dont_care_about_shutdown_completion(void *arg) {} + +static void finish_shutdown(grpc_tcp_server *s) { + size_t i; + + s->shutdown_complete(s->shutdown_complete_arg); + + /* Now that the accepts have been aborted, we can destroy the sockets. + The IOCP won't get notified on these, so we can flag them as already + closed by the system. */ + for (i = 0; i < s->nports; i++) { + server_port *sp = &s->ports[i]; + grpc_winsocket_destroy(sp->socket); + } + gpr_free(s->ports); + gpr_free(s); +} + /* Public function. Stops and destroys a grpc_tcp_server. */ void grpc_tcp_server_destroy(grpc_tcp_server *s, - void (*shutdown_done)(void *shutdown_done_arg), - void *shutdown_done_arg) { + void (*shutdown_complete)(void *shutdown_done_arg), + void *shutdown_complete_arg) { size_t i; + int immediately_done = 0; gpr_mu_lock(&s->mu); + + s->shutdown_complete = shutdown_complete + ? shutdown_complete + : dont_care_about_shutdown_completion; + s->shutdown_complete_arg = shutdown_complete_arg; + /* First, shutdown all fd's. This will queue abortion calls for all of the pending accepts due to the normal operation mechanism. */ + if (s->active_ports == 0) { + immediately_done = 1; + } for (i = 0; i < s->nports; i++) { server_port *sp = &s->ports[i]; sp->shutting_down = 1; - s->iomgr_callbacks_pending += grpc_winsocket_shutdown(sp->socket); - } - /* This happens asynchronously. Wait while that happens. */ - while (s->active_ports || s->iomgr_callbacks_pending) { - gpr_cv_wait(&s->cv, &s->mu, gpr_inf_future(GPR_CLOCK_REALTIME)); + grpc_winsocket_shutdown(sp->socket); } gpr_mu_unlock(&s->mu); - /* Now that the accepts have been aborted, we can destroy the sockets. - The IOCP won't get notified on these, so we can flag them as already - closed by the system. */ - for (i = 0; i < s->nports; i++) { - server_port *sp = &s->ports[i]; - grpc_winsocket_orphan(sp->socket); - } - gpr_free(s->ports); - gpr_free(s); - - if (shutdown_done) { - shutdown_done(shutdown_done_arg); + if (immediately_done) { + finish_shutdown(s); } } @@ -188,14 +201,17 @@ error: } static void decrement_active_ports_and_notify(server_port *sp) { + int notify = 0; sp->shutting_down = 0; - sp->socket->read_info.outstanding = 0; gpr_mu_lock(&sp->server->mu); GPR_ASSERT(sp->server->active_ports > 0); - if (0 == --sp->server->active_ports) { - gpr_cv_broadcast(&sp->server->cv); + if (0 == --sp->server->active_ports && sp->server->shutdown_complete != NULL) { + notify = 1; } gpr_mu_unlock(&sp->server->mu); + if (notify) { + finish_shutdown(sp->server); + } } /* start_accept will reference that for the IOCP notification request. */ @@ -280,12 +296,6 @@ static void on_accept(void *arg, int from_iocp) { this is necessary in the read/write case, it's useless for the accept case. We only need to adjust the pending callback count */ if (!from_iocp) { - gpr_mu_lock(&sp->server->mu); - GPR_ASSERT(sp->server->iomgr_callbacks_pending > 0); - if (0 == --sp->server->iomgr_callbacks_pending) { - gpr_cv_broadcast(&sp->server->cv); - } - gpr_mu_unlock(&sp->server->mu); return; } @@ -462,7 +472,6 @@ void grpc_tcp_server_start(grpc_tcp_server *s, grpc_pollset **pollset, s->cb = cb; s->cb_arg = cb_arg; for (i = 0; i < s->nports; i++) { - s->ports[i].socket->read_info.outstanding = 1; start_accept(s->ports + i); s->active_ports++; } diff --git a/src/core/iomgr/tcp_windows.c b/src/core/iomgr/tcp_windows.c index 901793ec43..fe3673c607 100644 --- a/src/core/iomgr/tcp_windows.c +++ b/src/core/iomgr/tcp_windows.c @@ -82,13 +82,11 @@ typedef struct grpc_tcp { /* Refcounting how many operations are in progress. */ gpr_refcount refcount; - grpc_endpoint_read_cb read_cb; - void *read_user_data; + grpc_iomgr_closure *read_cb; + grpc_iomgr_closure *write_cb; gpr_slice read_slice; - - grpc_endpoint_write_cb write_cb; - void *write_user_data; - gpr_slice_buffer write_slices; + gpr_slice_buffer *write_slices; + gpr_slice_buffer *read_slices; /* The IO Completion Port runs from another thread. We need some mechanism to protect ourselves when requesting a shutdown. */ @@ -98,82 +96,91 @@ typedef struct grpc_tcp { char *peer_string; } grpc_tcp; -static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); } +static void tcp_free(grpc_tcp *tcp) { + grpc_winsocket_destroy(tcp->socket); + gpr_mu_destroy(&tcp->mu); + gpr_free(tcp->peer_string); + gpr_free(tcp); +} +/*#define GRPC_TCP_REFCOUNT_DEBUG*/ +#ifdef GRPC_TCP_REFCOUNT_DEBUG +#define TCP_UNREF(tcp, reason) tcp_unref((tcp), (reason), __FILE__, __LINE__) +#define TCP_REF(tcp, reason) tcp_ref((tcp), (reason), __FILE__, __LINE__) +static void tcp_unref(grpc_tcp *tcp, const char *reason, const char *file, + int line) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP unref %p : %s %d -> %d", tcp, + reason, tcp->refcount.count, tcp->refcount.count - 1); + if (gpr_unref(&tcp->refcount)) { + tcp_free(tcp); + } +} + +static void tcp_ref(grpc_tcp *tcp, const char *reason, const char *file, + int line) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP ref %p : %s %d -> %d", tcp, + reason, tcp->refcount.count, tcp->refcount.count + 1); + gpr_ref(&tcp->refcount); +} +#else +#define TCP_UNREF(tcp, reason) tcp_unref((tcp)) +#define TCP_REF(tcp, reason) tcp_ref((tcp)) static void tcp_unref(grpc_tcp *tcp) { if (gpr_unref(&tcp->refcount)) { - gpr_slice_buffer_destroy(&tcp->write_slices); - grpc_winsocket_orphan(tcp->socket); - gpr_mu_destroy(&tcp->mu); - gpr_free(tcp->peer_string); - gpr_free(tcp); + tcp_free(tcp); } } +static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); } +#endif + /* Asynchronous callback from the IOCP, or the background thread. */ -static void on_read(void *tcpp, int from_iocp) { - grpc_tcp *tcp = (grpc_tcp *)tcpp; +static int on_read(grpc_tcp *tcp, int success) { grpc_winsocket *socket = tcp->socket; gpr_slice sub; gpr_slice *slice = NULL; size_t nslices = 0; - grpc_endpoint_cb_status status; - grpc_endpoint_read_cb cb; grpc_winsocket_callback_info *info = &socket->read_info; - void *opaque = tcp->read_user_data; int do_abort = 0; - gpr_mu_lock(&tcp->mu); - cb = tcp->read_cb; - tcp->read_cb = NULL; - if (!from_iocp || tcp->shutting_down) { - /* If we are here with from_iocp set to true, it means we got raced to - shutting down the endpoint. No actual abort callback will happen - though, so we're going to do it from here. */ - do_abort = 1; - } - gpr_mu_unlock(&tcp->mu); - - if (do_abort) { - if (from_iocp) { - tcp->socket->read_info.outstanding = 0; + if (success) { + if (socket->read_info.wsa_error != 0) { + if (socket->read_info.wsa_error != WSAECONNRESET) { + char *utf8_message = gpr_format_message(info->wsa_error); + gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message); + gpr_free(utf8_message); + } + success = 0; gpr_slice_unref(tcp->read_slice); - } - tcp_unref(tcp); - if (cb) cb(opaque, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN); - return; - } - - GPR_ASSERT(tcp->socket->read_info.outstanding); - - if (socket->read_info.wsa_error != 0) { - if (socket->read_info.wsa_error != WSAECONNRESET) { - char *utf8_message = gpr_format_message(info->wsa_error); - gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message); - gpr_free(utf8_message); - } - gpr_slice_unref(tcp->read_slice); - status = GRPC_ENDPOINT_CB_ERROR; - } else { - if (info->bytes_transfered != 0) { - sub = gpr_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered); - status = GRPC_ENDPOINT_CB_OK; - slice = ⊂ - nslices = 1; } else { - gpr_slice_unref(tcp->read_slice); - status = GRPC_ENDPOINT_CB_EOF; + if (info->bytes_transfered != 0) { + sub = gpr_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered); + gpr_slice_buffer_add(tcp->read_slices, sub); + success = 1; + } else { + gpr_slice_unref(tcp->read_slice); + success = 0; + } } } - tcp->socket->read_info.outstanding = 0; + return success; +} - tcp_unref(tcp); - cb(opaque, slice, nslices, status); +static void on_read_cb(void *tcpp, int from_iocp) { + grpc_tcp *tcp = tcpp; + grpc_iomgr_closure *cb = tcp->read_cb; + int success = on_read(tcp, from_iocp); + tcp->read_cb = NULL; + TCP_UNREF(tcp, "read"); + if (cb) { + cb->cb(cb->cb_arg, success); + } } -static void win_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb, - void *arg) { +static grpc_endpoint_op_status win_read(grpc_endpoint *ep, + gpr_slice_buffer *read_slices, + grpc_iomgr_closure *cb) { grpc_tcp *tcp = (grpc_tcp *)ep; grpc_winsocket *handle = tcp->socket; grpc_winsocket_callback_info *info = &handle->read_info; @@ -182,15 +189,13 @@ static void win_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb, DWORD flags = 0; WSABUF buffer; - GPR_ASSERT(!tcp->socket->read_info.outstanding); if (tcp->shutting_down) { - cb(arg, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN); - return; + return GRPC_ENDPOINT_ERROR; } - tcp_ref(tcp); - tcp->socket->read_info.outstanding = 1; + tcp->read_cb = cb; - tcp->read_user_data = arg; + tcp->read_slices = read_slices; + gpr_slice_buffer_reset_and_unref(read_slices); tcp->read_slice = gpr_slice_malloc(8192); @@ -204,12 +209,14 @@ static void win_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb, /* Did we get data immediately ? Yay. */ if (info->wsa_error != WSAEWOULDBLOCK) { + int ok; info->bytes_transfered = bytes_read; - /* This might heavily recurse. */ - on_read(tcp, 1); - return; + ok = on_read(tcp, 1); + return ok ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR; } + TCP_REF(tcp, "read"); + /* Otherwise, let's retry, by queuing a read. */ memset(&tcp->socket->read_info.overlapped, 0, sizeof(OVERLAPPED)); status = WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags, @@ -218,71 +225,51 @@ static void win_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb, if (status != 0) { int wsa_error = WSAGetLastError(); if (wsa_error != WSA_IO_PENDING) { + int ok; info->wsa_error = wsa_error; - on_read(tcp, 1); - return; + ok = on_read(tcp, 1); + return ok ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR; } } - grpc_socket_notify_on_read(tcp->socket, on_read, tcp); + grpc_socket_notify_on_read(tcp->socket, on_read_cb, tcp); + return GRPC_ENDPOINT_PENDING; } /* Asynchronous callback from the IOCP, or the background thread. */ -static void on_write(void *tcpp, int from_iocp) { +static void on_write(void *tcpp, int success) { grpc_tcp *tcp = (grpc_tcp *)tcpp; grpc_winsocket *handle = tcp->socket; grpc_winsocket_callback_info *info = &handle->write_info; - grpc_endpoint_cb_status status = GRPC_ENDPOINT_CB_OK; - grpc_endpoint_write_cb cb; - void *opaque = tcp->write_user_data; + grpc_iomgr_closure *cb; int do_abort = 0; gpr_mu_lock(&tcp->mu); cb = tcp->write_cb; tcp->write_cb = NULL; - if (!from_iocp || tcp->shutting_down) { - /* If we are here with from_iocp set to true, it means we got raced to - shutting down the endpoint. No actual abort callback will happen - though, so we're going to do it from here. */ - do_abort = 1; - } gpr_mu_unlock(&tcp->mu); - if (do_abort) { - if (from_iocp) { - tcp->socket->write_info.outstanding = 0; - gpr_slice_buffer_reset_and_unref(&tcp->write_slices); - } - tcp_unref(tcp); - if (cb) cb(opaque, GRPC_ENDPOINT_CB_SHUTDOWN); - return; - } - - GPR_ASSERT(tcp->socket->write_info.outstanding); - - if (info->wsa_error != 0) { - if (info->wsa_error != WSAECONNRESET) { - char *utf8_message = gpr_format_message(info->wsa_error); - gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message); - gpr_free(utf8_message); + if (success) { + if (info->wsa_error != 0) { + if (info->wsa_error != WSAECONNRESET) { + char *utf8_message = gpr_format_message(info->wsa_error); + gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message); + gpr_free(utf8_message); + } + success = 0; + } else { + GPR_ASSERT(info->bytes_transfered == tcp->write_slices->length); } - status = GRPC_ENDPOINT_CB_ERROR; - } else { - GPR_ASSERT(info->bytes_transfered == tcp->write_slices.length); } - gpr_slice_buffer_reset_and_unref(&tcp->write_slices); - tcp->socket->write_info.outstanding = 0; - - tcp_unref(tcp); - cb(opaque, status); + TCP_UNREF(tcp, "write"); + cb->cb(cb->cb_arg, success); } /* Initiates a write. */ -static grpc_endpoint_write_status win_write(grpc_endpoint *ep, - gpr_slice *slices, size_t nslices, - grpc_endpoint_write_cb cb, - void *arg) { +static grpc_endpoint_op_status win_write(grpc_endpoint *ep, + gpr_slice_buffer *slices, + grpc_iomgr_closure *cb) { grpc_tcp *tcp = (grpc_tcp *)ep; grpc_winsocket *socket = tcp->socket; grpc_winsocket_callback_info *info = &socket->write_info; @@ -293,30 +280,25 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep, WSABUF *allocated = NULL; WSABUF *buffers = local_buffers; - GPR_ASSERT(!tcp->socket->write_info.outstanding); if (tcp->shutting_down) { - return GRPC_ENDPOINT_WRITE_ERROR; + return GRPC_ENDPOINT_ERROR; } - tcp_ref(tcp); - tcp->socket->write_info.outstanding = 1; tcp->write_cb = cb; - tcp->write_user_data = arg; + tcp->write_slices = slices; - gpr_slice_buffer_addn(&tcp->write_slices, slices, nslices); - - if (tcp->write_slices.count > GPR_ARRAY_SIZE(local_buffers)) { - buffers = (WSABUF *)gpr_malloc(sizeof(WSABUF) * tcp->write_slices.count); + if (tcp->write_slices->count > GPR_ARRAY_SIZE(local_buffers)) { + buffers = (WSABUF *)gpr_malloc(sizeof(WSABUF) * tcp->write_slices->count); allocated = buffers; } - for (i = 0; i < tcp->write_slices.count; i++) { - buffers[i].len = GPR_SLICE_LENGTH(tcp->write_slices.slices[i]); - buffers[i].buf = (char *)GPR_SLICE_START_PTR(tcp->write_slices.slices[i]); + for (i = 0; i < tcp->write_slices->count; i++) { + buffers[i].len = GPR_SLICE_LENGTH(tcp->write_slices->slices[i]); + buffers[i].buf = (char *)GPR_SLICE_START_PTR(tcp->write_slices->slices[i]); } /* First, let's try a synchronous, non-blocking write. */ - status = WSASend(socket->socket, buffers, tcp->write_slices.count, + status = WSASend(socket->socket, buffers, tcp->write_slices->count, &bytes_sent, 0, NULL, NULL); info->wsa_error = status == 0 ? 0 : WSAGetLastError(); @@ -324,10 +306,10 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep, connection that has its send queue filled up. But if we don't, then we can avoid doing an async write operation at all. */ if (info->wsa_error != WSAEWOULDBLOCK) { - grpc_endpoint_write_status ret = GRPC_ENDPOINT_WRITE_ERROR; + grpc_endpoint_op_status ret = GRPC_ENDPOINT_ERROR; if (status == 0) { - ret = GRPC_ENDPOINT_WRITE_DONE; - GPR_ASSERT(bytes_sent == tcp->write_slices.length); + ret = GRPC_ENDPOINT_DONE; + GPR_ASSERT(bytes_sent == tcp->write_slices->length); } else { if (socket->read_info.wsa_error != WSAECONNRESET) { char *utf8_message = gpr_format_message(info->wsa_error); @@ -336,33 +318,30 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep, } } if (allocated) gpr_free(allocated); - gpr_slice_buffer_reset_and_unref(&tcp->write_slices); - tcp->socket->write_info.outstanding = 0; - tcp_unref(tcp); return ret; } + TCP_REF(tcp, "write"); + /* If we got a WSAEWOULDBLOCK earlier, then we need to re-do the same operation, this time asynchronously. */ memset(&socket->write_info.overlapped, 0, sizeof(OVERLAPPED)); - status = WSASend(socket->socket, buffers, tcp->write_slices.count, + status = WSASend(socket->socket, buffers, tcp->write_slices->count, &bytes_sent, 0, &socket->write_info.overlapped, NULL); if (allocated) gpr_free(allocated); if (status != 0) { int wsa_error = WSAGetLastError(); if (wsa_error != WSA_IO_PENDING) { - gpr_slice_buffer_reset_and_unref(&tcp->write_slices); - tcp->socket->write_info.outstanding = 0; - tcp_unref(tcp); - return GRPC_ENDPOINT_WRITE_ERROR; + TCP_UNREF(tcp, "write"); + return GRPC_ENDPOINT_ERROR; } } /* As all is now setup, we can now ask for the IOCP notification. It may trigger the callback immediately however, but no matter. */ grpc_socket_notify_on_write(socket, on_write, tcp); - return GRPC_ENDPOINT_WRITE_PENDING; + return GRPC_ENDPOINT_PENDING; } static void win_add_to_pollset(grpc_endpoint *ep, grpc_pollset *ps) { @@ -387,19 +366,17 @@ static void win_add_to_pollset_set(grpc_endpoint *ep, grpc_pollset_set *pss) { concurrent access of the data structure in that regard. */ static void win_shutdown(grpc_endpoint *ep) { grpc_tcp *tcp = (grpc_tcp *)ep; - int extra_refs = 0; gpr_mu_lock(&tcp->mu); /* At that point, what may happen is that we're already inside the IOCP callback. See the comments in on_read and on_write. */ tcp->shutting_down = 1; - extra_refs = grpc_winsocket_shutdown(tcp->socket); - while (extra_refs--) tcp_ref(tcp); + grpc_winsocket_shutdown(tcp->socket); gpr_mu_unlock(&tcp->mu); } static void win_destroy(grpc_endpoint *ep) { grpc_tcp *tcp = (grpc_tcp *)ep; - tcp_unref(tcp); + TCP_UNREF(tcp, "destroy"); } static char *win_get_peer(grpc_endpoint *ep) { @@ -408,8 +385,8 @@ static char *win_get_peer(grpc_endpoint *ep) { } static grpc_endpoint_vtable vtable = { - win_notify_on_read, win_write, win_add_to_pollset, win_add_to_pollset_set, - win_shutdown, win_destroy, win_get_peer}; + win_read, win_write, win_add_to_pollset, win_add_to_pollset_set, + win_shutdown, win_destroy, win_get_peer}; grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket, char *peer_string) { grpc_tcp *tcp = (grpc_tcp *)gpr_malloc(sizeof(grpc_tcp)); @@ -417,7 +394,6 @@ grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket, char *peer_string) { tcp->base.vtable = &vtable; tcp->socket = socket; gpr_mu_init(&tcp->mu); - gpr_slice_buffer_init(&tcp->write_slices); gpr_ref_init(&tcp->refcount, 1); tcp->peer_string = gpr_strdup(peer_string); return &tcp->base; diff --git a/src/core/security/client_auth_filter.c b/src/core/security/client_auth_filter.c index 8e63978b82..f3ecfd0e60 100644 --- a/src/core/security/client_auth_filter.c +++ b/src/core/security/client_auth_filter.c @@ -153,7 +153,8 @@ static void send_security_metadata(grpc_call_element *elem, } if (channel_creds_has_md && call_creds_has_md) { - calld->creds = grpc_composite_credentials_create(channel_creds, ctx->creds); + calld->creds = + grpc_composite_credentials_create(channel_creds, ctx->creds, NULL); if (calld->creds == NULL) { bubble_up_error(elem, GRPC_STATUS_INVALID_ARGUMENT, "Incompatible credentials set on channel and call."); diff --git a/src/core/security/credentials.c b/src/core/security/credentials.c index 8852cab3e7..a764413300 100644 --- a/src/core/security/credentials.c +++ b/src/core/security/credentials.c @@ -87,7 +87,10 @@ grpc_credentials *grpc_credentials_ref(grpc_credentials *creds) { void grpc_credentials_unref(grpc_credentials *creds) { if (creds == NULL) return; - if (gpr_unref(&creds->refcount)) creds->vtable->destroy(creds); + if (gpr_unref(&creds->refcount)) { + creds->vtable->destruct(creds); + gpr_free(creds); + } } void grpc_credentials_release(grpc_credentials *creds) { @@ -135,9 +138,26 @@ grpc_security_status grpc_credentials_create_security_connector( creds, target, args, request_metadata_creds, sc, new_args); } -void grpc_server_credentials_release(grpc_server_credentials *creds) { +grpc_server_credentials *grpc_server_credentials_ref( + grpc_server_credentials *creds) { + if (creds == NULL) return NULL; + gpr_ref(&creds->refcount); + return creds; +} + +void grpc_server_credentials_unref(grpc_server_credentials *creds) { if (creds == NULL) return; - creds->vtable->destroy(creds); + if (gpr_unref(&creds->refcount)) { + creds->vtable->destruct(creds); + if (creds->processor.destroy != NULL && creds->processor.state != NULL) { + creds->processor.destroy(creds->processor.state); + } + gpr_free(creds); + } +} + +void grpc_server_credentials_release(grpc_server_credentials *creds) { + grpc_server_credentials_unref(creds); } grpc_security_status grpc_server_credentials_create_security_connector( @@ -152,20 +172,22 @@ grpc_security_status grpc_server_credentials_create_security_connector( void grpc_server_credentials_set_auth_metadata_processor( grpc_server_credentials *creds, grpc_auth_metadata_processor processor) { if (creds == NULL) return; + if (creds->processor.destroy != NULL && creds->processor.state != NULL) { + creds->processor.destroy(creds->processor.state); + } creds->processor = processor; } /* -- Ssl credentials. -- */ -static void ssl_destroy(grpc_credentials *creds) { +static void ssl_destruct(grpc_credentials *creds) { grpc_ssl_credentials *c = (grpc_ssl_credentials *)creds; if (c->config.pem_root_certs != NULL) gpr_free(c->config.pem_root_certs); if (c->config.pem_private_key != NULL) gpr_free(c->config.pem_private_key); if (c->config.pem_cert_chain != NULL) gpr_free(c->config.pem_cert_chain); - gpr_free(creds); } -static void ssl_server_destroy(grpc_server_credentials *creds) { +static void ssl_server_destruct(grpc_server_credentials *creds) { grpc_ssl_server_credentials *c = (grpc_ssl_server_credentials *)creds; size_t i; for (i = 0; i < c->config.num_key_cert_pairs; i++) { @@ -185,7 +207,6 @@ static void ssl_server_destroy(grpc_server_credentials *creds) { gpr_free(c->config.pem_cert_chains_sizes); } if (c->config.pem_root_certs != NULL) gpr_free(c->config.pem_root_certs); - gpr_free(creds); } static int ssl_has_request_metadata(const grpc_credentials *creds) { return 0; } @@ -231,11 +252,11 @@ static grpc_security_status ssl_server_create_security_connector( } static grpc_credentials_vtable ssl_vtable = { - ssl_destroy, ssl_has_request_metadata, ssl_has_request_metadata_only, NULL, + ssl_destruct, ssl_has_request_metadata, ssl_has_request_metadata_only, NULL, ssl_create_security_connector}; static grpc_server_credentials_vtable ssl_server_vtable = { - ssl_server_destroy, ssl_server_create_security_connector}; + ssl_server_destruct, ssl_server_create_security_connector}; static void ssl_copy_key_material(const char *input, unsigned char **output, size_t *output_size) { @@ -298,8 +319,10 @@ static void ssl_build_server_config( } grpc_credentials *grpc_ssl_credentials_create( - const char *pem_root_certs, grpc_ssl_pem_key_cert_pair *pem_key_cert_pair) { + const char *pem_root_certs, grpc_ssl_pem_key_cert_pair *pem_key_cert_pair, + void *reserved) { grpc_ssl_credentials *c = gpr_malloc(sizeof(grpc_ssl_credentials)); + GPR_ASSERT(reserved == NULL); memset(c, 0, sizeof(grpc_ssl_credentials)); c->base.type = GRPC_CREDENTIALS_TYPE_SSL; c->base.vtable = &ssl_vtable; @@ -310,11 +333,13 @@ grpc_credentials *grpc_ssl_credentials_create( grpc_server_credentials *grpc_ssl_server_credentials_create( const char *pem_root_certs, grpc_ssl_pem_key_cert_pair *pem_key_cert_pairs, - size_t num_key_cert_pairs, int force_client_auth) { + size_t num_key_cert_pairs, int force_client_auth, void *reserved) { grpc_ssl_server_credentials *c = gpr_malloc(sizeof(grpc_ssl_server_credentials)); + GPR_ASSERT(reserved == NULL); memset(c, 0, sizeof(grpc_ssl_server_credentials)); c->base.type = GRPC_CREDENTIALS_TYPE_SSL; + gpr_ref_init(&c->base.refcount, 1); c->base.vtable = &ssl_server_vtable; ssl_build_server_config(pem_root_certs, pem_key_cert_pairs, num_key_cert_pairs, force_client_auth, &c->config); @@ -335,13 +360,12 @@ static void jwt_reset_cache(grpc_service_account_jwt_access_credentials *c) { c->cached.jwt_expiration = gpr_inf_past(GPR_CLOCK_REALTIME); } -static void jwt_destroy(grpc_credentials *creds) { +static void jwt_destruct(grpc_credentials *creds) { grpc_service_account_jwt_access_credentials *c = (grpc_service_account_jwt_access_credentials *)creds; grpc_auth_json_key_destruct(&c->key); jwt_reset_cache(c); gpr_mu_destroy(&c->cache_mu); - gpr_free(c); } static int jwt_has_request_metadata(const grpc_credentials *creds) { return 1; } @@ -406,7 +430,7 @@ static void jwt_get_request_metadata(grpc_credentials *creds, } static grpc_credentials_vtable jwt_vtable = { - jwt_destroy, jwt_has_request_metadata, jwt_has_request_metadata_only, + jwt_destruct, jwt_has_request_metadata, jwt_has_request_metadata_only, jwt_get_request_metadata, NULL}; grpc_credentials * @@ -430,20 +454,20 @@ grpc_service_account_jwt_access_credentials_create_from_auth_json_key( } grpc_credentials *grpc_service_account_jwt_access_credentials_create( - const char *json_key, gpr_timespec token_lifetime) { + const char *json_key, gpr_timespec token_lifetime, void *reserved) { + GPR_ASSERT(reserved == NULL); return grpc_service_account_jwt_access_credentials_create_from_auth_json_key( grpc_auth_json_key_create_from_string(json_key), token_lifetime); } /* -- Oauth2TokenFetcher credentials -- */ -static void oauth2_token_fetcher_destroy(grpc_credentials *creds) { +static void oauth2_token_fetcher_destruct(grpc_credentials *creds) { grpc_oauth2_token_fetcher_credentials *c = (grpc_oauth2_token_fetcher_credentials *)creds; grpc_credentials_md_store_unref(c->access_token_md); gpr_mu_destroy(&c->mu); grpc_httpcli_context_destroy(&c->httpcli_context); - gpr_free(c); } static int oauth2_token_fetcher_has_request_metadata( @@ -613,10 +637,10 @@ static void init_oauth2_token_fetcher(grpc_oauth2_token_fetcher_credentials *c, grpc_httpcli_context_init(&c->httpcli_context); } -/* -- ComputeEngine credentials. -- */ +/* -- GoogleComputeEngine credentials. -- */ static grpc_credentials_vtable compute_engine_vtable = { - oauth2_token_fetcher_destroy, oauth2_token_fetcher_has_request_metadata, + oauth2_token_fetcher_destruct, oauth2_token_fetcher_has_request_metadata, oauth2_token_fetcher_has_request_metadata_only, oauth2_token_fetcher_get_request_metadata, NULL}; @@ -635,94 +659,27 @@ static void compute_engine_fetch_oauth2( metadata_req); } -grpc_credentials *grpc_compute_engine_credentials_create(void) { +grpc_credentials *grpc_google_compute_engine_credentials_create( + void *reserved) { grpc_oauth2_token_fetcher_credentials *c = gpr_malloc(sizeof(grpc_oauth2_token_fetcher_credentials)); + GPR_ASSERT(reserved == NULL); init_oauth2_token_fetcher(c, compute_engine_fetch_oauth2); c->base.vtable = &compute_engine_vtable; return &c->base; } -/* -- ServiceAccount credentials. -- */ - -static void service_account_destroy(grpc_credentials *creds) { - grpc_service_account_credentials *c = - (grpc_service_account_credentials *)creds; - if (c->scope != NULL) gpr_free(c->scope); - grpc_auth_json_key_destruct(&c->key); - oauth2_token_fetcher_destroy(&c->base.base); -} - -static grpc_credentials_vtable service_account_vtable = { - service_account_destroy, oauth2_token_fetcher_has_request_metadata, - oauth2_token_fetcher_has_request_metadata_only, - oauth2_token_fetcher_get_request_metadata, NULL}; - -static void service_account_fetch_oauth2( - grpc_credentials_metadata_request *metadata_req, - grpc_httpcli_context *httpcli_context, grpc_pollset *pollset, - grpc_httpcli_response_cb response_cb, gpr_timespec deadline) { - grpc_service_account_credentials *c = - (grpc_service_account_credentials *)metadata_req->creds; - grpc_httpcli_header header = {"Content-Type", - "application/x-www-form-urlencoded"}; - grpc_httpcli_request request; - char *body = NULL; - char *jwt = grpc_jwt_encode_and_sign(&c->key, GRPC_JWT_OAUTH2_AUDIENCE, - c->token_lifetime, c->scope); - if (jwt == NULL) { - grpc_httpcli_response response; - memset(&response, 0, sizeof(grpc_httpcli_response)); - response.status = 400; /* Invalid request. */ - gpr_log(GPR_ERROR, "Could not create signed jwt."); - /* Do not even send the request, just call the response callback. */ - response_cb(metadata_req, &response); - return; - } - gpr_asprintf(&body, "%s%s", GRPC_SERVICE_ACCOUNT_POST_BODY_PREFIX, jwt); - memset(&request, 0, sizeof(grpc_httpcli_request)); - request.host = GRPC_GOOGLE_OAUTH2_SERVICE_HOST; - request.path = GRPC_GOOGLE_OAUTH2_SERVICE_TOKEN_PATH; - request.hdr_count = 1; - request.hdrs = &header; - request.handshaker = &grpc_httpcli_ssl; - grpc_httpcli_post(httpcli_context, pollset, &request, body, strlen(body), - deadline, response_cb, metadata_req); - gpr_free(body); - gpr_free(jwt); -} - -grpc_credentials *grpc_service_account_credentials_create( - const char *json_key, const char *scope, gpr_timespec token_lifetime) { - grpc_service_account_credentials *c; - grpc_auth_json_key key = grpc_auth_json_key_create_from_string(json_key); +/* -- GoogleRefreshToken credentials. -- */ - if (scope == NULL || (strlen(scope) == 0) || - !grpc_auth_json_key_is_valid(&key)) { - gpr_log(GPR_ERROR, - "Invalid input for service account credentials creation"); - return NULL; - } - c = gpr_malloc(sizeof(grpc_service_account_credentials)); - memset(c, 0, sizeof(grpc_service_account_credentials)); - init_oauth2_token_fetcher(&c->base, service_account_fetch_oauth2); - c->base.base.vtable = &service_account_vtable; - c->scope = gpr_strdup(scope); - c->key = key; - c->token_lifetime = token_lifetime; - return &c->base.base; -} - -/* -- RefreshToken credentials. -- */ - -static void refresh_token_destroy(grpc_credentials *creds) { - grpc_refresh_token_credentials *c = (grpc_refresh_token_credentials *)creds; +static void refresh_token_destruct(grpc_credentials *creds) { + grpc_google_refresh_token_credentials *c = + (grpc_google_refresh_token_credentials *)creds; grpc_auth_refresh_token_destruct(&c->refresh_token); - oauth2_token_fetcher_destroy(&c->base.base); + oauth2_token_fetcher_destruct(&c->base.base); } static grpc_credentials_vtable refresh_token_vtable = { - refresh_token_destroy, oauth2_token_fetcher_has_request_metadata, + refresh_token_destruct, oauth2_token_fetcher_has_request_metadata, oauth2_token_fetcher_has_request_metadata_only, oauth2_token_fetcher_get_request_metadata, NULL}; @@ -730,8 +687,8 @@ static void refresh_token_fetch_oauth2( grpc_credentials_metadata_request *metadata_req, grpc_httpcli_context *httpcli_context, grpc_pollset *pollset, grpc_httpcli_response_cb response_cb, gpr_timespec deadline) { - grpc_refresh_token_credentials *c = - (grpc_refresh_token_credentials *)metadata_req->creds; + grpc_google_refresh_token_credentials *c = + (grpc_google_refresh_token_credentials *)metadata_req->creds; grpc_httpcli_header header = {"Content-Type", "application/x-www-form-urlencoded"}; grpc_httpcli_request request; @@ -750,33 +707,34 @@ static void refresh_token_fetch_oauth2( gpr_free(body); } -grpc_credentials *grpc_refresh_token_credentials_create_from_auth_refresh_token( +grpc_credentials * +grpc_refresh_token_credentials_create_from_auth_refresh_token( grpc_auth_refresh_token refresh_token) { - grpc_refresh_token_credentials *c; + grpc_google_refresh_token_credentials *c; if (!grpc_auth_refresh_token_is_valid(&refresh_token)) { gpr_log(GPR_ERROR, "Invalid input for refresh token credentials creation"); return NULL; } - c = gpr_malloc(sizeof(grpc_refresh_token_credentials)); - memset(c, 0, sizeof(grpc_refresh_token_credentials)); + c = gpr_malloc(sizeof(grpc_google_refresh_token_credentials)); + memset(c, 0, sizeof(grpc_google_refresh_token_credentials)); init_oauth2_token_fetcher(&c->base, refresh_token_fetch_oauth2); c->base.base.vtable = &refresh_token_vtable; c->refresh_token = refresh_token; return &c->base.base; } -grpc_credentials *grpc_refresh_token_credentials_create( - const char *json_refresh_token) { +grpc_credentials *grpc_google_refresh_token_credentials_create( + const char *json_refresh_token, void *reserved) { + GPR_ASSERT(reserved == NULL); return grpc_refresh_token_credentials_create_from_auth_refresh_token( grpc_auth_refresh_token_create_from_string(json_refresh_token)); } /* -- Metadata-only credentials. -- */ -static void md_only_test_destroy(grpc_credentials *creds) { +static void md_only_test_destruct(grpc_credentials *creds) { grpc_md_only_test_credentials *c = (grpc_md_only_test_credentials *)creds; grpc_credentials_md_store_unref(c->md_store); - gpr_free(c); } static int md_only_test_has_request_metadata(const grpc_credentials *creds) { @@ -817,7 +775,7 @@ static void md_only_test_get_request_metadata(grpc_credentials *creds, } static grpc_credentials_vtable md_only_test_vtable = { - md_only_test_destroy, md_only_test_has_request_metadata, + md_only_test_destruct, md_only_test_has_request_metadata, md_only_test_has_request_metadata_only, md_only_test_get_request_metadata, NULL}; @@ -838,10 +796,9 @@ grpc_credentials *grpc_md_only_test_credentials_create(const char *md_key, /* -- Oauth2 Access Token credentials. -- */ -static void access_token_destroy(grpc_credentials *creds) { +static void access_token_destruct(grpc_credentials *creds) { grpc_access_token_credentials *c = (grpc_access_token_credentials *)creds; grpc_credentials_md_store_unref(c->access_token_md); - gpr_free(c); } static int access_token_has_request_metadata(const grpc_credentials *creds) { @@ -863,15 +820,16 @@ static void access_token_get_request_metadata(grpc_credentials *creds, } static grpc_credentials_vtable access_token_vtable = { - access_token_destroy, access_token_has_request_metadata, + access_token_destruct, access_token_has_request_metadata, access_token_has_request_metadata_only, access_token_get_request_metadata, NULL}; -grpc_credentials *grpc_access_token_credentials_create( - const char *access_token) { +grpc_credentials *grpc_access_token_credentials_create(const char *access_token, + void *reserved) { grpc_access_token_credentials *c = gpr_malloc(sizeof(grpc_access_token_credentials)); char *token_md_value; + GPR_ASSERT(reserved == NULL); memset(c, 0, sizeof(grpc_access_token_credentials)); c->base.type = GRPC_CREDENTIALS_TYPE_OAUTH2; c->base.vtable = &access_token_vtable; @@ -886,14 +844,14 @@ grpc_credentials *grpc_access_token_credentials_create( /* -- Fake transport security credentials. -- */ -static void fake_transport_security_credentials_destroy( +static void fake_transport_security_credentials_destruct( grpc_credentials *creds) { - gpr_free(creds); + /* Nothing to do here. */ } -static void fake_transport_security_server_credentials_destroy( +static void fake_transport_security_server_credentials_destruct( grpc_server_credentials *creds) { - gpr_free(creds); + /* Nothing to do here. */ } static int fake_transport_security_has_request_metadata( @@ -922,14 +880,14 @@ fake_transport_security_server_create_security_connector( } static grpc_credentials_vtable fake_transport_security_credentials_vtable = { - fake_transport_security_credentials_destroy, + fake_transport_security_credentials_destruct, fake_transport_security_has_request_metadata, fake_transport_security_has_request_metadata_only, NULL, fake_transport_security_create_security_connector}; static grpc_server_credentials_vtable fake_transport_security_server_credentials_vtable = { - fake_transport_security_server_credentials_destroy, + fake_transport_security_server_credentials_destruct, fake_transport_security_server_create_security_connector}; grpc_credentials *grpc_fake_transport_security_credentials_create(void) { @@ -946,6 +904,7 @@ grpc_server_credentials *grpc_fake_transport_security_server_credentials_create( grpc_server_credentials *c = gpr_malloc(sizeof(grpc_server_credentials)); memset(c, 0, sizeof(grpc_server_credentials)); c->type = GRPC_CREDENTIALS_TYPE_FAKE_TRANSPORT_SECURITY; + gpr_ref_init(&c->refcount, 1); c->vtable = &fake_transport_security_server_credentials_vtable; return c; } @@ -962,14 +921,13 @@ typedef struct { grpc_credentials_metadata_cb cb; } grpc_composite_credentials_metadata_context; -static void composite_destroy(grpc_credentials *creds) { +static void composite_destruct(grpc_credentials *creds) { grpc_composite_credentials *c = (grpc_composite_credentials *)creds; size_t i; for (i = 0; i < c->inner.num_creds; i++) { grpc_credentials_unref(c->inner.creds_array[i]); } gpr_free(c->inner.creds_array); - gpr_free(creds); } static int composite_has_request_metadata(const grpc_credentials *creds) { @@ -1085,7 +1043,7 @@ static grpc_security_status composite_create_security_connector( } static grpc_credentials_vtable composite_credentials_vtable = { - composite_destroy, composite_has_request_metadata, + composite_destruct, composite_has_request_metadata, composite_has_request_metadata_only, composite_get_request_metadata, composite_create_security_connector}; @@ -1101,12 +1059,14 @@ static grpc_credentials_array get_creds_array(grpc_credentials **creds_addr) { } grpc_credentials *grpc_composite_credentials_create(grpc_credentials *creds1, - grpc_credentials *creds2) { + grpc_credentials *creds2, + void *reserved) { size_t i; size_t creds_array_byte_size; grpc_credentials_array creds1_array; grpc_credentials_array creds2_array; grpc_composite_credentials *c; + GPR_ASSERT(reserved == NULL); GPR_ASSERT(creds1 != NULL); GPR_ASSERT(creds2 != NULL); c = gpr_malloc(sizeof(grpc_composite_credentials)); @@ -1182,10 +1142,9 @@ grpc_credentials *grpc_credentials_contains_type( /* -- IAM credentials. -- */ -static void iam_destroy(grpc_credentials *creds) { - grpc_iam_credentials *c = (grpc_iam_credentials *)creds; +static void iam_destruct(grpc_credentials *creds) { + grpc_google_iam_credentials *c = (grpc_google_iam_credentials *)creds; grpc_credentials_md_store_unref(c->iam_md); - gpr_free(c); } static int iam_has_request_metadata(const grpc_credentials *creds) { return 1; } @@ -1199,22 +1158,23 @@ static void iam_get_request_metadata(grpc_credentials *creds, const char *service_url, grpc_credentials_metadata_cb cb, void *user_data) { - grpc_iam_credentials *c = (grpc_iam_credentials *)creds; + grpc_google_iam_credentials *c = (grpc_google_iam_credentials *)creds; cb(user_data, c->iam_md->entries, c->iam_md->num_entries, GRPC_CREDENTIALS_OK); } static grpc_credentials_vtable iam_vtable = { - iam_destroy, iam_has_request_metadata, iam_has_request_metadata_only, + iam_destruct, iam_has_request_metadata, iam_has_request_metadata_only, iam_get_request_metadata, NULL}; -grpc_credentials *grpc_iam_credentials_create(const char *token, - const char *authority_selector) { - grpc_iam_credentials *c; +grpc_credentials *grpc_google_iam_credentials_create( + const char *token, const char *authority_selector, void *reserved) { + grpc_google_iam_credentials *c; + GPR_ASSERT(reserved == NULL); GPR_ASSERT(token != NULL); GPR_ASSERT(authority_selector != NULL); - c = gpr_malloc(sizeof(grpc_iam_credentials)); - memset(c, 0, sizeof(grpc_iam_credentials)); + c = gpr_malloc(sizeof(grpc_google_iam_credentials)); + memset(c, 0, sizeof(grpc_google_iam_credentials)); c->base.type = GRPC_CREDENTIALS_TYPE_IAM; c->base.vtable = &iam_vtable; gpr_ref_init(&c->base.refcount, 1); diff --git a/src/core/security/credentials.h b/src/core/security/credentials.h index 29cd1ac87f..8e4fed7615 100644 --- a/src/core/security/credentials.h +++ b/src/core/security/credentials.h @@ -129,7 +129,7 @@ typedef void (*grpc_credentials_metadata_cb)(void *user_data, grpc_credentials_status status); typedef struct { - void (*destroy)(grpc_credentials *c); + void (*destruct)(grpc_credentials *c); int (*has_request_metadata)(const grpc_credentials *c); int (*has_request_metadata_only)(const grpc_credentials *c); void (*get_request_metadata)(grpc_credentials *c, grpc_pollset *pollset, @@ -210,20 +210,28 @@ grpc_credentials *grpc_refresh_token_credentials_create_from_auth_refresh_token( /* --- grpc_server_credentials. --- */ typedef struct { - void (*destroy)(grpc_server_credentials *c); + void (*destruct)(grpc_server_credentials *c); grpc_security_status (*create_security_connector)( grpc_server_credentials *c, grpc_security_connector **sc); } grpc_server_credentials_vtable; + +/* TODO(jboeuf): Add a refcount. */ struct grpc_server_credentials { const grpc_server_credentials_vtable *vtable; const char *type; + gpr_refcount refcount; grpc_auth_metadata_processor processor; }; grpc_security_status grpc_server_credentials_create_security_connector( grpc_server_credentials *creds, grpc_security_connector **sc); +grpc_server_credentials *grpc_server_credentials_ref( + grpc_server_credentials *creds); + +void grpc_server_credentials_unref(grpc_server_credentials *creds); + /* -- Ssl credentials. -- */ typedef struct { @@ -277,21 +285,12 @@ typedef struct { grpc_fetch_oauth2_func fetch_func; } grpc_oauth2_token_fetcher_credentials; -/* -- ServiceAccount credentials. -- */ - -typedef struct { - grpc_oauth2_token_fetcher_credentials base; - grpc_auth_json_key key; - char *scope; - gpr_timespec token_lifetime; -} grpc_service_account_credentials; - -/* -- RefreshToken credentials. -- */ +/* -- GoogleRefreshToken credentials. -- */ typedef struct { grpc_oauth2_token_fetcher_credentials base; grpc_auth_refresh_token refresh_token; -} grpc_refresh_token_credentials; +} grpc_google_refresh_token_credentials; /* -- Oauth2 Access Token credentials. -- */ @@ -308,12 +307,12 @@ typedef struct { int is_async; } grpc_md_only_test_credentials; -/* -- IAM credentials. -- */ +/* -- GoogleIAM credentials. -- */ typedef struct { grpc_credentials base; grpc_credentials_md_store *iam_md; -} grpc_iam_credentials; +} grpc_google_iam_credentials; /* -- Composite credentials. -- */ diff --git a/src/core/security/google_default_credentials.c b/src/core/security/google_default_credentials.c index 3631de867a..874dd59e84 100644 --- a/src/core/security/google_default_credentials.c +++ b/src/core/security/google_default_credentials.c @@ -194,7 +194,7 @@ grpc_credentials *grpc_google_default_credentials_create(void) { int need_compute_engine_creds = is_stack_running_on_compute_engine(); compute_engine_detection_done = 1; if (need_compute_engine_creds) { - result = grpc_compute_engine_credentials_create(); + result = grpc_google_compute_engine_credentials_create(NULL); } } @@ -202,9 +202,9 @@ end: if (!serving_cached_credentials && result != NULL) { /* Blend with default ssl credentials and add a global reference so that it can be cached and re-served. */ - grpc_credentials *ssl_creds = grpc_ssl_credentials_create(NULL, NULL); + grpc_credentials *ssl_creds = grpc_ssl_credentials_create(NULL, NULL, NULL); default_credentials = grpc_credentials_ref( - grpc_composite_credentials_create(ssl_creds, result)); + grpc_composite_credentials_create(ssl_creds, result, NULL)); GPR_ASSERT(default_credentials != NULL); grpc_credentials_unref(ssl_creds); grpc_credentials_unref(result); diff --git a/src/core/security/secure_endpoint.c b/src/core/security/secure_endpoint.c index 81b3e33cb2..b696e384fc 100644 --- a/src/core/security/secure_endpoint.c +++ b/src/core/security/secure_endpoint.c @@ -49,15 +49,15 @@ typedef struct { struct tsi_frame_protector *protector; gpr_mu protector_mu; /* saved upper level callbacks and user_data. */ - grpc_endpoint_read_cb read_cb; - void *read_user_data; - grpc_endpoint_write_cb write_cb; - void *write_user_data; + grpc_iomgr_closure *read_cb; + grpc_iomgr_closure *write_cb; + grpc_iomgr_closure on_read; + gpr_slice_buffer *read_buffer; + gpr_slice_buffer source_buffer; /* saved handshaker leftover data to unprotect. */ gpr_slice_buffer leftover_bytes; /* buffers for read and write */ gpr_slice read_staging_buffer; - gpr_slice_buffer input_buffer; gpr_slice write_staging_buffer; gpr_slice_buffer output_buffer; @@ -67,62 +67,91 @@ typedef struct { int grpc_trace_secure_endpoint = 0; -static void secure_endpoint_ref(secure_endpoint *ep) { gpr_ref(&ep->ref); } - static void destroy(secure_endpoint *secure_ep) { secure_endpoint *ep = secure_ep; grpc_endpoint_destroy(ep->wrapped_ep); tsi_frame_protector_destroy(ep->protector); gpr_slice_buffer_destroy(&ep->leftover_bytes); gpr_slice_unref(ep->read_staging_buffer); - gpr_slice_buffer_destroy(&ep->input_buffer); gpr_slice_unref(ep->write_staging_buffer); gpr_slice_buffer_destroy(&ep->output_buffer); + gpr_slice_buffer_destroy(&ep->source_buffer); gpr_mu_destroy(&ep->protector_mu); gpr_free(ep); } +/*#define GRPC_SECURE_ENDPOINT_REFCOUNT_DEBUG*/ +#ifdef GRPC_SECURE_ENDPOINT_REFCOUNT_DEBUG +#define SECURE_ENDPOINT_UNREF(ep, reason) \ + secure_endpoint_unref((ep), (reason), __FILE__, __LINE__) +#define SECURE_ENDPOINT_REF(ep, reason) \ + secure_endpoint_ref((ep), (reason), __FILE__, __LINE__) +static void secure_endpoint_unref(secure_endpoint *ep, const char *reason, + const char *file, int line) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "SECENDP unref %p : %s %d -> %d", + ep, reason, ep->ref.count, ep->ref.count - 1); + if (gpr_unref(&ep->ref)) { + destroy(ep); + } +} + +static void secure_endpoint_ref(secure_endpoint *ep, const char *reason, + const char *file, int line) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "SECENDP ref %p : %s %d -> %d", + ep, reason, ep->ref.count, ep->ref.count + 1); + gpr_ref(&ep->ref); +} +#else +#define SECURE_ENDPOINT_UNREF(ep, reason) secure_endpoint_unref((ep)) +#define SECURE_ENDPOINT_REF(ep, reason) secure_endpoint_ref((ep)) static void secure_endpoint_unref(secure_endpoint *ep) { if (gpr_unref(&ep->ref)) { destroy(ep); } } +static void secure_endpoint_ref(secure_endpoint *ep) { gpr_ref(&ep->ref); } +#endif + static void flush_read_staging_buffer(secure_endpoint *ep, gpr_uint8 **cur, gpr_uint8 **end) { - gpr_slice_buffer_add(&ep->input_buffer, ep->read_staging_buffer); + gpr_slice_buffer_add(ep->read_buffer, ep->read_staging_buffer); ep->read_staging_buffer = gpr_slice_malloc(STAGING_BUFFER_SIZE); *cur = GPR_SLICE_START_PTR(ep->read_staging_buffer); *end = GPR_SLICE_END_PTR(ep->read_staging_buffer); } -static void call_read_cb(secure_endpoint *ep, gpr_slice *slices, size_t nslices, - grpc_endpoint_cb_status error) { +static void call_read_cb(secure_endpoint *ep, int success) { if (grpc_trace_secure_endpoint) { size_t i; - for (i = 0; i < nslices; i++) { - char *data = gpr_dump_slice(slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII); + for (i = 0; i < ep->read_buffer->count; i++) { + char *data = gpr_dump_slice(ep->read_buffer->slices[i], + GPR_DUMP_HEX | GPR_DUMP_ASCII); gpr_log(GPR_DEBUG, "READ %p: %s", ep, data); gpr_free(data); } } - ep->read_cb(ep->read_user_data, slices, nslices, error); - secure_endpoint_unref(ep); + ep->read_buffer = NULL; + ep->read_cb->cb(ep->read_cb->cb_arg, success); + SECURE_ENDPOINT_UNREF(ep, "read"); } -static void on_read(void *user_data, gpr_slice *slices, size_t nslices, - grpc_endpoint_cb_status error) { +static int on_read(void *user_data, int success) { unsigned i; gpr_uint8 keep_looping = 0; - size_t input_buffer_count = 0; tsi_result result = TSI_OK; secure_endpoint *ep = (secure_endpoint *)user_data; gpr_uint8 *cur = GPR_SLICE_START_PTR(ep->read_staging_buffer); gpr_uint8 *end = GPR_SLICE_END_PTR(ep->read_staging_buffer); + if (!success) { + gpr_slice_buffer_reset_and_unref(ep->read_buffer); + return 0; + } + /* TODO(yangg) check error, maybe bail out early */ - for (i = 0; i < nslices; i++) { - gpr_slice encrypted = slices[i]; + for (i = 0; i < ep->source_buffer.count; i++) { + gpr_slice encrypted = ep->source_buffer.slices[i]; gpr_uint8 *message_bytes = GPR_SLICE_START_PTR(encrypted); size_t message_size = GPR_SLICE_LENGTH(encrypted); @@ -161,7 +190,7 @@ static void on_read(void *user_data, gpr_slice *slices, size_t nslices, if (cur != GPR_SLICE_START_PTR(ep->read_staging_buffer)) { gpr_slice_buffer_add( - &ep->input_buffer, + ep->read_buffer, gpr_slice_split_head( &ep->read_staging_buffer, (size_t)(cur - GPR_SLICE_START_PTR(ep->read_staging_buffer)))); @@ -169,38 +198,53 @@ static void on_read(void *user_data, gpr_slice *slices, size_t nslices, /* TODO(yangg) experiment with moving this block after read_cb to see if it helps latency */ - for (i = 0; i < nslices; i++) { - gpr_slice_unref(slices[i]); - } + gpr_slice_buffer_reset_and_unref(&ep->source_buffer); if (result != TSI_OK) { - gpr_slice_buffer_reset_and_unref(&ep->input_buffer); - call_read_cb(ep, NULL, 0, GRPC_ENDPOINT_CB_ERROR); - return; + gpr_slice_buffer_reset_and_unref(ep->read_buffer); + return 0; } - /* The upper level will unref the slices. */ - input_buffer_count = ep->input_buffer.count; - ep->input_buffer.count = 0; - call_read_cb(ep, ep->input_buffer.slices, input_buffer_count, error); + + return 1; +} + +static void on_read_cb(void *user_data, int success) { + call_read_cb(user_data, on_read(user_data, success)); } -static void endpoint_notify_on_read(grpc_endpoint *secure_ep, - grpc_endpoint_read_cb cb, void *user_data) { +static grpc_endpoint_op_status endpoint_read(grpc_endpoint *secure_ep, + gpr_slice_buffer *slices, + grpc_iomgr_closure *cb) { secure_endpoint *ep = (secure_endpoint *)secure_ep; + int immediate_read_success = -1; ep->read_cb = cb; - ep->read_user_data = user_data; - - secure_endpoint_ref(ep); + ep->read_buffer = slices; + gpr_slice_buffer_reset_and_unref(ep->read_buffer); if (ep->leftover_bytes.count) { - size_t leftover_nslices = ep->leftover_bytes.count; - ep->leftover_bytes.count = 0; - on_read(ep, ep->leftover_bytes.slices, leftover_nslices, - GRPC_ENDPOINT_CB_OK); - return; + gpr_slice_buffer_swap(&ep->leftover_bytes, &ep->source_buffer); + GPR_ASSERT(ep->leftover_bytes.count == 0); + return on_read(ep, 1) ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR; } - grpc_endpoint_notify_on_read(ep->wrapped_ep, on_read, ep); + SECURE_ENDPOINT_REF(ep, "read"); + + switch ( + grpc_endpoint_read(ep->wrapped_ep, &ep->source_buffer, &ep->on_read)) { + case GRPC_ENDPOINT_DONE: + immediate_read_success = on_read(ep, 1); + break; + case GRPC_ENDPOINT_PENDING: + return GRPC_ENDPOINT_PENDING; + case GRPC_ENDPOINT_ERROR: + immediate_read_success = on_read(ep, 0); + break; + } + + GPR_ASSERT(immediate_read_success != -1); + SECURE_ENDPOINT_UNREF(ep, "read"); + + return immediate_read_success ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR; } static void flush_write_staging_buffer(secure_endpoint *ep, gpr_uint8 **cur, @@ -211,36 +255,28 @@ static void flush_write_staging_buffer(secure_endpoint *ep, gpr_uint8 **cur, *end = GPR_SLICE_END_PTR(ep->write_staging_buffer); } -static void on_write(void *data, grpc_endpoint_cb_status error) { - secure_endpoint *ep = data; - ep->write_cb(ep->write_user_data, error); - secure_endpoint_unref(ep); -} - -static grpc_endpoint_write_status endpoint_write(grpc_endpoint *secure_ep, - gpr_slice *slices, - size_t nslices, - grpc_endpoint_write_cb cb, - void *user_data) { +static grpc_endpoint_op_status endpoint_write(grpc_endpoint *secure_ep, + gpr_slice_buffer *slices, + grpc_iomgr_closure *cb) { unsigned i; - size_t output_buffer_count = 0; tsi_result result = TSI_OK; secure_endpoint *ep = (secure_endpoint *)secure_ep; gpr_uint8 *cur = GPR_SLICE_START_PTR(ep->write_staging_buffer); gpr_uint8 *end = GPR_SLICE_END_PTR(ep->write_staging_buffer); - grpc_endpoint_write_status status; - GPR_ASSERT(ep->output_buffer.count == 0); + + gpr_slice_buffer_reset_and_unref(&ep->output_buffer); if (grpc_trace_secure_endpoint) { - for (i = 0; i < nslices; i++) { - char *data = gpr_dump_slice(slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII); + for (i = 0; i < slices->count; i++) { + char *data = + gpr_dump_slice(slices->slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII); gpr_log(GPR_DEBUG, "WRITE %p: %s", ep, data); gpr_free(data); } } - for (i = 0; i < nslices; i++) { - gpr_slice plain = slices[i]; + for (i = 0; i < slices->count; i++) { + gpr_slice plain = slices->slices[i]; gpr_uint8 *message_bytes = GPR_SLICE_START_PTR(plain); size_t message_size = GPR_SLICE_LENGTH(plain); while (message_size > 0) { @@ -290,29 +326,13 @@ static grpc_endpoint_write_status endpoint_write(grpc_endpoint *secure_ep, } } - for (i = 0; i < nslices; i++) { - gpr_slice_unref(slices[i]); - } - if (result != TSI_OK) { /* TODO(yangg) do different things according to the error type? */ gpr_slice_buffer_reset_and_unref(&ep->output_buffer); - return GRPC_ENDPOINT_WRITE_ERROR; + return GRPC_ENDPOINT_ERROR; } - /* clear output_buffer and let the lower level handle its slices. */ - output_buffer_count = ep->output_buffer.count; - ep->output_buffer.count = 0; - ep->write_cb = cb; - ep->write_user_data = user_data; - /* Need to keep the endpoint alive across a transport */ - secure_endpoint_ref(ep); - status = grpc_endpoint_write(ep->wrapped_ep, ep->output_buffer.slices, - output_buffer_count, on_write, ep); - if (status != GRPC_ENDPOINT_WRITE_PENDING) { - secure_endpoint_unref(ep); - } - return status; + return grpc_endpoint_write(ep->wrapped_ep, &ep->output_buffer, cb); } static void endpoint_shutdown(grpc_endpoint *secure_ep) { @@ -320,9 +340,9 @@ static void endpoint_shutdown(grpc_endpoint *secure_ep) { grpc_endpoint_shutdown(ep->wrapped_ep); } -static void endpoint_unref(grpc_endpoint *secure_ep) { +static void endpoint_destroy(grpc_endpoint *secure_ep) { secure_endpoint *ep = (secure_endpoint *)secure_ep; - secure_endpoint_unref(ep); + SECURE_ENDPOINT_UNREF(ep, "destroy"); } static void endpoint_add_to_pollset(grpc_endpoint *secure_ep, @@ -343,9 +363,9 @@ static char *endpoint_get_peer(grpc_endpoint *secure_ep) { } static const grpc_endpoint_vtable vtable = { - endpoint_notify_on_read, endpoint_write, + endpoint_read, endpoint_write, endpoint_add_to_pollset, endpoint_add_to_pollset_set, - endpoint_shutdown, endpoint_unref, + endpoint_shutdown, endpoint_destroy, endpoint_get_peer}; grpc_endpoint *grpc_secure_endpoint_create( @@ -363,8 +383,10 @@ grpc_endpoint *grpc_secure_endpoint_create( } ep->write_staging_buffer = gpr_slice_malloc(STAGING_BUFFER_SIZE); ep->read_staging_buffer = gpr_slice_malloc(STAGING_BUFFER_SIZE); - gpr_slice_buffer_init(&ep->input_buffer); gpr_slice_buffer_init(&ep->output_buffer); + gpr_slice_buffer_init(&ep->source_buffer); + ep->read_buffer = NULL; + grpc_iomgr_closure_init(&ep->on_read, on_read_cb, ep); gpr_mu_init(&ep->protector_mu); gpr_ref_init(&ep->ref, 1); return &ep->base; diff --git a/src/core/security/secure_transport_setup.c b/src/core/security/secure_transport_setup.c index 0c3572b53c..bf0079577e 100644 --- a/src/core/security/secure_transport_setup.c +++ b/src/core/security/secure_transport_setup.c @@ -50,16 +50,17 @@ typedef struct { grpc_endpoint *wrapped_endpoint; grpc_endpoint *secure_endpoint; gpr_slice_buffer left_overs; + gpr_slice_buffer incoming; + gpr_slice_buffer outgoing; grpc_secure_transport_setup_done_cb cb; void *user_data; + grpc_iomgr_closure on_handshake_data_sent_to_peer; + grpc_iomgr_closure on_handshake_data_received_from_peer; } grpc_secure_transport_setup; -static void on_handshake_data_received_from_peer(void *setup, gpr_slice *slices, - size_t nslices, - grpc_endpoint_cb_status error); +static void on_handshake_data_received_from_peer(void *setup, int success); -static void on_handshake_data_sent_to_peer(void *setup, - grpc_endpoint_cb_status error); +static void on_handshake_data_sent_to_peer(void *setup, int success); static void secure_transport_setup_done(grpc_secure_transport_setup *s, int is_success) { @@ -78,6 +79,8 @@ static void secure_transport_setup_done(grpc_secure_transport_setup *s, if (s->handshaker != NULL) tsi_handshaker_destroy(s->handshaker); if (s->handshake_buffer != NULL) gpr_free(s->handshake_buffer); gpr_slice_buffer_destroy(&s->left_overs); + gpr_slice_buffer_destroy(&s->outgoing); + gpr_slice_buffer_destroy(&s->incoming); GRPC_SECURITY_CONNECTOR_UNREF(s->connector, "secure_transport_setup"); gpr_free(s); } @@ -102,6 +105,8 @@ static void on_peer_checked(void *user_data, grpc_security_status status) { s->secure_endpoint = grpc_secure_endpoint_create(protector, s->wrapped_endpoint, s->left_overs.slices, s->left_overs.count); + s->left_overs.count = 0; + s->left_overs.length = 0; secure_transport_setup_done(s, 1); return; } @@ -132,7 +137,6 @@ static void send_handshake_bytes_to_peer(grpc_secure_transport_setup *s) { size_t offset = 0; tsi_result result = TSI_OK; gpr_slice to_send; - grpc_endpoint_write_status write_status; do { size_t to_send_size = s->handshake_buffer_size - offset; @@ -155,28 +159,25 @@ static void send_handshake_bytes_to_peer(grpc_secure_transport_setup *s) { to_send = gpr_slice_from_copied_buffer((const char *)s->handshake_buffer, offset); + gpr_slice_buffer_reset_and_unref(&s->outgoing); + gpr_slice_buffer_add(&s->outgoing, to_send); /* TODO(klempner,jboeuf): This should probably use the client setup deadline */ - write_status = grpc_endpoint_write(s->wrapped_endpoint, &to_send, 1, - on_handshake_data_sent_to_peer, s); - if (write_status == GRPC_ENDPOINT_WRITE_ERROR) { - gpr_log(GPR_ERROR, "Could not send handshake data to peer."); - secure_transport_setup_done(s, 0); - } else if (write_status == GRPC_ENDPOINT_WRITE_DONE) { - on_handshake_data_sent_to_peer(s, GRPC_ENDPOINT_CB_OK); - } -} - -static void cleanup_slices(gpr_slice *slices, size_t num_slices) { - size_t i; - for (i = 0; i < num_slices; i++) { - gpr_slice_unref(slices[i]); + switch (grpc_endpoint_write(s->wrapped_endpoint, &s->outgoing, + &s->on_handshake_data_sent_to_peer)) { + case GRPC_ENDPOINT_ERROR: + gpr_log(GPR_ERROR, "Could not send handshake data to peer."); + secure_transport_setup_done(s, 0); + break; + case GRPC_ENDPOINT_DONE: + on_handshake_data_sent_to_peer(s, 1); + break; + case GRPC_ENDPOINT_PENDING: + break; } } -static void on_handshake_data_received_from_peer( - void *setup, gpr_slice *slices, size_t nslices, - grpc_endpoint_cb_status error) { +static void on_handshake_data_received_from_peer(void *setup, int success) { grpc_secure_transport_setup *s = setup; size_t consumed_slice_size = 0; tsi_result result = TSI_OK; @@ -184,32 +185,37 @@ static void on_handshake_data_received_from_peer( size_t num_left_overs; int has_left_overs_in_current_slice = 0; - if (error != GRPC_ENDPOINT_CB_OK) { + if (!success) { gpr_log(GPR_ERROR, "Read failed."); - cleanup_slices(slices, nslices); secure_transport_setup_done(s, 0); return; } - for (i = 0; i < nslices; i++) { - consumed_slice_size = GPR_SLICE_LENGTH(slices[i]); + for (i = 0; i < s->incoming.count; i++) { + consumed_slice_size = GPR_SLICE_LENGTH(s->incoming.slices[i]); result = tsi_handshaker_process_bytes_from_peer( - s->handshaker, GPR_SLICE_START_PTR(slices[i]), &consumed_slice_size); + s->handshaker, GPR_SLICE_START_PTR(s->incoming.slices[i]), + &consumed_slice_size); if (!tsi_handshaker_is_in_progress(s->handshaker)) break; } if (tsi_handshaker_is_in_progress(s->handshaker)) { /* We may need more data. */ if (result == TSI_INCOMPLETE_DATA) { - /* TODO(klempner,jboeuf): This should probably use the client setup - deadline */ - grpc_endpoint_notify_on_read(s->wrapped_endpoint, - on_handshake_data_received_from_peer, setup); - cleanup_slices(slices, nslices); + switch (grpc_endpoint_read(s->wrapped_endpoint, &s->incoming, + &s->on_handshake_data_received_from_peer)) { + case GRPC_ENDPOINT_DONE: + on_handshake_data_received_from_peer(s, 1); + break; + case GRPC_ENDPOINT_ERROR: + on_handshake_data_received_from_peer(s, 0); + break; + case GRPC_ENDPOINT_PENDING: + break; + } return; } else { send_handshake_bytes_to_peer(s); - cleanup_slices(slices, nslices); return; } } @@ -217,42 +223,40 @@ static void on_handshake_data_received_from_peer( if (result != TSI_OK) { gpr_log(GPR_ERROR, "Handshake failed with error %s", tsi_result_to_string(result)); - cleanup_slices(slices, nslices); secure_transport_setup_done(s, 0); return; } /* Handshake is done and successful this point. */ has_left_overs_in_current_slice = - (consumed_slice_size < GPR_SLICE_LENGTH(slices[i])); - num_left_overs = (has_left_overs_in_current_slice ? 1 : 0) + nslices - i - 1; + (consumed_slice_size < GPR_SLICE_LENGTH(s->incoming.slices[i])); + num_left_overs = + (has_left_overs_in_current_slice ? 1 : 0) + s->incoming.count - i - 1; if (num_left_overs == 0) { - cleanup_slices(slices, nslices); check_peer(s); return; } - cleanup_slices(slices, nslices - num_left_overs); - /* Put the leftovers in our buffer (ownership transfered). */ if (has_left_overs_in_current_slice) { - gpr_slice_buffer_add(&s->left_overs, - gpr_slice_split_tail(&slices[i], consumed_slice_size)); - gpr_slice_unref(slices[i]); /* split_tail above increments refcount. */ + gpr_slice_buffer_add( + &s->left_overs, + gpr_slice_split_tail(&s->incoming.slices[i], consumed_slice_size)); + gpr_slice_unref( + s->incoming.slices[i]); /* split_tail above increments refcount. */ } gpr_slice_buffer_addn( - &s->left_overs, &slices[i + 1], + &s->left_overs, &s->incoming.slices[i + 1], num_left_overs - (size_t)has_left_overs_in_current_slice); check_peer(s); } /* If setup is NULL, the setup is done. */ -static void on_handshake_data_sent_to_peer(void *setup, - grpc_endpoint_cb_status error) { +static void on_handshake_data_sent_to_peer(void *setup, int success) { grpc_secure_transport_setup *s = setup; /* Make sure that write is OK. */ - if (error != GRPC_ENDPOINT_CB_OK) { - gpr_log(GPR_ERROR, "Write failed with error %d.", error); + if (!success) { + gpr_log(GPR_ERROR, "Write failed."); if (setup != NULL) secure_transport_setup_done(s, 0); return; } @@ -261,8 +265,17 @@ static void on_handshake_data_sent_to_peer(void *setup, if (tsi_handshaker_is_in_progress(s->handshaker)) { /* TODO(klempner,jboeuf): This should probably use the client setup deadline */ - grpc_endpoint_notify_on_read(s->wrapped_endpoint, - on_handshake_data_received_from_peer, setup); + switch (grpc_endpoint_read(s->wrapped_endpoint, &s->incoming, + &s->on_handshake_data_received_from_peer)) { + case GRPC_ENDPOINT_ERROR: + on_handshake_data_received_from_peer(s, 0); + break; + case GRPC_ENDPOINT_PENDING: + break; + case GRPC_ENDPOINT_DONE: + on_handshake_data_received_from_peer(s, 1); + break; + } } else { check_peer(s); } @@ -288,6 +301,12 @@ void grpc_setup_secure_transport(grpc_security_connector *connector, s->wrapped_endpoint = nonsecure_endpoint; s->user_data = user_data; s->cb = cb; + grpc_iomgr_closure_init(&s->on_handshake_data_sent_to_peer, + on_handshake_data_sent_to_peer, s); + grpc_iomgr_closure_init(&s->on_handshake_data_received_from_peer, + on_handshake_data_received_from_peer, s); gpr_slice_buffer_init(&s->left_overs); + gpr_slice_buffer_init(&s->outgoing); + gpr_slice_buffer_init(&s->incoming); send_handshake_bytes_to_peer(s); } diff --git a/src/core/security/security_context.c b/src/core/security/security_context.c index c1b434f302..95d80ba122 100644 --- a/src/core/security/security_context.c +++ b/src/core/security/security_context.c @@ -42,19 +42,6 @@ #include <grpc/support/log.h> #include <grpc/support/string_util.h> -/* --- grpc_process_auth_metadata_func --- */ - -static grpc_auth_metadata_processor server_processor = {NULL, NULL}; - -grpc_auth_metadata_processor grpc_server_get_auth_metadata_processor(void) { - return server_processor; -} - -void grpc_server_register_auth_metadata_processor( - grpc_auth_metadata_processor processor) { - server_processor = processor; -} - /* --- grpc_call --- */ grpc_call_error grpc_call_set_credentials(grpc_call *call, diff --git a/src/core/security/server_auth_filter.c b/src/core/security/server_auth_filter.c index 6e831431fa..d134201e87 100644 --- a/src/core/security/server_auth_filter.c +++ b/src/core/security/server_auth_filter.c @@ -50,6 +50,7 @@ typedef struct call_data { handling it. */ grpc_iomgr_closure auth_on_recv; grpc_transport_stream_op transport_op; + grpc_metadata_array md; const grpc_metadata *consumed_md; size_t num_consumed_md; grpc_stream_op *md_op; @@ -90,13 +91,17 @@ static grpc_mdelem *remove_consumed_md(void *user_data, grpc_mdelem *md) { call_data *calld = elem->call_data; size_t i; for (i = 0; i < calld->num_consumed_md; i++) { + const grpc_metadata *consumed_md = &calld->consumed_md[i]; /* Maybe we could do a pointer comparison but we do not have any guarantee that the metadata processor used the same pointers for consumed_md in the callback. */ - if (memcmp(GPR_SLICE_START_PTR(md->key->slice), calld->consumed_md[i].key, + if (GPR_SLICE_LENGTH(md->key->slice) != strlen(consumed_md->key) || + GPR_SLICE_LENGTH(md->value->slice) != consumed_md->value_length) { + continue; + } + if (memcmp(GPR_SLICE_START_PTR(md->key->slice), consumed_md->key, GPR_SLICE_LENGTH(md->key->slice)) == 0 && - memcmp(GPR_SLICE_START_PTR(md->value->slice), - calld->consumed_md[i].value, + memcmp(GPR_SLICE_START_PTR(md->value->slice), consumed_md->value, GPR_SLICE_LENGTH(md->value->slice)) == 0) { return NULL; /* Delete. */ } @@ -123,9 +128,11 @@ static void on_md_processing_done( calld->num_consumed_md = num_consumed_md; grpc_metadata_batch_filter(&calld->md_op->data.metadata, remove_consumed_md, elem); + grpc_metadata_array_destroy(&calld->md); calld->on_done_recv->cb(calld->on_done_recv->cb_arg, 1); } else { gpr_slice message; + grpc_metadata_array_destroy(&calld->md); error_details = error_details != NULL ? error_details : "Authentication metadata processing failed."; @@ -145,17 +152,15 @@ static void auth_on_recv(void *user_data, int success) { size_t nops = calld->recv_ops->nops; grpc_stream_op *ops = calld->recv_ops->ops; for (i = 0; i < nops; i++) { - grpc_metadata_array md_array; grpc_stream_op *op = &ops[i]; if (op->type != GRPC_OP_METADATA || calld->got_client_metadata) continue; calld->got_client_metadata = 1; if (chand->processor.process == NULL) continue; calld->md_op = op; - md_array = metadata_batch_to_md_array(&op->data.metadata); + calld->md = metadata_batch_to_md_array(&op->data.metadata); chand->processor.process(chand->processor.state, calld->auth_context, - md_array.metadata, md_array.count, + calld->md.metadata, calld->md.count, on_md_processing_done, elem); - grpc_metadata_array_destroy(&md_array); return; } } diff --git a/src/core/security/server_secure_chttp2.c b/src/core/security/server_secure_chttp2.c index 8d9d036d80..4749f5f516 100644 --- a/src/core/security/server_secure_chttp2.c +++ b/src/core/security/server_secure_chttp2.c @@ -61,7 +61,7 @@ typedef struct grpc_server_secure_state { grpc_server *server; grpc_tcp_server *tcp; grpc_security_connector *sc; - grpc_auth_metadata_processor processor; + grpc_server_credentials *creds; tcp_endpoint_list *handshaking_tcp_endpoints; int is_shutdown; gpr_mu mu; @@ -79,6 +79,7 @@ static void state_unref(grpc_server_secure_state *state) { gpr_mu_unlock(&state->mu); /* clean up */ GRPC_SECURITY_CONNECTOR_UNREF(state->sc, "server"); + grpc_server_credentials_unref(state->creds); gpr_free(state); } } @@ -91,7 +92,8 @@ static void setup_transport(void *statep, grpc_transport *transport, grpc_channel_args *args_copy; grpc_arg args_to_add[2]; args_to_add[0] = grpc_security_connector_to_arg(state->sc); - args_to_add[1] = grpc_auth_metadata_processor_to_arg(&state->processor); + args_to_add[1] = + grpc_auth_metadata_processor_to_arg(&state->creds->processor); args_copy = grpc_channel_args_copy_and_add( grpc_server_get_channel_args(state->server), args_to_add, GPR_ARRAY_SIZE(args_to_add)); @@ -262,7 +264,8 @@ int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr, state->server = server; state->tcp = tcp; state->sc = sc; - state->processor = creds->processor; + state->creds = grpc_server_credentials_ref(creds); + state->handshaking_tcp_endpoints = NULL; state->is_shutdown = 0; gpr_mu_init(&state->mu); diff --git a/src/core/support/log_win32.c b/src/core/support/log_win32.c index 629781da8a..b68239f8f5 100644 --- a/src/core/support/log_win32.c +++ b/src/core/support/log_win32.c @@ -81,10 +81,18 @@ void gpr_log(const char *file, int line, gpr_log_severity severity, /* Simple starter implementation */ void gpr_default_log(gpr_log_func_args *args) { + char *final_slash; + const char *display_file; char time_buffer[64]; gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); struct tm tm; + final_slash = strrchr(args->file, '\\'); + if (final_slash == NULL) + display_file = args->file; + else + display_file = final_slash + 1; + if (localtime_s(&tm, &now.tv_sec)) { strcpy(time_buffer, "error:localtime"); } else if (0 == @@ -94,7 +102,7 @@ void gpr_default_log(gpr_log_func_args *args) { fprintf(stderr, "%s%s.%09u %5lu %s:%d] %s\n", gpr_log_severity_string(args->severity), time_buffer, - (int)(now.tv_nsec), GetCurrentThreadId(), args->file, args->line, + (int)(now.tv_nsec), GetCurrentThreadId(), display_file, args->line, args->message); } diff --git a/src/core/support/slice_buffer.c b/src/core/support/slice_buffer.c index 987d5cb9b5..6482ef9c9f 100644 --- a/src/core/support/slice_buffer.c +++ b/src/core/support/slice_buffer.c @@ -207,3 +207,25 @@ void gpr_slice_buffer_move_into(gpr_slice_buffer *src, gpr_slice_buffer *dst) { src->count = 0; src->length = 0; } + +void gpr_slice_buffer_trim_end(gpr_slice_buffer *sb, size_t n) { + GPR_ASSERT(n <= sb->length); + sb->length -= n; + for (;;) { + size_t idx = sb->count - 1; + gpr_slice slice = sb->slices[idx]; + size_t slice_len = GPR_SLICE_LENGTH(slice); + if (slice_len > n) { + sb->slices[idx] = gpr_slice_sub_no_ref(slice, 0, slice_len - n); + return; + } else if (slice_len == n) { + gpr_slice_unref(slice); + sb->count = idx; + return; + } else { + gpr_slice_unref(slice); + n -= slice_len; + sb->count = idx; + } + } +} diff --git a/src/core/support/sync_win32.c b/src/core/support/sync_win32.c index df23492171..f546477067 100644 --- a/src/core/support/sync_win32.c +++ b/src/core/support/sync_win32.c @@ -88,9 +88,9 @@ int gpr_cv_wait(gpr_cv *cv, gpr_mu *mu, gpr_timespec abs_deadline) { SleepConditionVariableCS(cv, &mu->cs, INFINITE); } else { gpr_timespec now = gpr_now(abs_deadline.clock_type); - gpr_int64 now_ms = now.tv_sec * 1000 + now.tv_nsec / 1000000; + gpr_int64 now_ms = (gpr_int64)now.tv_sec * 1000 + now.tv_nsec / 1000000; gpr_int64 deadline_ms = - abs_deadline.tv_sec * 1000 + abs_deadline.tv_nsec / 1000000; + (gpr_int64)abs_deadline.tv_sec * 1000 + abs_deadline.tv_nsec / 1000000; if (now_ms >= deadline_ms) { timeout = 1; } else { diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 4426bbbce9..6ece56371b 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -61,18 +61,6 @@ - status/close recv (depending on client/server) */ #define MAX_CONCURRENT_COMPLETIONS 6 -typedef enum { REQ_INITIAL = 0, REQ_READY, REQ_DONE } req_state; - -typedef enum { - SEND_NOTHING, - SEND_INITIAL_METADATA, - SEND_BUFFERED_INITIAL_METADATA, - SEND_MESSAGE, - SEND_BUFFERED_MESSAGE, - SEND_TRAILING_METADATA_AND_FINISH, - SEND_FINISH -} send_action; - typedef struct { grpc_ioreq_completion_func on_complete; void *user_data; @@ -630,9 +618,6 @@ static void unlock(grpc_call *call) { call->cancel_alarm = 0; if (!call->receiving && need_more_data(call)) { - op.recv_ops = &call->recv_ops; - op.recv_state = &call->recv_state; - op.on_done_recv = &call->on_done_recv; if (grpc_bbq_empty(&call->incoming_queue) && call->reading_message) { op.max_recv_bytes = call->incoming_message_length - call->incoming_message.length + MAX_RECV_PEEK_AHEAD; @@ -644,9 +629,16 @@ static void unlock(grpc_call *call) { op.max_recv_bytes = MAX_RECV_PEEK_AHEAD - buffered_bytes; } } - call->receiving = 1; - GRPC_CALL_INTERNAL_REF(call, "receiving"); - start_op = 1; + /* TODO(ctiller): 1024 is basically to cover a bug + I don't understand yet */ + if (op.max_recv_bytes > 1024) { + op.recv_ops = &call->recv_ops; + op.recv_state = &call->recv_state; + op.on_done_recv = &call->on_done_recv; + call->receiving = 1; + GRPC_CALL_INTERNAL_REF(call, "receiving"); + start_op = 1; + } } if (!call->sending) { diff --git a/src/core/surface/completion_queue.h b/src/core/surface/completion_queue.h index 8de024aaea..74dc09e36e 100644 --- a/src/core/surface/completion_queue.h +++ b/src/core/surface/completion_queue.h @@ -34,7 +34,7 @@ #ifndef GRPC_INTERNAL_CORE_SURFACE_COMPLETION_QUEUE_H #define GRPC_INTERNAL_CORE_SURFACE_COMPLETION_QUEUE_H -/* Internal API for completion channels */ +/* Internal API for completion queues */ #include "src/core/iomgr/pollset.h" #include <grpc/grpc.h> diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c index c68ba0e44b..9b554eeb70 100644 --- a/src/core/surface/secure_channel_create.c +++ b/src/core/surface/secure_channel_create.c @@ -185,7 +185,8 @@ static const grpc_subchannel_factory_vtable subchannel_factory_vtable = { - perform handshakes */ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds, const char *target, - const grpc_channel_args *args) { + const grpc_channel_args *args, + void *reserved) { grpc_channel *channel; grpc_arg connector_arg; grpc_channel_args *args_copy; @@ -198,6 +199,7 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds, const grpc_channel_filter *filters[MAX_FILTERS]; size_t n = 0; + GPR_ASSERT(reserved == NULL); if (grpc_find_security_connector_in_args(args) != NULL) { gpr_log(GPR_ERROR, "Cannot set security context in channel args."); return grpc_lame_client_channel_create( diff --git a/src/core/surface/version.c b/src/core/surface/version.c index d7aaba3868..4b90e06a04 100644 --- a/src/core/surface/version.c +++ b/src/core/surface/version.c @@ -37,5 +37,5 @@ #include <grpc/grpc.h> const char *grpc_version_string(void) { - return "0.10.1.0"; + return "0.11.0.0"; } diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h index 42cf0ecd5b..7a42de9245 100644 --- a/src/core/transport/chttp2/internal.h +++ b/src/core/transport/chttp2/internal.h @@ -214,6 +214,8 @@ typedef struct { grpc_chttp2_hpack_compressor hpack_compressor; /** is this a client? */ gpr_uint8 is_client; + /** callback for when writing is done */ + grpc_iomgr_closure done_cb; } grpc_chttp2_transport_writing; struct grpc_chttp2_transport_parsing { @@ -291,6 +293,9 @@ struct grpc_chttp2_transport { gpr_refcount refs; char *peer_string; + /** when this drops to zero it's safe to shutdown the endpoint */ + gpr_refcount shutdown_ep_refs; + gpr_mu mu; /** is the transport destroying itself? */ @@ -329,8 +334,11 @@ struct grpc_chttp2_transport { /** closure to execute writing */ grpc_iomgr_closure writing_action; - /** closure to start reading from the endpoint */ - grpc_iomgr_closure reading_action; + /** closure to finish reading from the endpoint */ + grpc_iomgr_closure recv_data; + + /** incoming read bytes */ + gpr_slice_buffer read_buffer; /** address to place a newly accepted stream - set and unset by grpc_chttp2_parsing_accept_stream; used by init_stream to @@ -463,8 +471,7 @@ int grpc_chttp2_unlocking_check_writes(grpc_chttp2_transport_global *global, grpc_chttp2_transport_writing *writing); void grpc_chttp2_perform_writes( grpc_chttp2_transport_writing *transport_writing, grpc_endpoint *endpoint); -void grpc_chttp2_terminate_writing( - grpc_chttp2_transport_writing *transport_writing, int success); +void grpc_chttp2_terminate_writing(void *transport_writing, int success); void grpc_chttp2_cleanup_writing(grpc_chttp2_transport_global *global, grpc_chttp2_transport_writing *writing); diff --git a/src/core/transport/chttp2/stream_lists.c b/src/core/transport/chttp2/stream_lists.c index 38c6052f9c..781db7b0d6 100644 --- a/src/core/transport/chttp2/stream_lists.c +++ b/src/core/transport/chttp2/stream_lists.c @@ -177,8 +177,10 @@ int grpc_chttp2_list_pop_writable_stream( grpc_chttp2_stream *stream; int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, GRPC_CHTTP2_LIST_WRITABLE); - *stream_global = &stream->global; - *stream_writing = &stream->writing; + if (r != 0) { + *stream_global = &stream->global; + *stream_writing = &stream->writing; + } return r; } @@ -210,7 +212,9 @@ int grpc_chttp2_list_pop_writing_stream( grpc_chttp2_stream *stream; int r = stream_list_pop(TRANSPORT_FROM_WRITING(transport_writing), &stream, GRPC_CHTTP2_LIST_WRITING); - *stream_writing = &stream->writing; + if (r != 0) { + *stream_writing = &stream->writing; + } return r; } @@ -230,8 +234,10 @@ int grpc_chttp2_list_pop_written_stream( grpc_chttp2_stream *stream; int r = stream_list_pop(TRANSPORT_FROM_WRITING(transport_writing), &stream, GRPC_CHTTP2_LIST_WRITTEN); - *stream_global = &stream->global; - *stream_writing = &stream->writing; + if (r != 0) { + *stream_global = &stream->global; + *stream_writing = &stream->writing; + } return r; } @@ -251,8 +257,10 @@ int grpc_chttp2_list_pop_parsing_seen_stream( grpc_chttp2_stream *stream; int r = stream_list_pop(TRANSPORT_FROM_PARSING(transport_parsing), &stream, GRPC_CHTTP2_LIST_PARSING_SEEN); - *stream_global = &stream->global; - *stream_parsing = &stream->parsing; + if (r != 0) { + *stream_global = &stream->global; + *stream_parsing = &stream->parsing; + } return r; } @@ -270,7 +278,9 @@ int grpc_chttp2_list_pop_waiting_for_concurrency( grpc_chttp2_stream *stream; int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, GRPC_CHTTP2_LIST_WAITING_FOR_CONCURRENCY); - *stream_global = &stream->global; + if (r != 0) { + *stream_global = &stream->global; + } return r; } @@ -288,7 +298,9 @@ int grpc_chttp2_list_pop_closed_waiting_for_parsing( grpc_chttp2_stream *stream; int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING); - *stream_global = &stream->global; + if (r != 0) { + *stream_global = &stream->global; + } return r; } @@ -306,7 +318,9 @@ int grpc_chttp2_list_pop_cancelled_waiting_for_writing( grpc_chttp2_stream *stream; int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, GRPC_CHTTP2_LIST_CANCELLED_WAITING_FOR_WRITING); - *stream_global = &stream->global; + if (r != 0) { + *stream_global = &stream->global; + } return r; } @@ -326,8 +340,10 @@ int grpc_chttp2_list_pop_incoming_window_updated( grpc_chttp2_stream *stream; int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, GRPC_CHTTP2_LIST_INCOMING_WINDOW_UPDATED); - *stream_global = &stream->global; - *stream_parsing = &stream->parsing; + if (r != 0) { + *stream_global = &stream->global; + *stream_parsing = &stream->parsing; + } return r; } @@ -353,7 +369,9 @@ int grpc_chttp2_list_pop_read_write_state_changed( grpc_chttp2_stream *stream; int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, GRPC_CHTTP2_LIST_READ_WRITE_STATE_CHANGED); - *stream_global = &stream->global; + if (r != 0) { + *stream_global = &stream->global; + } return r; } diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c index 123061b3fc..ac79044e08 100644 --- a/src/core/transport/chttp2/writing.c +++ b/src/core/transport/chttp2/writing.c @@ -37,7 +37,6 @@ #include <grpc/support/log.h> static void finalize_outbuf(grpc_chttp2_transport_writing *transport_writing); -static void finish_write_cb(void *tw, grpc_endpoint_cb_status write_status); int grpc_chttp2_unlocking_check_writes( grpc_chttp2_transport_global *transport_global, @@ -114,6 +113,10 @@ int grpc_chttp2_unlocking_check_writes( if (!stream_global->read_closed && stream_global->unannounced_incoming_window > 0) { + GPR_ASSERT(stream_writing->announce_window == 0); + GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( + "write", transport_writing, stream_writing, announce_window, + stream_global->unannounced_incoming_window); stream_writing->announce_window = stream_global->unannounced_incoming_window; GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( @@ -165,16 +168,15 @@ void grpc_chttp2_perform_writes( GPR_ASSERT(transport_writing->outbuf.count > 0); GPR_ASSERT(endpoint); - switch (grpc_endpoint_write(endpoint, transport_writing->outbuf.slices, - transport_writing->outbuf.count, finish_write_cb, - transport_writing)) { - case GRPC_ENDPOINT_WRITE_DONE: + switch (grpc_endpoint_write(endpoint, &transport_writing->outbuf, + &transport_writing->done_cb)) { + case GRPC_ENDPOINT_DONE: grpc_chttp2_terminate_writing(transport_writing, 1); break; - case GRPC_ENDPOINT_WRITE_ERROR: + case GRPC_ENDPOINT_ERROR: grpc_chttp2_terminate_writing(transport_writing, 0); break; - case GRPC_ENDPOINT_WRITE_PENDING: + case GRPC_ENDPOINT_PENDING: break; } } @@ -198,6 +200,9 @@ static void finalize_outbuf(grpc_chttp2_transport_writing *transport_writing) { &transport_writing->outbuf, grpc_chttp2_window_update_create(stream_writing->id, stream_writing->announce_window)); + GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( + "write", transport_writing, stream_writing, announce_window, + -(gpr_int64)stream_writing->announce_window); stream_writing->announce_window = 0; } if (stream_writing->send_closed == GRPC_SEND_CLOSED_WITH_RST_STREAM) { @@ -209,12 +214,6 @@ static void finalize_outbuf(grpc_chttp2_transport_writing *transport_writing) { } } -static void finish_write_cb(void *tw, grpc_endpoint_cb_status write_status) { - grpc_chttp2_transport_writing *transport_writing = tw; - grpc_chttp2_terminate_writing(transport_writing, - write_status == GRPC_ENDPOINT_CB_OK); -} - void grpc_chttp2_cleanup_writing( grpc_chttp2_transport_global *transport_global, grpc_chttp2_transport_writing *transport_writing) { @@ -243,6 +242,5 @@ void grpc_chttp2_cleanup_writing( grpc_chttp2_list_add_read_write_state_changed(transport_global, stream_global); } - transport_writing->outbuf.count = 0; - transport_writing->outbuf.length = 0; + gpr_slice_buffer_reset_and_unref(&transport_writing->outbuf); } diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 1bbd210e46..9e3d7dd551 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -84,15 +84,13 @@ static void unlock_check_read_write_state(grpc_chttp2_transport *t); /* forward declarations of various callbacks that we'll build closures around */ static void writing_action(void *t, int iomgr_success_ignored); -static void reading_action(void *t, int iomgr_success_ignored); /** Set a transport level setting, and push it to our peer */ static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id, gpr_uint32 value); /** Endpoint callback to process incoming data */ -static void recv_data(void *tp, gpr_slice *slices, size_t nslices, - grpc_endpoint_cb_status error); +static void recv_data(void *tp, int success); /** Start disconnection chain */ static void drop_connection(grpc_chttp2_transport *t); @@ -143,6 +141,7 @@ static void destruct_transport(grpc_chttp2_transport *t) { grpc_chttp2_hpack_compressor_destroy(&t->writing.hpack_compressor); gpr_slice_buffer_destroy(&t->parsing.qbuf); + gpr_slice_buffer_destroy(&t->read_buffer); grpc_chttp2_hpack_parser_destroy(&t->parsing.hpack_parser); grpc_chttp2_goaway_parser_destroy(&t->parsing.goaway_parser); @@ -223,6 +222,8 @@ static void init_transport(grpc_chttp2_transport *t, t->ep = ep; /* one ref is for destroy, the other for when ep becomes NULL */ gpr_ref_init(&t->refs, 2); + /* ref is dropped at transport close() */ + gpr_ref_init(&t->shutdown_ep_refs, 1); gpr_mu_init(&t->mu); grpc_mdctx_ref(mdctx); t->peer_string = grpc_endpoint_get_peer(ep); @@ -249,12 +250,16 @@ static void init_transport(grpc_chttp2_transport *t, gpr_slice_buffer_init(&t->writing.outbuf); grpc_chttp2_hpack_compressor_init(&t->writing.hpack_compressor, mdctx); grpc_iomgr_closure_init(&t->writing_action, writing_action, t); - grpc_iomgr_closure_init(&t->reading_action, reading_action, t); gpr_slice_buffer_init(&t->parsing.qbuf); grpc_chttp2_goaway_parser_init(&t->parsing.goaway_parser); grpc_chttp2_hpack_parser_init(&t->parsing.hpack_parser, t->metadata_context); + grpc_iomgr_closure_init(&t->writing.done_cb, grpc_chttp2_terminate_writing, + &t->writing); + grpc_iomgr_closure_init(&t->recv_data, recv_data, t); + gpr_slice_buffer_init(&t->read_buffer); + if (is_client) { gpr_slice_buffer_add( &t->global.qbuf, @@ -333,13 +338,45 @@ static void destroy_transport(grpc_transport *gt) { UNREF_TRANSPORT(t, "destroy"); } +/** block grpc_endpoint_shutdown being called until a paired + allow_endpoint_shutdown is made */ +static void prevent_endpoint_shutdown(grpc_chttp2_transport *t) { + GPR_ASSERT(t->ep); + gpr_ref(&t->shutdown_ep_refs); +} + +static void allow_endpoint_shutdown_locked(grpc_chttp2_transport *t) { + if (gpr_unref(&t->shutdown_ep_refs)) { + if (t->ep) { + grpc_endpoint_shutdown(t->ep); + } + } +} + +static void allow_endpoint_shutdown_unlocked(grpc_chttp2_transport *t) { + if (gpr_unref(&t->shutdown_ep_refs)) { + gpr_mu_lock(&t->mu); + if (t->ep) { + grpc_endpoint_shutdown(t->ep); + } + gpr_mu_unlock(&t->mu); + } +} + +static void destroy_endpoint(grpc_chttp2_transport *t) { + grpc_endpoint_destroy(t->ep); + t->ep = NULL; + UNREF_TRANSPORT( + t, "disconnect"); /* safe because we'll still have the ref for write */ +} + static void close_transport_locked(grpc_chttp2_transport *t) { if (!t->closed) { t->closed = 1; connectivity_state_set(&t->global, GRPC_CHANNEL_FATAL_FAILURE, "close_transport"); if (t->ep) { - grpc_endpoint_shutdown(t->ep); + allow_endpoint_shutdown_locked(t); } } } @@ -468,6 +505,7 @@ static void unlock(grpc_chttp2_transport *t) { t->writing_active = 1; REF_TRANSPORT(t, "writing"); grpc_chttp2_schedule_closure(&t->global, &t->writing_action, 1); + prevent_endpoint_shutdown(t); } run_closures = t->global.pending_closures_head; @@ -502,12 +540,14 @@ static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id, } } -void grpc_chttp2_terminate_writing( - grpc_chttp2_transport_writing *transport_writing, int success) { +void grpc_chttp2_terminate_writing(void *transport_writing_ptr, int success) { + grpc_chttp2_transport_writing *transport_writing = transport_writing_ptr; grpc_chttp2_transport *t = TRANSPORT_FROM_WRITING(transport_writing); lock(t); + allow_endpoint_shutdown_locked(t); + if (!success) { drop_connection(t); } @@ -519,10 +559,7 @@ void grpc_chttp2_terminate_writing( from starting */ t->writing_active = 0; if (t->ep && !t->endpoint_reading) { - grpc_endpoint_destroy(t->ep); - t->ep = NULL; - UNREF_TRANSPORT( - t, "disconnect"); /* safe because we'll still have the ref for write */ + destroy_endpoint(t); } unlock(t); @@ -658,9 +695,9 @@ static void perform_stream_op_locked( } grpc_chttp2_incoming_metadata_live_op_buffer_end( &stream_global->outstanding_metadata); + grpc_chttp2_list_add_read_write_state_changed(transport_global, + stream_global); if (stream_global->id != 0) { - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); grpc_chttp2_list_add_writable_stream(transport_global, stream_global); } } @@ -1052,82 +1089,90 @@ static void update_global_window(void *args, gpr_uint32 id, void *stream) { static void read_error_locked(grpc_chttp2_transport *t) { t->endpoint_reading = 0; if (!t->writing_active && t->ep) { - grpc_endpoint_destroy(t->ep); - t->ep = NULL; - /* safe as we still have a ref for read */ - UNREF_TRANSPORT(t, "disconnect"); + destroy_endpoint(t); } } /* tcp read callback */ -static void recv_data(void *tp, gpr_slice *slices, size_t nslices, - grpc_endpoint_cb_status error) { - grpc_chttp2_transport *t = tp; +static int recv_data_loop(grpc_chttp2_transport *t, int *success) { size_t i; - int unref = 0; + int keep_reading = 0; - switch (error) { - case GRPC_ENDPOINT_CB_SHUTDOWN: - case GRPC_ENDPOINT_CB_EOF: - case GRPC_ENDPOINT_CB_ERROR: - lock(t); + lock(t); + i = 0; + GPR_ASSERT(!t->parsing_active); + if (!t->closed) { + t->parsing_active = 1; + /* merge stream lists */ + grpc_chttp2_stream_map_move_into(&t->new_stream_map, + &t->parsing_stream_map); + grpc_chttp2_prepare_to_read(&t->global, &t->parsing); + gpr_mu_unlock(&t->mu); + for (; i < t->read_buffer.count && + grpc_chttp2_perform_read(&t->parsing, t->read_buffer.slices[i]); + i++) + ; + gpr_mu_lock(&t->mu); + if (i != t->read_buffer.count) { drop_connection(t); - read_error_locked(t); - unlock(t); - unref = 1; - for (i = 0; i < nslices; i++) gpr_slice_unref(slices[i]); - break; - case GRPC_ENDPOINT_CB_OK: - lock(t); - i = 0; - GPR_ASSERT(!t->parsing_active); - if (!t->closed) { - t->parsing_active = 1; - /* merge stream lists */ - grpc_chttp2_stream_map_move_into(&t->new_stream_map, - &t->parsing_stream_map); - grpc_chttp2_prepare_to_read(&t->global, &t->parsing); - gpr_mu_unlock(&t->mu); - for (; i < nslices && grpc_chttp2_perform_read(&t->parsing, slices[i]); - i++) { - gpr_slice_unref(slices[i]); - } - gpr_mu_lock(&t->mu); - if (i != nslices) { - drop_connection(t); - } - /* merge stream lists */ - grpc_chttp2_stream_map_move_into(&t->new_stream_map, - &t->parsing_stream_map); - t->global.concurrent_stream_count = - grpc_chttp2_stream_map_size(&t->parsing_stream_map); - if (t->parsing.initial_window_update != 0) { - grpc_chttp2_stream_map_for_each(&t->parsing_stream_map, - update_global_window, t); - t->parsing.initial_window_update = 0; - } - /* handle higher level things */ - grpc_chttp2_publish_reads(&t->global, &t->parsing); - t->parsing_active = 0; - } - if (i == nslices) { - grpc_chttp2_schedule_closure(&t->global, &t->reading_action, 1); - } else { - read_error_locked(t); - unref = 1; - } - unlock(t); - for (; i < nslices; i++) gpr_slice_unref(slices[i]); - break; + } + /* merge stream lists */ + grpc_chttp2_stream_map_move_into(&t->new_stream_map, + &t->parsing_stream_map); + t->global.concurrent_stream_count = + grpc_chttp2_stream_map_size(&t->parsing_stream_map); + if (t->parsing.initial_window_update != 0) { + grpc_chttp2_stream_map_for_each(&t->parsing_stream_map, + update_global_window, t); + t->parsing.initial_window_update = 0; + } + /* handle higher level things */ + grpc_chttp2_publish_reads(&t->global, &t->parsing); + t->parsing_active = 0; + } + if (!*success || i != t->read_buffer.count) { + drop_connection(t); + read_error_locked(t); + } else if (!t->closed) { + keep_reading = 1; + REF_TRANSPORT(t, "keep_reading"); + prevent_endpoint_shutdown(t); } - if (unref) { + gpr_slice_buffer_reset_and_unref(&t->read_buffer); + unlock(t); + + if (keep_reading) { + int ret = -1; + switch (grpc_endpoint_read(t->ep, &t->read_buffer, &t->recv_data)) { + case GRPC_ENDPOINT_DONE: + *success = 1; + ret = 1; + break; + case GRPC_ENDPOINT_ERROR: + *success = 0; + ret = 1; + break; + case GRPC_ENDPOINT_PENDING: + ret = 0; + break; + } + allow_endpoint_shutdown_unlocked(t); + UNREF_TRANSPORT(t, "keep_reading"); + return ret; + } else { UNREF_TRANSPORT(t, "recv_data"); + return 0; } + + gpr_log(GPR_ERROR, "should never reach here"); + abort(); } -static void reading_action(void *pt, int iomgr_success_ignored) { - grpc_chttp2_transport *t = pt; - grpc_endpoint_notify_on_read(t->ep, recv_data, t); +static void recv_data(void *tp, int success) { + grpc_chttp2_transport *t = tp; + + while (recv_data_loop(t, &success)) + ; } /* @@ -1240,5 +1285,6 @@ void grpc_chttp2_transport_start_reading(grpc_transport *transport, gpr_slice *slices, size_t nslices) { grpc_chttp2_transport *t = (grpc_chttp2_transport *)transport; REF_TRANSPORT(t, "recv_data"); /* matches unref inside recv_data */ - recv_data(t, slices, nslices, GRPC_ENDPOINT_CB_OK); + gpr_slice_buffer_addn(&t->read_buffer, slices, nslices); + recv_data(t, 1); } diff --git a/src/core/transport/metadata.c b/src/core/transport/metadata.c index 3fd21a2f5d..61638764a6 100644 --- a/src/core/transport/metadata.c +++ b/src/core/transport/metadata.c @@ -703,7 +703,7 @@ int grpc_mdstr_is_legal_header(grpc_mdstr *s) { int grpc_mdstr_is_legal_nonbin_header(grpc_mdstr *s) { static const gpr_uint8 legal_header_bits[256 / 8] = { - 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, + 0x00, 0x00, 0x00, 0x00, 0xff, 0xef, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x7f, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}; return conforms_to(s, legal_header_bits); |