Merge branch 'master' into php_interop_ssl_fix

baggiozhong · Mar 25, 2015 · b2863f5 · b2863f5
2 parents c1da8f2 + ad0dd69
commit b2863f5
Show file tree

Hide file tree

Showing 18 changed files with 118 additions and 73 deletions.
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);

diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c
@@ -117,7 +117,16 @@ void grpc_iomgr_shutdown(void) {
       gpr_mu_lock(&g_mu);
     }
     if (g_refs) {
-      if (gpr_cv_wait(&g_rcv, &g_mu, shutdown_deadline) && g_cbs_head == NULL) {
+      int timeout = 0;
+      gpr_timespec short_deadline = gpr_time_add(gpr_now(),
+                                                 gpr_time_from_millis(100));
+      while (gpr_cv_wait(&g_rcv, &g_mu, short_deadline) && g_cbs_head == NULL) {
+        if (gpr_time_cmp(gpr_now(), shutdown_deadline) > 0) {
+          timeout = 1;
+          break;
+        }
+      }
+      if (timeout) {
         gpr_log(GPR_DEBUG,
                 "Failed to free %d iomgr objects before shutdown deadline: "
                 "memory leaks are likely",

diff --git a/src/core/iomgr/socket_windows.c b/src/core/iomgr/socket_windows.c
@@ -55,7 +55,7 @@ grpc_winsocket *grpc_winsocket_create(SOCKET socket) {
   return r;
 }
 
-void shutdown_op(grpc_winsocket_callback_info *info) {
+static void shutdown_op(grpc_winsocket_callback_info *info) {
   if (!info->cb) return;
   grpc_iomgr_add_delayed_callback(info->cb, info->opaque, 0);
 }
@@ -68,8 +68,13 @@ void grpc_winsocket_shutdown(grpc_winsocket *socket) {
 
 void grpc_winsocket_orphan(grpc_winsocket *socket) {
   gpr_log(GPR_DEBUG, "grpc_winsocket_orphan");
+  grpc_iocp_socket_orphan(socket);
+  socket->orphan = 1;
   grpc_iomgr_unref();
   closesocket(socket->socket);
+}
+
+void grpc_winsocket_destroy(grpc_winsocket *socket) {
   gpr_mu_destroy(&socket->state_mu);
   gpr_free(socket);
 }

diff --git a/src/core/iomgr/socket_windows.h b/src/core/iomgr/socket_windows.h
@@ -57,12 +57,13 @@ typedef struct grpc_winsocket_callback_info {
 typedef struct grpc_winsocket {
   SOCKET socket;
 
-  int added_to_iocp;
-
   grpc_winsocket_callback_info write_info;
   grpc_winsocket_callback_info read_info;
 
   gpr_mu state_mu;
+
+  int added_to_iocp;
+  int orphan;
 } grpc_winsocket;
 
 /* Create a wrapped windows handle.
@@ -71,5 +72,6 @@ grpc_winsocket *grpc_winsocket_create(SOCKET socket);
 
 void grpc_winsocket_shutdown(grpc_winsocket *socket);
 void grpc_winsocket_orphan(grpc_winsocket *socket);
+void grpc_winsocket_destroy(grpc_winsocket *socket);
 
 #endif  /* GRPC_INTERNAL_CORE_IOMGR_SOCKET_WINDOWS_H */
diff --git a/src/core/tsi/ssl_transport_security.c b/src/core/tsi/ssl_transport_security.c
@@ -567,7 +567,8 @@ static tsi_result populate_ssl_context(
     EC_KEY* ecdh = EC_KEY_new_by_curve_name(NID_X9_62_prime256v1);
     if (!SSL_CTX_set_tmp_ecdh(context, ecdh)) {
       gpr_log(GPR_ERROR, "Could not set ephemeral ECDH key.");
-      result = TSI_INTERNAL_ERROR;
+      EC_KEY_free(ecdh);
+      return TSI_INTERNAL_ERROR;
     }
     SSL_CTX_set_options(context, SSL_OP_SINGLE_ECDH_USE);
     EC_KEY_free(ecdh);
@@ -604,6 +605,7 @@ static tsi_result build_alpn_protocol_name_list(
   unsigned char* current;
   *protocol_name_list = NULL;
   *protocol_name_list_length = 0;
+  if (num_alpn_protocols == 0) return TSI_INVALID_ARGUMENT;
   for (i = 0; i < num_alpn_protocols; i++) {
     if (alpn_protocols_lengths[i] == 0) {
       gpr_log(GPR_ERROR, "Invalid 0-length protocol name.");

diff --git a/src/cpp/common/call.cc b/src/cpp/common/call.cc
@@ -148,7 +148,7 @@ void FillMetadataMap(grpc_metadata_array* arr,
     // TODO(yangg) handle duplicates?
     metadata->insert(std::pair<grpc::string, grpc::string>(
         arr->metadata[i].key,
-        {arr->metadata[i].value, arr->metadata[i].value_length}));
+        grpc::string(arr->metadata[i].value, arr->metadata[i].value_length)));
   }
   grpc_metadata_array_destroy(arr);
   grpc_metadata_array_init(arr);

diff --git a/src/cpp/server/secure_server_credentials.cc b/src/cpp/server/secure_server_credentials.cc
@@ -59,9 +59,12 @@ class SecureServerCredentials GRPC_FINAL : public ServerCredentials {
 std::shared_ptr<ServerCredentials> SslServerCredentials(
     const SslServerCredentialsOptions& options) {
   std::vector<grpc_ssl_pem_key_cert_pair> pem_key_cert_pairs;
-  for (const auto& key_cert_pair : options.pem_key_cert_pairs) {
-    pem_key_cert_pairs.push_back(
-        {key_cert_pair.private_key.c_str(), key_cert_pair.cert_chain.c_str()});
+  for (auto key_cert_pair = options.pem_key_cert_pairs.begin();
+       key_cert_pair != options.pem_key_cert_pairs.end();
+       key_cert_pair++) {
+    grpc_ssl_pem_key_cert_pair p = {key_cert_pair->private_key.c_str(),
+				    key_cert_pair->cert_chain.c_str()};
+    pem_key_cert_pairs.push_back(p);
   }
   grpc_server_credentials* c_creds = grpc_ssl_server_credentials_create(
       options.pem_root_certs.empty() ? nullptr : options.pem_root_certs.c_str(),

diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
@@ -247,8 +247,8 @@ bool Server::Start() {
 
   // Start processing rpcs.
   if (!sync_methods_.empty()) {
-    for (auto& m : sync_methods_) {
-      m.Request(server_);
+    for (auto m = sync_methods_.begin(); m != sync_methods_.end(); m++) {
+      m->Request(server_);
     }
 
     ScheduleCallback();

diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc
@@ -86,24 +86,26 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
     thread_pool_owned = true;
   }
   std::unique_ptr<Server> server(new Server(thread_pool_, thread_pool_owned));
-  for (auto* service : services_) {
-    if (!server->RegisterService(service)) {
+  for (auto service = services_.begin(); service != services_.end();
+       service++) {
+    if (!server->RegisterService(*service)) {
       return nullptr;
     }
   }
-  for (auto* service : async_services_) {
-    if (!server->RegisterAsyncService(service)) {
+  for (auto service = async_services_.begin();
+       service != async_services_.end(); service++) {
+    if (!server->RegisterAsyncService(*service)) {
       return nullptr;
     }
   }
   if (generic_service_) {
     server->RegisterAsyncGenericService(generic_service_);
   }
-  for (auto& port : ports_) {
-    int r = server->AddListeningPort(port.addr, port.creds.get());
+  for (auto port = ports_.begin(); port != ports_.end(); port++) {
+    int r = server->AddListeningPort(port->addr, port->creds.get());
     if (!r) return nullptr;
-    if (port.selected_port != nullptr) {
-      *port.selected_port = r;
+    if (port->selected_port != nullptr) {
+      *port->selected_port = r;
     }
   }
   if (!server->Start()) {

diff --git a/src/cpp/server/thread_pool.cc b/src/cpp/server/thread_pool.cc
@@ -66,8 +66,8 @@ ThreadPool::~ThreadPool() {
     shutdown_ = true;
     cv_.notify_all();
   }
-  for (auto& t : threads_) {
-    t.join();
+  for (auto t = threads_.begin(); t != threads_.end(); t++) {
+    t->join();
   }
 }
 

diff --git a/test/cpp/end2end/generic_end2end_test.cc b/test/cpp/end2end/generic_end2end_test.cc
@@ -83,8 +83,8 @@ bool ParseFromByteBuffer(ByteBuffer* buffer, grpc::protobuf::Message* message) {
   buffer->Dump(&slices);
   grpc::string buf;
   buf.reserve(buffer->Length());
-  for (const Slice& s : slices) {
-    buf.append(reinterpret_cast<const char*>(s.begin()), s.size());
+  for (auto s = slices.begin(); s != slices.end(); s++) {
+    buf.append(reinterpret_cast<const char*>(s->begin()), s->size());
   }
   return message->ParseFromString(buf);
 }

diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
@@ -144,15 +144,16 @@ class AsyncClient GRPC_FINAL : public Client {
 
     int t = 0;
     for (int i = 0; i < config.outstanding_rpcs_per_channel(); i++) {
-      for (auto& channel : channels_) {
+      for (auto channel = channels_.begin(); channel != channels_.end();
+	   channel++) {
         auto* cq = cli_cqs_[t].get();
         t = (t + 1) % cli_cqs_.size();
         auto start_req = [cq](TestService::Stub* stub, grpc::ClientContext* ctx,
                               const SimpleRequest& request, void* tag) {
           return stub->AsyncUnaryCall(ctx, request, cq, tag);
         };
 
-        TestService::Stub* stub = channel.get_stub();
+        TestService::Stub* stub = channel->get_stub();
         const SimpleRequest& request = request_;
         new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
             stub, request, start_req, check_done);
@@ -165,11 +166,11 @@ class AsyncClient GRPC_FINAL : public Client {
   ~AsyncClient() GRPC_OVERRIDE {
     EndThreads();
 
-    for (auto& cq : cli_cqs_) {
-      cq->Shutdown();
+    for (auto cq = cli_cqs_.begin(); cq != cli_cqs_.end(); cq++) {
+      (*cq)->Shutdown();
       void* got_tag;
       bool ok;
-      while (cq->Next(&got_tag, &ok)) {
+      while ((*cq)->Next(&got_tag, &ok)) {
         delete ClientRpcContext::detag(got_tag);
       }
     }