refactor timer locks to provide more diagnostic information. (#165)

Also fix couple of bugs in the related logic

Author: meiliang86

src/main/java/com/uber/cadence/client/WorkflowClientOptions.java

@@ -29,11 +29,17 @@ public final class WorkflowClientOptions {
   public static final class Builder {
 
     private DataConverter dataConverter = JsonDataConverter.getInstance();
-
     private WorkflowClientInterceptor[] interceptors = EMPTY_INTERCEPTOR_ARRAY;
-
     private Scope metricsScope;
 
+    public Builder() {}
+
+    public Builder(WorkflowClientOptions options) {
+      dataConverter = options.getDataConverter();
+      interceptors = options.getInterceptors();
+      metricsScope = options.getMetricsScope();
+    }
+
     /**
      * Used to override default (JSON) data converter implementation.
      *

src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java

@@ -105,7 +105,7 @@ public TestWorkflowEnvironmentInternal(TestEnvironmentOptions options) {
       this.testEnvironmentOptions = options;
     }
     service = new WorkflowServiceWrapper();
-    service.lockTimeSkipping();
+    service.lockTimeSkipping("TestWorkflowEnvironmentInternal constructor");
   }
 
   @Override
@@ -137,7 +137,15 @@ public WorkflowClient newWorkflowClient() {
 
   @Override
   public WorkflowClient newWorkflowClient(WorkflowClientOptions options) {
-    return WorkflowClientInternal.newInstance(service, testEnvironmentOptions.getDomain(), options);
+    WorkflowClientInterceptor[] existingInterceptors = options.getInterceptors();
+    WorkflowClientInterceptor[] interceptors =
+        new WorkflowClientInterceptor[existingInterceptors.length + 1];
+    System.arraycopy(existingInterceptors, 0, interceptors, 0, existingInterceptors.length);
+    interceptors[interceptors.length - 1] = new TimeLockingInterceptor(service);
+    WorkflowClientOptions newOptions =
+        new WorkflowClientOptions.Builder(options).setInterceptors(interceptors).build();
+    return WorkflowClientInternal.newInstance(
+        service, testEnvironmentOptions.getDomain(), newOptions);
   }
 
   @Override
@@ -559,12 +567,12 @@ public void registerDelayedCallback(Duration delay, Runnable r) {
       impl.registerDelayedCallback(delay, r);
     }
 
-    public void lockTimeSkipping() {
-      impl.lockTimeSkipping();
+    public void lockTimeSkipping(String caller) {
+      impl.lockTimeSkipping(caller);
     }
 
-    public void unlockTimeSkipping() {
-      impl.unlockTimeSkipping();
+    public void unlockTimeSkipping(String caller) {
+      impl.unlockTimeSkipping(caller);
     }
 
     public void sleep(Duration duration) {
@@ -629,11 +637,11 @@ public WorkflowExecution getExecution() {
 
       @Override
       public <R> R getResult(Class<R> returnType) {
-        service.unlockTimeSkipping();
+        service.unlockTimeSkipping("TimeLockingWorkflowStub getResult");
         try {
           return next.getResult(returnType);
         } finally {
-          service.lockTimeSkipping();
+          service.lockTimeSkipping("TimeLockingWorkflowStub getResult");
         }
       }
 
@@ -645,11 +653,11 @@ public <R> CompletableFuture<R> getResultAsync(Class<R> returnType) {
       @Override
       public <R> R getResult(long timeout, TimeUnit unit, Class<R> returnType)
           throws TimeoutException {
-        service.unlockTimeSkipping();
+        service.unlockTimeSkipping("TimeLockingWorkflowStub getResult");
         try {
           return next.getResult(timeout, unit, returnType);
         } finally {
-          service.lockTimeSkipping();
+          service.lockTimeSkipping("TimeLockingWorkflowStub getResult");
         }
       }
 
@@ -681,7 +689,8 @@ public TimeLockingFuture(CompletableFuture<R> resultAsync) {
           CompletableFuture<R> ignored =
               resultAsync.whenComplete(
                   (r, e) -> {
-                    service.lockTimeSkipping();
+                    service.lockTimeSkipping(
+                        "TimeLockingWorkflowStub TimeLockingFuture constructor");
                     if (e == null) {
                       this.complete(r);
                     } else {
@@ -692,25 +701,28 @@ public TimeLockingFuture(CompletableFuture<R> resultAsync) {
 
         @Override
         public R get() throws InterruptedException, ExecutionException {
-          service.unlockTimeSkipping();
-          return super.get();
+          service.unlockTimeSkipping("TimeLockingWorkflowStub TimeLockingFuture get");
+          try {
+            return super.get();
+          } finally {
+            service.lockTimeSkipping("TimeLockingWorkflowStub TimeLockingFuture get");
+          }
         }
 
         @Override
         public R get(long timeout, TimeUnit unit)
             throws InterruptedException, ExecutionException, TimeoutException {
-          service.unlockTimeSkipping();
+          service.unlockTimeSkipping("TimeLockingWorkflowStub TimeLockingFuture get");
           try {
             return super.get(timeout, unit);
-          } catch (TimeoutException | InterruptedException e) {
-            service.lockTimeSkipping();
-            throw e;
+          } finally {
+            service.lockTimeSkipping("TimeLockingWorkflowStub TimeLockingFuture get");
           }
         }
 
         @Override
         public R join() {
-          service.unlockTimeSkipping();
+          service.unlockTimeSkipping("TimeLockingWorkflowStub TimeLockingFuture join");
           return super.join();
         }
       }

src/main/java/com/uber/cadence/internal/testservice/RequestContext.java

@@ -41,10 +41,12 @@ static final class Timer {
 
     private final long delaySeconds;
     private final Runnable callback;
+    private final String taskInfo;
 
-    Timer(long delaySeconds, Runnable callback) {
+    Timer(long delaySeconds, Runnable callback, String taskInfo) {
       this.delaySeconds = delaySeconds;
       this.callback = callback;
+      this.taskInfo = taskInfo;
     }
 
     long getDelaySeconds() {
@@ -54,6 +56,10 @@ long getDelaySeconds() {
     Runnable getCallback() {
       return callback;
     }
+
+    String getTaskInfo() {
+      return taskInfo;
+    }
   }
 
   private static final long NANOS_PER_MILLIS = 1_000_000;
@@ -173,8 +179,8 @@ void addActivityTask(ActivityTask activityTask) {
     this.activityTasks.add(activityTask);
   }
 
-  void addTimer(long delaySeconds, Runnable callback) {
-    Timer timer = new Timer(delaySeconds, callback);
+  void addTimer(long delaySeconds, Runnable callback, String name) {
+    Timer timer = new Timer(delaySeconds, callback, name);
     this.timers.add(timer);
   }
 

src/main/java/com/uber/cadence/internal/testservice/SelfAdvancingTimer.java

@@ -22,7 +22,7 @@
 
 /**
  * Timer service that automatically forwards current time to the next task time when is not locked
- * through {@link #lockTimeSkipping()}.
+ * through {@link #lockTimeSkipping(String)}.
  */
 interface SelfAdvancingTimer {
 
@@ -32,24 +32,32 @@ interface SelfAdvancingTimer {
    */
   void schedule(Duration delay, Runnable task);
 
+  void schedule(Duration delay, Runnable task, String taskInfo);
+
   /** Supplier that returns current time of the timer when called. */
   LongSupplier getClock();
 
   /**
-   * Prohibit automatic time skipping until {@link #unlockTimeSkipping()} is called. Locks and
+   * Prohibit automatic time skipping until {@link #unlockTimeSkipping(String)} is called. Locks and
    * unlocks are counted.
    */
-  void lockTimeSkipping();
+  LockHandle lockTimeSkipping(String caller);
 
-  void unlockTimeSkipping();
+  void unlockTimeSkipping(String caller);
 
   /**
    * Update lock count. The same as calling lockTimeSkipping count number of times for positive
    * count and unlockTimeSkipping for negative count.
    *
    * @param count
    */
-  void updateLocks(int count);
+  void updateLocks(int count, String caller);
+
+  void getDiagnostics(StringBuilder result);
 
   void shutdown();
 }
+
+interface LockHandle {
+  void unlock();
+}