Merge branch 'master' of github.com:grpc/grpc into cpp_string_ref

12 files changed, 382 insertions, 65 deletions

diff --git a/src/cpp/server/async_generic_service.cc b/src/cpp/server/async_generic_service.cc
index 2e99afcb5f..6b9ea532b6 100644
--- a/src/cpp/server/async_generic_service.cc
+++ b/src/cpp/server/async_generic_service.cc
@@ -31,7 +31,7 @@
  *
  */
 
-#include <grpc++/async_generic_service.h>
+#include <grpc++/generic/async_generic_service.h>
 
 #include <grpc++/server.h>
 
diff --git a/src/cpp/server/create_default_thread_pool.cc b/src/cpp/server/create_default_thread_pool.cc
index 81c84474d8..f3b07ec8ce 100644
--- a/src/cpp/server/create_default_thread_pool.cc
+++ b/src/cpp/server/create_default_thread_pool.cc
@@ -32,16 +32,17 @@
  */
 
 #include <grpc/support/cpu.h>
-#include <grpc++/dynamic_thread_pool.h>
+
+#include "src/cpp/server/dynamic_thread_pool.h"
 
 #ifndef GRPC_CUSTOM_DEFAULT_THREAD_POOL
 
 namespace grpc {
 
 ThreadPoolInterface* CreateDefaultThreadPool() {
-   int cores = gpr_cpu_num_cores();
-   if (!cores) cores = 4;
-   return new DynamicThreadPool(cores);
+  int cores = gpr_cpu_num_cores();
+  if (!cores) cores = 4;
+  return new DynamicThreadPool(cores);
 }
 
 }  // namespace grpc
diff --git a/src/cpp/server/dynamic_thread_pool.cc b/src/cpp/server/dynamic_thread_pool.cc
index f58d0420df..4b226c2992 100644
--- a/src/cpp/server/dynamic_thread_pool.cc
+++ b/src/cpp/server/dynamic_thread_pool.cc
@@ -33,13 +33,14 @@
 
 #include <grpc++/impl/sync.h>
 #include <grpc++/impl/thd.h>
