Merge branch 'master' of github.com:grpc/grpc into the-ultimate-showdown

Conflicts:
	include/grpc/grpc.h
	src/core/surface/channel.c
	src/core/surface/channel_create.c
	src/core/surface/completion_queue.c
	src/cpp/client/channel.cc
	src/cpp/client/insecure_credentials.cc
	src/csharp/ext/grpc_csharp_ext.c
	src/node/ext/call.cc
	src/node/ext/channel.cc
	src/php/ext/grpc/call.c
	src/php/ext/grpc/channel.c
	src/python/grpcio/grpc/_adapter/_c/types/channel.c
	src/ruby/ext/grpc/rb_channel.c
	test/core/end2end/dualstack_socket_test.c
	test/core/end2end/fixtures/chttp2_fullstack.c
	test/core/end2end/fixtures/chttp2_fullstack_compression.c
	test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c
	test/core/end2end/fixtures/chttp2_fullstack_with_poll.c
	test/core/end2end/multiple_server_queues_test.c
	test/core/end2end/no_server_test.c
	test/core/end2end/tests/bad_hostname.c
	test/core/end2end/tests/cancel_after_accept.c
	test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
	test/core/end2end/tests/cancel_after_invoke.c
	test/core/end2end/tests/cancel_before_invoke.c
	test/core/end2end/tests/cancel_in_a_vacuum.c
	test/core/end2end/tests/census_simple_request.c
	test/core/end2end/tests/disappearing_server.c
	test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
	test/core/end2end/tests/empty_batch.c
	test/core/end2end/tests/graceful_server_shutdown.c
	test/core/end2end/tests/invoke_large_request.c
	test/core/end2end/tests/max_concurrent_streams.c
	test/core/end2end/tests/max_message_length.c
	test/core/end2end/tests/ping_pong_streaming.c
	test/core/end2end/tests/registered_call.c
	test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
	test/core/end2end/tests/request_response_with_metadata_and_payload.c
	test/core/end2end/tests/request_response_with_payload.c
	test/core/end2end/tests/request_response_with_payload_and_call_creds.c
	test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c
	test/core/end2end/tests/request_with_compressed_payload.c
	test/core/end2end/tests/request_with_flags.c
	test/core/end2end/tests/request_with_large_metadata.c
	test/core/end2end/tests/request_with_payload.c
	test/core/end2end/tests/server_finishes_request.c
	test/core/end2end/tests/simple_delayed_request.c
	test/core/end2end/tests/simple_request.c
	test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
	test/core/fling/client.c
	test/core/fling/server.c
	test/core/surface/lame_client_test.c

5 files changed, 172 insertions, 6 deletions

diff --git a/src/cpp/server/create_default_thread_pool.cc b/src/cpp/server/create_default_thread_pool.cc
index cc182f59f4..81c84474d8 100644
--- a/src/cpp/server/create_default_thread_pool.cc
+++ b/src/cpp/server/create_default_thread_pool.cc
@@ -32,7 +32,7 @@
  */
 
 #include <grpc/support/cpu.h>
-#include <grpc++/fixed_size_thread_pool.h>
+#include <grpc++/dynamic_thread_pool.h>
 
 #ifndef GRPC_CUSTOM_DEFAULT_THREAD_POOL
 
@@ -41,7 +41,7 @@ namespace grpc {
 ThreadPoolInterface* CreateDefaultThreadPool() {
    int cores = gpr_cpu_num_cores();
    if (!cores) cores = 4;
-   return new FixedSizeThreadPool(cores);
+   return new DynamicThreadPool(cores);
 }
 
 }  // namespace grpc
