Address comments and update grpc plugins
diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h
index 9439825..fe4d9f9 100644
--- a/include/grpc/grpc.h
+++ b/include/grpc/grpc.h
@@ -355,12 +355,15 @@
} data;
} grpc_op;
+/** Registers a plugin to be initialized and destroyed with the library.
-/** Registers a plugin to be initialized and deinitialized with the library.
-
- It is safe to pass NULL to either argument. The initialization and
- deinitialization order isn't guaranteed. */
-void grpc_register_plugin(void (*init)(void), void (*deinit)(void));
+ The \a init and \a destroy functions will be invoked as part of
+ \a grpc_init() and \a grpc_shutdown(), respectively.
+ Note that these functions can be invoked an arbitrary number of times
+ (and hence so will \a init and \a destroy).
+ It is safe to pass NULL to either argument. Plugins are destroyed in
+ the reverse order they were initialized. */
+void grpc_register_plugin(void (*init)(void), void (*destroy)(void));
/** Frees the memory used by all the plugin information.
@@ -381,8 +384,8 @@
/* Default propagation mask: clients of the core API are encouraged to encode
deltas from this in their implementations... ie write:
- GRPC_PROPAGATE_DEFAULTS & ~GRPC_PROPAGATE_DEADLINE to disable deadline
- propagation. Doing so gives flexibility in the future to define new
+ GRPC_PROPAGATE_DEFAULTS & ~GRPC_PROPAGATE_DEADLINE to disable deadline
+ propagation. Doing so gives flexibility in the future to define new
propagation types that are default inherited or not. */
#define GRPC_PROPAGATE_DEFAULTS \
((gpr_uint32)(( \
@@ -429,8 +432,8 @@
otherwise a grpc_event describing the event that occurred.
Callers must not call grpc_completion_queue_next and
- grpc_completion_queue_pluck simultaneously on the same completion queue.
-
+ grpc_completion_queue_pluck simultaneously on the same completion queue.
+
Completion queues support a maximum of GRPC_MAX_COMPLETION_QUEUE_PLUCKERS
concurrently executing plucks at any time. */
grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cq, void *tag,
@@ -470,7 +473,7 @@
completions are sent to 'completion_queue'. 'method' and 'host' need only
live through the invocation of this function.
If parent_call is non-NULL, it must be a server-side call. It will be used
- to propagate properties from the server call to this new client call.
+ to propagate properties from the server call to this new client call.
*/
grpc_call *grpc_channel_create_call(grpc_channel *channel,
grpc_call *parent_call,
diff --git a/include/grpc/grpc_zookeeper.h b/include/grpc/grpc_zookeeper.h
index d1eeff3..2b195c1 100644
--- a/include/grpc/grpc_zookeeper.h
+++ b/include/grpc/grpc_zookeeper.h
@@ -31,6 +31,17 @@
*
*/
+/** Support zookeeper as alternative name system in addition to DNS
+ * Zookeeper name in gRPC is represented as a URI:
+ * zookeeper://host:port/path/service/instance
+ *
+ * Where zookeeper is the name system scheme
+ * host:port is the address of a zookeeper server
+ * /path/service/instance is the zookeeper name to be resolved
+ *
+ * Refer doc/naming.md for more details
+ */
+
#ifndef GRPC_GRPC_ZOOKEEPER_H
#define GRPC_GRPC_ZOOKEEPER_H
diff --git a/src/core/client_config/resolvers/zookeeper_resolver.c b/src/core/client_config/resolvers/zookeeper_resolver.c
index 96c59ea..3a99a42 100644
--- a/src/core/client_config/resolvers/zookeeper_resolver.c
+++ b/src/core/client_config/resolvers/zookeeper_resolver.c
@@ -142,7 +142,8 @@
gpr_mu_unlock(&r->mu);
}
-/** Zookeeper global watcher for connection management */
+/** Zookeeper global watcher for connection management
+ TODO: better connection management besides logs */
static void zookeeper_global_watcher(zhandle_t *zookeeper_handle, int type,
int state, const char *path,
void *watcher_ctx) {
@@ -155,7 +156,8 @@
}
}
-/** Zookeeper watcher for handling updates to watched nodes */
+/** Zookeeper watcher triggered by changes to watched nodes
+ Start to resolve again to get updated addresses */
static void zookeeper_watcher(zhandle_t *zookeeper_handle, int type, int state,
const char *path, void *watcher_ctx) {
if (watcher_ctx != NULL) {
@@ -170,6 +172,8 @@
}
}
+/** Callback function after getting all resolved addresses
+ Create a subchannel for each address */
static void zookeeper_on_resolved(void *arg,
grpc_resolved_addresses *addresses) {
zookeeper_resolver *r = arg;
@@ -239,16 +243,18 @@
}
}
-/** Parse json format address of a zookeeper node */
-static char *zookeeper_parse_address(char *buffer, int buffer_len) {
- const char *host;
- const char *port;
- char *address;
+/** Parse JSON format address of a zookeeper node */
+static char *zookeeper_parse_address(const char *value, int value_len) {
grpc_json *json;
grpc_json *cur;
+ const char *host;
+ const char *port;
+ char* buffer;
+ char *address = NULL;
- address = NULL;
- json = grpc_json_parse_string_with_len(buffer, buffer_len);
+ buffer = gpr_malloc(value_len);
+ memcpy(buffer, value, value_len);
+ json = grpc_json_parse_string_with_len(buffer, value_len);
if (json != NULL) {
host = NULL;
port = NULL;
@@ -270,6 +276,7 @@
}
grpc_json_destroy(json);
}
+ gpr_free(buffer);
return address;
}
@@ -279,7 +286,6 @@
const struct Stat *stat,
const void *arg) {
char *address = NULL;
- char *buffer = NULL;
zookeeper_resolver *r = (zookeeper_resolver *)arg;
int resolve_done = 0;
@@ -288,10 +294,7 @@
return;
}
- buffer = gpr_malloc(value_len);
- memcpy(buffer, value, value_len);
- address = zookeeper_parse_address(buffer, value_len);
- gpr_free(buffer);
+ address = zookeeper_parse_address(value, value_len);
if (address != NULL) {
/** Further resolve address by DNS */
grpc_resolve_address(address, NULL, zookeeper_dns_resolved, r);
@@ -330,6 +333,8 @@
r->resolved_addrs->naddrs = 0;
r->resolved_total = children->count;
+ /** TODO: Replace expensive heap allocation and free with stack
+ if we can get maximum allowed length of zookeeper path */
for (i = 0; i < children->count; i++) {
gpr_asprintf(&path, "%s/%s", r->name, children->data[i]);
status = zoo_awget(r->zookeeper_handle, path, zookeeper_watcher, r,
@@ -347,7 +352,6 @@
const void *arg) {
int status;
char *address = NULL;
- char *buffer = NULL;
zookeeper_resolver *r = (zookeeper_resolver *)arg;
r->resolved_addrs = NULL;
r->resolved_total = 0;
@@ -360,10 +364,7 @@
/** If zookeeper node of path r->name does not have address
(i.e. service node), get its children */
- buffer = gpr_malloc(value_len);
- memcpy(buffer, value, value_len);
- address = zookeeper_parse_address(buffer, value_len);
- gpr_free(buffer);
+ address = zookeeper_parse_address(value, value_len);
if (address != NULL) {
r->resolved_addrs = gpr_malloc(sizeof(grpc_resolved_addresses));
r->resolved_addrs->addrs = NULL;
@@ -428,22 +429,27 @@
size_t num_subchannels),
grpc_subchannel_factory *subchannel_factory) {
zookeeper_resolver *r;
- const char *path = uri->path;
+ size_t length;
+ char *path = uri->path;
if (0 == strcmp(uri->authority, "")) {
gpr_log(GPR_ERROR, "No authority specified in zookeeper uri");
return NULL;
}
+ /** Remove the trailing slash if exists */
+ length = strlen(path);
+ if (length > 1 && path[length - 1] == '/') {
+ path[length - 1] = 0;
+ }
+
r = gpr_malloc(sizeof(zookeeper_resolver));
memset(r, 0, sizeof(*r));
gpr_ref_init(&r->refs, 1);
gpr_mu_init(&r->mu);
grpc_resolver_init(&r->base, &zookeeper_resolver_vtable);
r->name = gpr_strdup(path);
- if (r->name[strlen(r->name) - 1] == '/') {
- r->name[strlen(r->name) - 1] = 0;
- }
+
r->subchannel_factory = subchannel_factory;
r->lb_policy_factory = lb_policy_factory;
grpc_subchannel_factory_ref(subchannel_factory);
diff --git a/src/core/surface/init.c b/src/core/surface/init.c
index 053878a..d904454 100644
--- a/src/core/surface/init.c
+++ b/src/core/surface/init.c
@@ -52,6 +52,8 @@
#include "src/core/transport/chttp2_transport.h"
#include "src/core/transport/connectivity_state.h"
+#define MAX_PLUGINS 128
+
static gpr_once g_basic_init = GPR_ONCE_INIT;
static gpr_mu g_init_mu;
static int g_initializations;
@@ -63,40 +65,21 @@
typedef struct grpc_plugin {
void (*init)();
- void (*deinit)();
- struct grpc_plugin *next;
+ void (*destroy)();
} grpc_plugin;
-static grpc_plugin *g_plugins_head = NULL;
+static grpc_plugin g_all_of_the_plugins[MAX_PLUGINS];
+static int g_number_of_plugins = 0;
-static grpc_plugin *new_plugin(void (*init)(void), void (*deinit)(void)) {
- grpc_plugin *plugin = gpr_malloc(sizeof(*plugin));
- memset(plugin, 0, sizeof(*plugin));
- plugin->init = init;
- plugin->deinit = deinit;
-
- return plugin;
-}
-
-void grpc_register_plugin(void (*init)(void), void (*deinit)(void)) {
- grpc_plugin *old_head = g_plugins_head;
- g_plugins_head = new_plugin(init, deinit);
- g_plugins_head->next = old_head;
-}
-
-void grpc_unregister_all_plugins() {
- grpc_plugin *plugin;
- grpc_plugin *next;
-
- for (plugin = g_plugins_head; plugin != NULL; plugin = next) {
- next = plugin->next;
- gpr_free(plugin);
- }
- g_plugins_head = NULL;
+void grpc_register_plugin(void (*init)(void), void (*destroy)(void)) {
+ GPR_ASSERT(g_number_of_plugins != MAX_PLUGINS);
+ g_all_of_the_plugins[g_number_of_plugins].init = init;
+ g_all_of_the_plugins[g_number_of_plugins].destroy = destroy;
+ g_number_of_plugins++;
}
void grpc_init(void) {
- grpc_plugin *plugin;
+ int i;
gpr_once_init(&g_basic_init, do_basic_init);
gpr_mu_lock(&g_init_mu);
@@ -125,9 +108,9 @@
}
}
grpc_timers_global_init();
- for (plugin = g_plugins_head; plugin != NULL; plugin = plugin->next) {
- if (plugin->init) {
- plugin->init();
+ for (i = 0; i < g_number_of_plugins; i++) {
+ if (g_all_of_the_plugins[i].init != NULL) {
+ g_all_of_the_plugins[i].init();
}
}
}
@@ -135,9 +118,7 @@
}
void grpc_shutdown(void) {
- grpc_plugin *plugin;
- grpc_plugin *next;
-
+ int i;
gpr_mu_lock(&g_init_mu);
if (--g_initializations == 0) {
grpc_iomgr_shutdown();
@@ -145,11 +126,10 @@
grpc_timers_global_destroy();
grpc_tracer_shutdown();
grpc_resolver_registry_shutdown();
- for (plugin = g_plugins_head; plugin != NULL; plugin = next) {
- if (plugin->deinit) {
- plugin->deinit();
+ for (i = 0; i < g_number_of_plugins; i++) {
+ if (g_all_of_the_plugins[i].destroy != NULL) {
+ g_all_of_the_plugins[i].destroy();
}
- next = plugin->next;
}
}
gpr_mu_unlock(&g_init_mu);
diff --git a/test/cpp/end2end/zookeeper_test.cc b/test/cpp/end2end/zookeeper_test.cc
index 27adcf9..a95ee95 100644
--- a/test/cpp/end2end/zookeeper_test.cc
+++ b/test/cpp/end2end/zookeeper_test.cc
@@ -85,12 +85,12 @@
// Register service instance /test/1 in zookeeper
string value =
- "{\"host\":\"localhost\",\"port\":\"" + std::to_string(port1) + "\"}";
+ "{\"host\":\"localhost\",\"port\":\"" + to_string(port1) + "\"}";
RegisterService("/test/1", value);
// Register service instance /test/2 in zookeeper
value =
- "{\"host\":\"localhost\",\"port\":\"" + std::to_string(port2) + "\"}";
+ "{\"host\":\"localhost\",\"port\":\"" + to_string(port2) + "\"}";
RegisterService("/test/2", value);
}
@@ -115,13 +115,10 @@
// Register zookeeper name resolver in grpc
grpc_zookeeper_register();
-
- // Unregister all plugins when exit
- atexit(grpc_unregister_all_plugins);
}
- std::unique_ptr<Server> SetUpServer(int port) {
- string server_address = "localhost:" + std::to_string(port);
+ std::unique_ptr<Server> SetUpServer(const int port) {
+ string server_address = "localhost:" + to_string(port);
ServerBuilder builder;
builder.AddListeningPort(server_address, InsecureServerCredentials());
@@ -130,7 +127,7 @@
return server;
}
- void RegisterService(string name, string value) {
+ void RegisterService(const string& name, const string& value) {
char* path = (char*)gpr_malloc(name.size());
int status = zoo_exists(zookeeper_handle_, name.c_str(), 0, NULL);
@@ -146,11 +143,16 @@
GPR_ASSERT(status == 0);
}
- void DeleteService(string name) {
+ void DeleteService(const string& name) {
int status = zoo_delete(zookeeper_handle_, name.c_str(), -1);
GPR_ASSERT(status == 0);
}
+ void ChangeZookeeperState() {
+ server1_->Shutdown();
+ DeleteService("/test/1");
+ }
+
void TearDown() GRPC_OVERRIDE {
server1_->Shutdown();
server2_->Shutdown();
@@ -163,6 +165,12 @@
stub_ = std::move(grpc::cpp::test::util::TestService::NewStub(channel_));
}
+ string to_string(const int number) {
+ std::stringstream strs;
+ strs << number;
+ return strs.str();
+ }
+
std::shared_ptr<ChannelInterface> channel_;
std::unique_ptr<grpc::cpp::test::util::TestService::Stub> stub_;
std::unique_ptr<Server> server1_;
@@ -173,7 +181,7 @@
};
// Test zookeeper state change between two RPCs
-// TODO(ctiller): leaked objects
+// TODO(ctiller): leaked objects in this test
TEST_F(ZookeeperTest, ZookeeperStateChangeTwoRpc) {
ResetStub();
@@ -188,8 +196,10 @@
EXPECT_TRUE(s1.ok());
// Zookeeper state change
- DeleteService("/test/1");
- gpr_log(GPR_DEBUG, "Zookeeper state change");
+ gpr_log(GPR_DEBUG, "Zookeeper state change");
+ ChangeZookeeperState();
+ // Wait for re-resolving addresses
+ // TODO(ctiller): RPC will probably fail if not waiting
sleep(1);
// Second RPC