Fix usages of some async utilities

spockframework · Jun 6, 2023 · 776db33 · 776db33
1 parent dedede9
commit 776db33
Show file tree

Hide file tree

Showing 5 changed files with 41 additions and 30 deletions.
diff --git a/...c/main/java/org/spockframework/runtime/extension/builtin/RepeatUntilFailureExtension.java b/...c/main/java/org/spockframework/runtime/extension/builtin/RepeatUntilFailureExtension.java
@@ -37,12 +37,8 @@ public void runIterations(IDataIterator dataIterator, IIterationRunner iteration
       dataIterator.forEachRemaining(data::add);
       for (int attempt = 0; attempt < maxAttempts; attempt++) {
         for (Object[] args : data) {
-          try {
-            ExecutionResult executionResult = iterationRunner.runIteration(args, maxIterations).get();
-            if (executionResult == ExecutionResult.FAILED) {
-              return;
-            }
-          } catch (InterruptedException | ExecutionException e) {
+          ExecutionResult executionResult = iterationRunner.runIteration(args, maxIterations).join();
+          if (executionResult == ExecutionResult.FAILED) {
             return;
           }
         }

diff --git a/...k-core/src/main/java/org/spockframework/runtime/extension/builtin/TimeoutInterceptor.java b/...k-core/src/main/java/org/spockframework/runtime/extension/builtin/TimeoutInterceptor.java
@@ -51,41 +51,57 @@ public void run() {
         long waitMillis = timeout.unit().toMillis(timeout.value());
         boolean synced = false;
 
+        boolean syncedWithFeature = false;
         try {
           startLatch.countDown();
-          startLatch.await(5, TimeUnit.SECONDS);
+          syncedWithFeature = startLatch.await(5, TimeUnit.SECONDS);
         } catch (InterruptedException ignored) {
+          // this is our own thread, so we can ignore the interruption safely
+        }
+        if (!syncedWithFeature) {
           System.out.printf("[spock.lang.Timeout] Could not sync with Feature for method '%s'", invocation.getMethod().getName());
         }
 
+        while (waitMillis > 0) {
+          long waitStart = System.nanoTime();
+          try {
+            synced = sync.offer(stackTrace, waitMillis, TimeUnit.MILLISECONDS);
+          } catch (InterruptedException ignored) {
+            // this is our own thread, so we can ignore the interruption safely and continue the remaining waiting
+            waitMillis -= TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStart);
+            continue;
+          }
+          break;
+        }
+        if (!synced) {
+          stackTrace = mainThread.getStackTrace();
+          waitMillis = 250;
+        }
         while (!synced) {
+          mainThread.interrupt();
           try {
             synced = sync.offer(stackTrace, waitMillis, TimeUnit.MILLISECONDS);
           } catch (InterruptedException ignored) {
             // The mission of this thread is to repeatedly interrupt the main thread until
             // the latter returns. Once this mission has been accomplished, this thread will die quickly
           }
           if (!synced) {
-            if (stackTrace.length == 0) {
-              stackTrace = mainThread.getStackTrace();
-              waitMillis = 250;
-            } else {
-              waitMillis *= 2;
-              System.out.printf("[spock.lang.Timeout] Method '%s' has not yet returned - interrupting. Next try in %1.2f seconds.\n",
-                  invocation.getMethod().getName(), waitMillis / 1000.);
-            }
-            mainThread.interrupt();
+            waitMillis *= 2;
+            System.out.printf("[spock.lang.Timeout] Method '%s' has not yet returned - interrupting. Next try in %1.2f seconds.\n",
+              invocation.getMethod().getName(), waitMillis / 1000.);
           }
         }
       }
     }.start();
 
-
+    boolean syncedWithWatcher = false;
     try {
       startLatch.countDown();
-      startLatch.await(5, TimeUnit.SECONDS);
-    } catch (InterruptedException ignored) {
-      System.out.printf("[spock.lang.Timeout] Could not sync with Watcher for method '%s'", invocation.getMethod().getName());
+      syncedWithWatcher = startLatch.await(5, TimeUnit.SECONDS);
+    } finally {
+      if (!syncedWithWatcher) {
+        System.out.printf("[spock.lang.Timeout] Could not sync with Watcher for method '%s'", invocation.getMethod().getName());
+      }
     }
 
     Throwable saved = null;

diff --git a/spock-core/src/main/java/spock/util/concurrent/AsyncConditions.java b/spock-core/src/main/java/spock/util/concurrent/AsyncConditions.java
@@ -131,15 +131,14 @@ public void await() throws Throwable {
    * @throws Throwable the first exception thrown by an evaluate block
    */
   public void await(double seconds) throws Throwable {
-    latch.await((long) (seconds * 1000), TimeUnit.MILLISECONDS);
+    boolean evalBlocksFinished = latch.await((long) (seconds * 1000), TimeUnit.MILLISECONDS);
     if (!exceptions.isEmpty())
       throw exceptions.poll();
 
-    long pendingEvalBlocks = latch.getCount();
-    if (pendingEvalBlocks > 0) {
+    if (!evalBlocksFinished) {
       String msg = String.format("Async conditions timed out " +
           "after %1.2f seconds; %d out of %d evaluate blocks did not complete in time",
-          seconds, pendingEvalBlocks, numEvalBlocks);
+          seconds, latch.getCount(), numEvalBlocks);
       throw new SpockTimeoutError(seconds, msg);
     }
   }

diff --git a/spock-specs/src/test/groovy/org/spockframework/smoke/extension/ParallelSpec.groovy b/spock-specs/src/test/groovy/org/spockframework/smoke/extension/ParallelSpec.groovy
@@ -265,7 +265,7 @@ class ParallelSpec extends EmbeddedSpecification {
         @ResourceLock(value = "a", mode = ResourceAccessMode.READ)
         def writeA() {
           when:
-          incrementAndBlock(atomicInteger, latch)
+          incrementAndBlock(atomicInteger, latch, 10000)
 
           then:
           atomicInteger.get() == 3
@@ -519,15 +519,15 @@ class ParallelSpec extends EmbeddedSpecification {
     throws InterruptedException {
     int value = sharedResource.incrementAndGet()
     countDownLatch.countDown()
-    countDownLatch.await(timeout, MILLISECONDS)
+    assert countDownLatch.await(timeout, MILLISECONDS) : 'Timeout expired'
     return value
   }
 
   static void storeAndBlockAndCheck(AtomicInteger sharedResource, CountDownLatch countDownLatch, long timeout = 100)
     throws InterruptedException {
     int value = sharedResource.get()
     countDownLatch.countDown()
-    countDownLatch.await(timeout, MILLISECONDS)
+    assert countDownLatch.await(timeout, MILLISECONDS) : 'Timeout expired'
     assert value == sharedResource.get()
   }
 }
diff --git a/...ecs/src/test/groovy/org/spockframework/smoke/mock/InvokingMocksFromMultipleThreads.groovy b/...ecs/src/test/groovy/org/spockframework/smoke/mock/InvokingMocksFromMultipleThreads.groovy
@@ -42,7 +42,7 @@ class InvokingMocksFromMultipleThreads extends Specification {
         }
       }
     }
-    latch.await(10, TimeUnit.SECONDS)
+    assert latch.await(10, TimeUnit.SECONDS) : 'Timeout expired'
 
     then:
     interaction {
@@ -68,7 +68,7 @@ class InvokingMocksFromMultipleThreads extends Specification {
         }
       }
     }
-    latch.await(10, TimeUnit.SECONDS)
+    assert latch.await(10, TimeUnit.SECONDS) : 'Timeout expired'
 
     then:
     interaction {
@@ -93,7 +93,7 @@ class InvokingMocksFromMultipleThreads extends Specification {
         }
       }
     }
-    latch.await(10, TimeUnit.SECONDS)
+    assert latch.await(10, TimeUnit.SECONDS) : 'Timeout expired'
 
     then:
     interaction {