Added parallelReduce

Author: cvb941

transformations/src/main/kotlin/ParallelMap.kt

@@ -0,0 +1,50 @@
+package com.lukaskusik.coroutines.transformations
+
+import kotlinx.coroutines.async
+import kotlinx.coroutines.coroutineScope
+
+/**
+ *  Performs map transformation on the iterable using coroutines.
+ */
+suspend fun <T, R> Iterable<T>.mapParallel(
+    transform: (T) -> R
+): List<R> = coroutineScope {
+    map { async { transform(it) } }.map { it.await() }
+}
+
+/**
+ *  Performs map transformation on the iterable using coroutines.
+ *  The chunkSize parameter is used to run multiple transformations on a single coroutine.
+ *
+ *  @param chunkSize Size of each sub-collection that will be reduced in each coroutine.
+ */
+suspend fun <T, R> Iterable<T>.mapParallelChunked(
+    chunkSize: Int,
+    transform: (T) -> R
+): List<R> = coroutineScope {
+    chunked(chunkSize).map { subChunk ->
+        async {
+            subChunk.map(transform)
+        }
+    }.flatMap {
+        it.await()
+    }
+}
+
+/**
+ *  Performs map transformation on the iterable using coroutines.
+ *
+ *  It can split the collection into multiple chunks using the chunksCount parameter.
+ *  Each chunk will then run on a single coroutine, minimizing thread management, etc.
+ *  The default and recommended chunksCount for multithreading is the number of CPU threads, e.g. 4 or 8.
+ *
+ *  @param chunksCount How many chunks should the collection be split into. Defaults to the number of available processors.
+ *
+ */
+suspend fun <T, E> Collection<T>.mapParallelChunked(
+    chunksCount: Int = Runtime.getRuntime().availableProcessors(),
+    transform: (T) -> E
+): List<E> {
+    val chunkSize = Math.ceil(size / chunksCount.toDouble()).toInt()
+    return asIterable().mapParallelChunked(chunkSize, transform)
+}

transformations/src/main/kotlin/com/lukaskusik/coroutines/transformations/ParallelMap.kt

@@ -0,0 +1,37 @@
+package com.lukaskusik.coroutines.transformations
+
+import kotlinx.coroutines.async
+import kotlinx.coroutines.coroutineScope
+
+
+/**
+ *  The reduce operation must be associative, since the reduce will be most likely out of order.
+ *  This method splits the original collection into chunks and reduces each of the chunks separately on their own coroutines and then reduces the result again.
+ *
+ *  @param chunkSize Size of each sub-collection that will be reduced in each coroutine.
+ */
+suspend fun <T> Iterable<T>.reduceParallel(
+    chunkSize: Int,
+    operation: (T, T) -> T
+): T = coroutineScope {
+    chunked(chunkSize).map { subChunk ->
+        async {
+            subChunk.reduce(operation)
+        }
+    }.map { it.await() }.reduce(operation)
+}
+
+/**
+ *  The operation must be associative, since the reduce will be most likely out of order.
+ *  This method splits the original collection into chunks and reduces each of the chunks separately on their own coroutines and then reduces the result again.
+ *  The recommended chunksCount for multithreading is the number of CPU threads, e.g. 4 or 8.
+ *
+ *  @param chunksCount How many chunks should the collection be split into. Defaults to the number of available processors.
+ */
+suspend fun <T> Collection<T>.reduceParallel(
+    chunksCount: Int = Runtime.getRuntime().availableProcessors(),
+    operation: (T, T) -> T
+): T {
+    val chunkSize = Math.ceil(size / chunksCount.toDouble()).toInt()
+    return asIterable().reduceParallel(chunkSize, operation)
+}

transformations/src/main/kotlin/com/lukaskusik/coroutines/transformations/ParallelReduce.kt

@@ -1,15 +1,19 @@
-package com.lukaskusik.coroutines.transformations.test
+package com.lukaskusik.coroutines.transformations.benchmark
 
 import com.carrotsearch.junitbenchmarks.AbstractBenchmark
+import com.carrotsearch.junitbenchmarks.annotation.AxisRange
+import com.carrotsearch.junitbenchmarks.annotation.BenchmarkMethodChart
 import com.lukaskusik.coroutines.transformations.mapParallel
+import com.lukaskusik.coroutines.transformations.mapParallelChunked
+import com.lukaskusik.coroutines.transformations.test.ParallelMapTest
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.runBlocking
 import org.junit.Test
 
 class ParallelMapBenchmark : AbstractBenchmark() {
 
     companion object {
-        const val LIST_SIZE = 100
+        const val LIST_SIZE = 1000
     }
 
     private val list = ParallelMapTest.getRandomListOfSize(LIST_SIZE)
@@ -34,11 +38,19 @@ class ParallelMapBenchmark : AbstractBenchmark() {
         }
     }
 
