Merge github.com:grpc/grpc into y12kdm3

crast · Aug 21, 2015 · b7e22b9 · b7e22b9
2 parents 849c7ca + b7e55a2
commit b7e22b9
Show file tree

Hide file tree

Showing 80 changed files with 2,071 additions and 377 deletions.
diff --git a/INSTALL b/INSTALL
@@ -9,25 +9,40 @@ wiki pages:
 * If you are in a hurry *
 *************************
 
+On Linux (Debian):
+
+ Note: you will need to add the Debian 'unstable' distribution to your sources
+ file first.
+
+ Add the following line to your `/etc/apt/sources.list` file:
+
+   deb http://ftp.us.debian.org/debian unstable main contrib non-free
+
+ Install the gRPC library:
+
+ $ [sudo] apt-get install libgrpc-dev
+
+OR
+
  $ git clone https://github.com/grpc/grpc.git
  $ cd grpc
  $ git submodule update --init
  $ make 
- $ sudo make install
+ $ [sudo] make install
 
 You don't need anything else than GNU Make, gcc and autotools. Under a Debian
 or Ubuntu system, this should boil down to the following packages:
 
-  $ apt-get install build-essential autoconf libtool
+ $ [sudo] apt-get install build-essential autoconf libtool
 
 Building the python wrapper requires the following:
 
-  # apt-get install python-all-dev python-virtualenv
+ $ [sudo] apt-get install python-all-dev python-virtualenv
 
 If you want to install in a different directory than the default /usr/lib, you can
 override it on the command line:
 
-  # make install prefix=/opt
+ $ [sudo] make install prefix=/opt
 
 
 *******************************

diff --git a/Makefile b/Makefile
@@ -889,6 +889,7 @@ reconnect_interop_server: $(BINDIR)/$(CONFIG)/reconnect_interop_server
 secure_auth_context_test: $(BINDIR)/$(CONFIG)/secure_auth_context_test
 server_crash_test: $(BINDIR)/$(CONFIG)/server_crash_test
 server_crash_test_client: $(BINDIR)/$(CONFIG)/server_crash_test_client
+shutdown_test: $(BINDIR)/$(CONFIG)/shutdown_test
 status_test: $(BINDIR)/$(CONFIG)/status_test
 sync_streaming_ping_pong_test: $(BINDIR)/$(CONFIG)/sync_streaming_ping_pong_test
 sync_unary_ping_pong_test: $(BINDIR)/$(CONFIG)/sync_unary_ping_pong_test
