Merge pull request #13974 from nathanielmanistaatgoogle/12531
Elide cygrpc.Timespec.
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index af46246..c850271 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -2,53 +2,89 @@
We definitely welcome your patches and contributions to gRPC!
-If you are new to github, please start by reading [Pull Request howto](https://help.github.com/articles/about-pull-requests/)
+If you are new to github, please start by reading [Pull Request
+howto](https://help.github.com/articles/about-pull-requests/)
## Legal requirements
In order to protect both you and ourselves, you will need to sign the
-[Contributor License Agreement](https://identity.linuxfoundation.org/projects/cncf).
+[Contributor License
+Agreement](https://identity.linuxfoundation.org/projects/cncf).
## Running tests
-Use `tools/run_tests/run_tests.py` script to run the unit tests.
-See [tools/run_tests](tools/run_tests) for how to run tests for a given language.
+Use `tools/run_tests/run_tests.py` script to run the unit tests. See
+[tools/run_tests](tools/run_tests) for how to run tests for a given language.
-Prerequisites for building and running tests are listed in [INSTALL.md](INSTALL.md)
-and in `src/YOUR-LANGUAGE` (e.g. `src/csharp`)
+Prerequisites for building and running tests are listed in
+[INSTALL.md](INSTALL.md) and in `src/YOUR-LANGUAGE` (e.g. `src/csharp`)
## Generated project files
-To ease maintenance of language- and platform- specific build systems,
-many projects files are generated using templates and should not be edited
-by hand.
-Run `tools/buildgen/generate_projects.sh` to regenerate.
-See [templates](templates) for details.
+To ease maintenance of language- and platform- specific build systems, many
+projects files are generated using templates and should not be edited by hand.
+Run `tools/buildgen/generate_projects.sh` to regenerate. See
+[templates](templates) for details.
-As a rule of thumb, if you see the "sanity tests" failing you've most likely edited generated files or you didn't regenerate the projects properly (or your code formatting doesn't match our code style).
+As a rule of thumb, if you see the "sanity tests" failing you've most likely
+edited generated files or you didn't regenerate the projects properly (or your
+code formatting doesn't match our code style).
## Guidelines for Pull Requests
How to get your contributions merged smoothly and quickly.
-- Create **small PRs** that are narrowly focused on **addressing a single concern**. We often times receive PRs that are trying to fix several things at a time, but only one fix is considered acceptable, nothing gets merged and both author's & review's time is wasted. Create more PRs to address different concerns and everyone will be happy.
+- Create **small PRs** that are narrowly focused on **addressing a single
+ concern**. We often times receive PRs that are trying to fix several things
+ at a time, but only one fix is considered acceptable, nothing gets merged and
+ both author's & review's time is wasted. Create more PRs to address different
+ concerns and everyone will be happy.
-- For speculative changes, consider opening an issue and discussing it first. If you are suggesting a behavioral or API change, consider starting with a [gRFC proposal](https://github.com/grpc/proposal).
+- For speculative changes, consider opening an issue and discussing it first.
+ If you are suggesting a behavioral or API change, consider starting with a
+ [gRFC proposal](https://github.com/grpc/proposal).
-- Provide a good **PR description** as a record of **what** change is being made and **why** it was made. Link to a github issue if it exists.
+- Provide a good **PR description** as a record of **what** change is being made
+ and **why** it was made. Link to a GitHub issue if it exists.
-- Don't fix code style and formatting unless you are already changing that line to address an issue. PRs with irrelevant changes won't be merged. If you do want to fix formatting or style, do that in a separate PR.
+- Don't fix code style and formatting unless you are already changing that line
+ to address an issue. PRs with irrelevant changes won't be merged. If you do
+ want to fix formatting or style, do that in a separate PR.
-- Unless your PR is trivial, you should expect there will be reviewer comments that you'll need to address before merging. We expect you to be reasonably responsive to those comments, otherwise the PR will be closed after 2-3 weeks of inactivity.
+- Unless your PR is trivial, you should expect there will be reviewer comments
+ that you'll need to address before merging. We expect you to be reasonably
+ responsive to those comments, otherwise the PR will be closed after 2-3 weeks
+ of inactivity.
+
+- If you have non-trivial contributions, please consider adding an entry to [the
+ AUTHORS file](https://github.com/grpc/grpc/blob/master/AUTHORS) listing the
+ copyright holder for the contribution (yourself, if you are signing the
+ individual CLA, or your company, for corporate CLAs) in the same PR as your
+ contribution. This needs to be done only once, for each company, or
+ individual.
-- Maintain **clean commit history** and use **meaningful commit messages**. PRs with messy commit history are difficult to review and won't be merged. Use `rebase -i upstream/master` to curate your commit history and/or to bring in latest changes from master (but avoid rebasing in the middle of a code review).
+- Maintain **clean commit history** and use **meaningful commit messages**.
+ PRs with messy commit history are difficult to review and won't be merged.
+ Use `rebase -i upstream/master` to curate your commit history and/or to
+ bring in latest changes from master (but avoid rebasing in the middle of
+ a code review).
-- Keep your PR up to date with upstream/master (if there are merge conflicts, we can't really merge your change).
+- Keep your PR up to date with upstream/master (if there are merge conflicts,
+ we can't really merge your change).
-- if you are regenerating the projects using `tools/buildgen/generate_projects.sh`, make changes to generated files a separate commit with commit message `regenerate projects`. Mixing changes to generated and hand-written files make your PR difficult to review.
+- If you are regenerating the projects using
+ `tools/buildgen/generate_projects.sh`, make changes to generated files a
+ separate commit with commit message `regenerate projects`. Mixing changes
+ to generated and hand-written files make your PR difficult to review.
-- **All tests need to be passing** before your change can be merged. We recommend you **run tests locally** before creating your PR to catch breakages early on (see [tools/run_tests](tools/run_tests). Ultimately, the green signal will be provided by our testing infrastructure. The reviewer will help you if there are test failures that seem not related to the change you are making.
+- **All tests need to be passing** before your change can be merged.
+ We recommend you **run tests locally** before creating your PR to catch
+ breakages early on (see [tools/run_tests](tools/run_tests). Ultimately, the
+ green signal will be provided by our testing infrastructure. The reviewer
+ will help you if there are test failures that seem not related to the change
+ you are making.
-- Exceptions to the rules can be made if there's a compelling reason for doing so.
+- Exceptions to the rules can be made if there's a compelling reason for doing
+ so.
diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc
index 3f3334d..e99022a 100644
--- a/src/core/ext/filters/client_channel/client_channel.cc
+++ b/src/core/ext/filters/client_channel/client_channel.cc
@@ -553,7 +553,6 @@
}
grpc_pollset_set_del_pollset_set(chand->lb_policy->interested_parties,
chand->interested_parties);
- grpc_lb_policy_shutdown_locked(chand->lb_policy, new_lb_policy);
GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel");
}
chand->lb_policy = new_lb_policy;
@@ -659,7 +658,6 @@
if (chand->lb_policy != nullptr) {
grpc_pollset_set_del_pollset_set(chand->lb_policy->interested_parties,
chand->interested_parties);
- grpc_lb_policy_shutdown_locked(chand->lb_policy, nullptr);
GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel");
chand->lb_policy = nullptr;
}
@@ -794,7 +792,6 @@
if (chand->lb_policy != nullptr) {
grpc_pollset_set_del_pollset_set(chand->lb_policy->interested_parties,
chand->interested_parties);
- grpc_lb_policy_shutdown_locked(chand->lb_policy, nullptr);
GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel");
}
gpr_free(chand->info_lb_policy_name);
@@ -855,10 +852,12 @@
grpc_subchannel_call* subchannel_call;
grpc_error* error;
- grpc_lb_policy_pick_state pick;
+ grpc_lb_policy* lb_policy; // Holds ref while LB pick is pending.
grpc_closure lb_pick_closure;
grpc_closure lb_pick_cancel_closure;
+ grpc_connected_subchannel* connected_subchannel;
+ grpc_call_context_element subchannel_call_context[GRPC_CONTEXT_COUNT];
grpc_polling_entity* pollent;
grpc_transport_stream_op_batch* waiting_for_pick_batches[MAX_WAITING_BATCHES];
@@ -867,6 +866,8 @@
grpc_transport_stream_op_batch* initial_metadata_batch;
+ grpc_linked_mdelem lb_token_mdelem;
+
grpc_closure on_complete;
grpc_closure* original_on_complete;
} call_data;
@@ -1004,16 +1005,16 @@
channel_data* chand = (channel_data*)elem->channel_data;
call_data* calld = (call_data*)elem->call_data;
const grpc_connected_subchannel_call_args call_args = {
- calld->pollent, // pollent
- calld->path, // path
- calld->call_start_time, // start_time
- calld->deadline, // deadline
- calld->arena, // arena
- calld->pick.subchannel_call_context, // context
- calld->call_combiner // call_combiner
+ calld->pollent, // pollent
+ calld->path, // path
+ calld->call_start_time, // start_time
+ calld->deadline, // deadline
+ calld->arena, // arena
+ calld->subchannel_call_context, // context
+ calld->call_combiner // call_combiner
};
grpc_error* new_error = grpc_connected_subchannel_create_call(
- calld->pick.connected_subchannel, &call_args, &calld->subchannel_call);
+ calld->connected_subchannel, &call_args, &calld->subchannel_call);
if (grpc_client_channel_trace.enabled()) {
gpr_log(GPR_DEBUG, "chand=%p calld=%p: create subchannel_call=%p: error=%s",
chand, calld, calld->subchannel_call, grpc_error_string(new_error));
@@ -1031,7 +1032,7 @@
static void pick_done_locked(grpc_call_element* elem, grpc_error* error) {
call_data* calld = (call_data*)elem->call_data;
channel_data* chand = (channel_data*)elem->channel_data;
- if (calld->pick.connected_subchannel == nullptr) {
+ if (calld->connected_subchannel == nullptr) {
// Failed to create subchannel.
GRPC_ERROR_UNREF(calld->error);
calld->error = error == GRPC_ERROR_NONE
@@ -1070,16 +1071,13 @@
grpc_call_element* elem = (grpc_call_element*)arg;
channel_data* chand = (channel_data*)elem->channel_data;
call_data* calld = (call_data*)elem->call_data;
- // Note: chand->lb_policy may have changed since we started our pick,
- // in which case we will be cancelling the pick on a policy other than
- // the one we started it on. However, this will just be a no-op.
- if (error != GRPC_ERROR_NONE && chand->lb_policy != nullptr) {
+ if (calld->lb_policy != nullptr) {
if (grpc_client_channel_trace.enabled()) {
gpr_log(GPR_DEBUG, "chand=%p calld=%p: cancelling pick from LB policy %p",
- chand, calld, chand->lb_policy);
+ chand, calld, calld->lb_policy);
}
- grpc_lb_policy_cancel_pick_locked(chand->lb_policy, &calld->pick,
- GRPC_ERROR_REF(error));
+ grpc_lb_policy_cancel_pick_locked(
+ calld->lb_policy, &calld->connected_subchannel, GRPC_ERROR_REF(error));
}
GRPC_CALL_STACK_UNREF(calld->owning_call, "pick_callback_cancel");
}
@@ -1094,6 +1092,9 @@
gpr_log(GPR_DEBUG, "chand=%p calld=%p: pick completed asynchronously",
chand, calld);
}
+ GPR_ASSERT(calld->lb_policy != nullptr);
+ GRPC_LB_POLICY_UNREF(calld->lb_policy, "pick_subchannel");
+ calld->lb_policy = nullptr;
async_pick_done_locked(elem, GRPC_ERROR_REF(error));
}
@@ -1127,21 +1128,26 @@
initial_metadata_flags &= ~GRPC_INITIAL_METADATA_WAIT_FOR_READY;
}
}
- calld->pick.initial_metadata =
+ const grpc_lb_policy_pick_args inputs = {
calld->initial_metadata_batch->payload->send_initial_metadata
- .send_initial_metadata;
- calld->pick.initial_metadata_flags = initial_metadata_flags;
+ .send_initial_metadata,
+ initial_metadata_flags, &calld->lb_token_mdelem};
+ // Keep a ref to the LB policy in calld while the pick is pending.
+ GRPC_LB_POLICY_REF(chand->lb_policy, "pick_subchannel");
+ calld->lb_policy = chand->lb_policy;
GRPC_CLOSURE_INIT(&calld->lb_pick_closure, pick_callback_done_locked, elem,
grpc_combiner_scheduler(chand->combiner));
- calld->pick.on_complete = &calld->lb_pick_closure;
- const bool pick_done =
- grpc_lb_policy_pick_locked(chand->lb_policy, &calld->pick);
+ const bool pick_done = grpc_lb_policy_pick_locked(
+ chand->lb_policy, &inputs, &calld->connected_subchannel,
+ calld->subchannel_call_context, nullptr, &calld->lb_pick_closure);
if (pick_done) {
/* synchronous grpc_lb_policy_pick call. Unref the LB policy. */
if (grpc_client_channel_trace.enabled()) {
gpr_log(GPR_DEBUG, "chand=%p calld=%p: pick completed synchronously",
chand, calld);
}
+ GRPC_LB_POLICY_UNREF(calld->lb_policy, "pick_subchannel");
+ calld->lb_policy = nullptr;
} else {
GRPC_CALL_STACK_REF(calld->owning_call, "pick_callback_cancel");
grpc_call_combiner_set_notify_on_cancel(
@@ -1283,7 +1289,7 @@
grpc_call_element* elem = (grpc_call_element*)arg;
call_data* calld = (call_data*)elem->call_data;
channel_data* chand = (channel_data*)elem->channel_data;
- GPR_ASSERT(calld->pick.connected_subchannel == nullptr);
+ GPR_ASSERT(calld->connected_subchannel == nullptr);
if (chand->lb_policy != nullptr) {
// We already have an LB policy, so ask it for a pick.
if (pick_callback_start_locked(elem)) {
@@ -1461,14 +1467,15 @@
GRPC_SUBCHANNEL_CALL_UNREF(calld->subchannel_call,
"client_channel_destroy_call");
}
+ GPR_ASSERT(calld->lb_policy == nullptr);
GPR_ASSERT(calld->waiting_for_pick_batches_count == 0);
- if (calld->pick.connected_subchannel != nullptr) {
- GRPC_CONNECTED_SUBCHANNEL_UNREF(calld->pick.connected_subchannel, "picked");
+ if (calld->connected_subchannel != nullptr) {
+ GRPC_CONNECTED_SUBCHANNEL_UNREF(calld->connected_subchannel, "picked");
}
for (size_t i = 0; i < GRPC_CONTEXT_COUNT; ++i) {
- if (calld->pick.subchannel_call_context[i].value != nullptr) {
- calld->pick.subchannel_call_context[i].destroy(
- calld->pick.subchannel_call_context[i].value);
+ if (calld->subchannel_call_context[i].value != nullptr) {
+ calld->subchannel_call_context[i].destroy(
+ calld->subchannel_call_context[i].value);
}
}
GRPC_CLOSURE_SCHED(then_schedule_closure, GRPC_ERROR_NONE);
diff --git a/src/core/ext/filters/client_channel/lb_policy.cc b/src/core/ext/filters/client_channel/lb_policy.cc
index cc4fe7e..7a5a8de 100644
--- a/src/core/ext/filters/client_channel/lb_policy.cc
+++ b/src/core/ext/filters/client_channel/lb_policy.cc
@@ -19,6 +19,8 @@
#include "src/core/ext/filters/client_channel/lb_policy.h"
#include "src/core/lib/iomgr/combiner.h"
+#define WEAK_REF_BITS 16
+
grpc_core::DebugOnlyTraceFlag grpc_trace_lb_policy_refcount(
false, "lb_policy_refcount");
@@ -26,60 +28,91 @@
const grpc_lb_policy_vtable* vtable,
grpc_combiner* combiner) {
policy->vtable = vtable;
- gpr_ref_init(&policy->refs, 1);
+ gpr_atm_no_barrier_store(&policy->ref_pair, 1 << WEAK_REF_BITS);
policy->interested_parties = grpc_pollset_set_create();
policy->combiner = GRPC_COMBINER_REF(combiner, "lb_policy");
}
#ifndef NDEBUG
-void grpc_lb_policy_ref(grpc_lb_policy* lb_policy, const char* file, int line,
- const char* reason) {
- if (grpc_trace_lb_policy_refcount.enabled()) {
- gpr_atm old_refs = gpr_atm_no_barrier_load(&lb_policy->refs.count);
- gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG,
- "LB_POLICY:%p ref %" PRIdPTR " -> %" PRIdPTR " %s", lb_policy,
- old_refs, old_refs + 1, reason);
- }
+#define REF_FUNC_EXTRA_ARGS , const char *file, int line, const char *reason
+#define REF_MUTATE_EXTRA_ARGS REF_FUNC_EXTRA_ARGS, const char* purpose
+#define REF_FUNC_PASS_ARGS(new_reason) , file, line, new_reason
+#define REF_MUTATE_PASS_ARGS(purpose) , file, line, reason, purpose
#else
-void grpc_lb_policy_ref(grpc_lb_policy* lb_policy) {
+#define REF_FUNC_EXTRA_ARGS
+#define REF_MUTATE_EXTRA_ARGS
+#define REF_FUNC_PASS_ARGS(new_reason)
+#define REF_MUTATE_PASS_ARGS(x)
#endif
- gpr_ref(&lb_policy->refs);
+
+static gpr_atm ref_mutate(grpc_lb_policy* c, gpr_atm delta,
+ int barrier REF_MUTATE_EXTRA_ARGS) {
+ gpr_atm old_val = barrier ? gpr_atm_full_fetch_add(&c->ref_pair, delta)
+ : gpr_atm_no_barrier_fetch_add(&c->ref_pair, delta);
+#ifndef NDEBUG
+ if (grpc_trace_lb_policy_refcount.enabled()) {
+ gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG,
+ "LB_POLICY: %p %12s 0x%" PRIxPTR " -> 0x%" PRIxPTR " [%s]", c,
+ purpose, old_val, old_val + delta, reason);
+ }
+#endif
+ return old_val;
}
-#ifndef NDEBUG
-void grpc_lb_policy_unref(grpc_lb_policy* lb_policy, const char* file, int line,
- const char* reason) {
- if (grpc_trace_lb_policy_refcount.enabled()) {
- gpr_atm old_refs = gpr_atm_no_barrier_load(&lb_policy->refs.count);
- gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG,
- "LB_POLICY:%p unref %" PRIdPTR " -> %" PRIdPTR " %s", lb_policy,
- old_refs, old_refs - 1, reason);
+void grpc_lb_policy_ref(grpc_lb_policy* policy REF_FUNC_EXTRA_ARGS) {
+ ref_mutate(policy, 1 << WEAK_REF_BITS, 0 REF_MUTATE_PASS_ARGS("STRONG_REF"));
+}
+
+static void shutdown_locked(void* arg, grpc_error* error) {
+ grpc_lb_policy* policy = (grpc_lb_policy*)arg;
+ policy->vtable->shutdown_locked(policy);
+ GRPC_LB_POLICY_WEAK_UNREF(policy, "strong-unref");
+}
+
+void grpc_lb_policy_unref(grpc_lb_policy* policy REF_FUNC_EXTRA_ARGS) {
+ gpr_atm old_val =
+ ref_mutate(policy, (gpr_atm)1 - (gpr_atm)(1 << WEAK_REF_BITS),
+ 1 REF_MUTATE_PASS_ARGS("STRONG_UNREF"));
+ gpr_atm mask = ~(gpr_atm)((1 << WEAK_REF_BITS) - 1);
+ gpr_atm check = 1 << WEAK_REF_BITS;
+ if ((old_val & mask) == check) {
+ GRPC_CLOSURE_SCHED(
+ GRPC_CLOSURE_CREATE(shutdown_locked, policy,
+ grpc_combiner_scheduler(policy->combiner)),
+ GRPC_ERROR_NONE);
+ } else {
+ grpc_lb_policy_weak_unref(policy REF_FUNC_PASS_ARGS("strong-unref"));
}
-#else
-void grpc_lb_policy_unref(grpc_lb_policy* lb_policy) {
-#endif
- if (gpr_unref(&lb_policy->refs)) {
- grpc_pollset_set_destroy(lb_policy->interested_parties);
- grpc_combiner* combiner = lb_policy->combiner;
- lb_policy->vtable->destroy(lb_policy);
+}
+
+void grpc_lb_policy_weak_ref(grpc_lb_policy* policy REF_FUNC_EXTRA_ARGS) {
+ ref_mutate(policy, 1, 0 REF_MUTATE_PASS_ARGS("WEAK_REF"));
+}
+
+void grpc_lb_policy_weak_unref(grpc_lb_policy* policy REF_FUNC_EXTRA_ARGS) {
+ gpr_atm old_val =
+ ref_mutate(policy, -(gpr_atm)1, 1 REF_MUTATE_PASS_ARGS("WEAK_UNREF"));
+ if (old_val == 1) {
+ grpc_pollset_set_destroy(policy->interested_parties);
+ grpc_combiner* combiner = policy->combiner;
+ policy->vtable->destroy(policy);
GRPC_COMBINER_UNREF(combiner, "lb_policy");
}
}
-void grpc_lb_policy_shutdown_locked(grpc_lb_policy* policy,
- grpc_lb_policy* new_policy) {
- policy->vtable->shutdown_locked(policy, new_policy);
-}
-
int grpc_lb_policy_pick_locked(grpc_lb_policy* policy,
- grpc_lb_policy_pick_state* pick) {
- return policy->vtable->pick_locked(policy, pick);
+ const grpc_lb_policy_pick_args* pick_args,
+ grpc_connected_subchannel** target,
+ grpc_call_context_element* context,
+ void** user_data, grpc_closure* on_complete) {
+ return policy->vtable->pick_locked(policy, pick_args, target, context,
+ user_data, on_complete);
}
void grpc_lb_policy_cancel_pick_locked(grpc_lb_policy* policy,
- grpc_lb_policy_pick_state* pick,
+ grpc_connected_subchannel** target,
grpc_error* error) {
- policy->vtable->cancel_pick_locked(policy, pick, error);
+ policy->vtable->cancel_pick_locked(policy, target, error);
}
void grpc_lb_policy_cancel_picks_locked(grpc_lb_policy* policy,
diff --git a/src/core/ext/filters/client_channel/lb_policy.h b/src/core/ext/filters/client_channel/lb_policy.h
index 1176a05..3572c97 100644
--- a/src/core/ext/filters/client_channel/lb_policy.h
+++ b/src/core/ext/filters/client_channel/lb_policy.h
@@ -33,7 +33,7 @@
struct grpc_lb_policy {
const grpc_lb_policy_vtable* vtable;
- gpr_refcount refs;
+ gpr_atm ref_pair;
/* owned pointer to interested parties in load balancing decisions */
grpc_pollset_set* interested_parties;
/* combiner under which lb_policy actions take place */
@@ -42,42 +42,32 @@
grpc_closure* request_reresolution;
};
-/// State used for an LB pick.
-typedef struct grpc_lb_policy_pick_state {
- /// Initial metadata associated with the picking call.
+/** Extra arguments for an LB pick */
+typedef struct grpc_lb_policy_pick_args {
+ /** Initial metadata associated with the picking call. */
grpc_metadata_batch* initial_metadata;
- /// Bitmask used for selective cancelling. See \a
- /// grpc_lb_policy_cancel_picks() and \a GRPC_INITIAL_METADATA_* in
- /// grpc_types.h.
+ /** Bitmask used for selective cancelling. See \a
+ * grpc_lb_policy_cancel_picks() and \a GRPC_INITIAL_METADATA_* in
+ * grpc_types.h */
uint32_t initial_metadata_flags;
- /// Storage for LB token in \a initial_metadata, or NULL if not used.
- grpc_linked_mdelem lb_token_mdelem_storage;
- /// Closure to run when pick is complete, if not completed synchronously.
- grpc_closure* on_complete;
- /// Will be set to the selected subchannel, or NULL on failure or when
- /// the LB policy decides to drop the call.
- grpc_connected_subchannel* connected_subchannel;
- /// Will be populated with context to pass to the subchannel call, if needed.
- grpc_call_context_element subchannel_call_context[GRPC_CONTEXT_COUNT];
- /// Upon success, \a *user_data will be set to whatever opaque information
- /// may need to be propagated from the LB policy, or NULL if not needed.
- void** user_data;
- /// Next pointer. For internal use by LB policy.
- struct grpc_lb_policy_pick_state* next;
-} grpc_lb_policy_pick_state;
+ /** Storage for LB token in \a initial_metadata, or NULL if not used */
+ grpc_linked_mdelem* lb_token_mdelem_storage;
+} grpc_lb_policy_pick_args;
struct grpc_lb_policy_vtable {
void (*destroy)(grpc_lb_policy* policy);
-
- /// \see grpc_lb_policy_shutdown_locked().
- void (*shutdown_locked)(grpc_lb_policy* policy, grpc_lb_policy* new_policy);
+ void (*shutdown_locked)(grpc_lb_policy* policy);
/** \see grpc_lb_policy_pick */
- int (*pick_locked)(grpc_lb_policy* policy, grpc_lb_policy_pick_state* pick);
+ int (*pick_locked)(grpc_lb_policy* policy,
+ const grpc_lb_policy_pick_args* pick_args,
+ grpc_connected_subchannel** target,
+ grpc_call_context_element* context, void** user_data,
+ grpc_closure* on_complete);
/** \see grpc_lb_policy_cancel_pick */
void (*cancel_pick_locked)(grpc_lb_policy* policy,
- grpc_lb_policy_pick_state* pick,
+ grpc_connected_subchannel** target,
grpc_error* error);
/** \see grpc_lb_policy_cancel_picks */
@@ -113,19 +103,37 @@
};
#ifndef NDEBUG
+
+/* Strong references: the policy will shutdown when they reach zero */
#define GRPC_LB_POLICY_REF(p, r) \
grpc_lb_policy_ref((p), __FILE__, __LINE__, (r))
#define GRPC_LB_POLICY_UNREF(p, r) \
grpc_lb_policy_unref((p), __FILE__, __LINE__, (r))
+
+/* Weak references: they don't prevent the shutdown of the LB policy. When no
+ * strong references are left but there are still weak ones, shutdown is called.
+ * Once the weak reference also reaches zero, the LB policy is destroyed. */
+#define GRPC_LB_POLICY_WEAK_REF(p, r) \
+ grpc_lb_policy_weak_ref((p), __FILE__, __LINE__, (r))
+#define GRPC_LB_POLICY_WEAK_UNREF(p, r) \
+ grpc_lb_policy_weak_unref((p), __FILE__, __LINE__, (r))
void grpc_lb_policy_ref(grpc_lb_policy* policy, const char* file, int line,
const char* reason);
void grpc_lb_policy_unref(grpc_lb_policy* policy, const char* file, int line,
const char* reason);
-#else // !NDEBUG
+void grpc_lb_policy_weak_ref(grpc_lb_policy* policy, const char* file, int line,
+ const char* reason);
+void grpc_lb_policy_weak_unref(grpc_lb_policy* policy, const char* file,
+ int line, const char* reason);
+#else
#define GRPC_LB_POLICY_REF(p, r) grpc_lb_policy_ref((p))
#define GRPC_LB_POLICY_UNREF(p, r) grpc_lb_policy_unref((p))
+#define GRPC_LB_POLICY_WEAK_REF(p, r) grpc_lb_policy_weak_ref((p))
+#define GRPC_LB_POLICY_WEAK_UNREF(p, r) grpc_lb_policy_weak_unref((p))
void grpc_lb_policy_ref(grpc_lb_policy* policy);
void grpc_lb_policy_unref(grpc_lb_policy* policy);
+void grpc_lb_policy_weak_ref(grpc_lb_policy* policy);
+void grpc_lb_policy_weak_unref(grpc_lb_policy* policy);
#endif
/** called by concrete implementations to initialize the base struct */
@@ -133,24 +141,28 @@
const grpc_lb_policy_vtable* vtable,
grpc_combiner* combiner);
-/// Shuts down \a policy.
-/// If \a new_policy is non-null, any pending picks will be restarted
-/// on that policy; otherwise, they will be failed.
-void grpc_lb_policy_shutdown_locked(grpc_lb_policy* policy,
- grpc_lb_policy* new_policy);
+/** Finds an appropriate subchannel for a call, based on \a pick_args.
-/** Finds an appropriate subchannel for a call, based on data in \a pick.
- \a pick must remain alive until the pick is complete.
+ \a target will be set to the selected subchannel, or NULL on failure
+ or when the LB policy decides to drop the call.
+
+ Upon success, \a user_data will be set to whatever opaque information
+ may need to be propagated from the LB policy, or NULL if not needed.
+ \a context will be populated with context to pass to the subchannel
+ call, if needed.
If the pick succeeds and a result is known immediately, a non-zero
- value will be returned. Otherwise, \a pick->on_complete will be invoked
+ value will be returned. Otherwise, \a on_complete will be invoked
once the pick is complete with its error argument set to indicate
success or failure.
Any IO should be done under the \a interested_parties \a grpc_pollset_set
in the \a grpc_lb_policy struct. */
int grpc_lb_policy_pick_locked(grpc_lb_policy* policy,
- grpc_lb_policy_pick_state* pick);
+ const grpc_lb_policy_pick_args* pick_args,
+ grpc_connected_subchannel** target,
+ grpc_call_context_element* context,
+ void** user_data, grpc_closure* on_complete);
/** Perform a connected subchannel ping (see \a grpc_connected_subchannel_ping)
against one of the connected subchannels managed by \a policy. */
@@ -158,11 +170,11 @@
grpc_closure* on_initiate,
grpc_closure* on_ack);
-/** Cancel picks for \a pick.
+/** Cancel picks for \a target.
The \a on_complete callback of the pending picks will be invoked with \a
*target set to NULL. */
void grpc_lb_policy_cancel_pick_locked(grpc_lb_policy* policy,
- grpc_lb_policy_pick_state* pick,
+ grpc_connected_subchannel** target,
grpc_error* error);
/** Cancel all pending picks for which their \a initial_metadata_flags (as given
diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc
index 5849ac9..1317cdc 100644
--- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc
+++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc
@@ -54,7 +54,7 @@
* operations in progress over the old RR instance. This is done by
* decreasing the reference count on the old policy. The moment no more
* references are held on the old RR policy, it'll be destroyed and \a
- * on_rr_connectivity_changed notified with a \a GRPC_CHANNEL_SHUTDOWN
+ * glb_rr_connectivity_changed notified with a \a GRPC_CHANNEL_SHUTDOWN
* state. At this point we can transition to a new RR instance safely, which
* is done once again via \a rr_handover_locked().
*
@@ -128,48 +128,187 @@
grpc_core::TraceFlag grpc_lb_glb_trace(false, "glb");
-struct glb_lb_policy;
+/* add lb_token of selected subchannel (address) to the call's initial
+ * metadata */
+static grpc_error* initial_metadata_add_lb_token(
+ grpc_metadata_batch* initial_metadata,
+ grpc_linked_mdelem* lb_token_mdelem_storage, grpc_mdelem lb_token) {
+ GPR_ASSERT(lb_token_mdelem_storage != nullptr);
+ GPR_ASSERT(!GRPC_MDISNULL(lb_token));
+ return grpc_metadata_batch_add_tail(initial_metadata, lb_token_mdelem_storage,
+ lb_token);
+}
+
+static void destroy_client_stats(void* arg) {
+ grpc_grpclb_client_stats_unref((grpc_grpclb_client_stats*)arg);
+}
+
+typedef struct wrapped_rr_closure_arg {
+ /* the closure instance using this struct as argument */
+ grpc_closure wrapper_closure;
+
+ /* the original closure. Usually a on_complete/notify cb for pick() and ping()
+ * calls against the internal RR instance, respectively. */
+ grpc_closure* wrapped_closure;
+
+ /* the pick's initial metadata, kept in order to append the LB token for the
+ * pick */
+ grpc_metadata_batch* initial_metadata;
+
+ /* the picked target, used to determine which LB token to add to the pick's
+ * initial metadata */
+ grpc_connected_subchannel** target;
+
+ /* the context to be populated for the subchannel call */
+ grpc_call_context_element* context;
+
+ /* Stats for client-side load reporting. Note that this holds a
+ * reference, which must be either passed on via context or unreffed. */
+ grpc_grpclb_client_stats* client_stats;
+
+ /* the LB token associated with the pick */
+ grpc_mdelem lb_token;
+
+ /* storage for the lb token initial metadata mdelem */
+ grpc_linked_mdelem* lb_token_mdelem_storage;
+
+ /* The RR instance related to the closure */
+ grpc_lb_policy* rr_policy;
+
+ /* The grpclb instance that created the wrapping. This instance is not owned,
+ * reference counts are untouched. It's used only for logging purposes. */
+ grpc_lb_policy* glb_policy;
+
+ /* heap memory to be freed upon closure execution. */
+ void* free_when_done;
+} wrapped_rr_closure_arg;
+
+/* The \a on_complete closure passed as part of the pick requires keeping a
+ * reference to its associated round robin instance. We wrap this closure in
+ * order to unref the round robin instance upon its invocation */
+static void wrapped_rr_closure(void* arg, grpc_error* error) {
+ wrapped_rr_closure_arg* wc_arg = (wrapped_rr_closure_arg*)arg;
+
+ GPR_ASSERT(wc_arg->wrapped_closure != nullptr);
+ GRPC_CLOSURE_SCHED(wc_arg->wrapped_closure, GRPC_ERROR_REF(error));
+
+ if (wc_arg->rr_policy != nullptr) {
+ /* if *target is nullptr, no pick has been made by the RR policy (eg, all
+ * addresses failed to connect). There won't be any user_data/token
+ * available */
+ if (*wc_arg->target != nullptr) {
+ if (!GRPC_MDISNULL(wc_arg->lb_token)) {
+ initial_metadata_add_lb_token(wc_arg->initial_metadata,
+ wc_arg->lb_token_mdelem_storage,
+ GRPC_MDELEM_REF(wc_arg->lb_token));
+ } else {
+ gpr_log(
+ GPR_ERROR,
+ "[grpclb %p] No LB token for connected subchannel pick %p (from RR "
+ "instance %p).",
+ wc_arg->glb_policy, *wc_arg->target, wc_arg->rr_policy);
+ abort();
+ }
+ // Pass on client stats via context. Passes ownership of the reference.
+ GPR_ASSERT(wc_arg->client_stats != nullptr);
+ wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].value = wc_arg->client_stats;
+ wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].destroy = destroy_client_stats;
+ } else {
+ grpc_grpclb_client_stats_unref(wc_arg->client_stats);
+ }
+ if (grpc_lb_glb_trace.enabled()) {
+ gpr_log(GPR_INFO, "[grpclb %p] Unreffing RR %p", wc_arg->glb_policy,
+ wc_arg->rr_policy);
+ }
+ GRPC_LB_POLICY_UNREF(wc_arg->rr_policy, "wrapped_rr_closure");
+ }
+ GPR_ASSERT(wc_arg->free_when_done != nullptr);
+ gpr_free(wc_arg->free_when_done);
+}
namespace {
-
-/// Linked list of pending pick requests. It stores all information needed to
-/// eventually call (Round Robin's) pick() on them. They mainly stay pending
-/// waiting for the RR policy to be created.
-///
-/// Note that when a pick is sent to the RR policy, we inject our own
-/// on_complete callback, so that we can intercept the result before
-/// invoking the original on_complete callback. This allows us to set the
-/// LB token metadata and add client_stats to the call context.
-/// See \a pending_pick_complete() for details.
+/* Linked list of pending pick requests. It stores all information needed to
+ * eventually call (Round Robin's) pick() on them. They mainly stay pending
+ * waiting for the RR policy to be created/updated.
+ *
+ * One particularity is the wrapping of the user-provided \a on_complete closure
+ * (in \a wrapped_on_complete and \a wrapped_on_complete_arg). This is needed in
+ * order to correctly unref the RR policy instance upon completion of the pick.
+ * See \a wrapped_rr_closure for details. */
struct pending_pick {
- // Our on_complete closure and the original one.
- grpc_closure on_complete;
- grpc_closure* original_on_complete;
- // The original pick.
- grpc_lb_policy_pick_state* pick;
- // Stats for client-side load reporting. Note that this holds a
- // reference, which must be either passed on via context or unreffed.
- grpc_grpclb_client_stats* client_stats;
- // The LB token associated with the pick. This is set via user_data in
- // the pick.
- grpc_mdelem lb_token;
- // The grpclb instance that created the wrapping. This instance is not owned,
- // reference counts are untouched. It's used only for logging purposes.
- glb_lb_policy* glb_policy;
- // Next pending pick.
struct pending_pick* next;
-};
-/// A linked list of pending pings waiting for the RR policy to be created.
-struct pending_ping {
- grpc_closure* on_initiate;
- grpc_closure* on_ack;
- struct pending_ping* next;
-};
+ /* original pick()'s arguments */
+ grpc_lb_policy_pick_args pick_args;
+ /* output argument where to store the pick()ed connected subchannel, or
+ * nullptr upon error. */
+ grpc_connected_subchannel** target;
+
+ /* args for wrapped_on_complete */
+ wrapped_rr_closure_arg wrapped_on_complete_arg;
+};
} // namespace
-struct glb_lb_policy {
+static void add_pending_pick(pending_pick** root,
+ const grpc_lb_policy_pick_args* pick_args,
+ grpc_connected_subchannel** target,
+ grpc_call_context_element* context,
+ grpc_closure* on_complete) {
+ pending_pick* pp = (pending_pick*)gpr_zalloc(sizeof(*pp));
+ pp->next = *root;
+ pp->pick_args = *pick_args;
+ pp->target = target;
+ pp->wrapped_on_complete_arg.wrapped_closure = on_complete;
+ pp->wrapped_on_complete_arg.target = target;
+ pp->wrapped_on_complete_arg.context = context;
+ pp->wrapped_on_complete_arg.initial_metadata = pick_args->initial_metadata;
+ pp->wrapped_on_complete_arg.lb_token_mdelem_storage =
+ pick_args->lb_token_mdelem_storage;
+ pp->wrapped_on_complete_arg.free_when_done = pp;
+ GRPC_CLOSURE_INIT(&pp->wrapped_on_complete_arg.wrapper_closure,
+ wrapped_rr_closure, &pp->wrapped_on_complete_arg,
+ grpc_schedule_on_exec_ctx);
+ *root = pp;
+}
+
+/* Same as the \a pending_pick struct but for ping operations */
+typedef struct pending_ping {
+ struct pending_ping* next;
+
+ /* args for sending the ping */
+ wrapped_rr_closure_arg* on_initiate;
+ wrapped_rr_closure_arg* on_ack;
+} pending_ping;
+
+static void add_pending_ping(pending_ping** root, grpc_closure* on_initiate,
+ grpc_closure* on_ack) {
+ pending_ping* pping = (pending_ping*)gpr_zalloc(sizeof(*pping));
+ if (on_initiate != nullptr) {
+ pping->on_initiate =
+ (wrapped_rr_closure_arg*)gpr_zalloc(sizeof(*pping->on_initiate));
+ pping->on_initiate->wrapped_closure = on_initiate;
+ pping->on_initiate->free_when_done = pping->on_initiate;
+ GRPC_CLOSURE_INIT(&pping->on_initiate->wrapper_closure, wrapped_rr_closure,
+ &pping->on_initiate, grpc_schedule_on_exec_ctx);
+ }
+ if (on_ack != nullptr) {
+ pping->on_ack = (wrapped_rr_closure_arg*)gpr_zalloc(sizeof(*pping->on_ack));
+ pping->on_ack->wrapped_closure = on_ack;
+ pping->on_ack->free_when_done = pping->on_ack;
+ GRPC_CLOSURE_INIT(&pping->on_ack->wrapper_closure, wrapped_rr_closure,
+ &pping->on_ack, grpc_schedule_on_exec_ctx);
+ }
+ pping->next = *root;
+ *root = pping;
+}
+
+/*
+ * glb_lb_policy
+ */
+typedef struct rr_connectivity_data rr_connectivity_data;
+
+typedef struct glb_lb_policy {
/** base policy: must be first */
grpc_lb_policy base;
@@ -194,9 +333,6 @@
/** the RR policy to use of the backend servers returned by the LB server */
grpc_lb_policy* rr_policy;
- grpc_closure on_rr_connectivity_changed;
- grpc_connectivity_state rr_connectivity_state;
-
bool started_picking;
/** our connectivity state tracker */
@@ -301,85 +437,15 @@
grpc_closure client_load_report_closure;
/* Client load report message payload. */
grpc_byte_buffer* client_load_report_payload;
+} glb_lb_policy;
+
+/* Keeps track and reacts to changes in connectivity of the RR instance */
+struct rr_connectivity_data {
+ grpc_closure on_change;
+ grpc_connectivity_state state;
+ glb_lb_policy* glb_policy;
};
-/* add lb_token of selected subchannel (address) to the call's initial
- * metadata */
-static grpc_error* initial_metadata_add_lb_token(
- grpc_metadata_batch* initial_metadata,
- grpc_linked_mdelem* lb_token_mdelem_storage, grpc_mdelem lb_token) {
- GPR_ASSERT(lb_token_mdelem_storage != nullptr);
- GPR_ASSERT(!GRPC_MDISNULL(lb_token));
- return grpc_metadata_batch_add_tail(initial_metadata, lb_token_mdelem_storage,
- lb_token);
-}
-
-static void destroy_client_stats(void* arg) {
- grpc_grpclb_client_stats_unref((grpc_grpclb_client_stats*)arg);
-}
-
-static void pending_pick_set_metadata_and_context(pending_pick* pp) {
- /* if connected_subchannel is nullptr, no pick has been made by the RR
- * policy (e.g., all addresses failed to connect). There won't be any
- * user_data/token available */
- if (pp->pick->connected_subchannel != nullptr) {
- if (!GRPC_MDISNULL(pp->lb_token)) {
- initial_metadata_add_lb_token(pp->pick->initial_metadata,
- &pp->pick->lb_token_mdelem_storage,
- GRPC_MDELEM_REF(pp->lb_token));
- } else {
- gpr_log(GPR_ERROR,
- "[grpclb %p] No LB token for connected subchannel pick %p",
- pp->glb_policy, pp->pick);
- abort();
- }
- // Pass on client stats via context. Passes ownership of the reference.
- GPR_ASSERT(pp->client_stats != nullptr);
- pp->pick->subchannel_call_context[GRPC_GRPCLB_CLIENT_STATS].value =
- pp->client_stats;
- pp->pick->subchannel_call_context[GRPC_GRPCLB_CLIENT_STATS].destroy =
- destroy_client_stats;
- } else {
- grpc_grpclb_client_stats_unref(pp->client_stats);
- }
-}
-
-/* The \a on_complete closure passed as part of the pick requires keeping a
- * reference to its associated round robin instance. We wrap this closure in
- * order to unref the round robin instance upon its invocation */
-static void pending_pick_complete(void* arg, grpc_error* error) {
- pending_pick* pp = (pending_pick*)arg;
- pending_pick_set_metadata_and_context(pp);
- GRPC_CLOSURE_SCHED(pp->original_on_complete, GRPC_ERROR_REF(error));
- gpr_free(pp);
-}
-
-static pending_pick* pending_pick_create(glb_lb_policy* glb_policy,
- grpc_lb_policy_pick_state* pick) {
- pending_pick* pp = (pending_pick*)gpr_zalloc(sizeof(*pp));
- pp->pick = pick;
- pp->glb_policy = glb_policy;
- GRPC_CLOSURE_INIT(&pp->on_complete, pending_pick_complete, pp,
- grpc_schedule_on_exec_ctx);
- pp->original_on_complete = pick->on_complete;
- pp->pick->on_complete = &pp->on_complete;
- return pp;
-}
-
-static void pending_pick_add(pending_pick** root, pending_pick* new_pp) {
- new_pp->next = *root;
- *root = new_pp;
-}
-
-static void pending_ping_add(pending_ping** root, grpc_closure* on_initiate,
- grpc_closure* on_ack) {
- pending_ping* pping = (pending_ping*)gpr_zalloc(sizeof(*pping));
- pping->on_initiate = on_initiate;
- pping->on_ack = on_ack;
- pping->next = *root;
- *root = pping;
-}
-
static bool is_server_valid(const grpc_grpclb_server* server, size_t idx,
bool log) {
if (server->drop) return false;
@@ -491,6 +557,7 @@
gpr_free(uri);
user_data = (void*)GRPC_MDELEM_LB_TOKEN_EMPTY.payload;
}
+
grpc_lb_addresses_set_address(lb_addresses, addr_idx, &addr.addr, addr.len,
false /* is_balancer */,
nullptr /* balancer_name */, user_data);
@@ -531,6 +598,7 @@
grpc_error* rr_state_error) {
const grpc_connectivity_state curr_glb_state =
grpc_connectivity_state_check(&glb_policy->state_tracker);
+
/* The new connectivity status is a function of the previous one and the new
* input coming from the status of the RR policy.
*
@@ -560,6 +628,7 @@
*
* (*) This function mustn't be called during shutting down. */
GPR_ASSERT(curr_glb_state != GRPC_CHANNEL_SHUTDOWN);
+
switch (rr_state) {
case GRPC_CHANNEL_TRANSIENT_FAILURE:
case GRPC_CHANNEL_SHUTDOWN:
@@ -570,6 +639,7 @@
case GRPC_CHANNEL_READY:
GPR_ASSERT(rr_state_error == GRPC_ERROR_NONE);
}
+
if (grpc_lb_glb_trace.enabled()) {
gpr_log(
GPR_INFO,
@@ -587,8 +657,10 @@
* cleanups this callback would otherwise be responsible for.
* If \a force_async is true, then we will manually schedule the
* completion callback even if the pick is available immediately. */
-static bool pick_from_internal_rr_locked(glb_lb_policy* glb_policy,
- bool force_async, pending_pick* pp) {
+static bool pick_from_internal_rr_locked(
+ glb_lb_policy* glb_policy, const grpc_lb_policy_pick_args* pick_args,
+ bool force_async, grpc_connected_subchannel** target,
+ wrapped_rr_closure_arg* wc_arg) {
// Check for drops if we are not using fallback backend addresses.
if (glb_policy->serverlist != nullptr) {
// Look at the index into the serverlist to see if we should drop this call.
@@ -598,36 +670,57 @@
glb_policy->serverlist_index = 0; // Wrap-around.
}
if (server->drop) {
+ // Not using the RR policy, so unref it.
+ if (grpc_lb_glb_trace.enabled()) {
+ gpr_log(GPR_INFO, "[grpclb %p] Unreffing RR %p for drop", glb_policy,
+ wc_arg->rr_policy);
+ }
+ GRPC_LB_POLICY_UNREF(wc_arg->rr_policy, "glb_pick_sync");
// Update client load reporting stats to indicate the number of
// dropped calls. Note that we have to do this here instead of in
// the client_load_reporting filter, because we do not create a
// subchannel call (and therefore no client_load_reporting filter)
// for dropped calls.
- GPR_ASSERT(glb_policy->client_stats != nullptr);
+ GPR_ASSERT(wc_arg->client_stats != nullptr);
grpc_grpclb_client_stats_add_call_dropped_locked(
- server->load_balance_token, glb_policy->client_stats);
+ server->load_balance_token, wc_arg->client_stats);
+ grpc_grpclb_client_stats_unref(wc_arg->client_stats);
if (force_async) {
- GRPC_CLOSURE_SCHED(pp->original_on_complete, GRPC_ERROR_NONE);
- gpr_free(pp);
+ GPR_ASSERT(wc_arg->wrapped_closure != nullptr);
+ GRPC_CLOSURE_SCHED(wc_arg->wrapped_closure, GRPC_ERROR_NONE);
+ gpr_free(wc_arg->free_when_done);
return false;
}
- gpr_free(pp);
+ gpr_free(wc_arg->free_when_done);
return true;
}
}
- // Set client_stats and user_data.
- pp->client_stats = grpc_grpclb_client_stats_ref(glb_policy->client_stats);
- GPR_ASSERT(pp->pick->user_data == nullptr);
- pp->pick->user_data = (void**)&pp->lb_token;
// Pick via the RR policy.
- bool pick_done = grpc_lb_policy_pick_locked(glb_policy->rr_policy, pp->pick);
+ const bool pick_done = grpc_lb_policy_pick_locked(
+ wc_arg->rr_policy, pick_args, target, wc_arg->context,
+ (void**)&wc_arg->lb_token, &wc_arg->wrapper_closure);
if (pick_done) {
- pending_pick_set_metadata_and_context(pp);
- if (force_async) {
- GRPC_CLOSURE_SCHED(pp->original_on_complete, GRPC_ERROR_NONE);
- pick_done = false;
+ /* synchronous grpc_lb_policy_pick call. Unref the RR policy. */
+ if (grpc_lb_glb_trace.enabled()) {
+ gpr_log(GPR_INFO, "[grpclb %p] Unreffing RR %p", glb_policy,
+ wc_arg->rr_policy);
}
- gpr_free(pp);
+ GRPC_LB_POLICY_UNREF(wc_arg->rr_policy, "glb_pick_sync");
+ /* add the load reporting initial metadata */
+ initial_metadata_add_lb_token(pick_args->initial_metadata,
+ pick_args->lb_token_mdelem_storage,
+ GRPC_MDELEM_REF(wc_arg->lb_token));
+ // Pass on client stats via context. Passes ownership of the reference.
+ GPR_ASSERT(wc_arg->client_stats != nullptr);
+ wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].value = wc_arg->client_stats;
+ wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].destroy = destroy_client_stats;
+ if (force_async) {
+ GPR_ASSERT(wc_arg->wrapped_closure != nullptr);
+ GRPC_CLOSURE_SCHED(wc_arg->wrapped_closure, GRPC_ERROR_NONE);
+ gpr_free(wc_arg->free_when_done);
+ return false;
+ }
+ gpr_free(wc_arg->free_when_done);
}
/* else, the pending pick will be registered and taken care of by the
* pending pick list inside the RR policy (glb_policy->rr_policy).
@@ -669,7 +762,7 @@
gpr_free(args);
}
-static void on_rr_connectivity_changed_locked(void* arg, grpc_error* error);
+static void glb_rr_connectivity_changed_locked(void* arg, grpc_error* error);
static void create_rr_locked(glb_lb_policy* glb_policy,
grpc_lb_policy_args* args) {
GPR_ASSERT(glb_policy->rr_policy == nullptr);
@@ -691,46 +784,72 @@
glb_policy->base.request_reresolution = nullptr;
glb_policy->rr_policy = new_rr_policy;
grpc_error* rr_state_error = nullptr;
- glb_policy->rr_connectivity_state = grpc_lb_policy_check_connectivity_locked(
- glb_policy->rr_policy, &rr_state_error);
+ const grpc_connectivity_state rr_state =
+ grpc_lb_policy_check_connectivity_locked(glb_policy->rr_policy,
+ &rr_state_error);
/* Connectivity state is a function of the RR policy updated/created */
- update_lb_connectivity_status_locked(
- glb_policy, glb_policy->rr_connectivity_state, rr_state_error);
+ update_lb_connectivity_status_locked(glb_policy, rr_state, rr_state_error);
/* Add the gRPC LB's interested_parties pollset_set to that of the newly
* created RR policy. This will make the RR policy progress upon activity on
* gRPC LB, which in turn is tied to the application's call */
grpc_pollset_set_add_pollset_set(glb_policy->rr_policy->interested_parties,
glb_policy->base.interested_parties);
- GRPC_CLOSURE_INIT(&glb_policy->on_rr_connectivity_changed,
- on_rr_connectivity_changed_locked, glb_policy,
+
+ /* Allocate the data for the tracking of the new RR policy's connectivity.
+ * It'll be deallocated in glb_rr_connectivity_changed() */
+ rr_connectivity_data* rr_connectivity =
+ (rr_connectivity_data*)gpr_zalloc(sizeof(rr_connectivity_data));
+ GRPC_CLOSURE_INIT(&rr_connectivity->on_change,
+ glb_rr_connectivity_changed_locked, rr_connectivity,
grpc_combiner_scheduler(glb_policy->base.combiner));
+ rr_connectivity->glb_policy = glb_policy;
+ rr_connectivity->state = rr_state;
+
/* Subscribe to changes to the connectivity of the new RR */
- GRPC_LB_POLICY_REF(&glb_policy->base, "glb_rr_connectivity_cb");
- grpc_lb_policy_notify_on_state_change_locked(
- glb_policy->rr_policy, &glb_policy->rr_connectivity_state,
- &glb_policy->on_rr_connectivity_changed);
+ GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "glb_rr_connectivity_cb");
+ grpc_lb_policy_notify_on_state_change_locked(glb_policy->rr_policy,
+ &rr_connectivity->state,
+ &rr_connectivity->on_change);
grpc_lb_policy_exit_idle_locked(glb_policy->rr_policy);
- // Send pending picks to RR policy.
+
+ /* Update picks and pings in wait */
pending_pick* pp;
while ((pp = glb_policy->pending_picks)) {
glb_policy->pending_picks = pp->next;
+ GRPC_LB_POLICY_REF(glb_policy->rr_policy, "rr_handover_pending_pick");
+ pp->wrapped_on_complete_arg.rr_policy = glb_policy->rr_policy;
+ pp->wrapped_on_complete_arg.client_stats =
+ grpc_grpclb_client_stats_ref(glb_policy->client_stats);
if (grpc_lb_glb_trace.enabled()) {
gpr_log(GPR_INFO,
"[grpclb %p] Pending pick about to (async) PICK from RR %p",
glb_policy, glb_policy->rr_policy);
}
- pick_from_internal_rr_locked(glb_policy, true /* force_async */, pp);
+ pick_from_internal_rr_locked(glb_policy, &pp->pick_args,
+ true /* force_async */, pp->target,
+ &pp->wrapped_on_complete_arg);
}
- // Send pending pings to RR policy.
+
pending_ping* pping;
while ((pping = glb_policy->pending_pings)) {
glb_policy->pending_pings = pping->next;
+ grpc_closure* on_initiate = nullptr;
+ grpc_closure* on_ack = nullptr;
+ if (pping->on_initiate != nullptr) {
+ GRPC_LB_POLICY_REF(glb_policy->rr_policy, "rr_handover_pending_ping");
+ pping->on_initiate->rr_policy = glb_policy->rr_policy;
+ on_initiate = &pping->on_initiate->wrapper_closure;
+ }
+ if (pping->on_ack != nullptr) {
+ GRPC_LB_POLICY_REF(glb_policy->rr_policy, "rr_handover_pending_ping");
+ pping->on_ack->rr_policy = glb_policy->rr_policy;
+ on_ack = &pping->on_ack->wrapper_closure;
+ }
if (grpc_lb_glb_trace.enabled()) {
gpr_log(GPR_INFO, "[grpclb %p] Pending ping about to PING from RR %p",
glb_policy, glb_policy->rr_policy);
}
- grpc_lb_policy_ping_one_locked(glb_policy->rr_policy, pping->on_initiate,
- pping->on_ack);
+ grpc_lb_policy_ping_one_locked(glb_policy->rr_policy, on_initiate, on_ack);
gpr_free(pping);
}
}
@@ -756,28 +875,31 @@
lb_policy_args_destroy(args);
}
-static void on_rr_connectivity_changed_locked(void* arg, grpc_error* error) {
- glb_lb_policy* glb_policy = (glb_lb_policy*)arg;
+static void glb_rr_connectivity_changed_locked(void* arg, grpc_error* error) {
+ rr_connectivity_data* rr_connectivity = (rr_connectivity_data*)arg;
+ glb_lb_policy* glb_policy = rr_connectivity->glb_policy;
if (glb_policy->shutting_down) {
- GRPC_LB_POLICY_UNREF(&glb_policy->base, "glb_rr_connectivity_cb");
+ GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, "glb_rr_connectivity_cb");
+ gpr_free(rr_connectivity);
return;
}
- if (glb_policy->rr_connectivity_state == GRPC_CHANNEL_SHUTDOWN) {
+ if (rr_connectivity->state == GRPC_CHANNEL_SHUTDOWN) {
/* An RR policy that has transitioned into the SHUTDOWN connectivity state
* should not be considered for picks or updates: the SHUTDOWN state is a
* sink, policies can't transition back from it. .*/
GRPC_LB_POLICY_UNREF(glb_policy->rr_policy, "rr_connectivity_shutdown");
glb_policy->rr_policy = nullptr;
- GRPC_LB_POLICY_UNREF(&glb_policy->base, "glb_rr_connectivity_cb");
+ GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, "glb_rr_connectivity_cb");
+ gpr_free(rr_connectivity);
return;
}
/* rr state != SHUTDOWN && !glb_policy->shutting down: biz as usual */
- update_lb_connectivity_status_locked(
- glb_policy, glb_policy->rr_connectivity_state, GRPC_ERROR_REF(error));
- /* Resubscribe. Reuse the "glb_rr_connectivity_cb" ref. */
- grpc_lb_policy_notify_on_state_change_locked(
- glb_policy->rr_policy, &glb_policy->rr_connectivity_state,
- &glb_policy->on_rr_connectivity_changed);
+ update_lb_connectivity_status_locked(glb_policy, rr_connectivity->state,
+ GRPC_ERROR_REF(error));
+ /* Resubscribe. Reuse the "glb_rr_connectivity_cb" weak ref. */
+ grpc_lb_policy_notify_on_state_change_locked(glb_policy->rr_policy,
+ &rr_connectivity->state,
+ &rr_connectivity->on_change);
}
static void destroy_balancer_name(void* balancer_name) {
@@ -885,17 +1007,22 @@
gpr_free(glb_policy);
}
-static void glb_shutdown_locked(grpc_lb_policy* pol,
- grpc_lb_policy* new_policy) {
+static void glb_shutdown_locked(grpc_lb_policy* pol) {
glb_lb_policy* glb_policy = (glb_lb_policy*)pol;
grpc_error* error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel shutdown");
glb_policy->shutting_down = true;
+
+ /* We need a copy of the lb_call pointer because we can't cancell the call
+ * while holding glb_policy->mu: lb_on_server_status_received, invoked due to
+ * the cancel, needs to acquire that same lock */
+ grpc_call* lb_call = glb_policy->lb_call;
+
/* glb_policy->lb_call and this local lb_call must be consistent at this point
* because glb_policy->lb_call is only assigned in lb_call_init_locked as part
* of query_for_backends_locked, which can only be invoked while
* glb_policy->shutting_down is false. */
- if (glb_policy->lb_call != nullptr) {
- grpc_call_cancel(glb_policy->lb_call, nullptr);
+ if (lb_call != nullptr) {
+ grpc_call_cancel(lb_call, nullptr);
/* lb_on_server_status_received will pick up the cancel and clean up */
}
if (glb_policy->retry_timer_callback_pending) {
@@ -904,8 +1031,12 @@
if (glb_policy->fallback_timer_callback_pending) {
grpc_timer_cancel(&glb_policy->lb_fallback_timer);
}
+
+ pending_pick* pp = glb_policy->pending_picks;
+ glb_policy->pending_picks = nullptr;
+ pending_ping* pping = glb_policy->pending_pings;
+ glb_policy->pending_pings = nullptr;
if (glb_policy->rr_policy != nullptr) {
- grpc_lb_policy_shutdown_locked(glb_policy->rr_policy, nullptr);
GRPC_LB_POLICY_UNREF(glb_policy->rr_policy, "glb_shutdown");
} else {
grpc_lb_policy_try_reresolve(pol, &grpc_lb_glb_trace, GRPC_ERROR_CANCELLED);
@@ -920,33 +1051,28 @@
}
grpc_connectivity_state_set(&glb_policy->state_tracker, GRPC_CHANNEL_SHUTDOWN,
GRPC_ERROR_REF(error), "glb_shutdown");
- // Clear pending picks.
- pending_pick* pp = glb_policy->pending_picks;
- glb_policy->pending_picks = nullptr;
+
while (pp != nullptr) {
pending_pick* next = pp->next;
- if (new_policy != nullptr) {
- // Hand pick over to new policy.
- grpc_grpclb_client_stats_unref(pp->client_stats);
- pp->pick->on_complete = pp->original_on_complete;
- if (grpc_lb_policy_pick_locked(new_policy, pp->pick)) {
- // Synchronous return; schedule callback.
- GRPC_CLOSURE_SCHED(pp->pick->on_complete, GRPC_ERROR_NONE);
- }
- gpr_free(pp);
- } else {
- pp->pick->connected_subchannel = nullptr;
- GRPC_CLOSURE_SCHED(&pp->on_complete, GRPC_ERROR_REF(error));
- }
+ *pp->target = nullptr;
+ GRPC_CLOSURE_SCHED(&pp->wrapped_on_complete_arg.wrapper_closure,
+ GRPC_ERROR_REF(error));
+ gpr_free(pp);
pp = next;
}
- // Clear pending pings.
- pending_ping* pping = glb_policy->pending_pings;
- glb_policy->pending_pings = nullptr;
+
while (pping != nullptr) {
pending_ping* next = pping->next;
- GRPC_CLOSURE_SCHED(pping->on_initiate, GRPC_ERROR_REF(error));
- GRPC_CLOSURE_SCHED(pping->on_ack, GRPC_ERROR_REF(error));
+ if (pping->on_initiate != nullptr) {
+ GRPC_CLOSURE_SCHED(&pping->on_initiate->wrapper_closure,
+ GRPC_ERROR_REF(error));
+ gpr_free(pping->on_initiate);
+ }
+ if (pping->on_ack != nullptr) {
+ GRPC_CLOSURE_SCHED(&pping->on_ack->wrapper_closure,
+ GRPC_ERROR_REF(error));
+ gpr_free(pping->on_ack);
+ }
gpr_free(pping);
pping = next;
}
@@ -964,16 +1090,16 @@
// level (grpclb), inside the glb_policy->pending_picks list. To cancel these,
// we invoke the completion closure and set *target to nullptr right here.
static void glb_cancel_pick_locked(grpc_lb_policy* pol,
- grpc_lb_policy_pick_state* pick,
+ grpc_connected_subchannel** target,
grpc_error* error) {
glb_lb_policy* glb_policy = (glb_lb_policy*)pol;
pending_pick* pp = glb_policy->pending_picks;
glb_policy->pending_picks = nullptr;
while (pp != nullptr) {
pending_pick* next = pp->next;
- if (pp->pick == pick) {
- pick->connected_subchannel = nullptr;
- GRPC_CLOSURE_SCHED(&pp->on_complete,
+ if (pp->target == target) {
+ *target = nullptr;
+ GRPC_CLOSURE_SCHED(&pp->wrapped_on_complete_arg.wrapper_closure,
GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
"Pick Cancelled", &error, 1));
} else {
@@ -983,7 +1109,7 @@
pp = next;
}
if (glb_policy->rr_policy != nullptr) {
- grpc_lb_policy_cancel_pick_locked(glb_policy->rr_policy, pick,
+ grpc_lb_policy_cancel_pick_locked(glb_policy->rr_policy, target,
GRPC_ERROR_REF(error));
}
GRPC_ERROR_UNREF(error);
@@ -1008,9 +1134,9 @@
glb_policy->pending_picks = nullptr;
while (pp != nullptr) {
pending_pick* next = pp->next;
- if ((pp->pick->initial_metadata_flags & initial_metadata_flags_mask) ==
+ if ((pp->pick_args.initial_metadata_flags & initial_metadata_flags_mask) ==
initial_metadata_flags_eq) {
- GRPC_CLOSURE_SCHED(&pp->on_complete,
+ GRPC_CLOSURE_SCHED(&pp->wrapped_on_complete_arg.wrapper_closure,
GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
"Pick Cancelled", &error, 1));
} else {
@@ -1036,7 +1162,7 @@
!glb_policy->fallback_timer_callback_pending) {
grpc_millis deadline =
grpc_core::ExecCtx::Get()->Now() + glb_policy->lb_fallback_timeout_ms;
- GRPC_LB_POLICY_REF(&glb_policy->base, "grpclb_fallback_timer");
+ GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "grpclb_fallback_timer");
GRPC_CLOSURE_INIT(&glb_policy->lb_on_fallback, lb_on_fallback_timer_locked,
glb_policy,
grpc_combiner_scheduler(glb_policy->base.combiner));
@@ -1058,9 +1184,19 @@
}
static int glb_pick_locked(grpc_lb_policy* pol,
- grpc_lb_policy_pick_state* pick) {
+ const grpc_lb_policy_pick_args* pick_args,
+ grpc_connected_subchannel** target,
+ grpc_call_context_element* context, void** user_data,
+ grpc_closure* on_complete) {
+ if (pick_args->lb_token_mdelem_storage == nullptr) {
+ *target = nullptr;
+ GRPC_CLOSURE_SCHED(on_complete,
+ GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+ "No mdelem storage for the LB token. Load reporting "
+ "won't work without it. Failing"));
+ return 0;
+ }
glb_lb_policy* glb_policy = (glb_lb_policy*)pol;
- pending_pick* pp = pending_pick_create(glb_policy, pick);
bool pick_done = false;
if (glb_policy->rr_policy != nullptr) {
const grpc_connectivity_state rr_connectivity_state =
@@ -1068,7 +1204,7 @@
nullptr);
// The glb_policy->rr_policy may have transitioned to SHUTDOWN but the
// callback registered to capture this event
- // (on_rr_connectivity_changed_locked) may not have been invoked yet. We
+ // (glb_rr_connectivity_changed_locked) may not have been invoked yet. We
// need to make sure we aren't trying to pick from a RR policy instance
// that's in shutdown.
if (rr_connectivity_state == GRPC_CHANNEL_SHUTDOWN) {
@@ -1078,16 +1214,32 @@
glb_policy, glb_policy->rr_policy,
grpc_connectivity_state_name(rr_connectivity_state));
}
- pending_pick_add(&glb_policy->pending_picks, pp);
+ add_pending_pick(&glb_policy->pending_picks, pick_args, target, context,
+ on_complete);
pick_done = false;
} else { // RR not in shutdown
if (grpc_lb_glb_trace.enabled()) {
gpr_log(GPR_INFO, "[grpclb %p] about to PICK from RR %p", glb_policy,
glb_policy->rr_policy);
}
+ GRPC_LB_POLICY_REF(glb_policy->rr_policy, "glb_pick");
+ wrapped_rr_closure_arg* wc_arg =
+ (wrapped_rr_closure_arg*)gpr_zalloc(sizeof(wrapped_rr_closure_arg));
+ GRPC_CLOSURE_INIT(&wc_arg->wrapper_closure, wrapped_rr_closure, wc_arg,
+ grpc_schedule_on_exec_ctx);
+ wc_arg->rr_policy = glb_policy->rr_policy;
+ wc_arg->target = target;
+ wc_arg->context = context;
GPR_ASSERT(glb_policy->client_stats != nullptr);
- pick_done =
- pick_from_internal_rr_locked(glb_policy, false /* force_async */, pp);
+ wc_arg->client_stats =
+ grpc_grpclb_client_stats_ref(glb_policy->client_stats);
+ wc_arg->wrapped_closure = on_complete;
+ wc_arg->lb_token_mdelem_storage = pick_args->lb_token_mdelem_storage;
+ wc_arg->initial_metadata = pick_args->initial_metadata;
+ wc_arg->free_when_done = wc_arg;
+ wc_arg->glb_policy = pol;
+ pick_done = pick_from_internal_rr_locked(
+ glb_policy, pick_args, false /* force_async */, target, wc_arg);
}
} else { // glb_policy->rr_policy == NULL
if (grpc_lb_glb_trace.enabled()) {
@@ -1095,7 +1247,8 @@
"[grpclb %p] No RR policy. Adding to grpclb's pending picks",
glb_policy);
}
- pending_pick_add(&glb_policy->pending_picks, pp);
+ add_pending_pick(&glb_policy->pending_picks, pick_args, target, context,
+ on_complete);
if (!glb_policy->started_picking) {
start_picking_locked(glb_policy);
}
@@ -1117,7 +1270,7 @@
if (glb_policy->rr_policy) {
grpc_lb_policy_ping_one_locked(glb_policy->rr_policy, on_initiate, on_ack);
} else {
- pending_ping_add(&glb_policy->pending_pings, on_initiate, on_ack);
+ add_pending_ping(&glb_policy->pending_pings, on_initiate, on_ack);
if (!glb_policy->started_picking) {
start_picking_locked(glb_policy);
}
@@ -1142,7 +1295,7 @@
}
query_for_backends_locked(glb_policy);
}
- GRPC_LB_POLICY_UNREF(&glb_policy->base, "grpclb_retry_timer");
+ GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, "grpclb_retry_timer");
}
static void maybe_restart_lb_call(glb_lb_policy* glb_policy) {
@@ -1168,7 +1321,7 @@
glb_policy);
}
}
- GRPC_LB_POLICY_REF(&glb_policy->base, "grpclb_retry_timer");
+ GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "grpclb_retry_timer");
GRPC_CLOSURE_INIT(&glb_policy->lb_on_call_retry,
lb_call_on_retry_timer_locked, glb_policy,
grpc_combiner_scheduler(glb_policy->base.combiner));
@@ -1176,8 +1329,8 @@
grpc_timer_init(&glb_policy->lb_call_retry_timer, next_try,
&glb_policy->lb_on_call_retry);
}
- GRPC_LB_POLICY_UNREF(&glb_policy->base,
- "lb_on_server_status_received_locked");
+ GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base,
+ "lb_on_server_status_received_locked");
}
static void send_client_load_report_locked(void* arg, grpc_error* error);
@@ -1200,7 +1353,7 @@
glb_policy->client_load_report_payload = nullptr;
if (error != GRPC_ERROR_NONE || glb_policy->lb_call == nullptr) {
glb_policy->client_load_report_timer_callback_pending = false;
- GRPC_LB_POLICY_UNREF(&glb_policy->base, "client_load_report");
+ GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, "client_load_report");
if (glb_policy->lb_call == nullptr) {
maybe_restart_lb_call(glb_policy);
}
@@ -1241,7 +1394,7 @@
glb_lb_policy* glb_policy = (glb_lb_policy*)arg;
if (error == GRPC_ERROR_CANCELLED || glb_policy->lb_call == nullptr) {
glb_policy->client_load_report_timer_callback_pending = false;
- GRPC_LB_POLICY_UNREF(&glb_policy->base, "client_load_report");
+ GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, "client_load_report");
if (glb_policy->lb_call == nullptr) {
maybe_restart_lb_call(glb_policy);
}
@@ -1394,8 +1547,10 @@
op->flags = 0;
op->reserved = nullptr;
op++;
- /* take a ref to be released in lb_on_sent_initial_request_locked() */
- GRPC_LB_POLICY_REF(&glb_policy->base, "lb_on_sent_initial_request_locked");
+ /* take a weak ref (won't prevent calling of \a glb_shutdown if the strong ref
+ * count goes to zero) to be unref'd in lb_on_sent_initial_request_locked() */
+ GRPC_LB_POLICY_WEAK_REF(&glb_policy->base,
+ "lb_on_sent_initial_request_locked");
call_error = grpc_call_start_batch_and_execute(
glb_policy->lb_call, ops, (size_t)(op - ops),
&glb_policy->lb_on_sent_initial_request);
@@ -1411,8 +1566,10 @@
op->flags = 0;
op->reserved = nullptr;
op++;
- /* take a ref to be released in lb_on_server_status_received_locked() */
- GRPC_LB_POLICY_REF(&glb_policy->base, "lb_on_server_status_received_locked");
+ /* take a weak ref (won't prevent calling of \a glb_shutdown if the strong ref
+ * count goes to zero) to be unref'd in lb_on_server_status_received_locked */
+ GRPC_LB_POLICY_WEAK_REF(&glb_policy->base,
+ "lb_on_server_status_received_locked");
call_error = grpc_call_start_batch_and_execute(
glb_policy->lb_call, ops, (size_t)(op - ops),
&glb_policy->lb_on_server_status_received);
@@ -1424,8 +1581,9 @@
op->flags = 0;
op->reserved = nullptr;
op++;
- /* take a ref to be unref'd/reused in lb_on_response_received_locked() */
- GRPC_LB_POLICY_REF(&glb_policy->base, "lb_on_response_received_locked");
+ /* take another weak ref to be unref'd/reused in
+ * lb_on_response_received_locked */
+ GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "lb_on_response_received_locked");
call_error = grpc_call_start_batch_and_execute(
glb_policy->lb_call, ops, (size_t)(op - ops),
&glb_policy->lb_on_response_received);
@@ -1440,7 +1598,8 @@
if (glb_policy->client_load_report_payload != nullptr) {
do_send_client_load_report_locked(glb_policy);
}
- GRPC_LB_POLICY_UNREF(&glb_policy->base, "lb_on_sent_initial_request_locked");
+ GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base,
+ "lb_on_sent_initial_request_locked");
}
static void lb_on_response_received_locked(void* arg, grpc_error* error) {
@@ -1472,9 +1631,11 @@
"client load reporting interval = %" PRIdPTR " milliseconds",
glb_policy, glb_policy->client_stats_report_interval);
}
- /* take a ref to be unref'd in send_client_load_report_locked() */
+ /* take a weak ref (won't prevent calling of \a glb_shutdown() if the
+ * strong ref count goes to zero) to be unref'd in
+ * send_client_load_report_locked() */
glb_policy->client_load_report_timer_callback_pending = true;
- GRPC_LB_POLICY_REF(&glb_policy->base, "client_load_report");
+ GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "client_load_report");
schedule_next_client_load_report(glb_policy);
} else if (grpc_lb_glb_trace.enabled()) {
gpr_log(GPR_INFO,
@@ -1556,21 +1717,21 @@
op->flags = 0;
op->reserved = nullptr;
op++;
- /* reuse the "lb_on_response_received_locked" ref taken in
+ /* reuse the "lb_on_response_received_locked" weak ref taken in
* query_for_backends_locked() */
const grpc_call_error call_error = grpc_call_start_batch_and_execute(
glb_policy->lb_call, ops, (size_t)(op - ops),
&glb_policy->lb_on_response_received); /* loop */
GPR_ASSERT(GRPC_CALL_OK == call_error);
} else {
- GRPC_LB_POLICY_UNREF(&glb_policy->base,
- "lb_on_response_received_locked_shutdown");
+ GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base,
+ "lb_on_response_received_locked_shutdown");
}
} else { /* empty payload: call cancelled. */
- /* dispose of the "lb_on_response_received_locked" ref taken in
+ /* dispose of the "lb_on_response_received_locked" weak ref taken in
* query_for_backends_locked() and reused in every reception loop */
- GRPC_LB_POLICY_UNREF(&glb_policy->base,
- "lb_on_response_received_locked_empty_payload");
+ GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base,
+ "lb_on_response_received_locked_empty_payload");
}
}
@@ -1590,7 +1751,7 @@
rr_handover_locked(glb_policy);
}
}
- GRPC_LB_POLICY_UNREF(&glb_policy->base, "grpclb_fallback_timer");
+ GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base, "grpclb_fallback_timer");
}
static void lb_on_server_status_received_locked(void* arg, grpc_error* error) {
@@ -1674,7 +1835,7 @@
grpc_channel_get_channel_stack(glb_policy->lb_channel));
GPR_ASSERT(client_channel_elem->filter == &grpc_client_channel_filter);
glb_policy->watching_lb_channel = true;
- GRPC_LB_POLICY_REF(&glb_policy->base, "watch_lb_channel_connectivity");
+ GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "watch_lb_channel_connectivity");
grpc_client_channel_watch_connectivity_state(
client_channel_elem,
grpc_polling_entity_create_from_pollset_set(
@@ -1730,8 +1891,8 @@
case GRPC_CHANNEL_SHUTDOWN:
done:
glb_policy->watching_lb_channel = false;
- GRPC_LB_POLICY_UNREF(&glb_policy->base,
- "watch_lb_channel_connectivity_cb_shutdown");
+ GRPC_LB_POLICY_WEAK_UNREF(&glb_policy->base,
+ "watch_lb_channel_connectivity_cb_shutdown");
break;
}
}
diff --git a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc
index 6038527..9ff40aa 100644
--- a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc
+++ b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc
@@ -31,6 +31,15 @@
grpc_core::TraceFlag grpc_lb_pick_first_trace(false, "pick_first");
+namespace {
+struct pending_pick {
+ struct pending_pick* next;
+ uint32_t initial_metadata_flags;
+ grpc_connected_subchannel** target;
+ grpc_closure* on_complete;
+};
+} // namespace
+
typedef struct {
/** base policy: must be first */
grpc_lb_policy base;
@@ -45,7 +54,7 @@
/** are we shut down? */
bool shutdown;
/** list of picks that are waiting on connectivity */
- grpc_lb_policy_pick_state* pending_picks;
+ pending_pick* pending_picks;
/** our connectivity state tracker */
grpc_connectivity_state_tracker state_tracker;
} pick_first_lb_policy;
@@ -63,27 +72,19 @@
}
}
-static void pf_shutdown_locked(grpc_lb_policy* pol,
- grpc_lb_policy* new_policy) {
+static void pf_shutdown_locked(grpc_lb_policy* pol) {
pick_first_lb_policy* p = (pick_first_lb_policy*)pol;
grpc_error* error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel shutdown");
if (grpc_lb_pick_first_trace.enabled()) {
gpr_log(GPR_DEBUG, "Pick First %p Shutting down", p);
}
p->shutdown = true;
- grpc_lb_policy_pick_state* pick;
- while ((pick = p->pending_picks) != nullptr) {
- p->pending_picks = pick->next;
- if (new_policy != nullptr) {
- // Hand off to new LB policy.
- if (grpc_lb_policy_pick_locked(new_policy, pick)) {
- // Synchronous return, schedule closure.
- GRPC_CLOSURE_SCHED(pick->on_complete, GRPC_ERROR_NONE);
- }
- } else {
- pick->connected_subchannel = nullptr;
- GRPC_CLOSURE_SCHED(pick->on_complete, GRPC_ERROR_REF(error));
- }
+ pending_pick* pp;
+ while ((pp = p->pending_picks) != nullptr) {
+ p->pending_picks = pp->next;
+ *pp->target = nullptr;
+ GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_REF(error));
+ gpr_free(pp);
}
grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_SHUTDOWN,
GRPC_ERROR_REF(error), "shutdown");
@@ -103,18 +104,19 @@
}
static void pf_cancel_pick_locked(grpc_lb_policy* pol,
- grpc_lb_policy_pick_state* pick,
+ grpc_connected_subchannel** target,
grpc_error* error) {
pick_first_lb_policy* p = (pick_first_lb_policy*)pol;
- grpc_lb_policy_pick_state* pp = p->pending_picks;
+ pending_pick* pp = p->pending_picks;
p->pending_picks = nullptr;
while (pp != nullptr) {
- grpc_lb_policy_pick_state* next = pp->next;
- if (pp == pick) {
- pick->connected_subchannel = nullptr;
- GRPC_CLOSURE_SCHED(pick->on_complete,
+ pending_pick* next = pp->next;
+ if (pp->target == target) {
+ *target = nullptr;
+ GRPC_CLOSURE_SCHED(pp->on_complete,
GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
"Pick Cancelled", &error, 1));
+ gpr_free(pp);
} else {
pp->next = p->pending_picks;
p->pending_picks = pp;
@@ -129,20 +131,21 @@
uint32_t initial_metadata_flags_eq,
grpc_error* error) {
pick_first_lb_policy* p = (pick_first_lb_policy*)pol;
- grpc_lb_policy_pick_state* pick = p->pending_picks;
+ pending_pick* pp = p->pending_picks;
p->pending_picks = nullptr;
- while (pick != nullptr) {
- grpc_lb_policy_pick_state* next = pick->next;
- if ((pick->initial_metadata_flags & initial_metadata_flags_mask) ==
+ while (pp != nullptr) {
+ pending_pick* next = pp->next;
+ if ((pp->initial_metadata_flags & initial_metadata_flags_mask) ==
initial_metadata_flags_eq) {
- GRPC_CLOSURE_SCHED(pick->on_complete,
+ GRPC_CLOSURE_SCHED(pp->on_complete,
GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
"Pick Cancelled", &error, 1));
+ gpr_free(pp);
} else {
- pick->next = p->pending_picks;
- p->pending_picks = pick;
+ pp->next = p->pending_picks;
+ p->pending_picks = pp;
}
- pick = next;
+ pp = next;
}
GRPC_ERROR_UNREF(error);
}
@@ -172,20 +175,27 @@
}
static int pf_pick_locked(grpc_lb_policy* pol,
- grpc_lb_policy_pick_state* pick) {
+ const grpc_lb_policy_pick_args* pick_args,
+ grpc_connected_subchannel** target,
+ grpc_call_context_element* context, void** user_data,
+ grpc_closure* on_complete) {
pick_first_lb_policy* p = (pick_first_lb_policy*)pol;
// If we have a selected subchannel already, return synchronously.
if (p->selected != nullptr) {
- pick->connected_subchannel = GRPC_CONNECTED_SUBCHANNEL_REF(
- p->selected->connected_subchannel, "picked");
+ *target = GRPC_CONNECTED_SUBCHANNEL_REF(p->selected->connected_subchannel,
+ "picked");
return 1;
}
// No subchannel selected yet, so handle asynchronously.
if (!p->started_picking) {
start_picking_locked(p);
}
- pick->next = p->pending_picks;
- p->pending_picks = pick;
+ pending_pick* pp = (pending_pick*)gpr_malloc(sizeof(*pp));
+ pp->next = p->pending_picks;
+ pp->target = target;
+ pp->initial_metadata_flags = pick_args->initial_metadata_flags;
+ pp->on_complete = on_complete;
+ p->pending_picks = pp;
return 0;
}
@@ -471,17 +481,18 @@
// Drop all other subchannels, since we are now connected.
destroy_unselected_subchannels_locked(p);
// Update any calls that were waiting for a pick.
- grpc_lb_policy_pick_state* pick;
- while ((pick = p->pending_picks)) {
- p->pending_picks = pick->next;
- pick->connected_subchannel = GRPC_CONNECTED_SUBCHANNEL_REF(
+ pending_pick* pp;
+ while ((pp = p->pending_picks)) {
+ p->pending_picks = pp->next;
+ *pp->target = GRPC_CONNECTED_SUBCHANNEL_REF(
p->selected->connected_subchannel, "picked");
if (grpc_lb_pick_first_trace.enabled()) {
gpr_log(GPR_INFO,
"Servicing pending pick with selected subchannel %p",
(void*)p->selected);
}
- GRPC_CLOSURE_SCHED(pick->on_complete, GRPC_ERROR_NONE);
+ GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_NONE);
+ gpr_free(pp);
}
// Renew notification.
grpc_lb_subchannel_data_start_connectivity_watch(sd);
diff --git a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc
index 92c7d5b..a964af0 100644
--- a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc
+++ b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc
@@ -41,6 +41,31 @@
grpc_core::TraceFlag grpc_lb_round_robin_trace(false, "round_robin");
+namespace {
+/** List of entities waiting for a pick.
+ *
+ * Once a pick is available, \a target is updated and \a on_complete called. */
+struct pending_pick {
+ pending_pick* next;
+
+ /* output argument where to store the pick()ed user_data. It'll be NULL if no
+ * such data is present or there's an error (the definite test for errors is
+ * \a target being NULL). */
+ void** user_data;
+
+ /* bitmask passed to pick() and used for selective cancelling. See
+ * grpc_lb_policy_cancel_picks() */
+ uint32_t initial_metadata_flags;
+
+ /* output argument where to store the pick()ed connected subchannel, or NULL
+ * upon error. */
+ grpc_connected_subchannel** target;
+
+ /* to be invoked once the pick() has completed (regardless of success) */
+ grpc_closure* on_complete;
+};
+} // namespace
+
typedef struct round_robin_lb_policy {
/** base policy: must be first */
grpc_lb_policy base;
@@ -52,7 +77,7 @@
/** are we shutting down? */
bool shutdown;
/** List of picks that are waiting on connectivity */
- grpc_lb_policy_pick_state* pending_picks;
+ pending_pick* pending_picks;
/** our connectivity state tracker */
grpc_connectivity_state_tracker state_tracker;
@@ -144,27 +169,19 @@
gpr_free(p);
}
-static void rr_shutdown_locked(grpc_lb_policy* pol,
- grpc_lb_policy* new_policy) {
+static void rr_shutdown_locked(grpc_lb_policy* pol) {
round_robin_lb_policy* p = (round_robin_lb_policy*)pol;
grpc_error* error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Channel shutdown");
if (grpc_lb_round_robin_trace.enabled()) {
gpr_log(GPR_DEBUG, "[RR %p] Shutting down", p);
}
p->shutdown = true;
- grpc_lb_policy_pick_state* pick;
- while ((pick = p->pending_picks) != nullptr) {
- p->pending_picks = pick->next;
- if (new_policy != nullptr) {
- // Hand off to new LB policy.
- if (grpc_lb_policy_pick_locked(new_policy, pick)) {
- // Synchronous return; schedule callback.
- GRPC_CLOSURE_SCHED(pick->on_complete, GRPC_ERROR_NONE);
- }
- } else {
- pick->connected_subchannel = nullptr;
- GRPC_CLOSURE_SCHED(pick->on_complete, GRPC_ERROR_REF(error));
- }
+ pending_pick* pp;
+ while ((pp = p->pending_picks) != nullptr) {
+ p->pending_picks = pp->next;
+ *pp->target = nullptr;
+ GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_REF(error));
+ gpr_free(pp);
}
grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_SHUTDOWN,
GRPC_ERROR_REF(error), "rr_shutdown");
@@ -184,18 +201,19 @@
}
static void rr_cancel_pick_locked(grpc_lb_policy* pol,
- grpc_lb_policy_pick_state* pick,
+ grpc_connected_subchannel** target,
grpc_error* error) {
round_robin_lb_policy* p = (round_robin_lb_policy*)pol;
- grpc_lb_policy_pick_state* pp = p->pending_picks;
+ pending_pick* pp = p->pending_picks;
p->pending_picks = nullptr;
while (pp != nullptr) {
- grpc_lb_policy_pick_state* next = pp->next;
- if (pp == pick) {
- pick->connected_subchannel = nullptr;
- GRPC_CLOSURE_SCHED(pick->on_complete,
+ pending_pick* next = pp->next;
+ if (pp->target == target) {
+ *target = nullptr;
+ GRPC_CLOSURE_SCHED(pp->on_complete,
GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
"Pick cancelled", &error, 1));
+ gpr_free(pp);
} else {
pp->next = p->pending_picks;
p->pending_picks = pp;
@@ -210,21 +228,22 @@
uint32_t initial_metadata_flags_eq,
grpc_error* error) {
round_robin_lb_policy* p = (round_robin_lb_policy*)pol;
- grpc_lb_policy_pick_state* pick = p->pending_picks;
+ pending_pick* pp = p->pending_picks;
p->pending_picks = nullptr;
- while (pick != nullptr) {
- grpc_lb_policy_pick_state* next = pick->next;
- if ((pick->initial_metadata_flags & initial_metadata_flags_mask) ==
+ while (pp != nullptr) {
+ pending_pick* next = pp->next;
+ if ((pp->initial_metadata_flags & initial_metadata_flags_mask) ==
initial_metadata_flags_eq) {
- pick->connected_subchannel = nullptr;
- GRPC_CLOSURE_SCHED(pick->on_complete,
+ *pp->target = nullptr;
+ GRPC_CLOSURE_SCHED(pp->on_complete,
GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
"Pick cancelled", &error, 1));
+ gpr_free(pp);
} else {
- pick->next = p->pending_picks;
- p->pending_picks = pick;
+ pp->next = p->pending_picks;
+ p->pending_picks = pp;
}
- pick = next;
+ pp = next;
}
GRPC_ERROR_UNREF(error);
}
@@ -249,10 +268,13 @@
}
static int rr_pick_locked(grpc_lb_policy* pol,
- grpc_lb_policy_pick_state* pick) {
+ const grpc_lb_policy_pick_args* pick_args,
+ grpc_connected_subchannel** target,
+ grpc_call_context_element* context, void** user_data,
+ grpc_closure* on_complete) {
round_robin_lb_policy* p = (round_robin_lb_policy*)pol;
if (grpc_lb_round_robin_trace.enabled()) {
- gpr_log(GPR_INFO, "[RR %p] Trying to pick (shutdown: %d)", pol,
+ gpr_log(GPR_INFO, "[RR %p] Trying to pick (shutdown: %d)", (void*)pol,
p->shutdown);
}
GPR_ASSERT(!p->shutdown);
@@ -262,18 +284,18 @@
/* readily available, report right away */
grpc_lb_subchannel_data* sd =
&p->subchannel_list->subchannels[next_ready_index];
- pick->connected_subchannel =
+ *target =
GRPC_CONNECTED_SUBCHANNEL_REF(sd->connected_subchannel, "rr_picked");
- if (pick->user_data != nullptr) {
- *pick->user_data = sd->user_data;
+ if (user_data != nullptr) {
+ *user_data = sd->user_data;
}
if (grpc_lb_round_robin_trace.enabled()) {
gpr_log(
GPR_DEBUG,
"[RR %p] Picked target <-- Subchannel %p (connected %p) (sl %p, "
- "index %" PRIuPTR ")",
- p, sd->subchannel, pick->connected_subchannel, sd->subchannel_list,
- next_ready_index);
+ "index %lu)",
+ (void*)p, (void*)sd->subchannel, (void*)*target,
+ (void*)sd->subchannel_list, (unsigned long)next_ready_index);
}
/* only advance the last picked pointer if the selection was used */
update_last_ready_subchannel_index_locked(p, next_ready_index);
@@ -284,8 +306,13 @@
if (!p->started_picking) {
start_picking_locked(p);
}
- pick->next = p->pending_picks;
- p->pending_picks = pick;
+ pending_pick* pp = (pending_pick*)gpr_malloc(sizeof(*pp));
+ pp->next = p->pending_picks;
+ pp->target = target;
+ pp->on_complete = on_complete;
+ pp->initial_metadata_flags = pick_args->initial_metadata_flags;
+ pp->user_data = user_data;
+ p->pending_picks = pp;
return 0;
}
@@ -468,13 +495,13 @@
// picks, update the last picked pointer
update_last_ready_subchannel_index_locked(p, next_ready_index);
}
- grpc_lb_policy_pick_state* pick;
- while ((pick = p->pending_picks)) {
- p->pending_picks = pick->next;
- pick->connected_subchannel = GRPC_CONNECTED_SUBCHANNEL_REF(
+ pending_pick* pp;
+ while ((pp = p->pending_picks)) {
+ p->pending_picks = pp->next;
+ *pp->target = GRPC_CONNECTED_SUBCHANNEL_REF(
selected->connected_subchannel, "rr_picked");
- if (pick->user_data != nullptr) {
- *pick->user_data = selected->user_data;
+ if (pp->user_data != nullptr) {
+ *pp->user_data = selected->user_data;
}
if (grpc_lb_round_robin_trace.enabled()) {
gpr_log(GPR_DEBUG,
@@ -483,7 +510,8 @@
(void*)p, (void*)selected->subchannel,
(void*)p->subchannel_list, (unsigned long)next_ready_index);
}
- GRPC_CLOSURE_SCHED(pick->on_complete, GRPC_ERROR_NONE);
+ GRPC_CLOSURE_SCHED(pp->on_complete, GRPC_ERROR_NONE);
+ gpr_free(pp);
}
}
// Renew notification.
diff --git a/src/core/ext/filters/client_channel/lb_policy/subchannel_list.cc b/src/core/ext/filters/client_channel/lb_policy/subchannel_list.cc
index 5ce1298..a3b4c8e 100644
--- a/src/core/ext/filters/client_channel/lb_policy/subchannel_list.cc
+++ b/src/core/ext/filters/client_channel/lb_policy/subchannel_list.cc
@@ -213,13 +213,13 @@
void grpc_lb_subchannel_list_ref_for_connectivity_watch(
grpc_lb_subchannel_list* subchannel_list, const char* reason) {
- GRPC_LB_POLICY_REF(subchannel_list->policy, reason);
+ GRPC_LB_POLICY_WEAK_REF(subchannel_list->policy, reason);
grpc_lb_subchannel_list_ref(subchannel_list, reason);
}
void grpc_lb_subchannel_list_unref_for_connectivity_watch(
grpc_lb_subchannel_list* subchannel_list, const char* reason) {
- GRPC_LB_POLICY_UNREF(subchannel_list->policy, reason);
+ GRPC_LB_POLICY_WEAK_UNREF(subchannel_list->policy, reason);
grpc_lb_subchannel_list_unref(subchannel_list, reason);
}
diff --git a/tools/interop_matrix/client_matrix.py b/tools/interop_matrix/client_matrix.py
index 71d3a79..5923109 100644
--- a/tools/interop_matrix/client_matrix.py
+++ b/tools/interop_matrix/client_matrix.py
@@ -103,6 +103,9 @@
{
'v1.8.2': None
},
+ {
+ 'v1.9.1': None
+ },
],
'java': [
{