diff options
Diffstat (limited to 'src/core/lib')
-rw-r--r-- | src/core/lib/http/parser.c | 4 | ||||
-rw-r--r-- | src/core/lib/iomgr/error.c | 2 | ||||
-rw-r--r-- | src/core/lib/iomgr/pollset_uv.c | 22 | ||||
-rw-r--r-- | src/core/lib/iomgr/sockaddr_utils.c | 12 | ||||
-rw-r--r-- | src/core/lib/iomgr/tcp_server_posix.c | 19 | ||||
-rw-r--r-- | src/core/lib/iomgr/udp_server.c | 6 | ||||
-rw-r--r-- | src/core/lib/support/sync.c | 12 | ||||
-rw-r--r-- | src/core/lib/surface/call.c | 4 | ||||
-rw-r--r-- | src/core/lib/transport/transport.c | 35 | ||||
-rw-r--r-- | src/core/lib/transport/transport.h | 6 | ||||
-rw-r--r-- | src/core/lib/tsi/test_creds/BUILD | 36 |
11 files changed, 142 insertions, 16 deletions
diff --git a/src/core/lib/http/parser.c b/src/core/lib/http/parser.c index 2f84adc187..b9c56c103c 100644 --- a/src/core/lib/http/parser.c +++ b/src/core/lib/http/parser.c @@ -284,9 +284,9 @@ static grpc_error *addbyte(grpc_http_parser *parser, uint8_t byte, case GRPC_HTTP_HEADERS: if (parser->cur_line_length >= GRPC_HTTP_PARSER_MAX_HEADER_LENGTH) { if (grpc_http1_trace) - gpr_log(GPR_ERROR, "HTTP client max line length (%d) exceeded", + gpr_log(GPR_ERROR, "HTTP header max line length (%d) exceeded", GRPC_HTTP_PARSER_MAX_HEADER_LENGTH); - return GRPC_ERROR_NONE; + return GRPC_ERROR_CREATE("HTTP header max line length exceeded"); } parser->cur_line[parser->cur_line_length] = byte; parser->cur_line_length++; diff --git a/src/core/lib/iomgr/error.c b/src/core/lib/iomgr/error.c index dbe5b139f9..6c4d0157c2 100644 --- a/src/core/lib/iomgr/error.c +++ b/src/core/lib/iomgr/error.c @@ -262,6 +262,8 @@ static grpc_error *copy_error_and_unref(grpc_error *in) { GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_CANCELLED); else out = GRPC_ERROR_CREATE("unknown"); + } else if (gpr_ref_is_unique(&in->refs)) { + return in; } else { out = gpr_malloc(sizeof(*out)); #ifdef GRPC_ERROR_REFCOUNT_DEBUG diff --git a/src/core/lib/iomgr/pollset_uv.c b/src/core/lib/iomgr/pollset_uv.c index a379ddf25f..af33949c69 100644 --- a/src/core/lib/iomgr/pollset_uv.c +++ b/src/core/lib/iomgr/pollset_uv.c @@ -57,14 +57,28 @@ int grpc_pollset_work_run_loop; gpr_mu grpc_polling_mu; +/* This is used solely to kick the uv loop, by setting a callback to be run + immediately in the next loop iteration. + Note: In the future, if there is a bug that involves missing wakeups in the + future, try adding a uv_async_t to kick the loop differently */ +uv_timer_t dummy_uv_handle; + size_t grpc_pollset_size() { return sizeof(grpc_pollset); } +void dummy_timer_cb(uv_timer_t *handle) {} + void grpc_pollset_global_init(void) { gpr_mu_init(&grpc_polling_mu); + uv_timer_init(uv_default_loop(), &dummy_uv_handle); grpc_pollset_work_run_loop = 1; } -void grpc_pollset_global_shutdown(void) { gpr_mu_destroy(&grpc_polling_mu); } +static void timer_close_cb(uv_handle_t *handle) { handle->data = (void *)1; } + +void grpc_pollset_global_shutdown(void) { + gpr_mu_destroy(&grpc_polling_mu); + uv_close((uv_handle_t *)&dummy_uv_handle, timer_close_cb); +} void grpc_pollset_init(grpc_pollset *pollset, gpr_mu **mu) { *mu = &grpc_polling_mu; @@ -72,8 +86,6 @@ void grpc_pollset_init(grpc_pollset *pollset, gpr_mu **mu) { pollset->shutting_down = 0; } -static void timer_close_cb(uv_handle_t *handle) { handle->data = (void *)1; } - void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_closure *closure) { GPR_ASSERT(!pollset->shutting_down); @@ -81,6 +93,9 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, if (grpc_pollset_work_run_loop) { // Drain any pending UV callbacks without blocking uv_run(uv_default_loop(), UV_RUN_NOWAIT); + } else { + // kick the loop once + uv_timer_start(&dummy_uv_handle, dummy_timer_cb, 0, 0); } grpc_closure_sched(exec_ctx, closure, GRPC_ERROR_NONE); } @@ -130,6 +145,7 @@ grpc_error *grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, grpc_error *grpc_pollset_kick(grpc_pollset *pollset, grpc_pollset_worker *specific_worker) { + uv_timer_start(&dummy_uv_handle, dummy_timer_cb, 0, 0); return GRPC_ERROR_NONE; } diff --git a/src/core/lib/iomgr/sockaddr_utils.c b/src/core/lib/iomgr/sockaddr_utils.c index ffa62cb53c..9d2732666b 100644 --- a/src/core/lib/iomgr/sockaddr_utils.c +++ b/src/core/lib/iomgr/sockaddr_utils.c @@ -162,6 +162,7 @@ int grpc_sockaddr_to_string(char **out, char ntop_buf[INET6_ADDRSTRLEN]; const void *ip = NULL; int port; + uint32_t sin6_scope_id = 0; int ret; *out = NULL; @@ -177,10 +178,19 @@ int grpc_sockaddr_to_string(char **out, const struct sockaddr_in6 *addr6 = (const struct sockaddr_in6 *)addr; ip = &addr6->sin6_addr; port = ntohs(addr6->sin6_port); + sin6_scope_id = addr6->sin6_scope_id; } if (ip != NULL && grpc_inet_ntop(addr->sa_family, ip, ntop_buf, sizeof(ntop_buf)) != NULL) { - ret = gpr_join_host_port(out, ntop_buf, port); + if (sin6_scope_id != 0) { + char *host_with_scope; + /* Enclose sin6_scope_id with the format defined in RFC 6784 section 2. */ + gpr_asprintf(&host_with_scope, "%s%%25%" PRIu32, ntop_buf, sin6_scope_id); + ret = gpr_join_host_port(out, host_with_scope, port); + gpr_free(host_with_scope); + } else { + ret = gpr_join_host_port(out, ntop_buf, port); + } } else { ret = gpr_asprintf(out, "(sockaddr family=%d)", addr->sa_family); } diff --git a/src/core/lib/iomgr/tcp_server_posix.c b/src/core/lib/iomgr/tcp_server_posix.c index 36f878fdd4..5f286a6723 100644 --- a/src/core/lib/iomgr/tcp_server_posix.c +++ b/src/core/lib/iomgr/tcp_server_posix.c @@ -114,6 +114,8 @@ struct grpc_tcp_server { /* is this server shutting down? */ bool shutdown; + /* have listeners been shutdown? */ + bool shutdown_listeners; /* use SO_REUSEPORT */ bool so_reuseport; /* expand wildcard addresses to a list of all local addresses */ @@ -161,7 +163,7 @@ grpc_error *grpc_tcp_server_create(grpc_exec_ctx *exec_ctx, grpc_tcp_server **server) { gpr_once_init(&check_init, init); - grpc_tcp_server *s = gpr_malloc(sizeof(grpc_tcp_server)); + grpc_tcp_server *s = gpr_zalloc(sizeof(grpc_tcp_server)); s->so_reuseport = has_so_reuseport; s->resource_quota = grpc_resource_quota_create(NULL); s->expand_wildcard_addrs = false; @@ -422,7 +424,14 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *err) { grpc_fd_notify_on_read(exec_ctx, sp->emfd, &sp->read_closure); return; default: - gpr_log(GPR_ERROR, "Failed accept4: %s", strerror(errno)); + gpr_mu_lock(&sp->server->mu); + if (!sp->server->shutdown_listeners) { + gpr_log(GPR_ERROR, "Failed accept4: %s", strerror(errno)); + } else { + /* if we have shutdown listeners, accept4 could fail, and we + needn't notify users */ + } + gpr_mu_unlock(&sp->server->mu); goto error; } } @@ -438,11 +447,6 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *err) { grpc_fd *fdobj = grpc_fd_create(fd, name); - if (read_notifier_pollset == NULL) { - gpr_log(GPR_ERROR, "Read notifier pollset is not set on the fd"); - goto error; - } - grpc_pollset_add_fd(exec_ctx, read_notifier_pollset, fdobj); // Create acceptor. @@ -941,6 +945,7 @@ void grpc_tcp_server_unref(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { void grpc_tcp_server_shutdown_listeners(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { gpr_mu_lock(&s->mu); + s->shutdown_listeners = true; /* shutdown all fd's */ if (s->active_ports) { grpc_tcp_listener *sp; diff --git a/src/core/lib/iomgr/udp_server.c b/src/core/lib/iomgr/udp_server.c index 2a1c8d39fa..d1bcd89af1 100644 --- a/src/core/lib/iomgr/udp_server.c +++ b/src/core/lib/iomgr/udp_server.c @@ -485,7 +485,11 @@ void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *s, grpc_schedule_on_exec_ctx); grpc_fd_notify_on_write(exec_ctx, sp->emfd, &sp->write_closure); - s->active_ports++; + /* Registered for both read and write callbacks: increment active_ports + * twice to account for this, and delay free-ing of memory until both + * on_read and on_write have fired. */ + s->active_ports += 2; + sp = sp->next; } diff --git a/src/core/lib/support/sync.c b/src/core/lib/support/sync.c index 44b83f8175..b52f004f74 100644 --- a/src/core/lib/support/sync.c +++ b/src/core/lib/support/sync.c @@ -37,6 +37,8 @@ #include <grpc/support/log.h> #include <grpc/support/sync.h> +#include <assert.h> + /* Number of mutexes to allocate for events, to avoid lock contention. Should be a prime. */ enum { event_sync_partitions = 31 }; @@ -99,8 +101,12 @@ void gpr_ref_init(gpr_refcount *r, int n) { gpr_atm_rel_store(&r->count, n); } void gpr_ref(gpr_refcount *r) { gpr_atm_no_barrier_fetch_add(&r->count, 1); } void gpr_ref_non_zero(gpr_refcount *r) { +#ifndef NDEBUG gpr_atm prior = gpr_atm_no_barrier_fetch_add(&r->count, 1); - GPR_ASSERT(prior > 0); + assert(prior > 0); +#else + gpr_ref(r); +#endif } void gpr_refn(gpr_refcount *r, int n) { @@ -113,6 +119,10 @@ int gpr_unref(gpr_refcount *r) { return prior == 1; } +int gpr_ref_is_unique(gpr_refcount *r) { + return gpr_atm_acq_load(&r->count) == 1; +} + void gpr_stats_init(gpr_stats_counter *c, intptr_t n) { gpr_atm_rel_store(&c->value, n); } diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c index 2f212686bc..cf4410f2c0 100644 --- a/src/core/lib/surface/call.c +++ b/src/core/lib/surface/call.c @@ -156,6 +156,7 @@ struct grpc_call { bool receiving_message; bool requested_final_op; bool received_final_op; + bool sent_any_op; /* have we received initial metadata */ bool has_initial_md_been_received; @@ -485,7 +486,7 @@ void grpc_call_destroy(grpc_call *c) { gpr_mu_lock(&c->mu); GPR_ASSERT(!c->destroy_called); c->destroy_called = 1; - cancel = !c->received_final_op; + cancel = c->sent_any_op && !c->received_final_op; gpr_mu_unlock(&c->mu); if (cancel) { cancel_with_error(&exec_ctx, c, STATUS_FROM_API_OVERRIDE, @@ -1685,6 +1686,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx, grpc_closure_init(&bctl->finish_batch, finish_batch, bctl, grpc_schedule_on_exec_ctx); stream_op->on_complete = &bctl->finish_batch; + call->sent_any_op = true; gpr_mu_unlock(&call->mu); execute_op(exec_ctx, call, stream_op); diff --git a/src/core/lib/transport/transport.c b/src/core/lib/transport/transport.c index 23410ef852..3b0d18c789 100644 --- a/src/core/lib/transport/transport.c +++ b/src/core/lib/transport/transport.c @@ -84,6 +84,39 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx, } } +#define STREAM_REF_FROM_SLICE_REF(p) \ + ((grpc_stream_refcount *)(((uint8_t *)p) - \ + offsetof(grpc_stream_refcount, slice_refcount))) + +static void slice_stream_ref(void *p) { +#ifdef GRPC_STREAM_REFCOUNT_DEBUG + grpc_stream_ref(STREAM_REF_FROM_SLICE_REF(p), "slice"); +#else + grpc_stream_ref(STREAM_REF_FROM_SLICE_REF(p)); +#endif +} + +static void slice_stream_unref(grpc_exec_ctx *exec_ctx, void *p) { +#ifdef GRPC_STREAM_REFCOUNT_DEBUG + grpc_stream_unref(exec_ctx, STREAM_REF_FROM_SLICE_REF(p), "slice"); +#else + grpc_stream_unref(exec_ctx, STREAM_REF_FROM_SLICE_REF(p)); +#endif +} + +grpc_slice grpc_slice_from_stream_owned_buffer(grpc_stream_refcount *refcount, + void *buffer, size_t length) { + slice_stream_ref(&refcount->slice_refcount); + return (grpc_slice){.refcount = &refcount->slice_refcount, + .data.refcounted = {.bytes = buffer, .length = length}}; +} + +static const grpc_slice_refcount_vtable stream_ref_slice_vtable = { + .ref = slice_stream_ref, + .unref = slice_stream_unref, + .eq = grpc_slice_default_eq_impl, + .hash = grpc_slice_default_hash_impl}; + #ifdef GRPC_STREAM_REFCOUNT_DEBUG void grpc_stream_ref_init(grpc_stream_refcount *refcount, int initial_refs, grpc_iomgr_cb_func cb, void *cb_arg, @@ -95,6 +128,8 @@ void grpc_stream_ref_init(grpc_stream_refcount *refcount, int initial_refs, #endif gpr_ref_init(&refcount->refs, initial_refs); grpc_closure_init(&refcount->destroy, cb, cb_arg, grpc_schedule_on_exec_ctx); + refcount->slice_refcount.vtable = &stream_ref_slice_vtable; + refcount->slice_refcount.sub_refcount = &refcount->slice_refcount; } static void move64(uint64_t *from, uint64_t *to) { diff --git a/src/core/lib/transport/transport.h b/src/core/lib/transport/transport.h index dc1f28106b..af3a90aa1e 100644 --- a/src/core/lib/transport/transport.h +++ b/src/core/lib/transport/transport.h @@ -64,6 +64,7 @@ typedef struct grpc_stream_refcount { #ifdef GRPC_STREAM_REFCOUNT_DEBUG const char *object_type; #endif + grpc_slice_refcount slice_refcount; } grpc_stream_refcount; #ifdef GRPC_STREAM_REFCOUNT_DEBUG @@ -84,6 +85,11 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx, grpc_stream_refcount *refcount); grpc_stream_ref_init(rc, ir, cb, cb_arg) #endif +/* Wrap a buffer that is owned by some stream object into a slice that shares + the same refcount */ +grpc_slice grpc_slice_from_stream_owned_buffer(grpc_stream_refcount *refcount, + void *buffer, size_t length); + typedef struct { uint64_t framing_bytes; uint64_t data_bytes; diff --git a/src/core/lib/tsi/test_creds/BUILD b/src/core/lib/tsi/test_creds/BUILD new file mode 100644 index 0000000000..5cf04caf17 --- /dev/null +++ b/src/core/lib/tsi/test_creds/BUILD @@ -0,0 +1,36 @@ +# Copyright 2017, 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. + +licenses(["notice"]) # 3-clause BSD + +exports_files([ + "ca.pem", + "server1.key", + "server1.pem", +]) |