Supress internal ThreadAbortedError in user threads (#526)

Author: eupp

bootstrap/src/sun/nio/ch/lincheck/EventTracker.java

@@ -23,6 +23,7 @@ public interface EventTracker {
     void beforeThreadStart();
     void afterThreadFinish();
     void threadJoin(Thread thread, boolean withTimeout);
+    void onThreadRunException(Throwable exception);
 
     void beforeLock(int codeLocation);
     void lock(Object monitor);

bootstrap/src/sun/nio/ch/lincheck/Injections.java

@@ -177,6 +177,21 @@ public static void afterThreadFinish() {
         tracker.afterThreadFinish();
     }
 
+    /**
+     * Called from thread's {@code run} method failed with an exception.
+     *
+     * @param exception the exception that was thrown in the thread.
+     */
+    public static void onThreadRunException(Throwable exception) {
+        Thread thread = Thread.currentThread();
+        // TestThread is handled separately
+        if (thread instanceof TestThread) return;
+        ThreadDescriptor descriptor = ThreadDescriptor.getCurrentThreadDescriptor();
+        if (descriptor == null) return;
+        EventTracker tracker = descriptor.getEventTracker();
+        tracker.onThreadRunException(exception);
+    }
+
     /**
      * Called from instrumented code instead of {@code thread.join()}.
      */

src/jvm/main/org/jetbrains/kotlinx/lincheck/Utils.kt

@@ -57,7 +57,7 @@ internal fun executeActor(
             throw invE
         // Exception thrown not during the method invocation should contain underlying exception
         return ExceptionResult.create(
-            invE.cause?.takeIf { exceptionCanBeValidExecutionResult(it) }
+            invE.cause?.takeIf { !isInternalException(it) }
                 ?: throw invE
         )
     } catch (e: Exception) {
@@ -187,12 +187,6 @@ internal fun collectThreadDump(runner: Runner) = Thread.getAllStackTraces().filt
 
 internal val String.canonicalClassName get() = this.replace('/', '.')
 
-@Suppress("DEPRECATION") // ThreadDeath
-internal fun exceptionCanBeValidExecutionResult(exception: Throwable): Boolean {
-    return exception !is ThreadDeath && // is used to stop thread in `FixedActiveThreadsExecutor` via `thread.stop()`
-           exception !is ThreadAbortedError // is used to abort thread in `ManagedStrategy`
-}
-
 internal val Throwable.text: String get() {
     val writer = StringWriter()
     printStackTrace(PrintWriter(writer))

src/jvm/main/org/jetbrains/kotlinx/lincheck/runner/ParallelThreadsRunner.kt

@@ -487,8 +487,6 @@ internal open class ParallelThreadsRunner(
     override fun isCurrentRunnerThread(thread: Thread): Boolean = executor.threads.any { it === thread }
 
     override fun onThreadFinish(iThread: Int) {}
-
-    override fun onThreadFailure(iThread: Int, e: Throwable) {}
 }
 
 internal enum class UseClocks { ALWAYS, RANDOM }

src/jvm/main/org/jetbrains/kotlinx/lincheck/runner/Runner.kt

@@ -13,6 +13,7 @@ import org.jetbrains.kotlinx.lincheck.*
 import org.jetbrains.kotlinx.lincheck.annotations.*
 import org.jetbrains.kotlinx.lincheck.execution.*
 import org.jetbrains.kotlinx.lincheck.strategy.*
+import org.jetbrains.kotlinx.lincheck.strategy.managed.*
 import java.io.*
 import java.lang.reflect.*
 import java.util.concurrent.atomic.*
@@ -59,12 +60,6 @@ abstract class Runner protected constructor(
      */
     abstract fun onThreadFinish(iThread: Int)
 
-    /**
-     * This method is invoked by the corresponding test thread
-     * when an unexpected exception is thrown.
-     */
-    abstract fun onThreadFailure(iThread: Int, e: Throwable)
-
     /**
      * This method is invoked by the corresponding test thread
      * when the current coroutine suspends.
@@ -99,13 +94,30 @@ abstract class Runner protected constructor(
     }
 
     /**
-     * Is invoked after each actor execution from the specified thread, even if a legal exception was thrown.
+     * Is invoked after each actor execution from the specified thread.
      * The invocations are inserted into the generated code.
      */
     fun onActorFinish() {
         strategy.onActorFinish()
     }
 
+    /**
+     * Is invoked if an actor execution has thrown an exception.
+     *
+     * Default implementation checks if the failure
+     * was caused by an internal exception (see [isInternalException]) and re-throw in this case,
+     * otherwise it treats the exception as a normal actor result.
+     *
+     * @param iThread the number of the invoking thread where the failure occurred.
+     * @param throwable the exception that caused the actor failure.
+     */
+    // used in byte-code generation
+    open fun onActorFailure(iThread: Int, throwable: Throwable) {
+        if (isInternalException(throwable)) {
+            throw throwable
+        }
+    }
+
     fun beforePart(part: ExecutionPart) {
         completedOrSuspendedThreads.set(0)
         currentExecutionPart = part
@@ -133,3 +145,15 @@ abstract class Runner protected constructor(
 enum class ExecutionPart {
     INIT, PARALLEL, POST, VALIDATION
 }
+
+/**
+ * Checks if the provided exception is considered an internal exception.
+ * Internal exceptions are those used by the Lincheck itself
+ * to control execution of the analyzed code.
+ */
+@Suppress("DEPRECATION") // ThreadDeath
+internal fun isInternalException(exception: Throwable): Boolean =
+    // is used to stop thread in `FixedActiveThreadsExecutor` via `thread.stop()`
+    exception is ThreadDeath ||
+    // is used to abort thread in `ManagedStrategy`
+    exception is LincheckAnalysisAbortedError

src/jvm/main/org/jetbrains/kotlinx/lincheck/runner/TestThreadExecution.java

@@ -11,8 +11,6 @@
 
 import org.jetbrains.kotlinx.lincheck.Result;
 
-import static org.jetbrains.kotlinx.lincheck.UtilsKt.*;
-
 
 /**
  * Instance of this class represents the test execution for ONE thread. Several instances should be run in parallel.
@@ -53,12 +51,4 @@ public void incClock() {
         curClock++;
     }
 
-    // used in byte-code generation
-    public void failOnExceptionIsUnexpected(int iThread, Throwable e) throws Throwable {
-        if (!exceptionCanBeValidExecutionResult(e)) {
-            runner.onThreadFailure(iThread, e);
-            throw e;
-        }
-    }
-
 }

src/jvm/main/org/jetbrains/kotlinx/lincheck/runner/TestThreadExecutionGenerator.java

@@ -70,7 +70,8 @@ public class TestThreadExecutionGenerator {
     private static final Method PARALLEL_THREADS_RUNNER_PROCESS_INVOCATION_RESULT_METHOD = new Method("processInvocationResult", RESULT_TYPE, new Type[]{ OBJECT_TYPE, INT_TYPE, INT_TYPE });
     private static final Method RUNNER_IS_PARALLEL_EXECUTION_COMPLETED_METHOD = new Method("isParallelExecutionCompleted", BOOLEAN_TYPE, new Type[]{});
 
-    private static final Method TEST_THREAD_EXECUTION_FAIL_ON_EXCEPTION_IF_UNEXPECTED = new Method("failOnExceptionIsUnexpected", VOID_TYPE, new Type[]{INT_TYPE, THROWABLE_TYPE});
+    private static final Method RUNNER_ON_ACTOR_FAILURE_METHOD = new Method("onActorFailure", VOID_TYPE, new Type[]{INT_TYPE, THROWABLE_TYPE});
+
     private static int generatedClassNumber = 0;
 
     static {
@@ -149,6 +150,14 @@ private static void generateRun(ClassVisitor cv, Type testType, int iThread, Lis
         mv.getField(TEST_THREAD_EXECUTION_TYPE, "runner", RUNNER_TYPE);
         mv.push(iThread);
         mv.invokeVirtual(RUNNER_TYPE, RUNNER_ON_THREAD_START_METHOD);
+
+        // wrap actor's running loop in try-finally
+        Label actorsRunningLoopBlockStart = mv.newLabel();
+        Label actorsRunningLoopBlockEnd = mv.newLabel();
+        Label actorsRunningLoopBlockFinally = mv.newLabel();
+        mv.visitTryCatchBlock(actorsRunningLoopBlockStart, actorsRunningLoopBlockEnd, actorsRunningLoopBlockFinally, null);
+        
+        mv.visitLabel(actorsRunningLoopBlockStart);
         // Number of current operation (starts with 0)
         int iLocal = mv.newLocal(INT_TYPE);
         mv.push(0);
@@ -232,11 +241,14 @@ private static void generateRun(ClassVisitor cv, Type testType, int iThread, Lis
             int eLocal = mv.newLocal(THROWABLE_TYPE);
             mv.storeLocal(eLocal);
 
+            // push the runner on stack to call its method
             mv.loadThis();
+            mv.getField(TEST_THREAD_EXECUTION_TYPE, "runner", RUNNER_TYPE);
+            // push iThread and exception on stack
             mv.push(iThread);
             mv.loadLocal(eLocal);
-            // Fail if this exception is not a valid execution result
-            mv.invokeVirtual(TEST_THREAD_EXECUTION_TYPE, TEST_THREAD_EXECUTION_FAIL_ON_EXCEPTION_IF_UNEXPECTED);
+            // Fail if this exception is an internal exception
+            mv.invokeVirtual(RUNNER_TYPE, RUNNER_ON_ACTOR_FAILURE_METHOD);
 
             mv.loadLocal(eLocal);
 
@@ -270,13 +282,28 @@ private static void generateRun(ClassVisitor cv, Type testType, int iThread, Lis
             mv.iinc(iLocal, 1);
             mv.visitLabel(launchNextActor);
         }
+        mv.visitInsn(ACONST_NULL); // push null exception value indicating normal method's termination
+        mv.goTo(actorsRunningLoopBlockFinally);
+        mv.visitLabel(actorsRunningLoopBlockEnd);
+        
+        mv.visitLabel(actorsRunningLoopBlockFinally);
         // Call runner's onThreadFinish(iThread) method
         mv.loadThis();
         mv.getField(TEST_THREAD_EXECUTION_TYPE, "runner", RUNNER_TYPE);
         mv.push(iThread);
         mv.invokeVirtual(RUNNER_TYPE, RUNNER_ON_THREAD_FINISH_METHOD);
+
+        // Check if an exception was thrown in the actors' running loop and re-throw it
+        Label methodReturnLabel = mv.newLabel();
+        mv.dup();
+        mv.ifNull(methodReturnLabel);
+        mv.throwException(); // re-throw exception
+
         // Complete the method
+        mv.visitLabel(methodReturnLabel);
+        mv.pop(); // pop null exception value indicating normal method's termination
         mv.visitInsn(RETURN);
+
         mv.visitMaxs(3, 4);
         mv.visitEnd();
     }

src/jvm/main/org/jetbrains/kotlinx/lincheck/strategy/managed/LoopDetector.kt

@@ -58,7 +58,7 @@ import java.util.ArrayList
  * - For instance, if the [currentInterleavingHistory] is [0: 2], [1: 3], [0: 3], [1: 3], [0: 3], ..., [1: 3], [0: 3] and a deadlock is detected,
  * the cycle is identified as [1: 3], [0: 3].
  * This means 2 executions in thread 0 and 3 executions in both threads 1 and 0 will be allowed.
- * - Execution is halted after the last execution in thread 0 using [ThreadAbortedError].
+ * - Execution is halted after the last execution in thread 0 using [LincheckAnalysisAbortedError].
  * - The logic for tracking executions and switches in replay mode is implemented in [ReplayModeLoopDetectorHelper].
  *
  * Note: An example of this behavior is detailed in the comments of the code itself.

src/jvm/main/org/jetbrains/kotlinx/lincheck/strategy/managed/ManagedStrategy.kt

@@ -523,6 +523,28 @@ abstract class ManagedStrategy(
         onThreadFinish(currentThreadId)
     }
 
+    /**
+     * Handles exceptions that occur in a specific thread.
+     * This method is called when a thread finishes with an exception.
+     *
+     * @param exception The exception that was thrown within the thread.
+     */
+    override fun onThreadRunException(exception: Throwable) = runInIgnoredSection {
+        val currentThreadId = threadScheduler.getCurrentThreadId()
+        // do not track unregistered threads
+        if (currentThreadId < 0) return
+        // scenario threads are handled separately by the runner itself
+        if (currentThreadId < scenario.nThreads) return
+        // check if the exception is internal
+        if (isInternalException(exception)) {
+            onInternalException(currentThreadId, exception)
+        } else {
+            // re-throw any non-internal exception,
+            // so it will be treated as the final result of `Thread::run`.
+            throw exception
+        }
+    }
+
     override fun threadJoin(thread: Thread?, withTimeout: Boolean) = runInIgnoredSection {
         if (withTimeout) return // timeouts occur instantly
         val currentThreadId = threadScheduler.getCurrentThreadId()
@@ -593,46 +615,51 @@ abstract class ManagedStrategy(
 
     /**
      * This method is executed as the first thread action.
-     * @param iThread the number of the executed thread according to the [scenario][ExecutionScenario].
+     *
+     * @param threadId the thread id of the started thread.
      */
-    open fun onThreadStart(iThread: Int) {
-        threadScheduler.awaitTurn(iThread)
-        threadScheduler.startThread(iThread)
+    open fun onThreadStart(threadId: ThreadId) {
+        threadScheduler.awaitTurn(threadId)
+        threadScheduler.startThread(threadId)
     }
 
     /**
-     * This method is executed as the last thread action if no exception has been thrown.
-     * @param iThread the number of the executed thread according to the [scenario][ExecutionScenario].
+     * This method is executed as the last thread action.
+     *
+     * @param threadId the thread id of the finished thread.
      */
-    open fun onThreadFinish(iThread: Int) {
-        threadScheduler.awaitTurn(iThread)
-        threadScheduler.finishThread(iThread)
-        loopDetector.onThreadFinish(iThread)
+    open fun onThreadFinish(threadId: ThreadId) {
+        threadScheduler.awaitTurn(threadId)
+        threadScheduler.finishThread(threadId)
+        loopDetector.onThreadFinish(threadId)
         traceCollector?.onThreadFinish()
-        unblockJoiningThreads(iThread)
-        val nextThread = chooseThreadSwitch(iThread, true)
+        unblockJoiningThreads(threadId)
+        val nextThread = chooseThreadSwitch(threadId, true)
         setCurrentThread(nextThread)
     }
 
     /**
-     * This method is executed if an illegal exception has been thrown (see [exceptionCanBeValidExecutionResult]).
-     * @param iThread the number of the executed thread according to the [scenario][ExecutionScenario].
-     * @param exception the exception that was thrown
+     * This method is executed if an internal exception has been thrown (see [isInternalException]).
+     *
+     * @param threadId the thread id of the thread where exception was thrown.
+     * @param exception the exception that was thrown.
      */
-    open fun onThreadFailure(iThread: Int, exception: Throwable) {
-        // This method is called only if exception can't be treated as a normal operation result,
+    open fun onInternalException(threadId: Int, exception: Throwable) {
+        check(isInternalException(exception))
+        // This method is called only if the exception cannot be treated as a normal result,
         // so we exit testing code to avoid trace collection resume or some bizarre bugs
         leaveTestingCode()
-        // skip abort exception
-        if (exception !== ThreadAbortedError) {
-            // Despite the fact that the corresponding failure will be detected by the runner,
-            // the managed strategy can construct a trace to reproduce this failure.
-            // Let's then store the corresponding failing result and construct the trace.
-            suddenInvocationResult = UnexpectedExceptionInvocationResult(exception, runner.collectExecutionResults())
-            threadScheduler.abortAllThreads()
-        }
-        // notify the scheduler that the thread is going to be finished
-        threadScheduler.finishThread(iThread)
+        // suppress `ThreadAbortedError`
+        if (exception is LincheckAnalysisAbortedError) return
+        // Though the corresponding failure will be detected by the runner,
+        // the managed strategy can construct a trace to reproduce this failure.
+        // Let's then store the corresponding failing result and construct the trace.
+        suddenInvocationResult = UnexpectedExceptionInvocationResult(
+            exception, 
+            runner.collectExecutionResults()
+        )
+        threadScheduler.abortAllThreads()
+        throw exception
     }
 
     override fun onActorStart(iThread: Int) {
@@ -1996,8 +2023,10 @@ internal class ManagedStrategyRunner(
         managedStrategy.onThreadFinish(iThread)
     }
 
-    override fun onThreadFailure(iThread: Int, e: Throwable) = runInIgnoredSection {
-        managedStrategy.onThreadFailure(iThread, e)
+    override fun onActorFailure(iThread: Int, throwable: Throwable) = runInIgnoredSection {
+        if (isInternalException(throwable)) {
+            managedStrategy.onInternalException(iThread, throwable)
+        }
     }
 
     override fun afterCoroutineSuspended(iThread: Int) = runInIgnoredSection {