@@ -1735,7 +1736,7 @@ buildtests_c: privatelibs_c $(BINDIR)/$(CONFIG)/alarm_heap_test $(BINDIR)/$(CONF
 buildtests_cxx: buildtests_zookeeper privatelibs_cxx $(BINDIR)/$(CONFIG)/async_end2end_test $(BINDIR)/$(CONFIG)/async_streaming_ping_pong_test $(BINDIR)/$(CONFIG)/async_unary_ping_pong_test $(BINDIR)/$(CONFIG)/auth_property_iterator_test $(BINDIR)/$(CONFIG)/channel_arguments_test $(BINDIR)/$(CONFIG)/cli_call_test $(BINDIR)/$(CONFIG)/client_crash_test $(BINDIR)/$(CONFIG)/client_crash_test_server $(BINDIR)/$(CONFIG)/credentials_test $(BINDIR)/$(CONFIG)/cxx_byte_buffer_test $(BINDIR)/$(CONFIG)/cxx_slice_test $(BINDIR)/$(CONFIG)/cxx_time_test $(BINDIR)/$(CONFIG)/dynamic_thread_pool_test $(BINDIR)/$(CONFIG)/end2end_test $(BINDIR)/$(CONFIG)/fixed_size_thread_pool_test $(BINDIR)/$(CONFIG)/generic_end2end_test $(BINDIR)/$(CONFIG)/grpc_cli $(BINDIR)/$(CONFIG)/interop_client $(BINDIR)/$(CONFIG)/interop_server $(BINDIR)/$(CONFIG)/interop_test $(BINDIR)/$(CONFIG)/mock_test $(BINDIR)/$(CONFIG)/qps_interarrival_test $(BINDIR)/$(CONFIG)/qps_openloop_test $(BINDIR)/$(CONFIG)/qps_test $(BINDIR)/$(CONFIG)/reconnect_interop_client $(BINDIR)/$(CONFIG)/reconnect_interop_server $(BINDIR)/$(CONFIG)/secure_auth_context_test $(BINDIR)/$(CONFIG)/server_crash_test $(BINDIR)/$(CONFIG)/server_crash_test_client $(BINDIR)/$(CONFIG)/status_test $(BINDIR)/$(CONFIG)/sync_streaming_ping_pong_test $(BINDIR)/$(CONFIG)/sync_unary_ping_pong_test $(BINDIR)/$(CONFIG)/thread_stress_test
 
 ifeq ($(HAS_ZOOKEEPER),true)
-buildtests_zookeeper: privatelibs_zookeeper $(BINDIR)/$(CONFIG)/zookeeper_test
+buildtests_zookeeper: privatelibs_zookeeper $(BINDIR)/$(CONFIG)/shutdown_test $(BINDIR)/$(CONFIG)/zookeeper_test
 else
 buildtests_zookeeper:
 endif
@@ -3363,6 +3364,8 @@ flaky_test_cxx: buildtests_cxx
 
 ifeq ($(HAS_ZOOKEEPER),true)
 test_zookeeper: buildtests_zookeeper
+	$(E) "[RUN]     Testing shutdown_test"
+	$(Q) $(BINDIR)/$(CONFIG)/shutdown_test || ( echo test shutdown_test failed ; exit 1 )
 	$(E) "[RUN]     Testing zookeeper_test"
 	$(Q) $(BINDIR)/$(CONFIG)/zookeeper_test || ( echo test zookeeper_test failed ; exit 1 )
 
@@ -10262,6 +10265,46 @@ endif
 endif
 
 
+SHUTDOWN_TEST_SRC = \
+    test/cpp/end2end/shutdown_test.cc \
+
+SHUTDOWN_TEST_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(SHUTDOWN_TEST_SRC))))
+ifeq ($(NO_SECURE),true)
+
+# You can't build secure targets if you don't have OpenSSL.
+
+$(BINDIR)/$(CONFIG)/shutdown_test: openssl_dep_error
+
+else
+
+
+ifeq ($(NO_PROTOBUF),true)
+
+# You can't build the protoc plugins or protobuf-enabled targets if you don't have protobuf 3.0.0+.
+
+$(BINDIR)/$(CONFIG)/shutdown_test: protobuf_dep_error
+
+else
+
+$(BINDIR)/$(CONFIG)/shutdown_test: $(PROTOBUF_DEP) $(SHUTDOWN_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc_zookeeper.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
+	$(E) "[LD]      Linking $@"
+	$(Q) mkdir -p `dirname $@`
+	$(Q) $(LDXX) $(LDFLAGS) $(SHUTDOWN_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc_zookeeper.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a -lzookeeper_mt $(LDLIBSXX) $(LDLIBS_PROTOBUF) $(LDLIBS) $(LDLIBS_SECURE) $(GTEST_LIB) -o $(BINDIR)/$(CONFIG)/shutdown_test
+
+endif
+
+endif
+
+$(OBJDIR)/$(CONFIG)/test/cpp/end2end/shutdown_test.o:  $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc_zookeeper.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
+deps_shutdown_test: $(SHUTDOWN_TEST_OBJS:.o=.dep)
+
+ifneq ($(NO_SECURE),true)
+ifneq ($(NO_DEPS),true)
+-include $(SHUTDOWN_TEST_OBJS:.o=.dep)
+endif
+endif
+
+
 STATUS_TEST_SRC = \
     test/cpp/util/status_test.cc \
 

diff --git a/build.json b/build.json
@@ -2488,6 +2488,9 @@
         "gpr",
         "grpc++_test_config"
       ],
+      "exclude_configs": [
+        "tsan"
+      ],
       "platforms": [
         "mac",
         "linux",
@@ -2610,6 +2613,26 @@
         "gpr"
       ]
     },