diff --git a/src/cpp/server/dynamic_thread_pool.cc b/src/cpp/server/dynamic_thread_pool.cc
new file mode 100644
index 0000000000..f58d0420df
--- /dev/null
+++ b/src/cpp/server/dynamic_thread_pool.cc
@@ -0,0 +1,131 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <grpc++/impl/sync.h>
+#include <grpc++/impl/thd.h>
+#include <grpc++/dynamic_thread_pool.h>
+
+namespace grpc {
+DynamicThreadPool::DynamicThread::DynamicThread(DynamicThreadPool *pool):
+  pool_(pool),
+  thd_(new grpc::thread(&DynamicThreadPool::DynamicThread::ThreadFunc, this)) {
+}
+DynamicThreadPool::DynamicThread::~DynamicThread() {
+  thd_->join();
+  thd_.reset();
+}
+
+void DynamicThreadPool::DynamicThread::ThreadFunc() {
+  pool_->ThreadFunc();
+  // Now that we have killed ourselves, we should reduce the thread count
+  grpc::unique_lock<grpc::mutex> lock(pool_->mu_);
+  pool_->nthreads_--;
+  // Move ourselves to dead list
+  pool_->dead_threads_.push_back(this);
+
+  if ((pool_->shutdown_) && (pool_->nthreads_ == 0)) {
+    pool_->shutdown_cv_.notify_one();
+  }
+}
+  
+void DynamicThreadPool::ThreadFunc() {
+  for (;;) {
+    // Wait until work is available or we are shutting down.
+    grpc::unique_lock<grpc::mutex> lock(mu_);
+    if (!shutdown_ && callbacks_.empty()) {
+      // If there are too many threads waiting, then quit this thread
+      if (threads_waiting_ >= reserve_threads_) {
+	break;
+      }
+      threads_waiting_++;
+      cv_.wait(lock);
+      threads_waiting_--;
+    }
+    // Drain callbacks before considering shutdown to ensure all work
+    // gets completed.
+    if (!callbacks_.empty()) {
+      auto cb = callbacks_.front();
+      callbacks_.pop();
+      lock.unlock();
+      cb();
+    } else if (shutdown_) {
+      break;
+    }
+  }
+}
+
+DynamicThreadPool::DynamicThreadPool(int reserve_threads) :
+  shutdown_(false), reserve_threads_(reserve_threads), nthreads_(0),
+  threads_waiting_(0) {
+  for (int i = 0; i < reserve_threads_; i++) {
+    grpc::lock_guard<grpc::mutex> lock(mu_);
+    nthreads_++;
+    new DynamicThread(this);
+  }
+}
+
+void DynamicThreadPool::ReapThreads(std::list<DynamicThread*>* tlist) {
+  for (auto t = tlist->begin(); t != tlist->end(); t = tlist->erase(t)) {
+    delete *t;    
+  }
+}
+  
+DynamicThreadPool::~DynamicThreadPool() {
+  grpc::unique_lock<grpc::mutex> lock(mu_);
+  shutdown_ = true;
+  cv_.notify_all();
+  while (nthreads_ != 0) {
+    shutdown_cv_.wait(lock);
+  }
+  ReapThreads(&dead_threads_);
+}
+
+void DynamicThreadPool::Add(const std::function<void()>& callback) {
+  grpc::lock_guard<grpc::mutex> lock(mu_);
+  // Add works to the callbacks list
+  callbacks_.push(callback);
+  // Increase pool size or notify as needed
+  if (threads_waiting_ == 0) {
+    // Kick off a new thread
+    nthreads_++;
+    new DynamicThread(this);
+  } else {
+    cv_.notify_one();
+  }
+  // Also use this chance to harvest dead threads
+  if (!dead_threads_.empty()) {
+    ReapThreads(&dead_threads_);
+  }
+}
+
+}  // namespace grpc
diff --git a/src/cpp/server/insecure_server_credentials.cc b/src/cpp/server/insecure_server_credentials.cc
index aca3568e59..800cd36caa 100644
--- a/src/cpp/server/insecure_server_credentials.cc
+++ b/src/cpp/server/insecure_server_credentials.cc
@@ -41,7 +41,7 @@ class InsecureServerCredentialsImpl GRPC_FINAL : public ServerCredentials {
  public:
   int AddPortToServer(const grpc::string& addr,
                       grpc_server* server) GRPC_OVERRIDE {
-    return grpc_server_add_http2_port(server, addr.c_str());
+    return grpc_server_add_insecure_http2_port(server, addr.c_str());
   }
 };
 }  // namespace
