[GR-18163] Fix Thread.handle_interrupt and implement Thread[.#]pending_interrupt? (#2219)

PullRequest: truffleruby/2355

Author: eregon

CHANGELOG.md

@@ -9,6 +9,7 @@ Bug fixes:
 * `Range#to_a` wasn't working for `long` ranges (#2198, @tomstuart and @LillianZ).
 * Show the interleaved host and guest stacktrace for host exceptions (#2226).
 * Fix the label of the first location reported by `Thread#backtrace_locations` (#2229).
+* Fix `Thread.handle_interrupt` to defer non-pure interrupts until the end of the `handle_interrupt` block (#2219).
 
 Compatibility:
 
@@ -41,6 +42,7 @@ Compatibility:
 * Support buffer argument for `UDPSocket#recvfrom_nonblock` (#2209, @HoneyryderChuck).
 * Fixed `Integer#digits` implementation to handle more bases (#2224, #2225).
 * Support the `inherit` parameter for `Module#{private, protected, public}_method_defined?`.
+* Implement `Thread.pending_interrupt?` and `Thread#pending_interrupt?` (#2219).
 
 Performance:
 

spec/ruby/core/thread/handle_interrupt_spec.rb

@@ -0,0 +1,120 @@
+require_relative '../../spec_helper'
+
+describe "Thread.handle_interrupt" do
+  def make_handle_interrupt_thread(interrupt_config, blocking = true)
+    interrupt_class = Class.new(RuntimeError)
+
+    ScratchPad.record []
+
+    in_handle_interrupt = Queue.new
+    can_continue = Queue.new
+
+    thread = Thread.new do
+      begin
+        Thread.handle_interrupt(interrupt_config) do
+          begin
+            in_handle_interrupt << true
+            if blocking
+              can_continue.pop
+            else
+              begin
+                can_continue.pop(true)
+              rescue ThreadError
+                Thread.pass
+                retry
+              end
+            end
+          rescue interrupt_class
+            ScratchPad << :interrupted
+          end
+        end
+      rescue interrupt_class
+        ScratchPad << :deferred
+      end
+    end
+
+    in_handle_interrupt.pop
+    thread.raise interrupt_class, "interrupt"
+    can_continue << true
+    thread.join
+
+    ScratchPad.recorded
+  end
+
+  before :each do
+    Thread.pending_interrupt?.should == false # sanity check
+  end
+
+  it "with :never defers interrupts until exiting the handle_interrupt block" do
+    make_handle_interrupt_thread(RuntimeError => :never).should == [:deferred]
+  end
+
+  it "with :on_blocking defers interrupts until the next blocking call" do
+    make_handle_interrupt_thread(RuntimeError => :on_blocking).should == [:interrupted]
+    make_handle_interrupt_thread({ RuntimeError => :on_blocking }, false).should == [:deferred]
+  end
+
+  it "with :immediate handles interrupts immediately" do
+    make_handle_interrupt_thread(RuntimeError => :immediate).should == [:interrupted]
+  end
+
+  it "with :immediate immediately runs pending interrupts, before the block" do
+    Thread.handle_interrupt(RuntimeError => :never) do
+      current = Thread.current
+      Thread.new {
+        current.raise "interrupt immediate"
+      }.join
+
+      Thread.pending_interrupt?.should == true
+      -> {
+        Thread.handle_interrupt(RuntimeError => :immediate) {
+          flunk "not reached"
+        }
+      }.should raise_error(RuntimeError, "interrupt immediate")
+      Thread.pending_interrupt?.should == false
+    end
+  end
+
+  it "also works with suspended Fibers and does not duplicate interrupts" do
+    fiber = Fiber.new { Fiber.yield }
+    fiber.resume
+
+    Thread.handle_interrupt(RuntimeError => :never) do
+      current = Thread.current
+      Thread.new {
+        current.raise "interrupt with fibers"
+      }.join
+
+      Thread.pending_interrupt?.should == true
+      -> {
+        Thread.handle_interrupt(RuntimeError => :immediate) {
+          flunk "not reached"
+        }
+      }.should raise_error(RuntimeError, "interrupt with fibers")
+      Thread.pending_interrupt?.should == false
+    end
+
+    fiber.resume
+  end
+
+  it "runs pending interrupts at the end of the block, even if there was an exception raised in the block" do
+    executed = false
+    -> {
+      Thread.handle_interrupt(RuntimeError => :never) do
+        current = Thread.current
+        Thread.new {
+          current.raise "interrupt exception"
+        }.join
+
+        Thread.pending_interrupt?.should == true
+        executed = true
+        raise "regular exception"
+      end
+    }.should raise_error(RuntimeError, "interrupt exception")
+    executed.should == true
+  end
+
+  it "supports multiple pairs in the Hash" do
+    make_handle_interrupt_thread(ArgumentError => :never, RuntimeError => :never).should == [:deferred]
+  end
+end

spec/ruby/core/thread/pending_interrupt_spec.rb

@@ -0,0 +1,32 @@
+require_relative '../../spec_helper'
+
+describe "Thread.pending_interrupt?" do
+  it "returns false if there are no pending interrupts, e.g., outside any Thread.handle_interrupt block" do
+    Thread.pending_interrupt?.should == false
+  end
+
+  it "returns true if there are pending interrupts, e.g., Thread#raise inside Thread.handle_interrupt" do
+    executed = false
+    -> {
+      Thread.handle_interrupt(RuntimeError => :never) do
+        Thread.pending_interrupt?.should == false
+
+        current = Thread.current
+        Thread.new {
+          current.raise "interrupt"
+        }.join
+
+        Thread.pending_interrupt?.should == true
+        executed = true
+      end
+    }.should raise_error(RuntimeError, "interrupt")
+    executed.should == true
+    Thread.pending_interrupt?.should == false
+  end
+end
+
+describe "Thread#pending_interrupt?" do
+  it "returns whether the given threads has pending interrupts" do
+    Thread.current.pending_interrupt?.should == false
+  end
+end

spec/tags/truffle/methods_tags.txt

@@ -57,7 +57,6 @@ fails:Public methods on Fiber should include to_s
 fails:Public methods on Mutex should not include marshal_dump
 fails:Public methods on Thread should not include freeze
 fails:Public methods on Thread should include add_trace_func
-fails:Public methods on Thread should include pending_interrupt?
 fails:Public methods on Thread should include set_trace_func
 fails:Public methods on Proc should not include block
 fails:Public methods on Proc should not include block=

src/main/java/org/truffleruby/core/VMPrimitiveNodes.java

@@ -52,7 +52,6 @@
 import org.truffleruby.core.cast.NameToJavaStringNode;
 import org.truffleruby.core.cast.ToRubyIntegerNode;
 import org.truffleruby.core.exception.RubyException;
-import org.truffleruby.core.fiber.FiberManager;
 import org.truffleruby.core.klass.RubyClass;
 import org.truffleruby.core.method.RubyMethod;
 import org.truffleruby.core.module.RubyModule;
@@ -64,8 +63,8 @@
 import org.truffleruby.core.string.RubyString;
 import org.truffleruby.core.string.StringNodes.MakeStringNode;
 import org.truffleruby.core.thread.RubyThread;
-import org.truffleruby.core.thread.ThreadManager;
 import org.truffleruby.language.RubyDynamicObject;
+import org.truffleruby.language.SafepointPredicate;
 import org.truffleruby.language.backtrace.Backtrace;
 import org.truffleruby.language.backtrace.BacktraceFormatter;
 import org.truffleruby.language.control.ExitException;
@@ -270,8 +269,6 @@ protected boolean watchSignalProc(Object signalString, RubyProc action,
                 }
 
                 final RubyThread rootThread = context.getThreadManager().getRootThread();
-                final FiberManager fiberManager = rootThread.fiberManager;
-                final ThreadManager threadManager = context.getThreadManager();
 
                 // Workaround: we need to register with Truffle (which means going multithreaded),
                 // so that NFI can get its context to call pthread_kill() (GR-7405).
@@ -296,14 +293,8 @@ protected boolean watchSignalProc(Object signalString, RubyProc action,
                 try {
                     context.getSafepointManager().pauseAllThreadsAndExecuteFromNonRubyThread(
                             "Handling of signal " + signal,
-                            true,
-                            (rubyThread, currentNode) -> {
-                                if (rubyThread == rootThread &&
-                                        threadManager.getRubyFiberFromCurrentJavaThread() == fiberManager
-                                                .getCurrentFiber()) {
-                                    ProcOperations.rootCall(action);
-                                }
-                            });
+                            SafepointPredicate.currentFiberOfThread(context, rootThread),
+                            (rubyThread, currentNode) -> ProcOperations.rootCall(action));
                 } finally {
                     truffleContext.leave(this, prev);
                 }

src/main/java/org/truffleruby/core/thread/RubyThread.java

@@ -11,11 +11,14 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Lock;
 
+import com.oracle.truffle.api.CompilerDirectives.TruffleBoundary;
 import org.truffleruby.RubyContext;
 import org.truffleruby.RubyLanguage;
 import org.truffleruby.core.InterruptMode;
@@ -29,6 +32,7 @@
 import org.truffleruby.core.tracepoint.TracePointState;
 import org.truffleruby.language.Nil;
 import org.truffleruby.language.RubyDynamicObject;
+import org.truffleruby.language.SafepointAction;
 import org.truffleruby.language.objects.ObjectGraph;
 import org.truffleruby.language.objects.ObjectGraphNode;
 import org.truffleruby.language.threadlocal.ThreadLocalGlobals;
@@ -37,28 +41,31 @@
 
 public class RubyThread extends RubyDynamicObject implements ObjectGraphNode {
 
-    public final ThreadLocalGlobals threadLocalGlobals;
-    public volatile InterruptMode interruptMode;
-    public volatile ThreadStatus status;
-    public final List<Lock> ownedLocks;
-    public FiberManager fiberManager;
-    CountDownLatch finishedLatch;
+    // Fields initialized here are initialized just after the super() call, and before the rest of the constructor
+    public final ThreadLocalGlobals threadLocalGlobals = new ThreadLocalGlobals();
+    public volatile InterruptMode interruptMode = InterruptMode.IMMEDIATE;
+    public volatile ThreadStatus status = ThreadStatus.RUN;
+    public final List<Lock> ownedLocks = new ArrayList<>();
+    public final FiberManager fiberManager;
+    CountDownLatch finishedLatch = new CountDownLatch(1);
     final RubyHash threadLocalVariables;
     final RubyHash recursiveObjects;
     final RubyHash recursiveObjectsSingle;
     final RubyRandomizer randomizer;
-    public final TracePointState tracePointState;
+    public final TracePointState tracePointState = new TracePointState();
     boolean reportOnException;
     boolean abortOnException;
-    public volatile Thread thread;
-    volatile RubyException exception;
-    volatile Object value;
-    public final AtomicBoolean wakeUp;
-    volatile int priority;
-    public ThreadLocalBuffer ioBuffer;
+    public volatile Thread thread = null;
+    volatile RubyException exception = null;
+    volatile Object value = null;
+    public final AtomicBoolean wakeUp = new AtomicBoolean(false);
+    volatile int priority = Thread.NORM_PRIORITY;
+    public ThreadLocalBuffer ioBuffer = ThreadLocalBuffer.NULL_BUFFER;
+    // Needs to be a thread-safe queue because multiple Fibers of the same Thread might enqueue concurrently
+    public final Queue<SafepointAction> pendingSafepointActions = newLinkedBlockingQueue();
     Object threadGroup;
     String sourceLocation;
-    Object name;
+    Object name = Nil.INSTANCE;
 
     public RubyThread(
             RubyClass rubyClass,
@@ -70,31 +77,16 @@ public RubyThread(
             Object threadGroup,
             String sourceLocation) {
         super(rubyClass, shape);
-        this.threadLocalGlobals = new ThreadLocalGlobals();
-        this.interruptMode = InterruptMode.IMMEDIATE;
-        this.status = ThreadStatus.RUN;
-        this.ownedLocks = new ArrayList<>();
-        this.finishedLatch = new CountDownLatch(1);
         this.threadLocalVariables = HashOperations.newEmptyHash(context, language);
         this.recursiveObjects = HashOperations.newEmptyHash(context, language);
         this.recursiveObjectsSingle = HashOperations.newEmptyHash(context, language);
         this.recursiveObjectsSingle.compareByIdentity = true;
-        this.randomizer = RandomizerNodes.newRandomizer(
-                context,
-                language,
-                false); // This random instance is only for this thread and thus does not need to be thread-safe
-        this.tracePointState = new TracePointState();
+        // This random instance is only for this thread and thus does not need to be thread-safe
+        this.randomizer = RandomizerNodes.newRandomizer(context, language, false);
         this.reportOnException = reportOnException;
         this.abortOnException = abortOnException;
-        this.thread = null;
-        this.exception = null;
-        this.value = null;
-        this.wakeUp = new AtomicBoolean(false);
-        this.priority = Thread.NORM_PRIORITY;
-        this.ioBuffer = ThreadLocalBuffer.NULL_BUFFER;
         this.threadGroup = threadGroup;
         this.sourceLocation = sourceLocation;
-        this.name = Nil.INSTANCE;
         // Initialized last as it captures `this`
         this.fiberManager = new FiberManager(language, context, this);
     }
@@ -105,4 +97,9 @@ public void getAdjacentObjects(Set<Object> reachable) {
         ObjectGraph.addProperty(reachable, name);
     }
 
+    @TruffleBoundary
+    private static LinkedBlockingQueue<SafepointAction> newLinkedBlockingQueue() {
+        return new LinkedBlockingQueue<>();
+    }
+
 }

src/main/java/org/truffleruby/core/thread/ThreadManager.java

@@ -667,17 +667,15 @@ private void doKillOtherThreads() {
         while (true) {
             try {
                 final String reason = "kill other threads for shutdown";
-                context.getSafepointManager().pauseAllThreadsAndExecute(reason, null, false, (thread, currentNode) -> {
-                    if (Thread.currentThread() != initiatingJavaThread) {
-                        final FiberManager fiberManager = thread.fiberManager;
-                        final RubyFiber fiber = getRubyFiberFromCurrentJavaThread();
-
-                        if (fiberManager.getCurrentFiber() == fiber) {
+                context.getSafepointManager().pauseAllThreadsAndExecute(
+                        reason,
+                        null,
+                        thread -> Thread.currentThread() != initiatingJavaThread &&
+                                getRubyFiberFromCurrentJavaThread() == thread.fiberManager.getCurrentFiber(),
+                        (thread, currentNode) -> {
                             thread.status = ThreadStatus.ABORTING;
                             throw new KillException();
-                        }
-                    }
-                });
+                        });
                 break; // Successfully executed the safepoint and sent the exceptions.
             } catch (RaiseException e) {
                 final RubyException rubyException = e.getException();