Merge branch 'shindig' of https://github.com/ctiller/grpc into shindig
Conflicts:
src/core/iomgr/tcp_client_windows.c
diff --git a/include/grpc++/security/auth_metadata_processor.h b/include/grpc++/security/auth_metadata_processor.h
index 18ad922..9b9c06e 100644
--- a/include/grpc++/security/auth_metadata_processor.h
+++ b/include/grpc++/security/auth_metadata_processor.h
@@ -45,7 +45,7 @@
class AuthMetadataProcessor {
public:
typedef std::multimap<grpc::string_ref, grpc::string_ref> InputMetadata;
- typedef std::multimap<grpc::string, grpc::string_ref> OutputMetadata;
+ typedef std::multimap<grpc::string, grpc::string> OutputMetadata;
virtual ~AuthMetadataProcessor() {}
diff --git a/include/grpc++/security/credentials.h b/include/grpc++/security/credentials.h
index fafcfdc..ff41bc5 100644
--- a/include/grpc++/security/credentials.h
+++ b/include/grpc++/security/credentials.h
@@ -180,7 +180,7 @@
// Gets the auth metatada produced by this plugin.
virtual Status GetMetadata(
grpc::string_ref service_url,
- std::multimap<grpc::string, grpc::string_ref>* metadata) = 0;
+ std::multimap<grpc::string, grpc::string>* metadata) = 0;
};
std::shared_ptr<Credentials> MetadataCredentialsFromPlugin(
diff --git a/src/core/channel/channel_args.h b/src/core/channel/channel_args.h
index f9e7b05..480cc9a 100644
--- a/src/core/channel/channel_args.h
+++ b/src/core/channel/channel_args.h
@@ -71,7 +71,7 @@
* compression algorithms are enabled. It's an error to disable an algorithm set
* by grpc_channel_args_set_compression_algorithm.
*
- * Returns an instance will the updated algorithm states. The \a a pointer is
+ * Returns an instance with the updated algorithm states. The \a a pointer is
* modified to point to the returned instance (which may be different from the
* input value of \a a). */
grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
diff --git a/src/core/client_config/resolvers/zookeeper_resolver.c b/src/core/client_config/resolvers/zookeeper_resolver.c
index 213d5a1..f640a00 100644
--- a/src/core/client_config/resolvers/zookeeper_resolver.c
+++ b/src/core/client_config/resolvers/zookeeper_resolver.c
@@ -87,24 +87,27 @@
int resolved_num;
} zookeeper_resolver;
-static void zookeeper_destroy(grpc_resolver *r);
+static void zookeeper_destroy(grpc_exec_ctx *exec_ctx, grpc_resolver *r);
static void zookeeper_start_resolving_locked(zookeeper_resolver *r);
-static grpc_closure *zookeeper_maybe_finish_next_locked(zookeeper_resolver *r)
- GRPC_MUST_USE_RESULT;
+static void zookeeper_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
+ zookeeper_resolver *r);
-static void zookeeper_shutdown(grpc_resolver *r);
-static void zookeeper_channel_saw_error(grpc_resolver *r,
+static void zookeeper_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *r);
+static void zookeeper_channel_saw_error(grpc_exec_ctx *exec_ctx,
+ grpc_resolver *r,
struct sockaddr *failing_address,
int failing_address_len);
-static void zookeeper_next(grpc_resolver *r, grpc_client_config **target_config,
+static void zookeeper_next(grpc_exec_ctx *exec_ctx, grpc_resolver *r,
+ grpc_client_config **target_config,
grpc_closure *on_complete);
static const grpc_resolver_vtable zookeeper_resolver_vtable = {
zookeeper_destroy, zookeeper_shutdown, zookeeper_channel_saw_error,
zookeeper_next};
-static void zookeeper_shutdown(grpc_resolver *resolver) {
+static void zookeeper_shutdown(grpc_exec_ctx *exec_ctx,
+ grpc_resolver *resolver) {
zookeeper_resolver *r = (zookeeper_resolver *)resolver;
grpc_closure *call = NULL;
gpr_mu_lock(&r->mu);
@@ -116,11 +119,12 @@
zookeeper_close(r->zookeeper_handle);
gpr_mu_unlock(&r->mu);
if (call != NULL) {
- call->cb(call->cb_arg, 1);
+ call->cb(exec_ctx, call->cb_arg, 1);
}
}
-static void zookeeper_channel_saw_error(grpc_resolver *resolver,
+static void zookeeper_channel_saw_error(grpc_exec_ctx *exec_ctx,
+ grpc_resolver *resolver,
struct sockaddr *sa, int len) {
zookeeper_resolver *r = (zookeeper_resolver *)resolver;
gpr_mu_lock(&r->mu);
@@ -130,11 +134,10 @@
gpr_mu_unlock(&r->mu);
}
-static void zookeeper_next(grpc_resolver *resolver,
+static void zookeeper_next(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver,
grpc_client_config **target_config,
grpc_closure *on_complete) {
zookeeper_resolver *r = (zookeeper_resolver *)resolver;
- grpc_closure *call;
gpr_mu_lock(&r->mu);
GPR_ASSERT(r->next_completion == NULL);
r->next_completion = on_complete;
@@ -142,10 +145,9 @@
if (r->resolved_version == 0 && r->resolving == 0) {
zookeeper_start_resolving_locked(r);
} else {
- call = zookeeper_maybe_finish_next_locked(r);
+ zookeeper_maybe_finish_next_locked(exec_ctx, r);
}
gpr_mu_unlock(&r->mu);
- if (call) call->cb(call->cb_arg, 1);
}
/** Zookeeper global watcher for connection management
@@ -180,14 +182,13 @@
/** Callback function after getting all resolved addresses
Creates a subchannel for each address */
-static void zookeeper_on_resolved(void *arg,
+static void zookeeper_on_resolved(grpc_exec_ctx *exec_ctx, void *arg,
grpc_resolved_addresses *addresses) {
zookeeper_resolver *r = arg;
grpc_client_config *config = NULL;
grpc_subchannel **subchannels;
grpc_subchannel_args args;
grpc_lb_policy *lb_policy;
- grpc_closure *call;
size_t i;
if (addresses != NULL) {
grpc_lb_policy_args lb_policy_args;
@@ -198,13 +199,13 @@
args.addr = (struct sockaddr *)(addresses->addrs[i].addr);
args.addr_len = addresses->addrs[i].len;
subchannels[i] = grpc_subchannel_factory_create_subchannel(
- r->subchannel_factory, &args);
+ exec_ctx, r->subchannel_factory, &args);
}
lb_policy_args.subchannels = subchannels;
lb_policy_args.num_subchannels = addresses->naddrs;
lb_policy = grpc_lb_policy_create(r->lb_policy_name, &lb_policy_args);
grpc_client_config_set_lb_policy(config, lb_policy);
- GRPC_LB_POLICY_UNREF(lb_policy, "construction");
+ GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "construction");
grpc_resolved_addresses_destroy(addresses);
gpr_free(subchannels);
}
@@ -212,20 +213,18 @@
GPR_ASSERT(r->resolving == 1);
r->resolving = 0;
if (r->resolved_config != NULL) {
- grpc_client_config_unref(r->resolved_config);
+ grpc_client_config_unref(exec_ctx, r->resolved_config);
}
r->resolved_config = config;
r->resolved_version++;
- call = zookeeper_maybe_finish_next_locked(r);
+ zookeeper_maybe_finish_next_locked(exec_ctx, r);
gpr_mu_unlock(&r->mu);
- if (call) call->cb(call->cb_arg, 1);
-
- GRPC_RESOLVER_UNREF(&r->base, "zookeeper-resolving");
+ GRPC_RESOLVER_UNREF(exec_ctx, &r->base, "zookeeper-resolving");
}
/** Callback function for each DNS resolved address */
-static void zookeeper_dns_resolved(void *arg,
+static void zookeeper_dns_resolved(grpc_exec_ctx *exec_ctx, void *arg,
grpc_resolved_addresses *addresses) {
size_t i;
zookeeper_resolver *r = arg;
@@ -251,7 +250,7 @@
resolve_done = (r->resolved_num == r->resolved_total);
gpr_mu_unlock(&r->mu);
if (resolve_done) {
- zookeeper_on_resolved(r, r->resolved_addrs);
+ zookeeper_on_resolved(exec_ctx, r, r->resolved_addrs);
}
}
@@ -300,9 +299,11 @@
char *address = NULL;
zookeeper_resolver *r = (zookeeper_resolver *)arg;
int resolve_done = 0;
+ grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
if (rc != 0) {
gpr_log(GPR_ERROR, "Error in getting a child node of %s", r->name);
+ grpc_exec_ctx_finish(&exec_ctx);
return;
}
@@ -318,9 +319,11 @@
resolve_done = (r->resolved_num == r->resolved_total);
gpr_mu_unlock(&r->mu);
if (resolve_done) {
- zookeeper_on_resolved(r, r->resolved_addrs);
+ zookeeper_on_resolved(&exec_ctx, r, r->resolved_addrs);
}
}
+
+ grpc_exec_ctx_finish(&exec_ctx);
}
static void zookeeper_get_children_completion(
@@ -411,28 +414,27 @@
zookeeper_resolve_address(r);
}
-static grpc_closure *zookeeper_maybe_finish_next_locked(zookeeper_resolver *r) {
- grpc_closure *call = NULL;
+static void zookeeper_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
+ zookeeper_resolver *r) {
if (r->next_completion != NULL &&
r->resolved_version != r->published_version) {
*r->target_config = r->resolved_config;
if (r->resolved_config != NULL) {
grpc_client_config_ref(r->resolved_config);
}
- call = r->next_completion;
+ grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, 1);
r->next_completion = NULL;
r->published_version = r->resolved_version;
}
- return call;
}
-static void zookeeper_destroy(grpc_resolver *gr) {
+static void zookeeper_destroy(grpc_exec_ctx *exec_ctx, grpc_resolver *gr) {
zookeeper_resolver *r = (zookeeper_resolver *)gr;
gpr_mu_destroy(&r->mu);
if (r->resolved_config != NULL) {
- grpc_client_config_unref(r->resolved_config);
+ grpc_client_config_unref(exec_ctx, r->resolved_config);
}
- grpc_subchannel_factory_unref(r->subchannel_factory);
+ grpc_subchannel_factory_unref(exec_ctx, r->subchannel_factory);
gpr_free(r->name);
gpr_free(r->lb_policy_name);
gpr_free(r);
diff --git a/src/core/iomgr/exec_ctx.h b/src/core/iomgr/exec_ctx.h
index 0cebc16..bfa08c4 100644
--- a/src/core/iomgr/exec_ctx.h
+++ b/src/core/iomgr/exec_ctx.h
@@ -36,6 +36,23 @@
#include "src/core/iomgr/closure.h"
+/** Execution context.
+ * A bag of data that collects information along a callstack.
+ * Generally created at public API entry points, and passed down as
+ * pointer to child functions that manipulate it.
+ *
+ * Specific responsibilities (this may grow in the future):
+ * - track a list of work that needs to be delayed until the top of the
+ * call stack (this provides a convenient mechanism to run callbacks
+ * without worrying about locking issues)
+ *
+ * CONVENTIONS:
+ * Instance of this must ALWAYS be constructed on the stack, never
+ * heap allocated. Instances and pointers to them must always be called
+ * exec_ctx. Instances are always passed as the first argument
+ * to a function that takes it, and always as a pointer (grpc_exec_ctx
+ * is never copied).
+ */
struct grpc_exec_ctx {
grpc_closure_list closure_list;
};
@@ -43,10 +60,17 @@
#define GRPC_EXEC_CTX_INIT \
{ GRPC_CLOSURE_LIST_INIT }
-void grpc_exec_ctx_finish(grpc_exec_ctx *exec_ctx);
+/** Flush any work that has been enqueued onto this grpc_exec_ctx.
+ * Caller must guarantee that no interfering locks are held. */
void grpc_exec_ctx_flush(grpc_exec_ctx *exec_ctx);
+/** Finish any pending work for a grpc_exec_ctx. Must be called before
+ * the instance is destroyed, or work may be lost. */
+void grpc_exec_ctx_finish(grpc_exec_ctx *exec_ctx);
+/** Add a closure to be executed at the next flush/finish point */
void grpc_exec_ctx_enqueue(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
int success);
+/** Add a list of closures to be executed at the next flush/finish point.
+ * Leaves \a list empty. */
void grpc_exec_ctx_enqueue_list(grpc_exec_ctx *exec_ctx,
grpc_closure_list *list);
diff --git a/src/core/iomgr/tcp_client_windows.c b/src/core/iomgr/tcp_client_windows.c
index 4dfab3b..15fe30f 100644
--- a/src/core/iomgr/tcp_client_windows.c
+++ b/src/core/iomgr/tcp_client_windows.c
@@ -179,7 +179,7 @@
socket = grpc_winsocket_create(sock, "client");
info = &socket->write_info;
- success = ConnectEx(sock, addr, addr_len, NULL, 0, NULL, &info->overlapped);
+ success = ConnectEx(sock, addr, (int)addr_len, NULL, 0, NULL, &info->overlapped);
/* It wouldn't be unusual to get a success immediately. But we'll still get
an IOCP notification, so let's ignore it. */
diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c
index 9881a41..55cd1a5 100644
--- a/src/core/iomgr/tcp_server_windows.c
+++ b/src/core/iomgr/tcp_server_windows.c
@@ -152,7 +152,7 @@
/* Prepare (bind) a recently-created socket for listening. */
static int prepare_socket(SOCKET sock, const struct sockaddr *addr,
- int addr_len) {
+ size_t addr_len) {
struct sockaddr_storage sockname_temp;
socklen_t sockname_len;
@@ -165,7 +165,7 @@
goto error;
}
- if (bind(sock, addr, addr_len) == SOCKET_ERROR) {
+ if (bind(sock, addr, (int)addr_len) == SOCKET_ERROR) {
char *addr_str;
char *utf8_message = gpr_format_message(WSAGetLastError());
grpc_sockaddr_to_string(&addr_str, addr, 0);
@@ -349,7 +349,7 @@
}
static int add_socket_to_server(grpc_tcp_server *s, SOCKET sock,
- const struct sockaddr *addr, int addr_len) {
+ const struct sockaddr *addr, size_t addr_len) {
server_port *sp;
int port;
int status;
diff --git a/src/cpp/client/secure_credentials.cc b/src/cpp/client/secure_credentials.cc
index 99b7468..1693cf7 100644
--- a/src/cpp/client/secure_credentials.cc
+++ b/src/cpp/client/secure_credentials.cc
@@ -173,7 +173,7 @@
void MetadataCredentialsPluginWrapper::InvokePlugin(
const char* service_url, grpc_credentials_plugin_metadata_cb cb,
void* user_data) {
- std::multimap<grpc::string, grpc::string_ref> metadata;
+ std::multimap<grpc::string, grpc::string> metadata;
Status status = plugin_->GetMetadata(service_url, &metadata);
std::vector<grpc_metadata> md;
for (auto it = metadata.begin(); it != metadata.end(); ++it) {
diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc
index 7f04799..2659b0e 100644
--- a/test/cpp/end2end/end2end_test.cc
+++ b/test/cpp/end2end/end2end_test.cc
@@ -121,7 +121,7 @@
bool IsBlocking() const GRPC_OVERRIDE { return is_blocking_; }
Status GetMetadata(grpc::string_ref service_url,
- std::multimap<grpc::string, grpc::string_ref>* metadata)
+ std::multimap<grpc::string, grpc::string>* metadata)
GRPC_OVERRIDE {
EXPECT_GT(service_url.length(), 0UL);
EXPECT_TRUE(metadata != nullptr);
@@ -175,9 +175,9 @@
if (auth_md_value == kGoodGuy) {
context->AddProperty(kIdentityPropName, kGoodGuy);
context->SetPeerIdentityPropertyName(kIdentityPropName);
- consumed_auth_metadata->insert(
- std::make_pair(string(auth_md->first.data(), auth_md->first.length()),
- auth_md->second));
+ consumed_auth_metadata->insert(std::make_pair(
+ string(auth_md->first.data(), auth_md->first.length()),
+ string(auth_md->second.data(), auth_md->second.length())));
return Status::OK;
} else {
return Status(StatusCode::UNAUTHENTICATED,