aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core
diff options
context:
space:
mode:
Diffstat (limited to 'src/core')
-rw-r--r--src/core/ext/client_channel/connector.h2
-rw-r--r--src/core/ext/client_channel/default_initial_connect_string.c38
-rw-r--r--src/core/ext/client_channel/initial_connect_string.c52
-rw-r--r--src/core/ext/client_channel/initial_connect_string.h50
-rw-r--r--src/core/ext/client_channel/subchannel.c7
-rw-r--r--src/core/ext/lb_policy/pick_first/pick_first.c6
-rw-r--r--src/core/ext/lb_policy/round_robin/round_robin.c6
-rw-r--r--src/core/ext/transport/chttp2/client/chttp2_connector.c37
-rw-r--r--src/core/ext/transport/cronet/transport/cronet_transport.c65
-rw-r--r--src/core/lib/iomgr/closure.c4
-rw-r--r--src/core/lib/iomgr/combiner.c2
-rw-r--r--src/core/lib/iomgr/error.h25
-rw-r--r--src/core/lib/iomgr/ev_poll_posix.c2
-rw-r--r--src/core/lib/iomgr/executor.c4
-rw-r--r--src/core/lib/profiling/basic_timers.c2
-rw-r--r--src/core/lib/security/transport/server_auth_filter.c3
16 files changed, 68 insertions, 237 deletions
diff --git a/src/core/ext/client_channel/connector.h b/src/core/ext/client_channel/connector.h
index 9bff41f003..94b5fb5c9e 100644
--- a/src/core/ext/client_channel/connector.h
+++ b/src/core/ext/client_channel/connector.h
@@ -48,8 +48,6 @@ struct grpc_connector {
typedef struct {
/** set of pollsets interested in this connection */
grpc_pollset_set *interested_parties;
- /** initial connect string to send */
- grpc_slice initial_connect_string;
/** deadline for connection */
gpr_timespec deadline;
/** channel arguments (to be passed to transport) */
diff --git a/src/core/ext/client_channel/default_initial_connect_string.c b/src/core/ext/client_channel/default_initial_connect_string.c
deleted file mode 100644
index 6db82d84ef..0000000000
--- a/src/core/ext/client_channel/default_initial_connect_string.c
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- *
- * 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/slice.h>
-#include "src/core/lib/iomgr/resolve_address.h"
-
-void grpc_set_default_initial_connect_string(grpc_resolved_address **addr,
- grpc_slice *initial_str) {}
diff --git a/src/core/ext/client_channel/initial_connect_string.c b/src/core/ext/client_channel/initial_connect_string.c
deleted file mode 100644
index 8ebd06c458..0000000000
--- a/src/core/ext/client_channel/initial_connect_string.c
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- *
- * 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 "src/core/ext/client_channel/initial_connect_string.h"
-
-#include <stddef.h>
-
-extern void grpc_set_default_initial_connect_string(
- grpc_resolved_address **addr, grpc_slice *initial_str);
-
-static grpc_set_initial_connect_string_func g_set_initial_connect_string_func =
- grpc_set_default_initial_connect_string;
-
-void grpc_test_set_initial_connect_string_function(
- grpc_set_initial_connect_string_func func) {
- g_set_initial_connect_string_func = func;
-}
-
-void grpc_set_initial_connect_string(grpc_resolved_address **addr,
- grpc_slice *initial_str) {
- g_set_initial_connect_string_func(addr, initial_str);
-}
diff --git a/src/core/ext/client_channel/initial_connect_string.h b/src/core/ext/client_channel/initial_connect_string.h
deleted file mode 100644
index 876abea40e..0000000000
--- a/src/core/ext/client_channel/initial_connect_string.h
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- *
- * 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.
- *
- */
-
-#ifndef GRPC_CORE_EXT_CLIENT_CHANNEL_INITIAL_CONNECT_STRING_H
-#define GRPC_CORE_EXT_CLIENT_CHANNEL_INITIAL_CONNECT_STRING_H
-
-#include <grpc/slice.h>
-#include "src/core/lib/iomgr/resolve_address.h"
-
-typedef void (*grpc_set_initial_connect_string_func)(
- grpc_resolved_address **addr, grpc_slice *initial_str);
-
-void grpc_test_set_initial_connect_string_function(
- grpc_set_initial_connect_string_func func);
-
-/** Set a string to be sent once connected. Optionally reset addr. */
-void grpc_set_initial_connect_string(grpc_resolved_address **addr,
- grpc_slice *connect_string);
-
-#endif /* GRPC_CORE_EXT_CLIENT_CHANNEL_INITIAL_CONNECT_STRING_H */
diff --git a/src/core/ext/client_channel/subchannel.c b/src/core/ext/client_channel/subchannel.c
index ed5029ea9a..e886fbc0cc 100644
--- a/src/core/ext/client_channel/subchannel.c
+++ b/src/core/ext/client_channel/subchannel.c
@@ -41,7 +41,6 @@
#include <grpc/support/string_util.h>
#include "src/core/ext/client_channel/client_channel.h"
-#include "src/core/ext/client_channel/initial_connect_string.h"
#include "src/core/ext/client_channel/parse_address.h"
#include "src/core/ext/client_channel/proxy_mapper_registry.h"
#include "src/core/ext/client_channel/subchannel_index.h"
@@ -103,9 +102,6 @@ struct grpc_subchannel {
grpc_subchannel_key *key;
- /** initial string to send to peer */
- grpc_slice initial_connect_string;
-
/** set during connection */
grpc_connect_out_args connecting_result;
@@ -215,7 +211,6 @@ static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg,
grpc_subchannel *c = arg;
gpr_free((void *)c->filters);
grpc_channel_args_destroy(exec_ctx, c->args);
- grpc_slice_unref_internal(exec_ctx, c->initial_connect_string);
grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker);
grpc_connector_unref(exec_ctx, c->connector);
grpc_pollset_set_destroy(exec_ctx, c->pollset_set);
@@ -333,7 +328,6 @@ grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx,
c->pollset_set = grpc_pollset_set_create();
grpc_resolved_address *addr = gpr_malloc(sizeof(*addr));
grpc_get_subchannel_address_arg(exec_ctx, args->args, addr);
- grpc_set_initial_connect_string(&addr, &c->initial_connect_string);
grpc_resolved_address *new_address = NULL;
grpc_channel_args *new_args = NULL;
if (grpc_proxy_mappers_map_address(exec_ctx, addr, args->args, &new_address,
@@ -404,7 +398,6 @@ static void continue_connect_locked(grpc_exec_ctx *exec_ctx,
args.interested_parties = c->pollset_set;
args.deadline = c->next_attempt;
args.channel_args = c->args;
- args.initial_connect_string = c->initial_connect_string;
grpc_connectivity_state_set(exec_ctx, &c->state_tracker,
GRPC_CHANNEL_CONNECTING, GRPC_ERROR_NONE,
diff --git a/src/core/ext/lb_policy/pick_first/pick_first.c b/src/core/ext/lb_policy/pick_first/pick_first.c
index e2a66d16bd..179dcd6eac 100644
--- a/src/core/ext/lb_policy/pick_first/pick_first.c
+++ b/src/core/ext/lb_policy/pick_first/pick_first.c
@@ -423,11 +423,13 @@ static grpc_lb_policy *create_pick_first(grpc_exec_ctx *exec_ctx,
"This LB policy doesn't support user data. It will be ignored");
}
+ static const char *keys_to_remove[] = {GRPC_ARG_SUBCHANNEL_ADDRESS};
memset(&sc_args, 0, sizeof(grpc_subchannel_args));
grpc_arg addr_arg =
grpc_create_subchannel_address_arg(&addresses->addresses[i].address);
- grpc_channel_args *new_args =
- grpc_channel_args_copy_and_add(args->args, &addr_arg, 1);
+ grpc_channel_args *new_args = grpc_channel_args_copy_and_add_and_remove(
+ args->args, keys_to_remove, GPR_ARRAY_SIZE(keys_to_remove), &addr_arg,
+ 1);
gpr_free(addr_arg.value.string);
sc_args.args = new_args;
grpc_subchannel *subchannel = grpc_client_channel_factory_create_subchannel(
diff --git a/src/core/ext/lb_policy/round_robin/round_robin.c b/src/core/ext/lb_policy/round_robin/round_robin.c
index f2d1d46179..09562d30ed 100644
--- a/src/core/ext/lb_policy/round_robin/round_robin.c
+++ b/src/core/ext/lb_policy/round_robin/round_robin.c
@@ -709,11 +709,13 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx,
/* Skip balancer addresses, since we only know how to handle backends. */
if (addresses->addresses[i].is_balancer) continue;
+ static const char *keys_to_remove[] = {GRPC_ARG_SUBCHANNEL_ADDRESS};
memset(&sc_args, 0, sizeof(grpc_subchannel_args));
grpc_arg addr_arg =
grpc_create_subchannel_address_arg(&addresses->addresses[i].address);
- grpc_channel_args *new_args =
- grpc_channel_args_copy_and_add(args->args, &addr_arg, 1);
+ grpc_channel_args *new_args = grpc_channel_args_copy_and_add_and_remove(
+ args->args, keys_to_remove, GPR_ARRAY_SIZE(keys_to_remove), &addr_arg,
+ 1);
gpr_free(addr_arg.value.string);
sc_args.args = new_args;
grpc_subchannel *subchannel = grpc_client_channel_factory_create_subchannel(
diff --git a/src/core/ext/transport/chttp2/client/chttp2_connector.c b/src/core/ext/transport/chttp2/client/chttp2_connector.c
index eae0145ecc..d49c32b671 100644
--- a/src/core/ext/transport/chttp2/client/chttp2_connector.c
+++ b/src/core/ext/transport/chttp2/client/chttp2_connector.c
@@ -63,8 +63,6 @@ typedef struct {
grpc_closure *notify;
grpc_connect_in_args args;
grpc_connect_out_args *result;
- grpc_closure initial_string_sent;
- grpc_slice_buffer initial_string_buffer;
grpc_endpoint *endpoint; // Non-NULL until handshaking starts.
@@ -82,7 +80,6 @@ static void chttp2_connector_unref(grpc_exec_ctx *exec_ctx,
grpc_connector *con) {
chttp2_connector *c = (chttp2_connector *)con;
if (gpr_unref(&c->refs)) {
- /* c->initial_string_buffer does not need to be destroyed */
gpr_mu_destroy(&c->mu);
// If handshaking is not yet in progress, destroy the endpoint.
// Otherwise, the handshaker will do this for us.
@@ -160,28 +157,6 @@ static void start_handshake_locked(grpc_exec_ctx *exec_ctx,
c->endpoint = NULL; // Endpoint handed off to handshake manager.
}
-static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg,
- grpc_error *error) {
- chttp2_connector *c = arg;
- gpr_mu_lock(&c->mu);
- if (error != GRPC_ERROR_NONE || c->shutdown) {
- if (error == GRPC_ERROR_NONE) {
- error = GRPC_ERROR_CREATE("connector shutdown");
- } else {
- error = GRPC_ERROR_REF(error);
- }
- memset(c->result, 0, sizeof(*c->result));
- grpc_closure *notify = c->notify;
- c->notify = NULL;
- grpc_closure_sched(exec_ctx, notify, error);
- gpr_mu_unlock(&c->mu);
- chttp2_connector_unref(exec_ctx, arg);
- } else {
- start_handshake_locked(exec_ctx, c);
- gpr_mu_unlock(&c->mu);
- }
-}
-
static void connected(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
chttp2_connector *c = arg;
gpr_mu_lock(&c->mu);
@@ -204,17 +179,7 @@ static void connected(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
chttp2_connector_unref(exec_ctx, arg);
} else {
GPR_ASSERT(c->endpoint != NULL);
- if (!GRPC_SLICE_IS_EMPTY(c->args.initial_connect_string)) {
- grpc_closure_init(&c->initial_string_sent, on_initial_connect_string_sent,
- c, grpc_schedule_on_exec_ctx);
- grpc_slice_buffer_init(&c->initial_string_buffer);
- grpc_slice_buffer_add(&c->initial_string_buffer,
- c->args.initial_connect_string);
- grpc_endpoint_write(exec_ctx, c->endpoint, &c->initial_string_buffer,
- &c->initial_string_sent);
- } else {
- start_handshake_locked(exec_ctx, c);
- }
+ start_handshake_locked(exec_ctx, c);
gpr_mu_unlock(&c->mu);
}
}
diff --git a/src/core/ext/transport/cronet/transport/cronet_transport.c b/src/core/ext/transport/cronet/transport/cronet_transport.c
index 450d9ab23a..36bb67869c 100644
--- a/src/core/ext/transport/cronet/transport/cronet_transport.c
+++ b/src/core/ext/transport/cronet/transport/cronet_transport.c
@@ -128,6 +128,7 @@ struct read_state {
int received_bytes;
int remaining_bytes;
int length_field;
+ bool compressed;
char grpc_header_bytes[GRPC_HEADER_SIZE_IN_BYTES];
char *payload_field;
bool read_stream_closed;
@@ -484,6 +485,16 @@ static void on_response_headers_received(
CRONET_LOG(GPR_DEBUG, "R: on_response_headers_received(%p, %p, %s)", stream,
headers, negotiated_protocol);
stream_obj *s = (stream_obj *)stream->annotation;
+
+ /* Identify if this is a header or a trailer (in a trailer-only response case)
+ */
+ for (size_t i = 0; i < headers->count; i++) {
+ if (0 == strcmp("grpc-status", headers->headers[i].key)) {
+ on_response_trailers_received(stream, headers);
+ return;
+ }
+ }
+
gpr_mu_lock(&s->mu);
memset(&s->state.rs.initial_metadata, 0,
sizeof(s->state.rs.initial_metadata));
@@ -507,6 +518,7 @@ static void on_response_headers_received(
is closed */
GPR_ASSERT(s->state.rs.length_field_received == false);
s->state.rs.read_buffer = s->state.rs.grpc_header_bytes;
+ s->state.rs.compressed = false;
s->state.rs.received_bytes = 0;
s->state.rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES;
CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs);
@@ -641,7 +653,7 @@ static void on_response_trailers_received(
*/
static void create_grpc_frame(grpc_slice_buffer *write_slice_buffer,
char **pp_write_buffer,
- size_t *p_write_buffer_size) {
+ size_t *p_write_buffer_size, uint32_t flags) {
grpc_slice slice = grpc_slice_buffer_take_first(write_slice_buffer);
size_t length = GRPC_SLICE_LENGTH(slice);
*p_write_buffer_size = length + GRPC_HEADER_SIZE_IN_BYTES;
@@ -650,7 +662,9 @@ static void create_grpc_frame(grpc_slice_buffer *write_slice_buffer,
*pp_write_buffer = write_buffer;
uint8_t *p = (uint8_t *)write_buffer;
/* Append 5 byte header */
- *p++ = 0;
+ /* Compressed flag */
+ *p++ = (flags & GRPC_WRITE_INTERNAL_COMPRESS) ? 1 : 0;
+ /* Message length */
*p++ = (uint8_t)(length >> 24);
*p++ = (uint8_t)(length >> 16);
*p++ = (uint8_t)(length >> 8);
@@ -728,14 +742,16 @@ static void convert_metadata_to_cronet_headers(
*p_num_headers = (size_t)num_headers;
}
-static int parse_grpc_header(const uint8_t *data) {
+static void parse_grpc_header(const uint8_t *data, int *length,
+ bool *compressed) {
+ const uint8_t c = *data;
const uint8_t *p = data + 1;
- int length = 0;
- length |= ((uint8_t)*p++) << 24;
- length |= ((uint8_t)*p++) << 16;
- length |= ((uint8_t)*p++) << 8;
- length |= ((uint8_t)*p++);
- return length;
+ *compressed = ((c & 0x01) == 0x01);
+ *length = 0;
+ *length |= ((uint8_t)*p++) << 24;
+ *length |= ((uint8_t)*p++) << 16;
+ *length |= ((uint8_t)*p++) << 8;
+ *length |= ((uint8_t)*p++);
}
static bool header_has_authority(grpc_linked_mdelem *head) {
@@ -788,7 +804,8 @@ static bool op_can_be_run(grpc_transport_stream_op *curr_op,
else if (!stream_state->state_callback_received[OP_SEND_INITIAL_METADATA])
result = false;
/* we haven't received headers yet. */
- else if (!stream_state->state_callback_received[OP_RECV_INITIAL_METADATA])
+ else if (!stream_state->state_callback_received[OP_RECV_INITIAL_METADATA] &&
+ !stream_state->state_op_done[OP_RECV_TRAILING_METADATA])
result = false;
} else if (op_id == OP_SEND_MESSAGE) {
/* already executed (note we're checking op specific state, not stream
@@ -801,7 +818,8 @@ static bool op_can_be_run(grpc_transport_stream_op *curr_op,
/* already executed */
if (op_state->state_op_done[OP_RECV_MESSAGE]) result = false;
/* we haven't received headers yet. */
- else if (!stream_state->state_callback_received[OP_RECV_INITIAL_METADATA])
+ else if (!stream_state->state_callback_received[OP_RECV_INITIAL_METADATA] &&
+ !stream_state->state_op_done[OP_RECV_TRAILING_METADATA])
result = false;
} else if (op_id == OP_RECV_TRAILING_METADATA) {
/* already executed */
@@ -955,12 +973,6 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
grpc_slice_buffer_init(&write_slice_buffer);
grpc_byte_stream_next(NULL, stream_op->send_message, &slice,
stream_op->send_message->length, NULL);
- /* Check that compression flag is OFF. We don't support compression yet.
- */
- if (stream_op->send_message->flags != 0) {
- gpr_log(GPR_ERROR, "Compression is not supported");
- GPR_ASSERT(stream_op->send_message->flags == 0);
- }
grpc_slice_buffer_add(&write_slice_buffer, slice);
if (write_slice_buffer.count != 1) {
/* Empty request not handled yet */
@@ -970,7 +982,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
if (write_slice_buffer.count > 0) {
size_t write_buffer_size;
create_grpc_frame(&write_slice_buffer, &stream_state->ws.write_buffer,
- &write_buffer_size);
+ &write_buffer_size, stream_op->send_message->flags);
CRONET_LOG(GPR_DEBUG, "bidirectional_stream_write (%p, %p)", s->cbs,
stream_state->ws.write_buffer);
stream_state->state_callback_received[OP_SEND_MESSAGE] = false;
@@ -1022,6 +1034,9 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
} else if (stream_state->state_callback_received[OP_FAILED]) {
grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
GRPC_ERROR_NONE);
+ } else if (stream_state->state_op_done[OP_RECV_TRAILING_METADATA]) {
+ grpc_closure_sched(exec_ctx, stream_op->recv_initial_metadata_ready,
+ GRPC_ERROR_NONE);
} else {
grpc_chttp2_incoming_metadata_buffer_publish(
exec_ctx, &oas->s->state.rs.initial_metadata,
@@ -1066,8 +1081,9 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
stream_state->rs.remaining_bytes == 0) {
/* Start a read operation for data */
stream_state->rs.length_field_received = true;
- stream_state->rs.length_field = stream_state->rs.remaining_bytes =
- parse_grpc_header((const uint8_t *)stream_state->rs.read_buffer);
+ parse_grpc_header((const uint8_t *)stream_state->rs.read_buffer,
+ &stream_state->rs.length_field,
+ &stream_state->rs.compressed);
CRONET_LOG(GPR_DEBUG, "length field = %d",
stream_state->rs.length_field);
if (stream_state->rs.length_field > 0) {
@@ -1089,6 +1105,9 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
grpc_slice_buffer_init(&stream_state->rs.read_slice_buffer);
grpc_slice_buffer_stream_init(&stream_state->rs.sbs,
&stream_state->rs.read_slice_buffer, 0);
+ if (stream_state->rs.compressed) {
+ stream_state->rs.sbs.base.flags |= GRPC_WRITE_INTERNAL_COMPRESS;
+ }
*((grpc_byte_buffer **)stream_op->recv_message) =
(grpc_byte_buffer *)&stream_state->rs.sbs;
grpc_closure_sched(exec_ctx, stream_op->recv_message_ready,
@@ -1100,6 +1119,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
stream_state->rs.read_buffer = stream_state->rs.grpc_header_bytes;
stream_state->rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES;
stream_state->rs.received_bytes = 0;
+ stream_state->rs.compressed = false;
stream_state->rs.length_field_received = false;
CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs);
stream_state->state_op_done[OP_READ_REQ_MADE] =
@@ -1114,6 +1134,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
stream_state->rs.read_buffer = stream_state->rs.grpc_header_bytes;
stream_state->rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES;
stream_state->rs.received_bytes = 0;
+ stream_state->rs.compressed = false;
CRONET_LOG(GPR_DEBUG, "bidirectional_stream_read(%p)", s->cbs);
stream_state->state_op_done[OP_READ_REQ_MADE] =
true; /* Indicates that at least one read request has been made */
@@ -1137,6 +1158,9 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
read_data_slice);
grpc_slice_buffer_stream_init(&stream_state->rs.sbs,
&stream_state->rs.read_slice_buffer, 0);
+ if (stream_state->rs.compressed) {
+ stream_state->rs.sbs.base.flags = GRPC_WRITE_INTERNAL_COMPRESS;
+ }
*((grpc_byte_buffer **)stream_op->recv_message) =
(grpc_byte_buffer *)&stream_state->rs.sbs;
grpc_closure_sched(exec_ctx, stream_op->recv_message_ready,
@@ -1146,6 +1170,7 @@ static enum e_op_result execute_stream_op(grpc_exec_ctx *exec_ctx,
/* Do an extra read to trigger on_succeeded() callback in case connection
is closed */
stream_state->rs.read_buffer = stream_state->rs.grpc_header_bytes;
+ stream_state->rs.compressed = false;
stream_state->rs.received_bytes = 0;
stream_state->rs.remaining_bytes = GRPC_HEADER_SIZE_IN_BYTES;
stream_state->rs.length_field_received = false;
diff --git a/src/core/lib/iomgr/closure.c b/src/core/lib/iomgr/closure.c
index 509c1ff95d..6633fb68ec 100644
--- a/src/core/lib/iomgr/closure.c
+++ b/src/core/lib/iomgr/closure.c
@@ -33,6 +33,7 @@
#include "src/core/lib/iomgr/closure.h"
+#include <assert.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
@@ -124,6 +125,7 @@ void grpc_closure_run(grpc_exec_ctx *exec_ctx, grpc_closure *c,
grpc_error *error) {
GPR_TIMER_BEGIN("grpc_closure_run", 0);
if (c != NULL) {
+ assert(c->cb);
c->scheduler->vtable->run(exec_ctx, c, error);
} else {
GRPC_ERROR_UNREF(error);
@@ -135,6 +137,7 @@ void grpc_closure_sched(grpc_exec_ctx *exec_ctx, grpc_closure *c,
grpc_error *error) {
GPR_TIMER_BEGIN("grpc_closure_sched", 0);
if (c != NULL) {
+ assert(c->cb);
c->scheduler->vtable->sched(exec_ctx, c, error);
} else {
GRPC_ERROR_UNREF(error);
@@ -146,6 +149,7 @@ void grpc_closure_list_sched(grpc_exec_ctx *exec_ctx, grpc_closure_list *list) {
grpc_closure *c = list->head;
while (c != NULL) {
grpc_closure *next = c->next_data.next;
+ assert(c->cb);
c->scheduler->vtable->sched(exec_ctx, c, c->error_data.error);
c = next;
}
diff --git a/src/core/lib/iomgr/combiner.c b/src/core/lib/iomgr/combiner.c
index fa9966c3a6..2bc476bbef 100644
--- a/src/core/lib/iomgr/combiner.c
+++ b/src/core/lib/iomgr/combiner.c
@@ -33,6 +33,7 @@
#include "src/core/lib/iomgr/combiner.h"
+#include <assert.h>
#include <string.h>
#include <grpc/support/alloc.h>
@@ -216,6 +217,7 @@ static void combiner_exec(grpc_exec_ctx *exec_ctx, grpc_combiner *lock,
GPR_DEBUG, "C:%p grpc_combiner_execute c=%p cov=%d last=%" PRIdPTR, lock,
cl, covered_by_poller, last));
GPR_ASSERT(last & STATE_UNORPHANED); // ensure lock has not been destroyed
+ assert(cl->cb);
cl->error_data.scratch =
pack_error_data((error_data){error, covered_by_poller});
if (covered_by_poller) {
diff --git a/src/core/lib/iomgr/error.h b/src/core/lib/iomgr/error.h
index eb953947ae..0e7a7b0a1e 100644
--- a/src/core/lib/iomgr/error.h
+++ b/src/core/lib/iomgr/error.h
@@ -45,28 +45,9 @@ extern "C" {
#endif
/// Opaque representation of an error.
-/// Errors are refcounted objects that represent the result of an operation.
-/// Ownership laws:
-/// if a grpc_error is returned by a function, the caller owns a ref to that
-/// instance
-/// if a grpc_error is passed to a grpc_closure callback function (functions
-/// with the signature:
-/// void (*f)(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error))
-/// then those functions do not own a ref to error (but are free to manually
-/// take a reference).
-/// if a grpc_error is passed to *ANY OTHER FUNCTION* then that function takes
-/// ownership of the error
-/// Errors have:
-/// a set of ints, strings, and timestamps that describe the error
-/// always present are:
-/// GRPC_ERROR_STR_FILE, GRPC_ERROR_INT_FILE_LINE - source location the error
-/// was generated
-/// GRPC_ERROR_STR_DESCRIPTION - a human readable description of the error
-/// GRPC_ERROR_TIME_CREATED - a timestamp indicating when the error happened
-/// an error can also have children; these are other errors that are believed
-/// to have contributed to this one. By accumulating children, we can begin
-/// to root cause high level failures from low level failures, without having
-/// to derive execution paths from log lines
+/// See https://github.com/grpc/grpc/blob/master/doc/core/grpc-error.md for a
+/// full write up of this object.
+
typedef struct grpc_error grpc_error;
typedef enum {
diff --git a/src/core/lib/iomgr/ev_poll_posix.c b/src/core/lib/iomgr/ev_poll_posix.c
index 5ddd5313e2..789ef22c55 100644
--- a/src/core/lib/iomgr/ev_poll_posix.c
+++ b/src/core/lib/iomgr/ev_poll_posix.c
@@ -1421,7 +1421,7 @@ static int cvfd_poll(struct pollfd *fds, nfds_t nfds, int timeout) {
g_cvfds.pollcount++;
opt = gpr_thd_options_default();
gpr_thd_options_set_detached(&opt);
- gpr_thd_new(&t_id, &run_poll, pargs, &opt);
+ GPR_ASSERT(gpr_thd_new(&t_id, &run_poll, pargs, &opt));
// We want the poll() thread to trigger the deadline, so wait forever here
gpr_cv_wait(pollcv, &g_cvfds.mu, gpr_inf_future(GPR_CLOCK_MONOTONIC));
if (gpr_atm_no_barrier_load(&pargs->status) == COMPLETED) {
diff --git a/src/core/lib/iomgr/executor.c b/src/core/lib/iomgr/executor.c
index a5b62aa888..ae3e2eabc3 100644
--- a/src/core/lib/iomgr/executor.c
+++ b/src/core/lib/iomgr/executor.c
@@ -115,8 +115,8 @@ static void maybe_spawn_locked() {
/* All previous instances of the thread should have been joined at this point.
* Spawn time! */
g_executor.busy = 1;
- gpr_thd_new(&g_executor.tid, closure_exec_thread_func, NULL,
- &g_executor.options);
+ GPR_ASSERT(gpr_thd_new(&g_executor.tid, closure_exec_thread_func, NULL,
+ &g_executor.options));
g_executor.pending_join = 1;
}
diff --git a/src/core/lib/profiling/basic_timers.c b/src/core/lib/profiling/basic_timers.c
index 1f1987fb8e..bc8e27714c 100644
--- a/src/core/lib/profiling/basic_timers.c
+++ b/src/core/lib/profiling/basic_timers.c
@@ -218,7 +218,7 @@ void gpr_timers_set_log_filename(const char *filename) {
static void init_output() {
gpr_thd_options options = gpr_thd_options_default();
gpr_thd_options_set_joinable(&options);
- gpr_thd_new(&g_writing_thread, writing_thread, NULL, &options);
+ GPR_ASSERT(gpr_thd_new(&g_writing_thread, writing_thread, NULL, &options));
atexit(finish_writing);
}
diff --git a/src/core/lib/security/transport/server_auth_filter.c b/src/core/lib/security/transport/server_auth_filter.c
index 01cb473177..c47fe56ba4 100644
--- a/src/core/lib/security/transport/server_auth_filter.c
+++ b/src/core/lib/security/transport/server_auth_filter.c
@@ -158,7 +158,7 @@ static void auth_on_recv(grpc_exec_ctx *exec_ctx, void *user_data,
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
if (error == GRPC_ERROR_NONE) {
- if (chand->creds->processor.process != NULL) {
+ if (chand->creds != NULL && chand->creds->processor.process != NULL) {
calld->md = metadata_batch_to_md_array(calld->recv_initial_metadata);
chand->creds->processor.process(
chand->creds->processor.state, calld->auth_context,
@@ -242,7 +242,6 @@ static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
GPR_ASSERT(!args->is_last);
GPR_ASSERT(auth_context != NULL);
- GPR_ASSERT(creds != NULL);
/* initialize members */
chand->auth_context =
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-13 12:35:14 +0000