Merge remote-tracking branch 'upstream/master' into fix_gcc_4_4

baggiozhong · Mar 25, 2015 · f865a6b · f865a6b
2 parents d5933b6 + 96e32aa
commit f865a6b
Show file tree

Hide file tree

Showing 25 changed files with 163 additions and 137 deletions.
diff --git a/examples/pubsub/main.cc b/examples/pubsub/main.cc
@@ -51,14 +51,14 @@
 #include "examples/pubsub/subscriber.h"
 
 DEFINE_int32(server_port, 443, "Server port.");
-DEFINE_string(server_host,
-              "pubsub-staging.googleapis.com", "Server host to connect to");
+DEFINE_string(server_host, "pubsub-staging.googleapis.com",
+              "Server host to connect to");
 DEFINE_string(project_id, "", "GCE project id such as stoked-keyword-656");
 
 // In some distros, gflags is in the namespace google, and in some others,
 // in gflags. This hack is enabling us to find both.
-namespace google { }
-namespace gflags { }
+namespace google {}
+namespace gflags {}
 using namespace google;
 using namespace gflags;
 
@@ -92,32 +92,32 @@ int main(int argc, char** argv) {
   grpc::string topic = ss.str();
 
   ss.str("");
-  ss << FLAGS_project_id << "/"  << kSubscriptionName;
+  ss << FLAGS_project_id << "/" << kSubscriptionName;
   grpc::string subscription_name = ss.str();
 
   // Clean up test topic and subcription if they exist before.
   grpc::string subscription_topic;
-  if (subscriber.GetSubscription(
-      subscription_name, &subscription_topic).IsOk()) {
+  if (subscriber.GetSubscription(subscription_name, &subscription_topic)
+          .IsOk()) {
     subscriber.DeleteSubscription(subscription_name);
   }
 
   if (publisher.GetTopic(topic).IsOk()) publisher.DeleteTopic(topic);
 
   grpc::Status s = publisher.CreateTopic(topic);
-  gpr_log(GPR_INFO, "Create topic returns code %d, %s",
-          s.code(), s.details().c_str());
+  gpr_log(GPR_INFO, "Create topic returns code %d, %s", s.code(),
+          s.details().c_str());
   GPR_ASSERT(s.IsOk());
 
   s = publisher.GetTopic(topic);
-  gpr_log(GPR_INFO, "Get topic returns code %d, %s",
-          s.code(), s.details().c_str());
+  gpr_log(GPR_INFO, "Get topic returns code %d, %s", s.code(),
+          s.details().c_str());
   GPR_ASSERT(s.IsOk());
 
   std::vector<grpc::string> topics;
   s = publisher.ListTopics(FLAGS_project_id, &topics);
-  gpr_log(GPR_INFO, "List topic returns code %d, %s",
-          s.code(), s.details().c_str());
+  gpr_log(GPR_INFO, "List topic returns code %d, %s", s.code(),
+          s.details().c_str());
   bool topic_found = false;
   for (unsigned int i = 0; i < topics.size(); i++) {
     if (topics[i] == topic) topic_found = true;
@@ -127,27 +127,27 @@ int main(int argc, char** argv) {
   GPR_ASSERT(topic_found);
 
   s = subscriber.CreateSubscription(topic, subscription_name);
-  gpr_log(GPR_INFO, "create subscrption returns code %d, %s",
-          s.code(), s.details().c_str());
+  gpr_log(GPR_INFO, "create subscrption returns code %d, %s", s.code(),
+          s.details().c_str());
   GPR_ASSERT(s.IsOk());
 
   s = publisher.Publish(topic, kMessageData);
-  gpr_log(GPR_INFO, "Publish %s returns code %d, %s",
-          kMessageData, s.code(), s.details().c_str());
+  gpr_log(GPR_INFO, "Publish %s returns code %d, %s", kMessageData, s.code(),
+          s.details().c_str());
   GPR_ASSERT(s.IsOk());
 
   grpc::string data;
   s = subscriber.Pull(subscription_name, &data);
   gpr_log(GPR_INFO, "Pull %s", data.c_str());
 
-  s =  subscriber.DeleteSubscription(subscription_name);
-  gpr_log(GPR_INFO, "Delete subscription returns code %d, %s",
-          s.code(), s.details().c_str());
+  s = subscriber.DeleteSubscription(subscription_name);
+  gpr_log(GPR_INFO, "Delete subscription returns code %d, %s", s.code(),
+          s.details().c_str());
   GPR_ASSERT(s.IsOk());
 
   s = publisher.DeleteTopic(topic);
-  gpr_log(GPR_INFO, "Delete topic returns code %d, %s",
-          s.code(), s.details().c_str());
+  gpr_log(GPR_INFO, "Delete topic returns code %d, %s", s.code(),
+          s.details().c_str());
   GPR_ASSERT(s.IsOk());
 
   subscriber.Shutdown();

diff --git a/examples/pubsub/publisher.cc b/examples/pubsub/publisher.cc
@@ -51,12 +51,9 @@ namespace examples {
 namespace pubsub {
 
 Publisher::Publisher(std::shared_ptr<ChannelInterface> channel)
-    : stub_(PublisherService::NewStub(channel)) {
-}
+    : stub_(PublisherService::NewStub(channel)) {}
 
-void Publisher::Shutdown() {
-  stub_.reset();
-}
+void Publisher::Shutdown() { stub_.reset(); }
 
 Status Publisher::CreateTopic(const grpc::string& topic) {
   Topic request;

diff --git a/examples/pubsub/publisher_test.cc b/examples/pubsub/publisher_test.cc
@@ -31,8 +31,6 @@
  *
  */
 
-#include <google/protobuf/stubs/common.h>
-
 #include <grpc++/channel_arguments.h>
 #include <grpc++/channel_interface.h>
 #include <grpc++/client_context.h>
@@ -84,20 +82,19 @@ class PublisherServiceImpl : public tech::pubsub::PublisherService::Service {
   Status ListTopics(
       ServerContext* context, const ::tech::pubsub::ListTopicsRequest* request,
       ::tech::pubsub::ListTopicsResponse* response) GRPC_OVERRIDE {
-   std::ostringstream ss;
-   ss << "cloud.googleapis.com/project in (/projects/" << kProjectId << ")";
-   EXPECT_EQ(request->query(), ss.str());
-   response->add_topic()->set_name(kTopic);
-   return Status::OK;
- }
-
- Status DeleteTopic(ServerContext* context,
-                    const ::tech::pubsub::DeleteTopicRequest* request,
-                    ::proto2::Empty* response) GRPC_OVERRIDE {
-    EXPECT_EQ(request->topic(), kTopic);
+    std::ostringstream ss;
+    ss << "cloud.googleapis.com/project in (/projects/" << kProjectId << ")";
+    EXPECT_EQ(request->query(), ss.str());
+    response->add_topic()->set_name(kTopic);
     return Status::OK;
- }
+  }
 
+  Status DeleteTopic(ServerContext* context,
+                     const ::tech::pubsub::DeleteTopicRequest* request,
+                     ::proto2::Empty* response) GRPC_OVERRIDE {
+    EXPECT_EQ(request->topic(), kTopic);
+    return Status::OK;
+  }
 };
 
 class PublisherTest : public ::testing::Test {
@@ -107,11 +104,13 @@ class PublisherTest : public ::testing::Test {
     int port = grpc_pick_unused_port_or_die();
     server_address_ << "localhost:" << port;
     ServerBuilder builder;
-    builder.AddListeningPort(server_address_.str(), grpc::InsecureServerCredentials());
+    builder.AddListeningPort(server_address_.str(),
+                             grpc::InsecureServerCredentials());
     builder.RegisterService(&service_);
     server_ = builder.BuildAndStart();
 
-    channel_ = CreateChannel(server_address_.str(), grpc::InsecureCredentials(), ChannelArguments());
+    channel_ = CreateChannel(server_address_.str(), grpc::InsecureCredentials(),
+                             ChannelArguments());
 
     publisher_.reset(new grpc::examples::pubsub::Publisher(channel_));
   }

diff --git a/examples/pubsub/subscriber.cc b/examples/pubsub/subscriber.cc
@@ -49,12 +49,9 @@ namespace examples {
 namespace pubsub {
 
 Subscriber::Subscriber(std::shared_ptr<ChannelInterface> channel)
-    : stub_(SubscriberService::NewStub(channel)) {
-}
+    : stub_(SubscriberService::NewStub(channel)) {}
 
-void Subscriber::Shutdown() {
-  stub_.reset();
-}
+void Subscriber::Shutdown() { stub_.reset(); }
 
 Status Subscriber::CreateSubscription(const grpc::string& topic,
                                       const grpc::string& name) {

diff --git a/examples/pubsub/subscriber_test.cc b/examples/pubsub/subscriber_test.cc
@@ -31,8 +31,6 @@
  *
  */
 
-#include <google/protobuf/stubs/common.h>
-
 #include <grpc++/channel_arguments.h>
 #include <grpc++/channel_interface.h>
 #include <grpc++/client_context.h>
@@ -95,7 +93,6 @@ class SubscriberServiceImpl : public tech::pubsub::SubscriberService::Service {
                      proto2::Empty* response) GRPC_OVERRIDE {
     return Status::OK;
   }
-
 };
 
 class SubscriberTest : public ::testing::Test {
@@ -105,11 +102,13 @@ class SubscriberTest : public ::testing::Test {
     int port = grpc_pick_unused_port_or_die();
     server_address_ << "localhost:" << port;
     ServerBuilder builder;
-    builder.AddListeningPort(server_address_.str(), grpc::InsecureServerCredentials());
+    builder.AddListeningPort(server_address_.str(),
+                             grpc::InsecureServerCredentials());
     builder.RegisterService(&service_);
     server_ = builder.BuildAndStart();
 
-    channel_ = CreateChannel(server_address_.str(), grpc::InsecureCredentials(), ChannelArguments());
+    channel_ = CreateChannel(server_address_.str(), grpc::InsecureCredentials(),
+                             ChannelArguments());
 
     subscriber_.reset(new grpc::examples::pubsub::Subscriber(channel_));
   }
@@ -129,17 +128,15 @@ class SubscriberTest : public ::testing::Test {
 };
 
 TEST_F(SubscriberTest, TestSubscriber) {
-  EXPECT_TRUE(subscriber_->CreateSubscription(kTopic,
-                                              kSubscriptionName).IsOk());
+  EXPECT_TRUE(
+      subscriber_->CreateSubscription(kTopic, kSubscriptionName).IsOk());
 
   grpc::string topic;
-  EXPECT_TRUE(subscriber_->GetSubscription(kSubscriptionName,
-                                           &topic).IsOk());
+  EXPECT_TRUE(subscriber_->GetSubscription(kSubscriptionName, &topic).IsOk());
   EXPECT_EQ(topic, kTopic);
 
   grpc::string data;
-  EXPECT_TRUE(subscriber_->Pull(kSubscriptionName,
-                                &data).IsOk());
+  EXPECT_TRUE(subscriber_->Pull(kSubscriptionName, &data).IsOk());
 
   EXPECT_TRUE(subscriber_->DeleteSubscription(kSubscriptionName).IsOk());
 }

diff --git a/src/compiler/python_generator.cc b/src/compiler/python_generator.cc
@@ -309,17 +309,20 @@ bool PrintServerFactory(const grpc::string& package_qualified_service_name,
           make_pair(method->name(), output_message_module_and_class));
     }
     out->Print("method_service_descriptions = {\n");
-    for (auto& name_and_description_constructor :
-         method_description_constructors) {
+    for (auto name_and_description_constructor =
+	   method_description_constructors.begin();
+	 name_and_description_constructor !=
+	   method_description_constructors.end();
+	 name_and_description_constructor++) {
       IndentScope raii_descriptions_indent(out);
-      const grpc::string method_name = name_and_description_constructor.first;
+      const grpc::string method_name = name_and_description_constructor->first;
       auto input_message_module_and_class =
           input_message_modules_and_classes.find(method_name);
       auto output_message_module_and_class =
           output_message_modules_and_classes.find(method_name);
       out->Print("\"$Method$\": utilities.$Constructor$(\n", "Method",
                  method_name, "Constructor",
-                 name_and_description_constructor.second);
+                 name_and_description_constructor->second);
       {
         IndentScope raii_description_arguments_indent(out);
         out->Print("servicer.$Method$,\n", "Method", method_name);
@@ -387,17 +390,20 @@ bool PrintStubFactory(const grpc::string& package_qualified_service_name,
           make_pair(method->name(), output_message_module_and_class));
     }
     out->Print("method_invocation_descriptions = {\n");
-    for (auto& name_and_description_constructor :
-         method_description_constructors) {
+    for (auto name_and_description_constructor =
+	   method_description_constructors.begin();
+	 name_and_description_constructor !=
+	   method_description_constructors.end();
+	 name_and_description_constructor++) {
       IndentScope raii_descriptions_indent(out);
-      const grpc::string method_name = name_and_description_constructor.first;
+      const grpc::string method_name = name_and_description_constructor->first;
       auto input_message_module_and_class =
           input_message_modules_and_classes.find(method_name);
       auto output_message_module_and_class =
           output_message_modules_and_classes.find(method_name);
       out->Print("\"$Method$\": utilities.$Constructor$(\n", "Method",
                  method_name, "Constructor",
-                 name_and_description_constructor.second);
+                 name_and_description_constructor->second);
       {
         IndentScope raii_description_arguments_indent(out);
         out->Print(

diff --git a/src/core/iomgr/iocp_windows.c b/src/core/iomgr/iocp_windows.c
@@ -52,10 +52,11 @@ static OVERLAPPED g_iocp_custom_overlap;
 
 static gpr_event g_shutdown_iocp;
 static gpr_event g_iocp_done;
+static gpr_atm g_orphans = 0;
 
 static HANDLE g_iocp;
 
-static int do_iocp_work() {
+static void do_iocp_work() {
   BOOL success;
   DWORD bytes = 0;
   DWORD flags = 0;
@@ -71,14 +72,14 @@ static int do_iocp_work() {
                                       gpr_time_to_millis(wait_time));
   if (!success && !overlapped) {
     /* The deadline got attained. */
-    return 0;
+    return;
   }
   GPR_ASSERT(completion_key && overlapped);
   if (overlapped == &g_iocp_custom_overlap) {
     if (completion_key == (ULONG_PTR) &g_iocp_kick_token) {
       /* We were awoken from a kick. */
       gpr_log(GPR_DEBUG, "do_iocp_work - got a kick");
-      return 1;
+      return;
     }
     gpr_log(GPR_ERROR, "Unknown custom completion key.");
     abort();
@@ -97,8 +98,13 @@ static int do_iocp_work() {
   }
   success = WSAGetOverlappedResult(socket->socket, &info->overlapped, &bytes,
                                    FALSE, &flags);
-  gpr_log(GPR_DEBUG, "bytes: %u, flags: %u - op %s", bytes, flags,
-          success ? "succeeded" : "failed");
+  gpr_log(GPR_DEBUG, "bytes: %u, flags: %u - op %s %s", bytes, flags,
+          success ? "succeeded" : "failed", socket->orphan ? "orphan" : "");
+  if (socket->orphan) {
+    grpc_winsocket_destroy(socket);
+    gpr_atm_full_fetch_add(&g_orphans, -1);
+    return;
+  }
   info->bytes_transfered = bytes;
   info->wsa_error = success ? 0 : WSAGetLastError();
   GPR_ASSERT(overlapped == &info->overlapped);
@@ -113,12 +119,10 @@ static int do_iocp_work() {
   }
   gpr_mu_unlock(&socket->state_mu);
   if (f) f(opaque, 1);
-
-  return 1;
 }
 
 static void iocp_loop(void *p) {
-  while (!gpr_event_get(&g_shutdown_iocp)) {
+  while (gpr_atm_acq_load(&g_orphans) || !gpr_event_get(&g_shutdown_iocp)) {
     grpc_maybe_call_delayed_callbacks(NULL, 1);
     do_iocp_work();
   }
@@ -138,13 +142,19 @@ void grpc_iocp_init(void) {
   gpr_thd_new(&id, iocp_loop, NULL, NULL);
 }
 
-void grpc_iocp_shutdown(void) {
+void grpc_iocp_kick(void) {
   BOOL success;
-  gpr_event_set(&g_shutdown_iocp, (void *)1);
+
   success = PostQueuedCompletionStatus(g_iocp, 0,
                                        (ULONG_PTR) &g_iocp_kick_token,
                                        &g_iocp_custom_overlap);
   GPR_ASSERT(success);
+}
+
+void grpc_iocp_shutdown(void) {
+  BOOL success;
+  gpr_event_set(&g_shutdown_iocp, (void *)1);
+  grpc_iocp_kick();
   gpr_event_wait(&g_iocp_done, gpr_inf_future);
   success = CloseHandle(g_iocp);
   GPR_ASSERT(success);
@@ -166,6 +176,10 @@ void grpc_iocp_add_socket(grpc_winsocket *socket) {
   GPR_ASSERT(ret == g_iocp);
 }
 
+void grpc_iocp_socket_orphan(grpc_winsocket *socket) {
+  gpr_atm_full_fetch_add(&g_orphans, 1);
+}
+
 static void socket_notify_on_iocp(grpc_winsocket *socket,
                                   void(*cb)(void *, int), void *opaque,
                                   grpc_winsocket_callback_info *info) {

diff --git a/src/core/iomgr/iocp_windows.h b/src/core/iomgr/iocp_windows.h
@@ -42,6 +42,7 @@
 void grpc_iocp_init(void);
 void grpc_iocp_shutdown(void);
 void grpc_iocp_add_socket(grpc_winsocket *);
+void grpc_iocp_socket_orphan(grpc_winsocket *);
 
 void grpc_socket_notify_on_write(grpc_winsocket *, void(*cb)(void *, int success),
                                  void *opaque);