Fix deadlock

silk-core/src/main/scala/org/silkframework/runtime/activity/ActivityContext.scala

@@ -55,10 +55,13 @@ trait ActivityContext[T] {
   def blockUntil(condition: () => Boolean): Unit
 
   /**
-    * Possibly executes other activities that are blocked.
+    * Possibly executes other activities that are blocked, for a bounded amount of time.
     * Can be called to avoid deadlocks if child activities are run in the background.
+    *
+    * @return True, if the pool is quiescent, i.e., there are no other activities left to help with.
+    *         Always true if the current thread is not running in our own pool.
     */
-  def helpQuiesce(): Unit
+  def helpQuiesce(): Boolean
 
   /**
     * The user that started the activity.

silk-core/src/main/scala/org/silkframework/runtime/activity/ActivityMonitor.scala

@@ -77,18 +77,16 @@ class ActivityMonitor[T](name: String,
     * @param condition Evaluates the condition to wait for. Will be called frequently.
     */
   def blockUntil(condition: () => Boolean): Unit = {
-    val sleepTime = ActivityMonitor.blockPollIntervalMs
     while(!condition() && !status().isInstanceOf[Canceling]) {
-      if(runningInOwnPool()) {
-        helpQuiesce()
-      } else {
+      if(helpQuiesce()) {
+        // The pool is quiescent, i.e., there are no other activities to help with, so we sleep before re-checking the condition.
         ForkJoinPool.managedBlock(
           new ManagedBlocker {
             @volatile
             private var releasable = false
 
             override def block(): Boolean = {
-              Thread.sleep(sleepTime)
+              Thread.sleep(ActivityMonitor.blockPollIntervalMs)
               releasable = true
               true
             }
@@ -103,19 +101,24 @@ class ActivityMonitor[T](name: String,
   }
 
   /**
-    * Possibly executes other activities that are blocked.
+    * Possibly executes other activities that are blocked, for at most [[ActivityMonitor.blockPollIntervalMs]] milliseconds.
     * Can be called to avoid deadlocks if child activities are run in the background.
+    *
+    * @return True, if the pool is quiescent, i.e., there are no other activities left to help with.
+    *         Always true if the current thread is not running in our own pool.
     */
-  def helpQuiesce(): Unit = {
-    // helpQuiesce() might execute another activity in this thread
+  def helpQuiesce(): Boolean = {
+    // Helping might execute another activity in this thread
     interruptEnabled.synchronized {
       interruptEnabled.set(false)
     }
     try {
-      // Only execute helpQuiesce() in our own thread pool
+      // Only help executing other activities in our own thread pool
       if(runningInOwnPool()) {
         Thread.currentThread().asInstanceOf[ForkJoinWorkerThread]
           .getPool.awaitQuiescence(ActivityMonitor.blockPollIntervalMs, TimeUnit.MILLISECONDS)
+      } else {
+        true
       }
     } finally {
       interruptEnabled.synchronized {

silk-core/src/test/scala/org/silkframework/runtime/activity/ActivityExecutionTest.scala

@@ -5,6 +5,7 @@ import org.scalatest.time.{Seconds, Span}
 import org.silkframework.runtime.activity.Status.{Finished, Running, Waiting}
 import org.silkframework.runtime.users.{User, UserActions}
 
+import java.util.concurrent.atomic.AtomicLong
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.Future
 import org.scalatest.flatspec.AnyFlatSpec
@@ -119,6 +120,50 @@ class ActivityExecutionTest extends AnyFlatSpec with Matchers with Eventually  {
     }
   }
 
+  it should "finish a cancelled blocking activity even if other activities keep the pool busy" in {
+    // Regression test for CMEM-7839: cancelled blocking activities used to be stuck in Canceling
+    // while helpQuiesce() waited for all other activities to finish.
+    val sleepingActivity = Activity(new SleepingActivity())
+    sleepingActivity.start()
+    eventually {
+      sleepingActivity.value() mustBe true
+    }
+
+    val blockingActivity = Activity(new BlockingActivity())
+    blockingActivity.start()
+    eventually {
+      blockingActivity.value() mustBe true
+    }
+    // Give the blocking activity time to enter helpQuiesce() before cancelling it
+    Thread.sleep(1000)
+
+    blockingActivity.cancel()
+    try {
+      eventually {
+        blockingActivity.status() mustBe a[Finished]
+      }
+    } finally {
+      stopActivities(Seq(sleepingActivity, blockingActivity))
+    }
+  }
+
+  it should "not busy-spin in blockUntil while the pool is quiescent" in {
+    val conditionEvaluations = new AtomicLong()
+    val activity = Activity(new Activity[Unit] {
+      override def run(context: ActivityContext[Unit])(implicit userContext: UserContext): Unit = {
+        val deadline = System.currentTimeMillis() + 2000
+        context.blockUntil { () =>
+          conditionEvaluations.incrementAndGet()
+          System.currentTimeMillis() > deadline
+        }
+      }
+    })
+    activity.start()
+    activity.waitUntilFinished()
+    // With a 500ms poll interval, only a few evaluations are expected within 2 seconds
+    conditionEvaluations.get() must be < 100L
+  }
+
   it should "allow activities to skip the waiting queue" in {
     val sleepingActivities =
       for (_ <- 0 until parallelism) yield {

silk-core/src/test/scala/org/silkframework/util/ActivityContextMock.scala

@@ -11,6 +11,6 @@ case class ActivityContextMock[T](initialValue: Option[T] = None) extends Activi
   override def log: Logger = Logger.getAnonymousLogger
   override def child[R](activity: Activity[R], progressContribution: Double): ActivityControl[R] = ???
   override def blockUntil(condition: () => Boolean): Unit = ???
-  override def helpQuiesce(): Unit = ???
+  override def helpQuiesce(): Boolean = ???
   override def startedBy: UserContext = ???
 }