-#include <grpc++/dynamic_thread_pool.h>
+
+#include "src/cpp/server/dynamic_thread_pool.h"
 
 namespace grpc {
-DynamicThreadPool::DynamicThread::DynamicThread(DynamicThreadPool *pool):
-  pool_(pool),
-  thd_(new grpc::thread(&DynamicThreadPool::DynamicThread::ThreadFunc, this)) {
-}
+DynamicThreadPool::DynamicThread::DynamicThread(DynamicThreadPool* pool)
+    : pool_(pool),
+      thd_(new grpc::thread(&DynamicThreadPool::DynamicThread::ThreadFunc,
+                            this)) {}
 DynamicThreadPool::DynamicThread::~DynamicThread() {
   thd_->join();
   thd_.reset();
@@ -57,7 +58,7 @@ void DynamicThreadPool::DynamicThread::ThreadFunc() {
     pool_->shutdown_cv_.notify_one();
   }
 }
-  
+
 void DynamicThreadPool::ThreadFunc() {
   for (;;) {
     // Wait until work is available or we are shutting down.
@@ -65,7 +66,7 @@ void DynamicThreadPool::ThreadFunc() {
     if (!shutdown_ && callbacks_.empty()) {
       // If there are too many threads waiting, then quit this thread
       if (threads_waiting_ >= reserve_threads_) {
-	break;
+        break;
       }
       threads_waiting_++;
       cv_.wait(lock);
@@ -84,9 +85,11 @@ void DynamicThreadPool::ThreadFunc() {
   }
 }
 
-DynamicThreadPool::DynamicThreadPool(int reserve_threads) :
-  shutdown_(false), reserve_threads_(reserve_threads), nthreads_(0),
-  threads_waiting_(0) {
+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_++;
@@ -96,10 +99,10 @@ DynamicThreadPool::DynamicThreadPool(int reserve_threads) :
 
 void DynamicThreadPool::ReapThreads(std::list<DynamicThread*>* tlist) {
   for (auto t = tlist->begin(); t != tlist->end(); t = tlist->erase(t)) {
-    delete *t;    
+    delete *t;
   }
 }
-  
+
 DynamicThreadPool::~DynamicThreadPool() {
   grpc::unique_lock<grpc::mutex> lock(mu_);
   shutdown_ = true;
diff --git a/src/cpp/server/dynamic_thread_pool.h b/src/cpp/server/dynamic_thread_pool.h
new file mode 100644
index 0000000000..5ba7533c05
--- /dev/null
+++ b/src/cpp/server/dynamic_thread_pool.h
@@ -0,0 +1,83 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#ifndef GRPC_INTERNAL_CPP_DYNAMIC_THREAD_POOL_H
+#define GRPC_INTERNAL_CPP_DYNAMIC_THREAD_POOL_H
+
+#include <list>
+#include <memory>
+#include <queue>
+
+#include <grpc++/impl/sync.h>
+#include <grpc++/impl/thd.h>
+#include <grpc++/support/config.h>
+
+#include "src/cpp/server/thread_pool_interface.h"
+
+namespace grpc {
+
+class DynamicThreadPool GRPC_FINAL : public ThreadPoolInterface {
+ public:
+  explicit DynamicThreadPool(int reserve_threads);
+  ~DynamicThreadPool();
+
+  void Add(const std::function<void()>& callback) GRPC_OVERRIDE;
+
+ private:
+  class DynamicThread {
+   public:
+    DynamicThread(DynamicThreadPool* pool);
+    ~DynamicThread();
+
+   private:
+    DynamicThreadPool* pool_;
+    std::unique_ptr<grpc::thread> thd_;
+    void ThreadFunc();
+  };
+  grpc::mutex mu_;
+  grpc::condition_variable cv_;
+  grpc::condition_variable shutdown_cv_;
+  bool shutdown_;
+  std::queue<std::function<void()>> callbacks_;
+  int reserve_threads_;
+  int nthreads_;
+  int threads_waiting_;
+  std::list<DynamicThread*> dead_threads_;
+
+  void ThreadFunc();
+  static void ReapThreads(std::list<DynamicThread*>* tlist);
+};
+
+}  // namespace grpc
+
+#endif  // GRPC_INTERNAL_CPP_DYNAMIC_THREAD_POOL_H
diff --git a/src/cpp/server/fixed_size_thread_pool.cc b/src/cpp/server/fixed_size_thread_pool.cc
index bafbc5802a..2bdc44be2e 100644
--- a/src/cpp/server/fixed_size_thread_pool.cc
+++ b/src/cpp/server/fixed_size_thread_pool.cc
@@ -33,7 +33,7 @@
 
 #include <grpc++/impl/sync.h>
 #include <grpc++/impl/thd.h>
-#include <grpc++/fixed_size_thread_pool.h>
+#include "src/cpp/server/fixed_size_thread_pool.h"
 
 namespace grpc {
 
diff --git a/src/cpp/server/fixed_size_thread_pool.h b/src/cpp/server/fixed_size_thread_pool.h
new file mode 100644
index 0000000000..394ae5821e
--- /dev/null
+++ b/src/cpp/server/fixed_size_thread_pool.h
@@ -0,0 +1,67 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#ifndef GRPC_INTERNAL_CPP_FIXED_SIZE_THREAD_POOL_H
+#define GRPC_INTERNAL_CPP_FIXED_SIZE_THREAD_POOL_H
+
+#include <queue>
+#include <vector>
+
+#include <grpc++/impl/sync.h>
+#include <grpc++/impl/thd.h>
+#include <grpc++/support/config.h>
+
+#include "src/cpp/server/thread_pool_interface.h"
+
+namespace grpc {
+
+class FixedSizeThreadPool GRPC_FINAL : public ThreadPoolInterface {
+ public:
+  explicit FixedSizeThreadPool(int num_threads);
+  ~FixedSizeThreadPool();
+
+  void Add(const std::function<void()>& callback) GRPC_OVERRIDE;
+
+ private:
+  grpc::mutex mu_;
+  grpc::condition_variable cv_;
+  bool shutdown_;
+  std::queue<std::function<void()>> callbacks_;
+  std::vector<grpc::thread*> threads_;
+
+  void ThreadFunc();
+};
+
+}  // namespace grpc
+
+#endif  // GRPC_INTERNAL_CPP_FIXED_SIZE_THREAD_POOL_H
diff --git a/src/cpp/server/secure_server_credentials.cc b/src/cpp/server/secure_server_credentials.cc
index 32c45e2280..f203cf7f49 100644
--- a/src/cpp/server/secure_server_credentials.cc
+++ b/src/cpp/server/secure_server_credentials.cc
@@ -35,8 +35,8 @@
 
 namespace grpc {
 
-int SecureServerCredentials::AddPortToServer(
-    const grpc::string& addr, grpc_server* server) {
+int SecureServerCredentials::AddPortToServer(const grpc::string& addr,
+                                             grpc_server* server) {
   return grpc_server_add_secure_http2_port(server, addr.c_str(), creds_);
 }
 
diff --git a/src/cpp/server/secure_server_credentials.h b/src/cpp/server/secure_server_credentials.h
index b9803f107e..d3d37b188d 100644
--- a/src/cpp/server/secure_server_credentials.h
+++ b/src/cpp/server/secure_server_credentials.h
@@ -34,10 +34,10 @@
 #ifndef GRPC_INTERNAL_CPP_SERVER_SECURE_SERVER_CREDENTIALS_H
 #define GRPC_INTERNAL_CPP_SERVER_SECURE_SERVER_CREDENTIALS_H
 
-#include <grpc/grpc_security.h>
-
 #include <grpc++/server_credentials.h>
 
+#include <grpc/grpc_security.h>
+
 namespace grpc {
 
 class SecureServerCredentials GRPC_FINAL : public ServerCredentials {
diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
index a70b555855..66cd27cc33 100644
--- a/src/cpp/server/server.cc
+++ b/src/cpp/server/server.cc
@@ -32,24 +32,71 @@
  */
 
 #include <grpc++/server.h>
+
 #include <utility>
 
 #include <grpc/grpc.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc++/completion_queue.h>
-#include <grpc++/async_generic_service.h>
+#include <grpc++/generic/async_generic_service.h>
 #include <grpc++/impl/rpc_service_method.h>
 #include <grpc++/impl/service_type.h>
 #include <grpc++/server_context.h>
 #include <grpc++/server_credentials.h>
-#include <grpc++/thread_pool_interface.h>
-#include <grpc++/time.h>
+#include <grpc++/support/time.h>
 
 #include "src/core/profiling/timers.h"
+#include "src/cpp/server/thread_pool_interface.h"
 
 namespace grpc {
 
+class Server::UnimplementedAsyncRequestContext {
+ protected:
+  UnimplementedAsyncRequestContext() : generic_stream_(&server_context_) {}
+
+  GenericServerContext server_context_;
+  GenericServerAsyncReaderWriter generic_stream_;
+};
+
+class Server::UnimplementedAsyncRequest GRPC_FINAL
+    : public UnimplementedAsyncRequestContext,
+      public GenericAsyncRequest {
+ public:
+  UnimplementedAsyncRequest(Server* server, ServerCompletionQueue* cq)
+      : GenericAsyncRequest(server, &server_context_, &generic_stream_, cq, cq,
+                            NULL, false),
+        server_(server),
+        cq_(cq) {}
+
+  bool FinalizeResult(void** tag, bool* status) GRPC_OVERRIDE;
+
+  ServerContext* context() { return &server_context_; }
+  GenericServerAsyncReaderWriter* stream() { return &generic_stream_; }
+
+ private:
+  Server* const server_;
+  ServerCompletionQueue* const cq_;
+};
+
+typedef SneakyCallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus>
+    UnimplementedAsyncResponseOp;
+class Server::UnimplementedAsyncResponse GRPC_FINAL
+    : public UnimplementedAsyncResponseOp {
+ public:
+  UnimplementedAsyncResponse(UnimplementedAsyncRequest* request);
+  ~UnimplementedAsyncResponse() { delete request_; }
+
+  bool FinalizeResult(void** tag, bool* status) GRPC_OVERRIDE {
+    bool r = UnimplementedAsyncResponseOp::FinalizeResult(tag, status);
+    delete this;
+    return r;
+  }
+
+ private:
+  UnimplementedAsyncRequest* const request_;
+};
+
 class Server::ShutdownRequest GRPC_FINAL : public CompletionQueueTag {
  public:
   bool FinalizeResult(void** tag, bool* status) {
@@ -90,6 +137,26 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
     return mrd;
   }
 
+  static bool AsyncWait(CompletionQueue* cq, SyncRequest** req, bool* ok,
+                        gpr_timespec deadline) {
+    void* tag = nullptr;
+    *ok = false;
+    switch (cq->AsyncNext(&tag, ok, deadline)) {
+      case CompletionQueue::TIMEOUT:
+        *req = nullptr;
+        return true;
+      case CompletionQueue::SHUTDOWN:
+        *req = nullptr;
+        return false;
+      case CompletionQueue::GOT_EVENT:
+        *req = static_cast<SyncRequest*>(tag);
+        GPR_ASSERT((*req)->in_flight_);
+        return true;
+    }
+    gpr_log(GPR_ERROR, "Should never reach here");
+    abort();
+  }
+
   void SetupRequest() { cq_ = grpc_completion_queue_create(nullptr); }
 
   void TeardownRequest() {
@@ -230,11 +297,11 @@ Server::~Server() {
   delete sync_methods_;
 }
 
-bool Server::RegisterService(const grpc::string *host, RpcService* service) {
+bool Server::RegisterService(const grpc::string* host, RpcService* service) {
   for (int i = 0; i < service->GetMethodCount(); ++i) {
     RpcServiceMethod* method = service->GetMethod(i);
-    void* tag = grpc_server_register_method(
-        server_, method->name(), host ? host->c_str() : nullptr);
+    void* tag = grpc_server_register_method(server_, method->name(),
+                                            host ? host->c_str() : nullptr);
     if (!tag) {
       gpr_log(GPR_DEBUG, "Attempt to register %s multiple times",
               method->name());
@@ -277,18 +344,23 @@ int Server::AddListeningPort(const grpc::string& addr,
   return creds->AddPortToServer(addr, server_);
 }
 
-bool Server::Start() {
+bool Server::Start(ServerCompletionQueue** cqs, size_t num_cqs) {
   GPR_ASSERT(!started_);
   started_ = true;
   grpc_server_start(server_);
 
   if (!has_generic_service_) {
-    unknown_method_.reset(new RpcServiceMethod(
-        "unknown", RpcMethod::BIDI_STREAMING, new UnknownMethodHandler));
-    // Use of emplace_back with just constructor arguments is not accepted here
-    // by gcc-4.4 because it can't match the anonymous nullptr with a proper
-    // constructor implicitly. Construct the object and use push_back.
-    sync_methods_->push_back(SyncRequest(unknown_method_.get(), nullptr));
+    if (!sync_methods_->empty()) {
+      unknown_method_.reset(new RpcServiceMethod(
+          "unknown", RpcMethod::BIDI_STREAMING, new UnknownMethodHandler));
+      // Use of emplace_back with just constructor arguments is not accepted
+      // here by gcc-4.4 because it can't match the anonymous nullptr with a 
+      // proper constructor implicitly. Construct the object and use push_back.
+      sync_methods_->push_back(SyncRequest(unknown_method_.get(), nullptr));
+    }
+    for (size_t i = 0; i < num_cqs; i++) {
+      new UnimplementedAsyncRequest(this, cqs[i]);
+    }
   }
   // Start processing rpcs.
   if (!sync_methods_->empty()) {
@@ -303,12 +375,27 @@ bool Server::Start() {
   return true;
 }
 
-void Server::Shutdown() {
+void Server::ShutdownInternal(gpr_timespec deadline) {
   grpc::unique_lock<grpc::mutex> lock(mu_);
   if (started_ && !shutdown_) {
     shutdown_ = true;
     grpc_server_shutdown_and_notify(server_, cq_.cq(), new ShutdownRequest());
     cq_.Shutdown();
+    // Spin, eating requests until the completion queue is completely shutdown.
+    // If the deadline expires then cancel anything that's pending and keep
+    // spinning forever until the work is actually drained.
+    // Since nothing else needs to touch state guarded by mu_, holding it 
+    // through this loop is fine.
+    SyncRequest* request;
+    bool ok;
+    while (SyncRequest::AsyncWait(&cq_, &request, &ok, deadline)) {
+      if (request == NULL) {  // deadline expired
+        grpc_server_cancel_all_calls(server_);
+        deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
+      } else if (ok) {
+        SyncRequest::CallData call_data(this, request);
+      }
+    }
 
     // Wait for running callbacks to finish.
     while (num_running_cb_ != 0) {
@@ -335,12 +422,14 @@ void Server::PerformOpsOnCall(CallOpSetInterface* ops, Call* call) {
 
 Server::BaseAsyncRequest::BaseAsyncRequest(
     Server* server, ServerContext* context,
-    ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, void* tag)
+    ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, void* tag,
+    bool delete_on_finalize)
     : server_(server),
       context_(context),
       stream_(stream),
       call_cq_(call_cq),
       tag_(tag),
+      delete_on_finalize_(delete_on_finalize),
       call_(nullptr) {
   memset(&initial_metadata_array_, 0, sizeof(initial_metadata_array_));
 }
@@ -367,14 +456,16 @@ bool Server::BaseAsyncRequest::FinalizeResult(void** tag, bool* status) {
   // just the pointers inside call are copied here
   stream_->BindCall(&call);
   *tag = tag_;
-  delete this;
+  if (delete_on_finalize_) {
+    delete this;
+  }
   return true;
 }
 
 Server::RegisteredAsyncRequest::RegisteredAsyncRequest(
     Server* server, ServerContext* context,
     ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, void* tag)
-    : BaseAsyncRequest(server, context, stream, call_cq, tag) {}
+    : BaseAsyncRequest(server, context, stream, call_cq, tag, true) {}
 
 void Server::RegisteredAsyncRequest::IssueRequest(
     void* registered_method, grpc_byte_buffer** payload,
@@ -388,8 +479,9 @@ void Server::RegisteredAsyncRequest::IssueRequest(
 Server::GenericAsyncRequest::GenericAsyncRequest(
     Server* server, GenericServerContext* context,
     ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
-    ServerCompletionQueue* notification_cq, void* tag)
-    : BaseAsyncRequest(server, context, stream, call_cq, tag) {
+    ServerCompletionQueue* notification_cq, void* tag, bool delete_on_finalize)
+    : BaseAsyncRequest(server, context, stream, call_cq, tag,
+                       delete_on_finalize) {
   grpc_call_details_init(&call_details_);
   GPR_ASSERT(notification_cq);
   GPR_ASSERT(call_cq);
@@ -410,6 +502,25 @@ bool Server::GenericAsyncRequest::FinalizeResult(void** tag, bool* status) {
   return BaseAsyncRequest::FinalizeResult(tag, status);
 }
 
+bool Server::UnimplementedAsyncRequest::FinalizeResult(void** tag,
+                                                       bool* status) {
+  if (GenericAsyncRequest::FinalizeResult(tag, status) && *status) {
+    new UnimplementedAsyncRequest(server_, cq_);
+    new UnimplementedAsyncResponse(this);
+  } else {
+    delete this;
+  }
+  return false;
+}
+
+Server::UnimplementedAsyncResponse::UnimplementedAsyncResponse(
+    UnimplementedAsyncRequest* request)
+    : request_(request) {
+  Status status(StatusCode::UNIMPLEMENTED, "");
+  UnknownMethodHandler::FillOps(request_->context(), this);
+  request_->stream()->call_.PerformOps(this);
+}
+
 void Server::ScheduleCallback() {
   {
     grpc::unique_lock<grpc::mutex> lock(mu_);
diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc
index 09118879f4..b739cbfe62 100644
--- a/src/cpp/server/server_builder.cc
+++ b/src/cpp/server/server_builder.cc
@@ -37,8 +37,8 @@
 #include <grpc/support/log.h>
 #include <grpc++/impl/service_type.h>
 #include <grpc++/server.h>
-#include <grpc++/thread_pool_interface.h>
-#include <grpc++/fixed_size_thread_pool.h>
+#include "src/cpp/server/thread_pool_interface.h"
+#include "src/cpp/server/fixed_size_thread_pool.h"
 
 namespace grpc {
 
@@ -59,14 +59,16 @@ void ServerBuilder::RegisterAsyncService(AsynchronousService* service) {
   async_services_.emplace_back(new NamedService<AsynchronousService>(service));
 }
 
-void ServerBuilder::RegisterService(
-    const grpc::string& addr, SynchronousService* service) {
-  services_.emplace_back(new NamedService<RpcService>(addr, service->service()));
+void ServerBuilder::RegisterService(const grpc::string& addr,
+                                    SynchronousService* service) {
+  services_.emplace_back(
+      new NamedService<RpcService>(addr, service->service()));
 }
 
-void ServerBuilder::RegisterAsyncService(
-    const grpc::string& addr, AsynchronousService* service) {
-  async_services_.emplace_back(new NamedService<AsynchronousService>(addr, service));
+void ServerBuilder::RegisterAsyncService(const grpc::string& addr,
+                                         AsynchronousService* service) {
+  async_services_.emplace_back(
+      new NamedService<AsynchronousService>(addr, service));
 }
 
 void ServerBuilder::RegisterAsyncGenericService(AsyncGenericService* service) {
@@ -87,10 +89,6 @@ void ServerBuilder::AddListeningPort(const grpc::string& addr,
   ports_.push_back(port);
 }
 
-void ServerBuilder::SetThreadPool(ThreadPoolInterface* thread_pool) {
-  thread_pool_ = thread_pool;
-}
-
 std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
   bool thread_pool_owned = false;
   if (!async_services_.empty() && !services_.empty()) {
@@ -101,12 +99,6 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
     thread_pool_ = CreateDefaultThreadPool();
     thread_pool_owned = true;
   }
-  // Async services only, create a thread pool to handle requests to unknown
-  // services.
-  if (!thread_pool_ && !generic_service_ && !async_services_.empty()) {
-    thread_pool_ = new FixedSizeThreadPool(1);
-    thread_pool_owned = true;
-  }
   std::unique_ptr<Server> server(
       new Server(thread_pool_, thread_pool_owned, max_message_size_));
   for (auto cq = cqs_.begin(); cq != cqs_.end(); ++cq) {
@@ -119,9 +111,10 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
       return nullptr;
     }
   }
-  for (auto service = async_services_.begin();
-       service != async_services_.end(); service++) {
-    if (!server->RegisterAsyncService((*service)->host.get(), (*service)->service)) {
+  for (auto service = async_services_.begin(); service != async_services_.end();
+       service++) {
+    if (!server->RegisterAsyncService((*service)->host.get(),
+                                      (*service)->service)) {
       return nullptr;
     }
   }
@@ -135,7 +128,7 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
       *port->selected_port = r;
     }
   }
-  if (!server->Start()) {
+  if (!server->Start(&cqs_[0], cqs_.size())) {
     return nullptr;
   }
   return server;
diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc
index bb34040a2f..acc163d6b5 100644
--- a/src/cpp/server/server_context.cc
+++ b/src/cpp/server/server_context.cc
@@ -38,7 +38,7 @@
 #include <grpc/support/log.h>
 #include <grpc++/impl/call.h>
 #include <grpc++/impl/sync.h>
-#include <grpc++/time.h>
+#include <grpc++/support/time.h>
 
 #include "src/core/channel/compress_filter.h"
 #include "src/cpp/common/create_auth_context.h"
@@ -50,7 +50,12 @@ namespace grpc {
 class ServerContext::CompletionOp GRPC_FINAL : public CallOpSetInterface {
  public:
   // initial refs: one in the server context, one in the cq
-  CompletionOp() : has_tag_(false), tag_(nullptr), 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;
diff --git a/src/cpp/server/thread_pool_interface.h b/src/cpp/server/thread_pool_interface.h
new file mode 100644
index 0000000000..1ebe30fe2a
--- /dev/null
+++ b/src/cpp/server/thread_pool_interface.h
@@ -0,0 +1,54 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#ifndef GRPC_INTERNAL_CPP_THREAD_POOL_INTERFACE_H
+#define GRPC_INTERNAL_CPP_THREAD_POOL_INTERFACE_H
+
+#include <functional>
+
+namespace grpc {
+
+// A thread pool interface for running callbacks.
+class ThreadPoolInterface {
+ public:
+  virtual ~ThreadPoolInterface() {}
+
+  // Schedule the given callback for execution.
+  virtual void Add(const std::function<void()>& callback) = 0;
+};
+
+ThreadPoolInterface* CreateDefaultThreadPool();
+
+}  // namespace grpc
+
+#endif  // GRPC_INTERNAL_CPP_THREAD_POOL_INTERFACE_H