[openwhisk] branch master updated: Replace remaining usages of parallel collections with explicit concurrency. (#4843)

[markusthoemmes]

This is an automated email from the ASF dual-hosted git repository.

markusthoemmes pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/openwhisk.git


The following commit(s) were added to refs/heads/master by this push:
     new 08efb58  Replace remaining usages of parallel collections with 
explicit concurrency. (#4843)
08efb58 is described below

commit 08efb58cf5eef3475bdb41e45db448250bcf9c17
Author: Markus Thömmes <markusthoem...@me.com>
AuthorDate: Thu Feb 27 16:52:11 2020 +0100

    Replace remaining usages of parallel collections with explicit concurrency. 
(#4843)
---
 .../core/database/test/CacheConcurrencyTests.scala | 43 ++++++++++++----------
 1 file changed, 23 insertions(+), 20 deletions(-)

diff --git 
a/tests/src/test/scala/org/apache/openwhisk/core/database/test/CacheConcurrencyTests.scala
 
b/tests/src/test/scala/org/apache/openwhisk/core/database/test/CacheConcurrencyTests.scala
index 90c8bb7..0026df8 100644
--- 
a/tests/src/test/scala/org/apache/openwhisk/core/database/test/CacheConcurrencyTests.scala
+++ 
b/tests/src/test/scala/org/apache/openwhisk/core/database/test/CacheConcurrencyTests.scala
@@ -17,9 +17,9 @@
 
 package org.apache.openwhisk.core.database.test
 
-import scala.collection.parallel._
 import scala.concurrent.duration.DurationInt
-import java.util.concurrent.ForkJoinPool
+import java.util.concurrent.Executors
+
 import org.junit.runner.RunWith
 import org.scalatest.BeforeAndAfterEach
 import org.scalatest.FlatSpec
@@ -32,9 +32,17 @@ import spray.json.JsString
 import org.apache.openwhisk.common.TransactionId
 import org.apache.openwhisk.utils.retry
 
-@RunWith(classOf[JUnitRunner])
-class CacheConcurrencyTests extends FlatSpec with WskTestHelpers with 
WskActorSystem with BeforeAndAfterEach {
+import scala.concurrent.ExecutionContext
 
+@RunWith(classOf[JUnitRunner])
+class CacheConcurrencyTests
+    extends FlatSpec
+    with WskTestHelpers
+    with WskActorSystem
+    with BeforeAndAfterEach
+    with ConcurrencyHelpers {
+
+  val timeout = 5.minutes
   println(s"Running tests on # proc: 
${Runtime.getRuntime.availableProcessors()}")
 
   implicit private val transId = TransactionId.testing
@@ -43,15 +51,14 @@ class CacheConcurrencyTests extends FlatSpec with 
WskTestHelpers with WskActorSy
 
   val nExternalIters = 1
   val nInternalIters = 5
-  val nThreads = nInternalIters * 30
+  val nThreads = nInternalIters * 30 // The maximum number of tasks running in 
parallel at any given time
+  val parallelismExecutionContext = 
ExecutionContext.fromExecutor(Executors.newFixedThreadPool(nThreads))
 
-  val parallel = (1 to nInternalIters).par
-  parallel.tasksupport = new ForkJoinTaskSupport(new ForkJoinPool(nThreads))
-
-  def run[W](phase: String)(block: String => W) = parallel.map { i =>
-    val name = s"testy${i}"
-    withClue(s"$phase: failed for $name") { (name, block(name)) }
-  }
+  def run[W](phase: String)(block: String => W) =
+    concurrently((1 to nInternalIters), timeout) { i =>
+      val name = s"testy${i}"
+      withClue(s"$phase: failed for $name") { (name, block(name)) }
+    }(parallelismExecutionContext)
 
   override def beforeEach() = {
     run("pre-test sanitize") { name =>
@@ -79,9 +86,7 @@ class CacheConcurrencyTests extends FlatSpec with 
WskTestHelpers with WskActorSy
 
       run("delete+get") { name =>
         // run 30 operations in parallel: 15 get, 1 delete, 14 more get
-        val para = (1 to 30).par
-        para.tasksupport = new ForkJoinTaskSupport(new ForkJoinPool(nThreads))
-        para.map { i =>
+        concurrently((1 to 30), timeout) { i =>
           if (i != 16) {
             val rr = wsk.action.get(name, expectedExitCode = DONTCARE_EXIT)
             withClue(s"expecting get to either succeed or fail with not found: 
$rr") {
@@ -91,7 +96,7 @@ class CacheConcurrencyTests extends FlatSpec with 
WskTestHelpers with WskActorSy
           } else {
             wsk.action.delete(name)
           }
-        }
+        }(parallelismExecutionContext)
       }
 
       // Give some time to replicate the state between the controllers
@@ -117,9 +122,7 @@ class CacheConcurrencyTests extends FlatSpec with 
WskTestHelpers with WskActorSy
 
       run("update+get") { name =>
         // run 30 operations in parallel: 15 get, 1 update, 14 more get
-        val para = (1 to 30).par
-        para.tasksupport = new ForkJoinTaskSupport(new ForkJoinPool(nThreads))
-        para.map { i =>
+        concurrently((1 to 30), timeout) { i =>
           if (i != 16) {
             val rr = wsk.action.get(name, expectedExitCode = DONTCARE_EXIT)
             withClue(s"expecting get to either succeed or fail with not found: 
$rr") {
@@ -129,7 +132,7 @@ class CacheConcurrencyTests extends FlatSpec with 
WskTestHelpers with WskActorSy
           } else {
             wsk.action.create(name, None, parameters = Map("color" -> 
JsString("blue")), update = true)
           }
-        }
+        }(parallelismExecutionContext)
       }
 
       // All controllers should have the correct action