aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/lib')
-rw-r--r--src/core/lib/http/parser.c4
-rw-r--r--src/core/lib/iomgr/sockaddr_utils.c12
-rw-r--r--src/core/lib/iomgr/tcp_server_posix.c19
-rw-r--r--src/core/lib/iomgr/udp_server.c6
-rw-r--r--src/core/lib/support/sync.c8
-rw-r--r--src/core/lib/surface/call.c4
-rw-r--r--src/core/lib/transport/transport.c35
-rw-r--r--src/core/lib/transport/transport.h6
-rw-r--r--src/core/lib/tsi/test_creds/BUILD2
9 files changed, 83 insertions, 13 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/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 9b5a4ac708..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) {
diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c
index cc57654ea4..c2547c5147 100644
--- a/src/core/lib/surface/call.c
+++ b/src/core/lib/surface/call.c
@@ -161,6 +161,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;
@@ -488,7 +489,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,
@@ -1678,6 +1679,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 004e748f25..165950e288 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 bb23c0225a..cc1c277b35 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
index dcd6d930a8..5cf04caf17 100644
--- a/src/core/lib/tsi/test_creds/BUILD
+++ b/src/core/lib/tsi/test_creds/BUILD
@@ -27,6 +27,8 @@
# (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",
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-26 01:06:11 +0000