Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #1870 (2018-07-04 15:37:01)

Adding "Not Start" state to Watchdog
The master branch is broken.  It appears that there's a discrepancy between "idle" and "not started" states that were breaking the tests.  This PR adds the notion of "not started" and also has some other minor touch ups.

igorbernstein2: Lgtm. I think the not started case used to handled by state being null, but I unwitteningly broke it addressing https://github.com/GoogleCloudPlatform/cloud-bigtable-client/pull/1862#discussion_r199209495.

I think having an explicit state is better
kevinsi4508: Not super familiar with watchdog. I don't see documentation such as state transition diagram. Up to Igor.

Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/io/Watchdog.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/io/Watchdog.java
index 64467821e..8788b258b 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/io/Watchdog.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/io/Watchdog.java
@@ -50,6 +50,7 @@ import javax.annotation.concurrent.GuardedBy;
 @InternalApi
 public class Watchdog implements Runnable {
   public enum State {
+    NOT_STARTED,
     /** Stream has been started, but doesn't have any outstanding requests. */
     IDLE,
     /** Stream is awaiting a response from upstream. */
@@ -126,7 +127,7 @@ public class Watchdog implements Runnable {
     private final Object lock = new Object();
 
     @GuardedBy("lock")
-    private State state = State.IDLE;
+    private State state = State.NOT_STARTED;
 
     @GuardedBy("lock")
     private int pendingCount = 0;
@@ -143,8 +144,8 @@ public class Watchdog implements Runnable {
       ClientCall<ReqT, RespT> call = delegate();
 
       synchronized (lock) {
-        Preconditions.checkState(state == null, "Already started");
-        lastActivityAt = clock.currentTimeMillis();
+        Preconditions.checkState(state == State.NOT_STARTED, "Already started");
+        setState(State.IDLE);
       }
 
       openStreams.put(this, PRESENT);
@@ -152,18 +153,14 @@ public class Watchdog implements Runnable {
       call.start(new SimpleForwardingClientCallListener<RespT>(listener) {
         @Override
         public void onMessage(RespT message) {
-          synchronized (lock) {
-            state = State.DELIVERING;
-          }
+          setState(State.DELIVERING);
 
           try {
             super.onMessage(message);
           } finally {
             synchronized (lock) {
               pendingCount--;
-              lastActivityAt = clock.currentTimeMillis();
-
-              state = (pendingCount > 0) ? State.WAITING : State.IDLE;
+              setState((pendingCount > 0) ? State.WAITING : State.IDLE);
             }
           }
         }
@@ -177,49 +174,65 @@ public class Watchdog implements Runnable {
       }, metadata);
     }
 
+    private void setState(State state) {
+      synchronized (lock) {
+        this.state = state;
+        this.lastActivityAt = clock.currentTimeMillis();
+      }
+    }
+
     @Override
     public void request(int count) {
       synchronized (lock) {
-        if (state == State.IDLE) {
-          state = State.WAITING;
-          lastActivityAt = clock.currentTimeMillis();
-        }
-
+        Preconditions.checkState(state != State.NOT_STARTED,
+            "The Call was not started");
         // Increment the request count without overflow
         int maxIncrement = Integer.MAX_VALUE - pendingCount;
-        count = Math.min(maxIncrement, count);
-        pendingCount += count;
+        pendingCount += Math.min(maxIncrement, count);
+
+        if (state == State.IDLE) {
+          setState(State.WAITING);
+        }
       }
 
       super.request(count);
     }
 
-    boolean cancelIfStale() {
+    /**
+     *
+     * @return true if this RPC was cancelled.
+     */
+    private boolean cancelIfStale() {
       synchronized (lock) {
         long waitTime = clock.currentTimeMillis() - lastActivityAt;
 
         switch (this.state) {
-          case IDLE:
-            if (waitTime >= idleTimeoutMs) {
-              delegate().cancel("Canceled due to idle connection", new StreamWaitTimeoutException(this.state, waitTime));
-              return true;
-            }
-            break;
-          case WAITING:
-            if (waitTime >= waitTimeoutMs) {
-              delegate().cancel("Canceled due to timeout waiting for next response", new StreamWaitTimeoutException(this.state, waitTime));
-              return true;
-            }
-            break;
-          case DELIVERING:
-            // Don't cancel the stream while it's results are being processed by user code.
-            break;
-          default:
-            throw new IllegalStateException("Unknown state: " + this.state);
+        case NOT_STARTED:
+        case IDLE:
+          if (waitTime >= idleTimeoutMs) {
+            delegate().cancel("Canceled due to idle connection",
+                new StreamWaitTimeoutException(this.state, waitTime));
+            return true;
+          }
+          break;
+
+        case WAITING:
+          if (waitTime >= waitTimeoutMs) {
+            delegate().cancel("Canceled due to timeout waiting for next response",
+                new StreamWaitTimeoutException(this.state, waitTime));
+            return true;
+          }
+          break;
+
+        case DELIVERING:
+          // Don't cancel the stream while it's results are being processed by user code.
+          break;
+
+        default:
+          throw new IllegalStateException("Unknown state: " + this.state);
         }
+        return false;
       }
-
-      return false;
     }
   }
 


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/io/WatchdogTest.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/io/WatchdogTest.java
index bee77f6fe..003639df4 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/io/WatchdogTest.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/io/WatchdogTest.java
@@ -81,7 +81,29 @@ public class WatchdogTest {
   }
 
   @Test
-  public void testIdleTimeout() throws InterruptedException {
+  public void testNotStarted() {
+    clock.setTime(idleTimeMs - 1);
+    watchdog.run();
+    Assert.assertNull(call.cancelMessage);
+
+    clock.setTime(idleTimeMs);
+    watchdog.run();
+    Assert.assertNotNull(call.cancelMessage);
+
+    call.listener
+        .onClose(
+            Status.CANCELLED.withDescription("Some upstream exception representing cancellation"),
+            new Metadata());
+
+    Assert.assertEquals(listener.closeStatus.getDescription(),
+        "Some upstream exception representing cancellation");
+  }
+
+  @Test
+  public void testIdleTimeout() {
+    call.request(1);
+    call.listener.onMessage("");
+
     clock.setTime(idleTimeMs - 1);
     watchdog.run();
     Assert.assertNull(call.cancelMessage);