+    {
+      "name": "shutdown_test",
+      "build": "test",
+      "language": "c++",
+      "src": [
+        "test/cpp/end2end/shutdown_test.cc"
+      ],
+      "deps": [
+        "grpc++_test_util",
+        "grpc_test_util",
+        "grpc++",
+        "grpc_zookeeper",
+        "grpc",
+        "gpr_test_util",
+        "gpr"
+      ],
+      "external_deps": [
+        "zookeeper"
+      ]
+    },
     {
       "name": "status_test",
       "build": "test",

diff --git a/doc/connection-backoff-interop-test-description.md b/doc/connection-backoff-interop-test-description.md
@@ -31,9 +31,9 @@ Clients should accept these arguments:
 * --server_retry_port=PORT
     * The server port to connect to for testing backoffs. For example, "8081"
 
-The client must connect to the control port without TLS. The client should
-either assert on the server returned backoff status or check the returned
-backoffs on its own.
+The client must connect to the control port without TLS. The client must connect
+to the retry port with TLS. The client should either assert on the server
+returned backoff status or check the returned backoffs on its own.
 
 Procedure of client:
 

diff --git a/doc/connection-backoff.md b/doc/connection-backoff.md
@@ -44,3 +44,12 @@ different jitter logic.
 Alternate implementations must ensure that connection backoffs started at the
 same time disperse, and must not attempt connections substantially more often
 than the above algorithm.
+
+## Reset Backoff
+
+The back off should be reset to INITIAL_BACKOFF at some time point, so that the
+reconnecting behavior is consistent no matter the connection is a newly started
+one or a previously disconnected one.
+
+We choose to reset the Backoff when the SETTINGS frame is received, at that time
+point, we know for sure that this connection was accepted by the server.
diff --git a/include/grpc++/server.h b/include/grpc++/server.h
@@ -63,7 +63,14 @@ class Server GRPC_FINAL : public GrpcLibrary, private CallHook {
   ~Server();
 
   // Shutdown the server, block until all rpc processing finishes.
-  void Shutdown();
+  // Forcefully terminate pending calls after deadline expires.
+  template <class T>
+  void Shutdown(const T& deadline) {
+    ShutdownInternal(TimePoint<T>(deadline).raw_time());
+  }
+
+  // Shutdown the server, waiting for all rpc processing to finish.
+  void Shutdown() { ShutdownInternal(gpr_inf_future(GPR_CLOCK_MONOTONIC)); }
 
   // Block waiting for all work to complete (the server must either
   // be shutting down or some other thread must call Shutdown for this
@@ -99,6 +106,8 @@ class Server GRPC_FINAL : public GrpcLibrary, private CallHook {
 
   void PerformOpsOnCall(CallOpSetInterface* ops, Call* call) GRPC_OVERRIDE;
 
+  void ShutdownInternal(gpr_timespec deadline);
+
   class BaseAsyncRequest : public CompletionQueueTag {
    public:
     BaseAsyncRequest(Server* server, ServerContext* context,

diff --git a/include/grpc/compression.h b/include/grpc/compression.h
@@ -67,7 +67,9 @@ int grpc_compression_algorithm_parse(const char *name, size_t name_length,
                                      grpc_compression_algorithm *algorithm);
 
 /** Updates \a name with the encoding name corresponding to a valid \a
- * algorithm.  Returns 1 upon success, 0 otherwise. */
+ * algorithm. Note that the string returned through \a name upon success is
+ * statically allocated and shouldn't be freed. Returns 1 upon success, 0
+ * otherwise. */
 int grpc_compression_algorithm_name(grpc_compression_algorithm algorithm,
                                     char **name);
 

diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h
@@ -386,13 +386,6 @@ typedef struct grpc_op {
     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.
-
-    While grpc_init and grpc_shutdown can be called multiple times, the plugins
-    won't be unregistered and their memory cleaned up unless you call that
-    function. Using atexit(grpc_unregister_all_plugins) is a valid method. */
-void grpc_unregister_all_plugins();
-
 /* Propagation bits: this can be bitwise or-ed to form propagation_mask for
  * grpc_call */
 /** Propagate deadline */

diff --git a/src/core/channel/compress_filter.c b/src/core/channel/compress_filter.c
@@ -216,7 +216,7 @@ static void process_send_ops(grpc_call_element *elem,
                                   [calld->compression_algorithm]));
 
           /* convey supported compression algorithms */
-          grpc_metadata_batch_add_head(
+          grpc_metadata_batch_add_tail(
               &(sop->data.metadata), &calld->accept_encoding_storage,
               GRPC_MDELEM_REF(channeld->mdelem_accept_encoding));
 

diff --git a/src/core/surface/server.c b/src/core/surface/server.c
@@ -975,6 +975,11 @@ void grpc_server_setup_transport(grpc_server *s, grpc_transport *transport,
   grpc_transport_perform_op(transport, &op);
 }
 
+void done_published_shutdown(void *done_arg, grpc_cq_completion *storage) {
+  (void) done_arg;
+  gpr_free(storage);
+}
+
 void grpc_server_shutdown_and_notify(grpc_server *server,
                                      grpc_completion_queue *cq, void *tag) {
   listener *l;
@@ -986,6 +991,12 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
   /* lock, and gather up some stuff to do */
   gpr_mu_lock(&server->mu_global);
   grpc_cq_begin_op(cq);
+  if (server->shutdown_published) {
+    grpc_cq_end_op(cq, tag, 1, done_published_shutdown, NULL,
+                   gpr_malloc(sizeof(grpc_cq_completion)));
+    gpr_mu_unlock(&server->mu_global);
+    return;
+  }
   server->shutdown_tags =
       gpr_realloc(server->shutdown_tags,
                   sizeof(shutdown_tag) * (server->num_shutdown_tags + 1));

diff --git a/src/cpp/client/channel.cc b/src/cpp/client/channel.cc
@@ -71,7 +71,7 @@ Call Channel::CreateCall(const RpcMethod& method, ClientContext* context,
   } else {
     const char* host_str = NULL;
     if (!context->authority().empty()) {
-      host_str = context->authority().c_str();
+      host_str = context->authority_.c_str();
     } else if (!host_.empty()) {
       host_str = host_.c_str();
     }

diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
@@ -136,6 +136,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() {
@@ -354,12 +374,27 @@ bool Server::Start(ServerCompletionQueue** cqs, size_t num_cqs) {
   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) {

diff --git a/src/csharp/Grpc.Core.Tests/ChannelTest.cs b/src/csharp/Grpc.Core.Tests/ChannelTest.cs
@@ -41,12 +41,6 @@ namespace Grpc.Core.Tests
 {
     public class ChannelTest
     {
-        [TestFixtureTearDown]
-        public void CleanupClass()
-        {
-            GrpcEnvironment.Shutdown();
-        }
-
         [Test]
         public void Constructor_RejectsInvalidParams()
         {
@@ -56,36 +50,33 @@ public void Constructor_RejectsInvalidParams()
         [Test]
         public void State_IdleAfterCreation()
         {
-            using (var channel = new Channel("localhost", Credentials.Insecure))
-            {
-                Assert.AreEqual(ChannelState.Idle, channel.State);
-            }
+            var channel = new Channel("localhost", Credentials.Insecure);
+            Assert.AreEqual(ChannelState.Idle, channel.State);
+            channel.ShutdownAsync().Wait();
         }
 
         [Test]
         public void WaitForStateChangedAsync_InvalidArgument()
         {
-            using (var channel = new Channel("localhost", Credentials.Insecure))
-            {
-                Assert.Throws(typeof(ArgumentException), () => channel.WaitForStateChangedAsync(ChannelState.FatalFailure));
-            }
+            var channel = new Channel("localhost", Credentials.Insecure);
+            Assert.Throws(typeof(ArgumentException), () => channel.WaitForStateChangedAsync(ChannelState.FatalFailure));
+            channel.ShutdownAsync().Wait();
         }
 
         [Test]
         public void ResolvedTarget()
         {
-            using (var channel = new Channel("127.0.0.1", Credentials.Insecure))
-            {
-                Assert.IsTrue(channel.ResolvedTarget.Contains("127.0.0.1"));
-            }
+            var channel = new Channel("127.0.0.1", Credentials.Insecure);
+            Assert.IsTrue(channel.ResolvedTarget.Contains("127.0.0.1"));
+            channel.ShutdownAsync().Wait();
         }
 
         [Test]
-        public void Dispose_IsIdempotent()
+        public void Shutdown_AllowedOnlyOnce()
         {
             var channel = new Channel("localhost", Credentials.Insecure);
-            channel.Dispose();
-            channel.Dispose();
+            channel.ShutdownAsync().Wait();
+            Assert.Throws(typeof(InvalidOperationException), () => channel.ShutdownAsync().GetAwaiter().GetResult());
         }
     }
 }