diff --git a/src/cpp/server/secure_server_credentials.cc b/src/cpp/server/secure_server_credentials.cc
index 3e262dd74f..32c45e2280 100644
--- a/src/cpp/server/secure_server_credentials.cc
+++ b/src/cpp/server/secure_server_credentials.cc
@@ -51,7 +51,8 @@ std::shared_ptr<ServerCredentials> SslServerCredentials(
   }
   grpc_server_credentials* c_creds = grpc_ssl_server_credentials_create(
       options.pem_root_certs.empty() ? nullptr : options.pem_root_certs.c_str(),
-      &pem_key_cert_pairs[0], pem_key_cert_pairs.size());
+      &pem_key_cert_pairs[0], pem_key_cert_pairs.size(),
+      options.force_client_auth);
   return std::shared_ptr<ServerCredentials>(
       new SecureServerCredentials(c_creds));
 }
diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc
index a178772548..bb34040a2f 100644
--- a/src/cpp/server/server_context.cc
+++ b/src/cpp/server/server_context.cc
@@ -34,6 +34,7 @@
 #include <grpc++/server_context.h>
 
 #include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc++/impl/call.h>
 #include <grpc++/impl/sync.h>
@@ -49,16 +50,23 @@ namespace grpc {
 class ServerContext::CompletionOp GRPC_FINAL : public CallOpSetInterface {
  public:
   // initial refs: one in the server context, one in the cq
-  CompletionOp() : refs_(2), finalized_(false), cancelled_(0) {}
+  CompletionOp() : has_tag_(false), tag_(nullptr), refs_(2), finalized_(false), cancelled_(0) {}
 
   void FillOps(grpc_op* ops, size_t* nops) GRPC_OVERRIDE;
   bool FinalizeResult(void** tag, bool* status) GRPC_OVERRIDE;
 
   bool CheckCancelled(CompletionQueue* cq);
 
+  void set_tag(void* tag) {
+    has_tag_ = true;
+    tag_ = tag;
+  }
+
   void Unref();
 
  private:
+  bool has_tag_;
+  void* tag_;
   grpc::mutex mu_;
   int refs_;
   bool finalized_;
@@ -90,18 +98,25 @@ void ServerContext::CompletionOp::FillOps(grpc_op* ops, size_t* nops) {
 bool ServerContext::CompletionOp::FinalizeResult(void** tag, bool* status) {
   grpc::unique_lock<grpc::mutex> lock(mu_);
   finalized_ = true;
+  bool ret = false;
+  if (has_tag_) {
+    *tag = tag_;
+    ret = true;
+  }
   if (!*status) cancelled_ = 1;
   if (--refs_ == 0) {
     lock.unlock();
     delete this;
   }
-  return false;
+  return ret;
 }
 
 // ServerContext body
 
 ServerContext::ServerContext()
     : completion_op_(nullptr),
+      has_notify_when_done_tag_(false),
+      async_notify_when_done_tag_(nullptr),
       call_(nullptr),
       cq_(nullptr),
       sent_initial_metadata_(false) {}
@@ -109,6 +124,8 @@ ServerContext::ServerContext()
 ServerContext::ServerContext(gpr_timespec deadline, grpc_metadata* metadata,
                              size_t metadata_count)
     : completion_op_(nullptr),
+      has_notify_when_done_tag_(false),
+      async_notify_when_done_tag_(nullptr),
       deadline_(deadline),
       call_(nullptr),
       cq_(nullptr),
@@ -133,6 +150,9 @@ ServerContext::~ServerContext() {
 void ServerContext::BeginCompletionOp(Call* call) {
   GPR_ASSERT(!completion_op_);
   completion_op_ = new CompletionOp();
+  if (has_notify_when_done_tag_) {
+    completion_op_->set_tag(async_notify_when_done_tag_);
+  }
   call->PerformOps(completion_op_);
 }
 
@@ -180,4 +200,18 @@ std::shared_ptr<const AuthContext> ServerContext::auth_context() const {
   return auth_context_;
 }
 
+grpc::string ServerContext::peer() const {
+  grpc::string peer;
+  if (call_) {
+    char* c_peer = grpc_call_get_peer(call_);
+    peer = c_peer;
+    gpr_free(c_peer);
+  }
+  return peer;
+}
+
+const struct census_context* ServerContext::census_context() const {
+  return grpc_census_call_get_context(call_);
+}
+
 }  // namespace grpc