aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core
diff options
context:
space:
mode:
authorGravatar Yuchen Zeng <zyc@google.com>2016-08-11 01:09:14 -0700
committerGravatar Yuchen Zeng <zyc@google.com>2016-08-11 01:09:14 -0700
commit2fbf6fd3d8ea496df50996d5a7d9099f146c25f1 (patch)
tree7f4757d76883bda310a8376498ae55350bcace9b /src/core
parent839fff5df009746108aec6fe867f662ef8e09a6e (diff)
parent2c779d64db4b4e460c697b59170f85135d552db5 (diff)
Merge remote-tracking branch 'upstream/master' into cares_csharp
Diffstat (limited to 'src/core')
-rw-r--r--src/core/ext/lb_policy/grpclb/grpclb.c3
-rw-r--r--src/core/ext/transport/chttp2/transport/chttp2_transport.c16
-rw-r--r--src/core/lib/iomgr/tcp_server_posix.c25
-rw-r--r--src/core/lib/surface/channel.h8
4 files changed, 40 insertions, 12 deletions
diff --git a/src/core/ext/lb_policy/grpclb/grpclb.c b/src/core/ext/lb_policy/grpclb/grpclb.c
index dec25efe61..af913d8a9d 100644
--- a/src/core/ext/lb_policy/grpclb/grpclb.c
+++ b/src/core/ext/lb_policy/grpclb/grpclb.c
@@ -767,6 +767,9 @@ static lb_client_data *lb_client_data_create(glb_lb_policy *glb_policy) {
lb_client->deadline = gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
gpr_time_from_seconds(3, GPR_TIMESPAN));
+ /* Note the following LB call progresses every time there's activity in \a
+ * glb_policy->base.interested_parties, which is comprised of the polling
+ * entities passed to glb_pick(). */
lb_client->lb_call = grpc_channel_create_pollset_set_call(
glb_policy->lb_channel, NULL, GRPC_PROPAGATE_DEFAULTS,
glb_policy->base.interested_parties, "/BalanceLoad",
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index 6bd65cea02..28768f57f0 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -94,7 +94,8 @@ static void initiate_writing(grpc_exec_ctx *exec_ctx, void *t,
static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t);
static void end_waiting_for_write(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t, grpc_error *error);
+ grpc_chttp2_transport *t, grpc_error *error,
+ const char *reason);
/** Set a transport level setting, and push it to our peer */
static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
@@ -876,7 +877,7 @@ static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) {
set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE,
"start_writing:nothing_to_write");
}
- end_waiting_for_write(exec_ctx, t, GRPC_ERROR_CREATE("Nothing to write"));
+ end_waiting_for_write(exec_ctx, t, GRPC_ERROR_NONE, "Nothing to write");
if (t->ep && !t->endpoint_reading) {
destroy_endpoint(exec_ctx, t);
}
@@ -925,11 +926,18 @@ static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
}
}
+/* error may be GRPC_ERROR_NONE if there is no error allocated yet.
+ In that case, use "reason" as the text for a new error. */
static void end_waiting_for_write(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t, grpc_error *error) {
+ grpc_chttp2_transport *t, grpc_error *error,
+ const char *reason) {
grpc_chttp2_stream_global *stream_global;
while (grpc_chttp2_list_pop_closed_waiting_for_writing(&t->global,
&stream_global)) {
+ if (error == GRPC_ERROR_NONE && reason != NULL) {
+ /* create error object. */
+ error = GRPC_ERROR_CREATE(reason);
+ }
fail_pending_writes(exec_ctx, &t->global, stream_global,
GRPC_ERROR_REF(error));
GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "finish_writes");
@@ -951,7 +959,7 @@ static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx,
grpc_chttp2_cleanup_writing(exec_ctx, &t->global, &t->writing);
- end_waiting_for_write(exec_ctx, t, error);
+ end_waiting_for_write(exec_ctx, t, error, NULL);
switch (t->executor.write_state) {
case GRPC_CHTTP2_WRITING_INACTIVE:
diff --git a/src/core/lib/iomgr/tcp_server_posix.c b/src/core/lib/iomgr/tcp_server_posix.c
index 38ebd2dbcb..ea8d241154 100644
--- a/src/core/lib/iomgr/tcp_server_posix.c
+++ b/src/core/lib/iomgr/tcp_server_posix.c
@@ -90,10 +90,12 @@ struct grpc_tcp_listener {
grpc_closure read_closure;
grpc_closure destroyed_closure;
struct grpc_tcp_listener *next;
- /* When we add a listener, more than one can be created, mainly because of
- IPv6. A sibling will still be in the normal list, but will be flagged
- as such. Any action, such as ref or unref, will affect all of the
- siblings in the list. */
+ /* sibling is a linked list of all listeners for a given port. add_port and
+ clone_port place all new listeners in the same sibling list. A member of
+ the 'sibling' list is also a member of the 'next' list. The head of each
+ sibling list has is_sibling==0, and subsequent members of sibling lists
+ have is_sibling==1. is_sibling allows separate sibling lists to be
+ identified while iterating through 'next'. */
struct grpc_tcp_listener *sibling;
int is_sibling;
};
@@ -306,7 +308,7 @@ static grpc_error *prepare_socket(int fd, const struct sockaddr *addr,
GPR_ASSERT(fd >= 0);
- if (so_reuseport) {
+ if (so_reuseport && !grpc_is_unix_socket(addr)) {
err = grpc_set_socket_reuse_port(fd, 1);
if (err != GRPC_ERROR_NONE) goto error;
}
@@ -480,6 +482,9 @@ static grpc_error *add_socket_to_server(grpc_tcp_server *s, int fd,
return err;
}
+/* Insert count new listeners after listener. Every new listener will have the
+ same listen address as listener (SO_REUSEPORT must be enabled). Every new
+ listener is a sibling of listener. */
static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) {
grpc_tcp_listener *sp = NULL;
char *addr_str;
@@ -506,6 +511,11 @@ static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) {
sp = gpr_malloc(sizeof(grpc_tcp_listener));
sp->next = listener->next;
listener->next = sp;
+ /* sp (the new listener) is a sibling of 'listener' (the original
+ listener). */
+ sp->is_sibling = 1;
+ sp->sibling = listener->sibling;
+ listener->sibling = sp;
sp->server = listener->server;
sp->fd = fd;
sp->emfd = grpc_fd_create(fd, name);
@@ -514,8 +524,6 @@ static grpc_error *clone_port(grpc_tcp_listener *listener, unsigned count) {
sp->port = port;
sp->port_index = listener->port_index;
sp->fd_index = listener->fd_index + count - i;
- sp->is_sibling = 1;
- sp->sibling = listener->is_sibling ? listener->sibling : listener;
GPR_ASSERT(sp->emfd);
while (listener->server->tail->next != NULL) {
listener->server->tail = listener->server->tail->next;
@@ -685,7 +693,8 @@ void grpc_tcp_server_start(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s,
s->pollset_count = pollset_count;
sp = s->head;
while (sp != NULL) {
- if (s->so_reuseport && pollset_count > 1) {
+ if (s->so_reuseport && !grpc_is_unix_socket(&sp->addr.sockaddr) &&
+ pollset_count > 1) {
GPR_ASSERT(GRPC_LOG_IF_ERROR(
"clone_port", clone_port(sp, (unsigned)(pollset_count - 1))));
for (i = 0; i < pollset_count; i++) {
diff --git a/src/core/lib/surface/channel.h b/src/core/lib/surface/channel.h
index 7eff7b8883..4c62974346 100644
--- a/src/core/lib/surface/channel.h
+++ b/src/core/lib/surface/channel.h
@@ -42,6 +42,14 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
grpc_channel_stack_type channel_stack_type,
grpc_transport *optional_transport);
+/** Create a call given a grpc_channel, in order to call \a method.
+ Progress is tied to activity on \a pollset_set. The returned call object is
+ meant to be used with \a grpc_call_start_batch_and_execute, which relies on
+ callbacks to signal completions. \a method and \a host need
+ only live through the invocation of this function. If \a parent_call is
+ non-NULL, it must be a server-side call. It will be used to propagate
+ properties from the server call to this new client call, depending on the
+ value of \a propagation_mask (see propagation_bits.h for possible values) */
grpc_call *grpc_channel_create_pollset_set_call(
grpc_channel *channel, grpc_call *parent_call, uint32_t propagation_mask,
grpc_pollset_set *pollset_set, const char *method, const char *host,