Prevent modifying the concurrency limit of the thread pools underlyin…

…g locked code thunks (#240)
criteo · Feb 27, 2018 · 0730ba4 · 0730ba4
1 parent ce09a82
commit 0730ba4
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 3 deletions.
diff --git a/core/src/main/scala/com/criteo/cuttle/Executor.scala b/core/src/main/scala/com/criteo/cuttle/Executor.scala
@@ -242,7 +242,7 @@ case class Execution[S <: Scheduling](
     *
     * While waiting for the lock, the [[Execution]] will be seen as __WAITING__ in the UI and the API.
     */
-  def withMaxParallelRuns[A, B](lock: A, concurrencyLimit: Int)(thunk: => Future[B]) = {
+  def withMaxParallelRuns[A, B](lock: A, concurrencyLimit: Int)(thunk: => Future[B]): Future[B] = {
     if (isWaiting.get) {
       sys.error(s"Already waiting")
     } else {
@@ -254,7 +254,11 @@ case class Execution[S <: Scheduling](
         }
         Execution.locks(lock)
       }
-      pool.run(this, s"Locked on ${lock}") { () =>
+      require(
+        pool.concurrencyLimit == concurrencyLimit,
+        s"Inconsistent concurrency limits defined for the execution pool for $lock"
+      )
+      pool.run(this, s"Locked on $lock") { () =>
         streams.debug(s"Resuming")
         isWaiting.set(false)
         thunk

diff --git a/core/src/main/scala/com/criteo/cuttle/platforms/ExecutionPool.scala b/core/src/main/scala/com/criteo/cuttle/platforms/ExecutionPool.scala
@@ -15,7 +15,7 @@ import io.circe.syntax._
   *
   * @param concurrencyLimit The maximum number of concurrent executions.
   */
-class ExecutionPool(concurrencyLimit: Int) extends WaitingExecutionQueue {
+case class ExecutionPool(concurrencyLimit: Int) extends WaitingExecutionQueue {
   def canRunNextCondition(implicit txn: InTxn) = _running().size < concurrencyLimit
   def doRunNext()(implicit txn: InTxn): Unit = ()