Replace parallel collections with future based concurrency in tests. (#4841)

Scala 2.13 puts parallel collections into a separate module that's not compatible with Scala 2.12. To avoid having to work around things and to keep cross-compilation compatibility this just exchanges the approach for concurrency in tests to not use parallel collections at all.

Author: markusthoemmes

tests/src/test/scala/common/ConcurrencyHelpers.scala

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package common
+
+import scala.concurrent.duration._
+import scala.concurrent.{Await, ExecutionContext, Future}
+
+trait ConcurrencyHelpers {
+  def concurrently[T](times: Int, timeout: FiniteDuration)(op: => T)(implicit ec: ExecutionContext): Iterable[T] =
+    Await.result(Future.sequence((1 to times).map(_ => Future(op))), timeout)
+
+  def concurrently[B, T](over: Iterable[B], timeout: FiniteDuration)(op: B => T)(
+    implicit ec: ExecutionContext): Iterable[T] =
+    Await.result(Future.sequence(over.map(v => Future(op(v)))), timeout)
+}

tests/src/test/scala/limits/ThrottleTests.scala

@@ -21,9 +21,7 @@ import java.time.Instant
 
 import akka.http.scaladsl.model.StatusCodes.TooManyRequests
 
-import scala.collection.parallel.immutable.ParSeq
-import scala.concurrent.Future
-import scala.concurrent.Promise
+import scala.concurrent.{Await, Future, Promise}
 import scala.concurrent.duration._
 import org.junit.runner.RunWith
 import org.scalatest.BeforeAndAfterAll
@@ -102,10 +100,15 @@ class ThrottleTests
    *
    * @param results the sequence of results from invocations or firings
    */
-  def waitForActivations(results: ParSeq[RunResult]) = results.foreach { result =>
-    if (result.exitCode == SUCCESS_EXIT) {
-      withActivation(wsk.activation, result, totalWait = 5.minutes)(identity)
+  def waitForActivations(results: Seq[RunResult]) = {
+    val done = results.map { result =>
+      if (result.exitCode == SUCCESS_EXIT) {
+        Future(withActivation(wsk.activation, result, totalWait = 5.minutes)(_ => ()))
+      } else {
+        Future.successful(())
+      }
     }
+    Await.result(Future.sequence(done), 5.minutes)
   }
 
   /**
@@ -201,7 +204,7 @@ class ThrottleTests
     // wait for the activations last, if these fail, the throttle should be settled
     // and this gives the activations time to complete and may avoid unnecessarily polling
     println("waiting for activations to complete")
-    waitForActivations(results.par)
+    waitForActivations(results)
   }
 
   it should "throttle multiple activations of one trigger" in withAssetCleaner(wskprops) { (wp, assetHelper) =>
@@ -286,7 +289,7 @@ class ThrottleTests
     // wait for the activations last, giving the activations time to complete and
     // may avoid unnecessarily polling; if these fail, the throttle may not be settled
     println("waiting for activations to complete")
-    waitForActivations(combinedResults.par)
+    waitForActivations(combinedResults)
   }
 }
 

tests/src/test/scala/org/apache/openwhisk/common/ForcibleSemaphoreTests.scala

@@ -17,12 +17,19 @@
 
 package org.apache.openwhisk.common
 
+import common.ConcurrencyHelpers
+import org.apache.openwhisk.utils.ExecutionContextFactory
 import org.junit.runner.RunWith
 import org.scalatest.{FlatSpec, Matchers}
 import org.scalatest.junit.JUnitRunner
 
+import scala.concurrent.duration.DurationInt
+
 @RunWith(classOf[JUnitRunner])
-class ForcibleSemaphoreTests extends FlatSpec with Matchers {
+class ForcibleSemaphoreTests extends FlatSpec with Matchers with ConcurrencyHelpers {
+  // use an infinite thread pool to allow for maximum concurrency
+  implicit val executionContext = ExecutionContextFactory.makeCachedThreadPoolExecutionContext()
+
   behavior of "ForcableSemaphore"
 
   it should "not allow to acquire, force or release negative amounts of permits" in {
@@ -79,7 +86,7 @@ class ForcibleSemaphoreTests extends FlatSpec with Matchers {
     (0 until 100).foreach { _ =>
       val s = new ForcibleSemaphore(32)
       // try to acquire more permits than allowed in parallel
-      val acquires = (0 until 64).par.map(_ => s.tryAcquire()).seq
+      val acquires = concurrently(64, 1.minute)(s.tryAcquire())
 
       val result = Seq.fill(32)(true) ++ Seq.fill(32)(false)
       acquires should contain theSameElementsAs result

tests/src/test/scala/org/apache/openwhisk/common/NestedSemaphoreTests.scala

@@ -17,13 +17,21 @@
 
 package org.apache.openwhisk.common
 
+import common.ConcurrencyHelpers
+import org.apache.openwhisk.utils.ExecutionContextFactory
 import org.junit.runner.RunWith
 import org.scalatest.FlatSpec
 import org.scalatest.Matchers
 import org.scalatest.junit.JUnitRunner
 
+import scala.concurrent.duration.DurationInt
+
 @RunWith(classOf[JUnitRunner])
-class NestedSemaphoreTests extends FlatSpec with Matchers {
+class NestedSemaphoreTests extends FlatSpec with Matchers with ConcurrencyHelpers {
+  // use an infinite thread pool to allow for maximum concurrency
+  implicit val executionContext = ExecutionContextFactory.makeCachedThreadPoolExecutionContext()
+  val acquireTimeout = 1.minute
+
   behavior of "NestedSemaphore"
 
   it should "allow acquire of concurrency permits before acquire of memory permits" in {
@@ -34,16 +42,17 @@ class NestedSemaphoreTests extends FlatSpec with Matchers {
     val actionConcurrency = 5
     val actionMemory = 3
     //use all concurrency on a single slot
-    (1 to 5).par.map { i =>
-      s.tryAcquireConcurrent(actionId, actionConcurrency, actionMemory) shouldBe true
-    }
+    concurrently(5, acquireTimeout) {
+      s.tryAcquireConcurrent(actionId, actionConcurrency, actionMemory)
+    } should contain only true
     s.availablePermits shouldBe 20 - 3 //we used a single container (memory == 3)
     s.concurrentState(actionId).availablePermits shouldBe 0
 
     //use up all the remaining memory (17) and concurrency slots (17 / 3 * 5 = 25)
-    (1 to 25).par.map { i =>
-      s.tryAcquireConcurrent(actionId, actionConcurrency, actionMemory) shouldBe true
-    }
+    concurrently(25, acquireTimeout) {
+      s.tryAcquireConcurrent(actionId, actionConcurrency, actionMemory)
+    } should contain only true
+
     s.availablePermits shouldBe 2 //we used 18 (20/3 = 6, 6*3=18)
     s.concurrentState(actionId).availablePermits shouldBe 0
     s.tryAcquireConcurrent("action1", actionConcurrency, actionMemory) shouldBe false
@@ -55,7 +64,7 @@ class NestedSemaphoreTests extends FlatSpec with Matchers {
     (0 until 100).foreach { _ =>
       val s = new NestedSemaphore(32)
       // try to acquire more permits than allowed in parallel
-      val acquires = (0 until 64).par.map(_ => s.tryAcquire()).seq
+      val acquires = concurrently(64, acquireTimeout)(s.tryAcquire())
 
       val result = Seq.fill(32)(true) ++ Seq.fill(32)(false)
       acquires should contain theSameElementsAs result

tests/src/test/scala/org/apache/openwhisk/common/ResizableSemaphoreTests.scala

@@ -17,13 +17,21 @@
 
 package org.apache.openwhisk.common
 
+import common.ConcurrencyHelpers
+import org.apache.openwhisk.utils.ExecutionContextFactory
 import org.junit.runner.RunWith
 import org.scalatest.FlatSpec
 import org.scalatest.Matchers
 import org.scalatest.junit.JUnitRunner
 
+import scala.concurrent.duration.DurationInt
+
 @RunWith(classOf[JUnitRunner])
-class ResizableSemaphoreTests extends FlatSpec with Matchers {
+class ResizableSemaphoreTests extends FlatSpec with Matchers with ConcurrencyHelpers {
+  // use an infinite thread pool to allow for maximum concurrency
+  implicit val executionContext = ExecutionContextFactory.makeCachedThreadPoolExecutionContext()
+  val acquireTimeout = 1.minute
+
   behavior of "ResizableSemaphore"
 
   it should "not allow to acquire, force or release negative amounts of permits" in {
@@ -163,7 +171,7 @@ class ResizableSemaphoreTests extends FlatSpec with Matchers {
     (0 until 100).foreach { _ =>
       val s = new ResizableSemaphore(32, 35)
       // try to acquire more permits than allowed in parallel
-      val acquires = (0 until 64).par.map(_ => s.tryAcquire()).seq
+      val acquires = concurrently(64, acquireTimeout)(s.tryAcquire())
 
       val result = Seq.fill(32)(true) ++ Seq.fill(32)(false)
       acquires should contain theSameElementsAs result
@@ -173,11 +181,10 @@ class ResizableSemaphoreTests extends FlatSpec with Matchers {
   it should "release permits even under concurrent load" in {
     val s = new ResizableSemaphore(32, 35)
     // try to acquire more permits than allowed in parallel
-    val acquires = (0 until 64).par.map(_ => s.tryAcquire()).seq
+    concurrently(64, acquireTimeout)(s.tryAcquire())
+    concurrently(32, acquireTimeout)(s.release(1, true))
 
-    (0 until 32).par.map(_ => s.release(1, true))
     s.counter shouldBe 0
-
   }
 
 }

tests/src/test/scala/org/apache/openwhisk/core/controller/test/ActionsApiTests.scala

@@ -1161,7 +1161,7 @@ class ActionsApiTests extends ControllerTestCommon with WhiskActionsApi {
       action.publish,
       action.annotations ++ systemAnnotations(kind))
 
-    (0 until 5).par.map { i =>
+    (0 until 5).map { i =>
       Get(s"$collectionPath/$name") ~> Route.seal(routes(creds)(transid())) ~> check {
         status should be(OK)
         val response = responseAs[WhiskAction]

tests/src/test/scala/org/apache/openwhisk/core/limits/MaxActionDurationTests.scala

@@ -18,18 +18,12 @@
 package org.apache.openwhisk.core.limits
 
 import java.io.File
-import scala.concurrent.duration.DurationInt
 
+import scala.concurrent.duration.DurationInt
 import org.junit.runner.RunWith
 import org.scalatest.junit.JUnitRunner
-
-import common.TestHelpers
-import common.TestUtils
+import common.{ConcurrencyHelpers, TestHelpers, TestUtils, WskActorSystem, WskProps, WskTestHelpers}
 import common.rest.WskRestOperations
-import common.WskProps
-import common.WskTestHelpers
-import common.WskActorSystem
-
 import org.apache.openwhisk.core.entity._
 import spray.json.DefaultJsonProtocol._
 import spray.json._
@@ -41,7 +35,7 @@ import org.scalatest.tagobjects.Slow
  * Tests for action duration limits. These tests require a deployed backend.
  */
 @RunWith(classOf[JUnitRunner])
-class MaxActionDurationTests extends TestHelpers with WskTestHelpers with WskActorSystem {
+class MaxActionDurationTests extends TestHelpers with WskTestHelpers with WskActorSystem with ConcurrencyHelpers {
 
   implicit val wskprops = WskProps()
   val wsk = new WskRestOperations
@@ -65,41 +59,41 @@ class MaxActionDurationTests extends TestHelpers with WskTestHelpers with WskAct
   "node-, python, and java-action" should s"run up to the max allowed duration (${TimeLimit.MAX_DURATION})" taggedAs (Slow) in withAssetCleaner(
     wskprops) { (wp, assetHelper) =>
     // When you add more runtimes, keep in mind, how many actions can be processed in parallel by the Invokers!
-    Map("node" -> "helloDeadline.js", "python" -> "sleep.py", "java" -> "sleep.jar")
+    val runtimes = Map("node" -> "helloDeadline.js", "python" -> "sleep.py", "java" -> "sleep.jar")
       .filter {
         case (_, name) =>
           new File(TestUtils.getTestActionFilename(name)).exists()
       }
-      .par
-      .map {
-        case (k, name) =>
-          println(s"Testing action kind '${k}' with action '${name}'")
-          assetHelper.withCleaner(wsk.action, name) { (action, _) =>
-            val main = if (k == "java") Some("Sleep") else None
-            action.create(
-              name,
-              Some(TestUtils.getTestActionFilename(name)),
-              timeout = Some(TimeLimit.MAX_DURATION),
-              main = main)
-          }
 
-          val run = wsk.action.invoke(
+    concurrently(runtimes.toSeq, TimeLimit.MAX_DURATION + 2.minutes) {
+      case (k, name) =>
+        println(s"Testing action kind '${k}' with action '${name}'")
+        assetHelper.withCleaner(wsk.action, name) { (action, _) =>
+          val main = if (k == "java") Some("Sleep") else None
+          action.create(
             name,
-            Map("forceHang" -> true.toJson, "sleepTimeInMs" -> (TimeLimit.MAX_DURATION + 30.seconds).toMillis.toJson))
-          withActivation(
-            wsk.activation,
-            run,
-            initialWait = 1.minute,
-            pollPeriod = 1.minute,
-            totalWait = TimeLimit.MAX_DURATION + 2.minutes) { activation =>
-            withClue("Activation result not as expected:") {
-              activation.response.status shouldBe ActivationResponse.messageForCode(ActivationResponse.DeveloperError)
-              activation.response.result shouldBe Some(
-                JsObject("error" -> Messages.timedoutActivation(TimeLimit.MAX_DURATION, init = false).toJson))
-              activation.duration.toInt should be >= TimeLimit.MAX_DURATION.toMillis.toInt
-            }
+            Some(TestUtils.getTestActionFilename(name)),
+            timeout = Some(TimeLimit.MAX_DURATION),
+            main = main)
+        }
+
+        val run = wsk.action.invoke(
+          name,
+          Map("forceHang" -> true.toJson, "sleepTimeInMs" -> (TimeLimit.MAX_DURATION + 30.seconds).toMillis.toJson))
+
+        withActivation(
+          wsk.activation,
+          run,
+          initialWait = 1.minute,
+          pollPeriod = 1.minute,
+          totalWait = TimeLimit.MAX_DURATION + 2.minutes) { activation =>
+          withClue("Activation result not as expected:") {
+            activation.response.status shouldBe ActivationResponse.messageForCode(ActivationResponse.DeveloperError)
+            activation.response.result shouldBe Some(
+              JsObject("error" -> Messages.timedoutActivation(TimeLimit.MAX_DURATION, init = false).toJson))
+            activation.duration.toInt should be >= TimeLimit.MAX_DURATION.toMillis.toInt
           }
-          () // explicitly map to Unit
-      }
+        }
+    }
   }
 }

tests/src/test/scala/org/apache/openwhisk/core/loadBalancer/test/ShardingContainerPoolBalancerTests.scala

@@ -497,7 +497,7 @@ class ShardingContainerPoolBalancerTests
     val stepSize = stepSizes(hash % stepSizes.size)
     val uuid = UUID()
     //initiate activation
-    val published = (0 until numActivations).par.map { _ =>
+    val published = (0 until numActivations).map { _ =>
       val aid = ActivationId.generate()
       val msg = ActivationMessage(
         TransactionId.testing,
@@ -545,12 +545,12 @@ class ShardingContainerPoolBalancerTests
     }
 
     //complete all
-    val acks = ids.par.map { aid =>
+    val acks = ids.map { aid =>
       val invoker = balancer.activationSlots(aid).invokerName
       completeActivation(invoker, balancer, aid)
     }
 
-    Await.ready(Future.sequence(acks.toList), 10.seconds)
+    Await.ready(Future.sequence(acks), 10.seconds)
 
     //verify invokers go back to unused state
     invokers.foreach { i =>