Clarify why we need to check Thread#interrupted after Condition#await

* Use the new helpers for all Condition#await.

Author: eregon

src/main/.checkstyle_checks.xml

@@ -149,6 +149,10 @@
       <property name="format" value="\.computeIfAbsent"/>
       <property name="message" value="Use ConcurrentOperations.getOrCompute() instead of ConcurrentHashMap.computeIfAbsent() which does not scale"/>
     </module>
+    <module name="RegexpSinglelineJava">
+      <property name="format" value="\b(?!\w*[Ll]atch)\w+\.await\(.*\)"/>
+      <property name="message" value="Use ConcurrentOperations.awaitAndCheckInterrupt() instead"/>
+    </module>
     <module name="RegexpSinglelineJava">
       <property name="format" value="@NodeChildren"/>
       <property name="message" value="Do not use @NodeChildren, use multiple @Child annotations directly"/>

src/main/java/org/truffleruby/collections/ConcurrentOperations.java

@@ -13,6 +13,10 @@
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.AbstractQueuedSynchronizer;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
 
 public abstract class ConcurrentOperations {
@@ -43,4 +47,40 @@ public static <K, V> boolean replace(Map<K, V> map, K key, V oldValue, V newValu
         return oldValue == null ? map.putIfAbsent(key, newValue) == null : map.replace(key, oldValue, newValue);
     }
 
+    /** {@link Condition#await()} might return and not throw InterruptedException if there was both a signal and an
+     * interrupt before the lock could be re-acquired by await(). We always want the InterruptedException if both a
+     * signal and an interrupt happen.
+     * <hr>
+     * {@link Condition#await()} when it sees both a signal and a {@link Thread#interrupt()} has to decide whether it
+     * prefers the signal or the interrupt. For a {@link ReentrantLock#newCondition()} condition, the implementation in
+     * {@link AbstractQueuedSynchronizer.ConditionObject#await()} sometimes throws InterruptedException and sometimes
+     * returns, depending on which happens first. Because await() needs to reacquire the lock, even if the
+     * InterruptedException happens before it might still become a situation of "both signal and interrupt". If it
+     * prefers the signal and just returns, it does {@code Thread.currentThread().interrupt()} to let us know there was
+     * an interrupt too. In any case, if there was any interrupt we want to throw InterruptedException, regardless of
+     * what the implementation prefers. */
+    public static void awaitAndCheckInterrupt(Condition condition) throws InterruptedException {
+        // Checkstyle: stop
+        condition.await();
+        // Checkstyle: resume
+
+        if (Thread.interrupted()) {
+            throw new InterruptedException();
+        }
+    }
+
+    /** See {@link ConcurrentOperations#awaitAndCheckInterrupt(java.util.concurrent.locks.Condition)} */
+    public static boolean awaitAndCheckInterrupt(Condition condition, long time, TimeUnit unit)
+            throws InterruptedException {
+        // Checkstyle: stop
+        boolean value = condition.await(time, unit);
+        // Checkstyle: resume
+
+        if (Thread.interrupted()) {
+            throw new InterruptedException();
+        }
+
+        return value;
+    }
+
 }

src/main/java/org/truffleruby/core/mutex/ConditionVariableNodes.java

@@ -21,6 +21,7 @@
 import org.truffleruby.annotations.CoreModule;
 import org.truffleruby.annotations.Primitive;
 import org.truffleruby.builtins.PrimitiveArrayArgumentsNode;
+import org.truffleruby.collections.ConcurrentOperations;
 import org.truffleruby.collections.Memo;
 import org.truffleruby.core.klass.RubyClass;
 import org.truffleruby.core.thread.RubyThread;
