Implemented "selectUnbiased"

Author: elizarov

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/Select.kt

@@ -118,6 +118,10 @@ public interface SelectInstance<in R> {
  * becomes the result of the select. If any clause _fails_, then the select invocation produces the
  * corresponding exception. No clause is selected in this case.
  *
+ * This select function is _biased_ to the first clause. When multiple clauses can be selected at the same time,
+ * the first one of them gets priority. Use [selectUnbiased] for an unbiased (randomized) selection among
+ * the clauses.
+
  * There is no `default` clause for select expression. Instead, each selectable suspending function has the
  * corresponding non-suspending version that can be used with a regular `when` expression to select one
  * of the alternatives or to perform default (`else`) action if none of them can be immediately selected.

kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/selects/SelectUnbiased.kt

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed 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 kotlinx.coroutines.experimental.selects
+
+import kotlinx.coroutines.experimental.Deferred
+import kotlinx.coroutines.experimental.Job
+import kotlinx.coroutines.experimental.channels.ReceiveChannel
+import kotlinx.coroutines.experimental.channels.SendChannel
+import java.util.*
+import kotlin.coroutines.experimental.Continuation
+import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
+
+/**
+ * Waits for the result of multiple suspending functions simultaneously like [select], but in an _unbiased_
+ * way when multiple clauses are selectable at the same time.
+ *
+ * This unbiased implementation of `select` expression randomly shuffles the clauses before checking
+ * if they are selectable, thus ensuring that there is no statistical bias to the selection of the first
+ * clauses.
+ *
+ * See [select] function description for all the other details.
+ */
+public inline suspend fun <R> selectUnbiased(crossinline builder: SelectBuilder<R>.() -> Unit): R =
+    suspendCoroutineOrReturn { cont ->
+        val scope = UnbiasedSelectBuilderImpl(cont)
+        try {
+            builder(scope)
+        } catch (e: Throwable) {
+            scope.handleBuilderException(e)
+        }
+        scope.initSelectResult()
+    }
+
+
+@PublishedApi
+internal class UnbiasedSelectBuilderImpl<in R>(cont: Continuation<R>) : SelectBuilder<R> {
+    val instance = SelectBuilderImpl(cont)
+    val clauses = arrayListOf<() -> Unit>()
+
+    @PublishedApi
+    internal fun handleBuilderException(e: Throwable) = instance.handleBuilderException(e)
+
+    @PublishedApi
+    internal fun initSelectResult(): Any? {
+        if (!instance.isSelected) {
+            try {
+                Collections.shuffle(clauses)
+                clauses.forEach { it.invoke() }
+            } catch (e: Throwable) {
+                instance.handleBuilderException(e)
+            }
+        }
+        return instance.initSelectResult()
+    }
+
+    override fun Job.onJoin(block: suspend () -> R) {
+        clauses += { registerSelectJoin(instance, block) }
+    }
+
+    override fun <T> Deferred<T>.onAwait(block: suspend (T) -> R) {
+        clauses += { registerSelectAwait(instance, block) }
+    }
+
+    override fun <E> SendChannel<E>.onSend(element: E, block: suspend () -> R) {
+        clauses += { registerSelectSend(instance, element, block) }
+    }
+
+    override fun <E> ReceiveChannel<E>.onReceive(block: suspend (E) -> R) {
+        clauses += { registerSelectReceive(instance, block) }
+    }
+
+    override fun <E> ReceiveChannel<E>.onReceiveOrNull(block: suspend (E?) -> R) {
+        clauses += { registerSelectReceiveOrNull(instance, block) }
+    }
+}

kotlinx-coroutines-core/src/test/kotlin/kotlinx/coroutines/experimental/selects/SelectBiasTest.kt

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * Licensed 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 kotlinx.coroutines.experimental.selects
+
+import kotlinx.coroutines.experimental.async
+import kotlinx.coroutines.experimental.runBlocking
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+class SelectBiasTest {
+    val n = 10_000
+
+    @Test
+    fun testBiased() = runBlocking<Unit> {
+        val d0 = async(context) { 0 }
+        val d1 = async(context) { 1 }
+        val counter = IntArray(2)
+        repeat(n) {
+            val selected = select<Int> {
+                d0.onAwait { 0 }
+                d1.onAwait { 1 }
+            }
+            counter[selected]++
+        }
+        assertEquals(n, counter[0])
+        assertEquals(0, counter[1])
+    }
+
+    @Test
+    fun testUnbiased() = runBlocking<Unit> {
+        val d0 = async(context) { 0 }
+        val d1 = async(context) { 1 }
+        val counter = IntArray(2)
+        repeat(n) {
+            val selected = selectUnbiased<Int> {
+                d0.onAwait { 0 }
+                d1.onAwait { 1 }
+            }
+            counter[selected]++
+        }
+        assertTrue(counter[0] >= n / 4)
+        assertTrue(counter[1] >= n / 4)
+    }
+}