Don't hold a lock in DelayedStream when calling realStream

Our current lock ordering rules prohibit holding a lock when calling the
channel and stream. This change avoids the lock for both
DelayedClientTransport and DelayedStream. It is effectively a rewrite of
DelayedStream.

The fixes to ClientCallImpl were to ensure sane state in DelayedStream.

Fixes #1510

Author: ejona86

core/src/main/java/io/grpc/internal/ClientCallImpl.java

@@ -31,6 +31,7 @@
 
 package io.grpc.internal;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
@@ -208,11 +209,11 @@ public void runInContext() {
       stream.setAuthority(callOptions.getAuthority());
     }
     stream.setCompressor(compressor);
+
+    stream.start(new ClientStreamListenerImpl(observer));
     if (compressor != Codec.Identity.NONE) {
       stream.setMessageCompression(true);
     }
-
-    stream.start(new ClientStreamListenerImpl(observer));
     // Delay any sources of cancellation after start(), because most of the transports are broken if
     // they receive cancel before start. Issue #1343 has more details
 
@@ -269,6 +270,7 @@ private static Long getRemainingTimeoutNanos(@Nullable Long deadlineNanoTime) {
   @Override
   public void request(int numMessages) {
     Preconditions.checkState(stream != null, "Not started");
+    checkArgument(numMessages >= 0, "Number requested must be non-negative");
     stream.request(numMessages);
   }
 

core/src/main/java/io/grpc/internal/DelayedClientTransport.java

@@ -80,44 +80,41 @@ public void start(Listener listener) {
   @Override
   public ClientStream newStream(MethodDescriptor<?, ?> method, Metadata headers) {
     Supplier<ClientTransport> supplier = transportSupplier;
+    if (supplier == null) {
+      synchronized (lock) {
+        // Check again, since it may have changed while waiting for lock
+        supplier = transportSupplier;
+        if (supplier == null && !shutdown) {
+          PendingStream pendingStream = new PendingStream(method, headers);
+          pendingStreams.add(pendingStream);
+          return pendingStream;
+        }
+      }
+    }
     if (supplier != null) {
       return supplier.get().newStream(method, headers);
     }
-    synchronized (lock) {
-      // Check again, since it may have changed while waiting for lock
-      supplier = transportSupplier;
-      if (supplier != null) {
-        return supplier.get().newStream(method, headers);
-      }
-      if (!shutdown) {
-        PendingStream pendingStream = new PendingStream(method, headers);
-        pendingStreams.add(pendingStream);
-        return pendingStream;
-      }
-    }
     return new FailingClientStream(Status.UNAVAILABLE.withDescription("transport shutdown"));
   }
 
   @Override
   public void ping(final PingCallback callback, Executor executor) {
     Supplier<ClientTransport> supplier = transportSupplier;
+    if (supplier == null) {
+      synchronized (lock) {
+        // Check again, since it may have changed while waiting for lock
+        supplier = transportSupplier;
+        if (supplier == null && !shutdown) {
+          PendingPing pendingPing = new PendingPing(callback, executor);
+          pendingPings.add(pendingPing);
+          return;
+        }
+      }
+    }
     if (supplier != null) {
       supplier.get().ping(callback, executor);
       return;
     }
-    synchronized (lock) {
-      // Check again, since it may have changed while waiting for lock
-      supplier = transportSupplier;
-      if (supplier != null) {
-        supplier.get().ping(callback, executor);
-        return;
-      }
-      if (!shutdown) {
-        PendingPing pendingPing = new PendingPing(callback, executor);
-        pendingPings.add(pendingPing);
-        return;
-      }
-    }
     executor.execute(new Runnable() {
         @Override public void run() {
           callback.onFailure(