aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/channel
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/channel')
-rw-r--r--src/core/channel/channel_args.c25
-rw-r--r--src/core/channel/channel_args.h18
-rw-r--r--src/core/channel/child_channel.c14
-rw-r--r--src/core/channel/client_channel.c12
-rw-r--r--src/core/channel/client_setup.c45
-rw-r--r--src/core/channel/client_setup.h6
6 files changed, 75 insertions, 45 deletions
diff --git a/src/core/channel/channel_args.c b/src/core/channel/channel_args.c
index 509ae0df89..166d559a45 100644
--- a/src/core/channel/channel_args.c
+++ b/src/core/channel/channel_args.c
@@ -36,6 +36,7 @@
#include "src/core/support/string.h"
#include <grpc/support/alloc.h>
+#include <grpc/support/string_util.h>
#include <string.h>
@@ -114,3 +115,27 @@ int grpc_channel_args_is_census_enabled(const grpc_channel_args *a) {
}
return 0;
}
+
+grpc_compression_level grpc_channel_args_get_compression_level(
+ const grpc_channel_args *a) {
+ size_t i;
+ if (a) {
+ for (i = 0; a && i < a->num_args; ++i) {
+ if (a->args[i].type == GRPC_ARG_INTEGER &&
+ !strcmp(GRPC_COMPRESSION_LEVEL_ARG, a->args[i].key)) {
+ return a->args[i].value.integer;
+ break;
+ }
+ }
+ }
+ return GRPC_COMPRESS_LEVEL_NONE;
+}
+
+void grpc_channel_args_set_compression_level(
+ grpc_channel_args **a, grpc_compression_level level) {
+ grpc_arg tmp;
+ tmp.type = GRPC_ARG_INTEGER;
+ tmp.key = GRPC_COMPRESSION_LEVEL_ARG;
+ tmp.value.integer = level;
+ *a = grpc_channel_args_copy_and_add(*a, &tmp);
+}
diff --git a/src/core/channel/channel_args.h b/src/core/channel/channel_args.h
index eb5bf63986..bf747b26e6 100644
--- a/src/core/channel/channel_args.h
+++ b/src/core/channel/channel_args.h
@@ -34,21 +34,31 @@
#ifndef GRPC_INTERNAL_CORE_CHANNEL_CHANNEL_ARGS_H
#define GRPC_INTERNAL_CORE_CHANNEL_CHANNEL_ARGS_H
+#include <grpc/compression.h>
#include <grpc/grpc.h>
/* Copy some arguments */
grpc_channel_args *grpc_channel_args_copy(const grpc_channel_args *src);
-/* Copy some arguments and add the to_add parameter in the end.
+/** Copy some arguments and add the to_add parameter in the end.
If to_add is NULL, it is equivalent to call grpc_channel_args_copy. */
grpc_channel_args *grpc_channel_args_copy_and_add(const grpc_channel_args *src,
const grpc_arg *to_add);
-/* Destroy arguments created by grpc_channel_args_copy */
+/** Destroy arguments created by grpc_channel_args_copy */
void grpc_channel_args_destroy(grpc_channel_args *a);
-/* Reads census_enabled settings from channel args. Returns 1 if census_enabled
- is specified in channel args, otherwise returns 0. */
+/** Reads census_enabled settings from channel args. Returns 1 if census_enabled
+ * is specified in channel args, otherwise returns 0. */
int grpc_channel_args_is_census_enabled(const grpc_channel_args *a);
+/** Returns the compression level set in \a a. */
+grpc_compression_level grpc_channel_args_get_compression_level(
+ const grpc_channel_args *a);
+
+/** Sets the compression level in \a a to \a level. Setting it to
+ * GRPC_COMPRESS_LEVEL_NONE disables compression for the channel. */
+void grpc_channel_args_set_compression_level(
+ grpc_channel_args **a, grpc_compression_level level);
+
#endif /* GRPC_INTERNAL_CORE_CHANNEL_CHANNEL_ARGS_H */
diff --git a/src/core/channel/child_channel.c b/src/core/channel/child_channel.c
index cdc1c11681..6690265d75 100644
--- a/src/core/channel/child_channel.c
+++ b/src/core/channel/child_channel.c
@@ -157,17 +157,9 @@ static void lb_destroy_channel_elem(grpc_channel_element *elem) {
}
const grpc_channel_filter grpc_child_channel_top_filter = {
- lb_start_transport_op,
- lb_channel_op,
-
- sizeof(lb_call_data),
- lb_init_call_elem,
- lb_destroy_call_elem,
-
- sizeof(lb_channel_data),
- lb_init_channel_elem,
- lb_destroy_channel_elem,
-
+ lb_start_transport_op, lb_channel_op,
+ sizeof(lb_call_data), lb_init_call_elem, lb_destroy_call_elem,
+ sizeof(lb_channel_data), lb_init_channel_elem, lb_destroy_channel_elem,
"child-channel",
};
diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index 31b1fc3bde..726196e996 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -102,10 +102,17 @@ struct call_data {
static int prepare_activate(grpc_call_element *elem,
grpc_child_channel *on_child) {
call_data *calld = elem->call_data;
+ channel_data *chand = elem->channel_data;
if (calld->state == CALL_CANCELLED) return 0;
/* no more access to calld->s.waiting allowed */
GPR_ASSERT(calld->state == CALL_WAITING);
+
+ if (calld->s.waiting_op.bind_pollset) {
+ grpc_transport_setup_del_interested_party(chand->transport_setup,
+ calld->s.waiting_op.bind_pollset);
+ }
+
calld->state = CALL_ACTIVE;
/* create a child call */
@@ -199,6 +206,7 @@ static void cc_start_transport_op(grpc_call_element *elem,
handle_op_after_cancellation(elem, op);
} else {
calld->state = CALL_WAITING;
+ calld->s.waiting_op.bind_pollset = NULL;
if (chand->active_child) {
/* channel is connected - use the connected stack */
if (prepare_activate(elem, chand->active_child)) {
@@ -230,14 +238,14 @@ static void cc_start_transport_op(grpc_call_element *elem,
}
calld->s.waiting_op = *op;
chand->waiting_children[chand->waiting_child_count++] = calld;
+ grpc_transport_setup_add_interested_party(chand->transport_setup,
+ op->bind_pollset);
gpr_mu_unlock(&chand->mu);
/* finally initiate transport setup if needed */
if (initiate_transport_setup) {
grpc_transport_setup_initiate(chand->transport_setup);
}
- grpc_transport_setup_add_interested_party(chand->transport_setup,
- op->bind_pollset);
}
}
break;
diff --git a/src/core/channel/client_setup.c b/src/core/channel/client_setup.c
index f305d8ba9e..5be8fa66e9 100644
--- a/src/core/channel/client_setup.c
+++ b/src/core/channel/client_setup.c
@@ -56,12 +56,14 @@ struct grpc_client_setup {
gpr_cv cv;
grpc_client_setup_request *active_request;
int refs;
+ /** The set of pollsets that are currently interested in this
+ connection being established */
+ grpc_pollset_set interested_parties;
};
struct grpc_client_setup_request {
/* pointer back to the setup object */
grpc_client_setup *setup;
- grpc_pollset_set interested_parties;
gpr_timespec deadline;
};
@@ -71,7 +73,7 @@ gpr_timespec grpc_client_setup_request_deadline(grpc_client_setup_request *r) {
grpc_pollset_set *grpc_client_setup_get_interested_parties(
grpc_client_setup_request *r) {
- return &r->interested_parties;
+ return &r->setup->interested_parties;
}
static void destroy_setup(grpc_client_setup *s) {
@@ -79,13 +81,11 @@ static void destroy_setup(grpc_client_setup *s) {
gpr_cv_destroy(&s->cv);
s->done(s->user_data);
grpc_channel_args_destroy(s->args);
+ grpc_pollset_set_destroy(&s->interested_parties);
gpr_free(s);
}
-static void destroy_request(grpc_client_setup_request *r) {
- grpc_pollset_set_destroy(&r->interested_parties);
- gpr_free(r);
-}
+static void destroy_request(grpc_client_setup_request *r) { gpr_free(r); }
/* initiate handshaking */
static void setup_initiate(grpc_transport_setup *sp) {
@@ -94,8 +94,6 @@ static void setup_initiate(grpc_transport_setup *sp) {
int in_alarm = 0;
r->setup = s;
- grpc_pollset_set_init(&r->interested_parties);
- /* TODO(klempner): Actually set a deadline */
r->deadline = gpr_time_add(gpr_now(), gpr_time_from_seconds(60));
gpr_mu_lock(&s->mu);
@@ -120,33 +118,23 @@ static void setup_initiate(grpc_transport_setup *sp) {
}
}
+/** implementation of add_interested_party for setup vtable */
static void setup_add_interested_party(grpc_transport_setup *sp,
grpc_pollset *pollset) {
grpc_client_setup *s = (grpc_client_setup *)sp;
gpr_mu_lock(&s->mu);
- if (!s->active_request) {
- gpr_mu_unlock(&s->mu);
- return;
- }
-
- grpc_pollset_set_add_pollset(&s->active_request->interested_parties, pollset);
-
+ grpc_pollset_set_add_pollset(&s->interested_parties, pollset);
gpr_mu_unlock(&s->mu);
}
+/** implementation of del_interested_party for setup vtable */
static void setup_del_interested_party(grpc_transport_setup *sp,
grpc_pollset *pollset) {
grpc_client_setup *s = (grpc_client_setup *)sp;
gpr_mu_lock(&s->mu);
- if (!s->active_request) {
- gpr_mu_unlock(&s->mu);
- return;
- }
-
- grpc_pollset_set_del_pollset(&s->active_request->interested_parties, pollset);
-
+ grpc_pollset_set_del_pollset(&s->interested_parties, pollset);
gpr_mu_unlock(&s->mu);
}
@@ -179,7 +167,8 @@ static void setup_cancel(grpc_transport_setup *sp) {
}
}
-int grpc_client_setup_cb_begin(grpc_client_setup_request *r, const char *reason) {
+int grpc_client_setup_cb_begin(grpc_client_setup_request *r,
+ const char *reason) {
gpr_mu_lock(&r->setup->mu);
if (r->setup->cancelled) {
gpr_mu_unlock(&r->setup->mu);
@@ -190,7 +179,8 @@ int grpc_client_setup_cb_begin(grpc_client_setup_request *r, const char *reason)
return 1;
}
-void grpc_client_setup_cb_end(grpc_client_setup_request *r, const char *reason) {
+void grpc_client_setup_cb_end(grpc_client_setup_request *r,
+ const char *reason) {
gpr_mu_lock(&r->setup->mu);
r->setup->in_cb--;
if (r->setup->cancelled) gpr_cv_signal(&r->setup->cv);
@@ -223,11 +213,13 @@ void grpc_client_setup_create_and_attach(
s->in_alarm = 0;
s->in_cb = 0;
s->cancelled = 0;
+ grpc_pollset_set_init(&s->interested_parties);
grpc_client_channel_set_transport_setup(newly_minted_channel, &s->base);
}
-int grpc_client_setup_request_should_continue(grpc_client_setup_request *r, const char *reason) {
+int grpc_client_setup_request_should_continue(grpc_client_setup_request *r,
+ const char *reason) {
int result;
if (gpr_time_cmp(gpr_now(), r->deadline) > 0) {
result = 0;
@@ -239,7 +231,8 @@ int grpc_client_setup_request_should_continue(grpc_client_setup_request *r, cons
return result;
}
-static void backoff_alarm_done(void *arg /* grpc_client_setup */, int success) {
+static void backoff_alarm_done(void *arg /* grpc_client_setup_request */,
+ int success) {
grpc_client_setup_request *r = arg;
grpc_client_setup *s = r->setup;
/* Handle status cancelled? */
diff --git a/src/core/channel/client_setup.h b/src/core/channel/client_setup.h
index cbabb510b1..7d40338840 100644
--- a/src/core/channel/client_setup.h
+++ b/src/core/channel/client_setup.h
@@ -52,7 +52,8 @@ void grpc_client_setup_create_and_attach(
/* Check that r is the active request: needs to be performed at each callback.
If this races, we'll have two connection attempts running at once and the
old one will get cleaned up in due course, which is fine. */
-int grpc_client_setup_request_should_continue(grpc_client_setup_request *r, const char *reason);
+int grpc_client_setup_request_should_continue(grpc_client_setup_request *r,
+ const char *reason);
void grpc_client_setup_request_finish(grpc_client_setup_request *r,
int was_successful);
const grpc_channel_args *grpc_client_setup_get_channel_args(
@@ -61,7 +62,8 @@ const grpc_channel_args *grpc_client_setup_get_channel_args(
/* Call before calling back into the setup listener, and call only if
this function returns 1. If it returns 1, also promise to call
grpc_client_setup_cb_end */
-int grpc_client_setup_cb_begin(grpc_client_setup_request *r, const char *reason);
+int grpc_client_setup_cb_begin(grpc_client_setup_request *r,
+ const char *reason);
void grpc_client_setup_cb_end(grpc_client_setup_request *r, const char *reason);
/* Get the deadline for a request passed in to initiate. Implementations should
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-22 14:30:53 +0000