Don't use thread interrupts to cancel executions (enso-org#1574)

MULXCODE · Mar 16, 2021 · ca90d98 · ca90d98
1 parent b0680d0
commit ca90d98
Show file tree

Hide file tree

Showing 5 changed files with 21 additions and 10 deletions.
diff --git a/engine/runtime/src/main/java/org/enso/interpreter/runtime/ThreadManager.java b/engine/runtime/src/main/java/org/enso/interpreter/runtime/ThreadManager.java
@@ -6,6 +6,7 @@
 import com.oracle.truffle.api.nodes.InvalidAssumptionException;
 import org.enso.interpreter.runtime.control.ThreadInterruptedException;
 
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Phaser;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -23,6 +24,7 @@ protected boolean onAdvance(int phase, int registeredParties) {
   private final ReentrantLock lock = new ReentrantLock();
 
   private volatile boolean safepoint = false;
+  private final ConcurrentHashMap<Thread, Boolean> interruptFlags = new ConcurrentHashMap<>();
 
   /**
    * Registers the current thread as running guest code.
@@ -35,6 +37,7 @@ protected boolean onAdvance(int phase, int registeredParties) {
    */
   public void enter() {
     safepointPhaser.register();
+    interruptFlags.put(Thread.currentThread(), false);
   }
 
   /**
@@ -44,32 +47,35 @@ public void enter() {
    */
   public void leave() {
     safepointPhaser.arriveAndDeregister();
+    interruptFlags.remove(Thread.currentThread());
   }
 
   /** Called from the interpreter to periodically perform a safepoint check. */
   public void poll() {
     if (safepoint) {
       CompilerDirectives.transferToInterpreter();
       safepointPhaser.arriveAndAwaitAdvance();
-      if (Thread.interrupted()) {
+      if (interruptFlags.get(Thread.currentThread())) {
+        interruptFlags.put(Thread.currentThread(), false);
         throw new ThreadInterruptedException();
       }
     }
   }
 
   /**
    * Forces all threads managed by this system to halt at the next safepoint (i.e. a {@link #poll()}
-   * call) and throw an exception if they were interrupted.
+   * call) and throw a {@link ThreadInterruptedException}.
    *
    * <p>This method is blocking, does not return until the last managed thread reports at a
    * safepoint.
    *
    * <p>This method may not be called from a thread that is itself managed by this system, as doing
    * so may result in a deadlock.
    */
-  public void checkInterrupts() {
+  public void interruptThreads() {
     lock.lock();
     try {
+      interruptFlags.replaceAll((t, b) -> true);
       enter();
       try {
         safepoint = true;

diff --git a/...runtime/src/main/scala/org/enso/interpreter/instrument/execution/JobExecutionEngine.scala b/...runtime/src/main/scala/org/enso/interpreter/instrument/execution/JobExecutionEngine.scala
@@ -91,7 +91,7 @@ class JobExecutionEngine(
       runningJob.future.cancel(runningJob.job.mayInterruptIfRunning)
     }
     runtimeContext.executionService.getContext.getThreadManager
-      .checkInterrupts()
+      .interruptThreads()
   }
 
   /** @inheritdoc */
@@ -104,15 +104,15 @@ class JobExecutionEngine(
       }
     }
     runtimeContext.executionService.getContext.getThreadManager
-      .checkInterrupts()
+      .interruptThreads()
   }
 
   /** @inheritdoc */
   override def stop(): Unit = {
     val allJobs = runningJobsRef.get()
     allJobs.foreach(_.future.cancel(true))
     runtimeContext.executionService.getContext.getThreadManager
-      .checkInterrupts()
+      .interruptThreads()
     jobExecutor.shutdownNow()
   }
 

diff --git a/engine/runtime/src/main/scala/org/enso/interpreter/instrument/job/ExecuteJob.scala b/engine/runtime/src/main/scala/org/enso/interpreter/instrument/job/ExecuteJob.scala
@@ -19,7 +19,13 @@ class ExecuteJob(
   stack: List[InstrumentFrame],
   updatedVisualisations: Seq[UUID],
   sendMethodCallUpdates: Boolean
-) extends Job[Unit](List(contextId), true, true)
+) extends Job[Unit](
+      List(contextId),
+      isCancellable = true,
+      // TODO[MK]: make this interruptible when https://github.com/oracle/graal/issues/3273
+      // is resolved
+      mayInterruptIfRunning = false
+    )
     with ProgramExecutionSupport {
 
   def this(exe: Executable) =

diff --git a/engine/runtime/src/test/scala/org/enso/interpreter/test/semantic/RuntimeManagementTest.scala b/engine/runtime/src/test/scala/org/enso/interpreter/test/semantic/RuntimeManagementTest.scala
@@ -51,8 +51,7 @@ class RuntimeManagementTest extends InterpreterTest {
           reportedCount += consumeOut.length
         }
         val expectedOut = List.fill(n)("Interrupted.")
-        threads.foreach(_.interrupt())
-        langCtx.getThreadManager.checkInterrupts()
+        langCtx.getThreadManager.interruptThreads()
         threads.foreach(_.join())
         consumeOut shouldEqual expectedOut
         threads.forall(!_.isAlive) shouldBe true

diff --git a/table/src/main/java/org/enso/table/data/index/HashIndex.java b/table/src/main/java/org/enso/table/data/index/HashIndex.java
@@ -44,7 +44,7 @@ public List<Integer> loc(Object item) {
 
   @Override
   public String ilocString(int loc) {
-    return iloc(loc).toString();
+    return String.valueOf(iloc(loc));
   }
 
   @Override