diff --git a/silk-core/src/main/scala/org/silkframework/runtime/activity/ActivityContext.scala b/silk-core/src/main/scala/org/silkframework/runtime/activity/ActivityContext.scala index 50b9878da2..f875b2d5b6 100644 --- a/silk-core/src/main/scala/org/silkframework/runtime/activity/ActivityContext.scala +++ b/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. diff --git a/silk-core/src/main/scala/org/silkframework/runtime/activity/ActivityMonitor.scala b/silk-core/src/main/scala/org/silkframework/runtime/activity/ActivityMonitor.scala index 7ea052284e..c62e7c58dd 100644 --- a/silk-core/src/main/scala/org/silkframework/runtime/activity/ActivityMonitor.scala +++ b/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 { diff --git a/silk-core/src/test/scala/org/silkframework/runtime/activity/ActivityExecutionTest.scala b/silk-core/src/test/scala/org/silkframework/runtime/activity/ActivityExecutionTest.scala index 54f571f0db..f316aaa6f1 100644 --- a/silk-core/src/test/scala/org/silkframework/runtime/activity/ActivityExecutionTest.scala +++ b/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 { diff --git a/silk-core/src/test/scala/org/silkframework/util/ActivityContextMock.scala b/silk-core/src/test/scala/org/silkframework/util/ActivityContextMock.scala index 481c7ad0da..52e6dac8eb 100644 --- a/silk-core/src/test/scala/org/silkframework/util/ActivityContextMock.scala +++ b/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 = ??? }