@@ -98,7 +99,7 @@ private void waitInternal(RubyConditionVariable conditionVariable, ReentrantLock
 
             // condLock must be locked before unlocking mutexLock, to avoid losing potential signals.
             // We must not change the Ruby Thread status and not consume a Java thread interrupt while locking condLock.
-            // If there is an interrupt, it should be consumed by condition.await() and the Ruby Thread sleep status
+            // If there is an interrupt, it should be consumed by Condition#await() and the Ruby Thread sleep status
             // must imply being ready to be interrupted by Thread#{run,wakeup}.
             condLock.lock();
             int holdCount = 0; // can be > 1 for MonitorMixin
@@ -164,9 +165,10 @@ private void awaitSignal(RubyConditionVariable self, RubyThread thread, long dur
                          * ReentrantLock exposed to Ruby, e.g., via a Ruby Mutex, as that could be held forever by
                          * another Ruby thread (which did {code mutex.lock} after the #wait), and we would never be able
                          * to interrupt both threads at the same time for a synchronous ThreadLocalAction). */
-                        condition.await(endNanoTime - currentTime, TimeUnit.NANOSECONDS);
+                        ConcurrentOperations.awaitAndCheckInterrupt(condition, endNanoTime - currentTime,
+                                TimeUnit.NANOSECONDS);
                     } else {
-                        condition.await();
+                        ConcurrentOperations.awaitAndCheckInterrupt(condition);
                     }
                     if (consumeSignal(self)) {
                         return BlockingAction.SUCCESS;

src/main/java/org/truffleruby/core/queue/SizedQueue.java

@@ -16,6 +16,7 @@
 import java.util.concurrent.locks.ReentrantLock;
 
 import com.oracle.truffle.api.CompilerDirectives.TruffleBoundary;
+import org.truffleruby.collections.ConcurrentOperations;
 
 public class SizedQueue {
 
@@ -112,7 +113,7 @@ public boolean put(Object item) throws InterruptedException {
                     return false;
                 }
 
-                canAdd.await();
+                ConcurrentOperations.awaitAndCheckInterrupt(canAdd);
             }
 
             if (closed) {
@@ -144,17 +145,8 @@ public Object poll(long timeoutMilliseconds) throws InterruptedException {
 
         try {
             if (size == 0) {
-                final boolean signalled = canTake.await(timeoutMilliseconds, TimeUnit.MILLISECONDS);
-
-                /* We need to check the interrupted flag here, while holding the lock and after the await(). Otherwise,
-                 * if another thread does {@code th.raise(exc); q.push(1)} like in core/thread/handle_interrupt_spec.rb
-                 * then we might miss the ThreadLocalAction and instead return, which would incorrectly delay the
-                 * ThreadLocalAction if under {@code Thread.handle_interrupt(Object => :on_blocking)}. The other thread
-                 * cannot wait on the Future, as that might block arbitrary long when side-effects are disabled on this
-                 * thread. */
-                if (Thread.interrupted()) {
-                    throw new InterruptedException();
-                }
+                final boolean signalled = ConcurrentOperations.awaitAndCheckInterrupt(canTake, timeoutMilliseconds,
+                        TimeUnit.MILLISECONDS);
 
                 if (!signalled) {
                     // Timed out.
@@ -197,7 +189,7 @@ public Object take() throws InterruptedException {
                     return CLOSED;
                 }
 
-                canTake.await();
+                ConcurrentOperations.awaitAndCheckInterrupt(canTake);
             }
 
             return doTake();

src/main/java/org/truffleruby/core/queue/UnsizedQueue.java

@@ -17,6 +17,7 @@
 import java.util.concurrent.locks.ReentrantLock;
 
 import com.oracle.truffle.api.CompilerDirectives.TruffleBoundary;
+import org.truffleruby.collections.ConcurrentOperations;
 
 public class UnsizedQueue {
 
@@ -95,17 +96,7 @@ public Object take() throws InterruptedException {
                     return CLOSED;
                 }
 
-                canTake.await();
-
-                /* We need to check the interrupted flag here, while holding the lock and after the await(). Otherwise,
-                 * if another thread does {@code th.raise(exc); q.push(1)} like in core/thread/handle_interrupt_spec.rb
-                 * then we might miss the ThreadLocalAction and instead return, which would incorrectly delay the
-                 * ThreadLocalAction if under {@code Thread.handle_interrupt(Object => :on_blocking)}. The other thread
-                 * cannot wait on the Future, as that might block arbitrary long when side-effects are disabled on this
-                 * thread. */
-                if (Thread.interrupted()) {
-                    throw new InterruptedException();
-                }
+                ConcurrentOperations.awaitAndCheckInterrupt(canTake);
             }
 
             return doTake();
@@ -120,11 +111,8 @@ public Object poll(long timeoutMilliseconds) throws InterruptedException {
 
         try {
             if (takeEnd == null) {
-                final boolean signalled = canTake.await(timeoutMilliseconds, TimeUnit.MILLISECONDS);
-
-                if (Thread.interrupted()) { // See comment of take()
-                    throw new InterruptedException();
-                }
+                final boolean signalled = ConcurrentOperations.awaitAndCheckInterrupt(canTake, timeoutMilliseconds,
+                        TimeUnit.MILLISECONDS);
 
                 if (!signalled) {
                     return null;