-//    @Test
-//    fun parallelJustThreads() {
-//        runBlocking(Dispatchers.Default) {
-//            list.mapParallel(LIST_SIZE, sumClosure)
-//        }
-//    }
+    @Test
+    fun coroutineOnThreadPoolChunked4() {
+        runBlocking(Dispatchers.Default) {
+            list.mapParallelChunked(4) { Thread.sleep(1); it / 2 }
+        }
+    }
+
+    @Test
+    fun coroutineOnThreadPoolChunked8() {
+        runBlocking(Dispatchers.Default) {
+            list.mapParallelChunked(8) { Thread.sleep(1); it / 2 }
+        }
+    }
+
 
 }

transformations/src/test/kotlin/ParallelMapBenchmark.kt renamed to transformations/src/test/kotlin/com/lukaskusik/coroutines/transformations/benchmark/ParallelMapBenchmark.kt

@@ -0,0 +1,54 @@
+package com.lukaskusik.coroutines.transformations.benchmark
+
+import com.carrotsearch.junitbenchmarks.AbstractBenchmark
+import com.lukaskusik.coroutines.transformations.reduceParallel
+import com.lukaskusik.coroutines.transformations.test.ParallelMapTest
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.runBlocking
+import org.junit.Test
+
+class ParallelReduceBenchmark : AbstractBenchmark() {
+
+    companion object {
+        const val LIST_SIZE = 1000
+    }
+
+    private val list = ParallelMapTest.getRandomListOfSize(LIST_SIZE)
+
+    private val operation = { acc: Int, i: Int -> Thread.sleep(1); acc + i }
+
+    @Test
+    fun sequential() {
+        list.reduce(operation)
+    }
+
+    @Test
+    fun coroutineOnMain() {
+        runBlocking {
+            list.reduceParallel(1, operation)
+        }
+    }
+
+    @Test
+    fun coroutineOnThreadPool() {
+        runBlocking(Dispatchers.Default) {
+            list.reduceParallel(1, operation)
+        }
+    }
+
+    @Test
+    fun coroutineOnThreadPoolChunked4() {
+        runBlocking(Dispatchers.Default) {
+            list.reduceParallel(4, operation)
+        }
+    }
+
+    @Test
+    fun coroutineOnThreadPoolChunked8() {
+        runBlocking(Dispatchers.Default) {
+            list.reduceParallel(8, operation)
+        }
+    }
+
+
+}

transformations/src/test/kotlin/com/lukaskusik/coroutines/transformations/benchmark/ParallelReduceBenchmark.kt

@@ -1,6 +1,7 @@
 package com.lukaskusik.coroutines.transformations.test
 
 import com.lukaskusik.coroutines.transformations.mapParallel
+import com.lukaskusik.coroutines.transformations.mapParallelChunked
 import kotlinx.coroutines.runBlocking
 import org.junit.Assert
 import org.junit.Test
@@ -33,5 +34,17 @@ class ParallelMapTest {
         Assert.assertEquals(listSequential, listParallel)
     }
 
+    @Test
+    fun parallelMap4Chunks() {
+        var listSequential = listOf(1, 3, 3, 4, 5)
+        var listParallel = listSequential.toList()
 
+        listSequential = listSequential.map { it * 2 }
+        runBlocking {
+            listParallel = listParallel.mapParallelChunked(4) { it * 2 }
+        }
+
+
+        Assert.assertEquals(listSequential, listParallel)
+    }
 }

transformations/src/test/kotlin/ParallelMapTest.kt renamed to transformations/src/test/kotlin/com/lukaskusik/coroutines/transformations/test/ParallelMapTest.kt

@@ -0,0 +1,58 @@
+package com.lukaskusik.coroutines.transformations.test
+
+import com.lukaskusik.coroutines.transformations.reduceParallel
+import kotlinx.coroutines.runBlocking
+import org.junit.Assert
+import org.junit.Test
+import kotlin.random.Random
+
+class ParallelReduceTest {
+    companion object {
+        fun getList() = listOf(1, 2, 3, 4, 5)
+
+        fun getRandomListOfSize(listSize: Int): List<Int> {
+            val random = Random(648)
+            val list = ArrayList<Int>(listSize)
+            repeat(listSize) {
+                list.add(random.nextInt())
+            }
+            return list
+        }
+    }
+
+    private fun theTest(chunks: Int) {
+        val listSequential = getList()
+        val listParallel = listSequential.toList()
+        val operation = { acc: Int, i: Int -> acc + i }
+
+        val sequentialResult = listSequential.reduce(operation)
+        var parallelResult: Int? = null
+        runBlocking {
+            parallelResult =
+                listParallel.reduceParallel(chunks, operation)
+        }
+
+        Assert.assertEquals(sequentialResult, parallelResult)
+    }
+
+    @Test
+    fun parallelReduceNoChunks() {
+        theTest(1)
+    }
+
+    @Test
+    fun parallelReduce4Chunks() {
+        theTest(4)
+    }
+
+
+    @Test(expected = IllegalArgumentException::class)
+    fun parallelReduce0ChunksError() {
+        theTest(0)
+    }
+
+   @Test(expected = IllegalArgumentException::class)
+    fun parallelReduceNegativeChunksError() {
+        theTest(-10)
+    }
+}