Couple more examples for "Shared mutable state and concurrency" section

Author: elizarov

coroutines-guide.md

@@ -78,8 +78,10 @@ This is a short guide on core features of `kotlinx.coroutines` with a series of
   * [Buffered channels](#buffered-channels)
 * [Shared mutable state and concurrency](#shared-mutable-state-and-concurrency)
   * [The problem](#the-problem)
+  * [Volatiles are of no help](#volatiles-are-of-no-help)
   * [Thread-safe data structures](#thread-safe-data-structures)
-  * [Thread confinement](#thread-confinement)
+  * [Thread confinement fine-grained](#thread-confinement-fine-grained)
+  * [Thread confinement coarse-grained](#thread-confinement-coarse-grained)
   * [Mutual exclusion](#mutual-exclusion)
   * [Actors](#actors)
 * [Select expression](#select-expression)
@@ -1484,49 +1486,52 @@ but others are unique.
 
 ### The problem
 
-Let us launch 100k coroutines all doing the same action. We'll also measure their completion time for 
-further comparisons:
+Let us launch a thousand coroutines all doing the same action thousand times (for a total of a million executions). 
+We'll also measure their completion time for further comparisons:
 
 <!--- INCLUDE .*/example-sync-([0-9]+).kt
+import kotlin.coroutines.experimental.CoroutineContext
 import kotlin.system.measureTimeMillis
 -->
 
-<!--- INCLUDE .*/example-sync-02.kt
+<!--- INCLUDE .*/example-sync-03.kt
 import java.util.concurrent.atomic.AtomicInteger
 -->
 
-<!--- INCLUDE .*/example-sync-04.kt
+<!--- INCLUDE .*/example-sync-06.kt
 import kotlinx.coroutines.experimental.sync.Mutex
 -->
 
-<!--- INCLUDE .*/example-sync-05.kt
+<!--- INCLUDE .*/example-sync-07.kt
 import kotlinx.coroutines.experimental.channels.*
 -->
 
 ```kotlin
-suspend fun massiveRun(action: suspend () -> Unit) {
-    val n = 100_000
+suspend fun massiveRun(context: CoroutineContext, action: suspend () -> Unit) {
+    val n = 1000 // number of coroutines to launch
+    val k = 1000 // times an action is repeated by each coroutine
     val time = measureTimeMillis {
         val jobs = List(n) {
-            launch(CommonPool) {
-                action()
+            launch(context) {
+                repeat(k) { action() }
             }
         }
         jobs.forEach { it.join() }
     }
-    println("Completed in $time ms")    
+    println("Completed ${n * k} actions in $time ms")    
 }
 ```
 
 <!--- INCLUDE .*/example-sync-([0-9]+).kt -->
 
-We start with a very simple action that increments a shared mutable variable. 
+We start with a very simple action that increments a shared mutable variable using 
+multi-threaded [CommonPool] context. 
 
 ```kotlin
 var counter = 0
 
 fun main(args: Array<String>) = runBlocking<Unit> {
-    massiveRun {
+    massiveRun(CommonPool) {
         counter++
     }
     println("Counter = $counter")
@@ -1535,40 +1540,73 @@ fun main(args: Array<String>) = runBlocking<Unit> {
 
 > You can get full code [here](kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-01.kt)
 
-<!--- TEST lines.size == 2 && lines[1].startsWith("Counter = ") -->
+<!--- TEST LINES_START
+Completed 1000000 actions in
+Counter =
+-->
+
+What does it print at the end? It is highly unlikely to ever print "Counter = 1000000", because a thousand coroutines 
+increment the `counter` concurrently from multiple threads without any synchronization.
+
+### Volatiles are of no help
 
-What does it print at the end? It is highly unlikely to ever print "100000", because all the 
-100k coroutines increment the `counter` concurrently without any synchronization.
+There is common misconception that making a variable `volatile` solves concurrency problem. Let us try it:
+
+```kotlin
+@Volatile // in Kotlin `volatile` is an annotation 
+var counter = 0
+
+fun main(args: Array<String>) = runBlocking<Unit> {
+    massiveRun(CommonPool) {
+        counter++
+    }
+    println("Counter = $counter")
+}
+```
+
+> You can get full code [here](kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-02.kt)
+
+<!--- TEST LINES_START
+Completed 1000000 actions in
+Counter =
+-->
+
+This code works slower, but we still don't get "Counter = 1000000" at the end, because volatile variables guarantee
+linearizable (this is a technical term for "atomic") reads and writes to the corresponding variable, but
+do not provide atomicity of larger actions (increment in our case).
 
 ### Thread-safe data structures
 
 The general solution that works both for threads and for coroutines is to use a thread-safe (aka synchronized,
 linearizable, or atomic) data structure that provides all the necessarily synchronization for the corresponding 
 operations that needs to be performed on a shared state. 
-In the case of a simple counter we can use `AtomicInteger` class:
+In the case of a simple counter we can use `AtomicInteger` class which has atomic `incrementAndGet` operations:
 
 ```kotlin
 var counter = AtomicInteger()
 
 fun main(args: Array<String>) = runBlocking<Unit> {
-    massiveRun {
+    massiveRun(CommonPool) {
         counter.incrementAndGet()
     }
     println("Counter = ${counter.get()}")
 }
 ```
 
-> You can get full code [here](kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-02.kt)
+> You can get full code [here](kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-03.kt)
 
-<!--- TEST lines.size == 2 && lines[1] == "Counter = 100000" -->
+<!--- TEST ARBITRARY_TIME
+Completed 1000000 actions in xxx ms
+Counter = 1000000
+-->
 
 This is the fastest solution for this particular problem. It works for plain counters, collections, queues and other
 standard data structures and basic operations on them. However, it does not easily scale to complex
 state or to complex operations that do not have ready-to-use thread-safe implementations. 
 
-### Thread confinement
+### Thread confinement fine-grained
 
-Thread confinement is an approach to the problem of shared mutable state where all access to the particular shared
+_Thread confinement_ is an approach to the problem of shared mutable state where all access to the particular shared
 state is confined to a single thread. It is typically used in UI applications, where all UI state is confined to 
 the single event-dispatch/application thread. It is easy to apply with coroutines by using a  
 single-threaded context:
@@ -1578,18 +1616,51 @@ val counterContext = newSingleThreadContext("CounterContext")
 var counter = 0
 
 fun main(args: Array<String>) = runBlocking<Unit> {
-    massiveRun {
-        run(counterContext) {
+    massiveRun(CommonPool) { // run each coroutine in CommonPool
+        run(counterContext) { // but confine each increment to the single-threaded context
             counter++
         }
     }
     println("Counter = $counter")
 }
 ```
 
-> You can get full code [here](kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-03.kt)
+> You can get full code [here](kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-04.kt)
+
+<!--- TEST ARBITRARY_TIME
+Completed 1000000 actions in xxx ms
+Counter = 1000000
+-->
+
+This code works very slowly, because it does _fine-grained_ thread-confinement. Each individual increment switches 
+from multi-threaded `CommonPool` context to the single-threaded context using [run] block. 
+
+### Thread confinement coarse-grained
+
+In practice, thread confinement is performed in large chunks, e.g. big pieces of state-updating business logic
+are confined to the single thread. The following example does it like that, running each coroutine in 
+the single-threaded context to start with.
 
-<!--- TEST lines.size == 2 && lines[1] == "Counter = 100000" -->
+```kotlin
+val counterContext = newSingleThreadContext("CounterContext")
+var counter = 0
+
+fun main(args: Array<String>) = runBlocking<Unit> {
+    massiveRun(counterContext) { // run each coroutine in the single-threaded context
+        counter++
+    }
+    println("Counter = $counter")
+}
+```
+
+> You can get full code [here](kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-05.kt)
+
+<!--- TEST ARBITRARY_TIME
+Completed 1000000 actions in xxx ms
+Counter = 1000000
+-->
+
+This now works much faster and produces correct result.
 
 ### Mutual exclusion
 
@@ -1603,7 +1674,7 @@ val mutex = Mutex()
 var counter = 0
 
 fun main(args: Array<String>) = runBlocking<Unit> {
-    massiveRun {
+    massiveRun(CommonPool) {
         mutex.lock()
         try { counter++ }
         finally { mutex.unlock() }
@@ -1612,9 +1683,16 @@ fun main(args: Array<String>) = runBlocking<Unit> {
 }
 ```
 
-> You can get full code [here](kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-04.kt)
+> You can get full code [here](kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-06.kt)
+
+<!--- TEST ARBITRARY_TIME
+Completed 1000000 actions in xxx ms
+Counter = 1000000
+-->
 
-<!--- TEST lines.size == 2 && lines[1] == "Counter = 100000" -->
+The locking in this example is fine-grained, so it pays the price. However, it is a good choice for some situations
+where you absolutely must modify some shared state periodically, but there is no natural thread that this state
+is confined to.
 
 ### Actors
 
@@ -1643,7 +1721,7 @@ fun counterActor(request: ReceiveChannel<CounterMsg>) = launch(CommonPool) {
 fun main(args: Array<String>) = runBlocking<Unit> {
     val request = Channel<CounterMsg>()
     counterActor(request)
-    massiveRun {
+    massiveRun(CommonPool) {
         request.send(IncCounter)
     }
     val response = Channel<Int>()
@@ -1652,14 +1730,20 @@ fun main(args: Array<String>) = runBlocking<Unit> {
 }
 ```
 
-> You can get full code [here](kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-05.kt)
+> You can get full code [here](kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-07.kt)
 
-<!--- TEST lines.size == 2 && lines[1] == "Counter = 100000" -->
+<!--- TEST ARBITRARY_TIME
+Completed 1000000 actions in xxx ms
+Counter = 1000000
+-->
 
 Notice, that it does not matter (for correctness) what context the actor itself is executed in. An actor is
 a coroutine and a coroutine is executed sequentially, so confinement of the state to the specific coroutine
 works as a solution to the problem of shared mutable state.
 
+Actor is more efficient than locking under load, because in this case it always has work to do and does not 
+have to switch at all.
+
 ## Select expression
 
 Select expression makes it possible to await multiple suspending functions simultaneously and _select_

knit/src/Knit.kt

@@ -41,6 +41,7 @@ const val SECTION_START = "##"
 
 const val PACKAGE_PREFIX = "package "
 const val STARTS_WITH_PREDICATE = "STARTS_WITH"
+const val ARBITRARY_TIME_PREDICATE = "ARBITRARY_TIME"
 const val FLEXIBLE_TIME_PREDICATE = "FLEXIBLE_TIME"
 const val FLEXIBLE_THREAD_PREDICATE = "FLEXIBLE_THREAD"
 const val LINES_START_UNORDERED_PREDICATE = "LINES_START_UNORDERED"
@@ -250,6 +251,7 @@ fun writeTest(testOut: PrintWriter, pgk: String, test: List<String>, predicate:
         when (predicate) {
             "" -> writeTestLines("verifyLines", test)
             STARTS_WITH_PREDICATE -> writeTestLines("verifyLinesStartWith", test)
+            ARBITRARY_TIME_PREDICATE -> writeTestLines("verifyLinesArbitraryTime", test)
             FLEXIBLE_TIME_PREDICATE -> writeTestLines("verifyLinesFlexibleTime", test)
             FLEXIBLE_THREAD_PREDICATE -> writeTestLines("verifyLinesFlexibleThread", test)
             LINES_START_UNORDERED_PREDICATE -> writeTestLines("verifyLinesStartUnordered", test)

kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-01.kt

@@ -18,25 +18,27 @@
 package guide.sync.example01
 
 import kotlinx.coroutines.experimental.*
+import kotlin.coroutines.experimental.CoroutineContext
 import kotlin.system.measureTimeMillis
 
-suspend fun massiveRun(action: suspend () -> Unit) {
-    val n = 100_000
+suspend fun massiveRun(context: CoroutineContext, action: suspend () -> Unit) {
+    val n = 1000 // number of coroutines to launch
+    val k = 1000 // times an action is repeated by each coroutine
     val time = measureTimeMillis {
         val jobs = List(n) {
-            launch(CommonPool) {
-                action()
+            launch(context) {
+                repeat(k) { action() }
             }
         }
         jobs.forEach { it.join() }
     }
-    println("Completed in $time ms")    
+    println("Completed ${n * k} actions in $time ms")    
 }
 
 var counter = 0
 
 fun main(args: Array<String>) = runBlocking<Unit> {
-    massiveRun {
+    massiveRun(CommonPool) {
         counter++
     }
     println("Counter = $counter")

kotlinx-coroutines-core/src/test/kotlin/guide/example-sync-02.kt

@@ -18,27 +18,29 @@
 package guide.sync.example02
 
 import kotlinx.coroutines.experimental.*
+import kotlin.coroutines.experimental.CoroutineContext
 import kotlin.system.measureTimeMillis
-import java.util.concurrent.atomic.AtomicInteger
 
-suspend fun massiveRun(action: suspend () -> Unit) {
-    val n = 100_000
+suspend fun massiveRun(context: CoroutineContext, action: suspend () -> Unit) {
+    val n = 1000 // number of coroutines to launch
+    val k = 1000 // times an action is repeated by each coroutine
     val time = measureTimeMillis {
         val jobs = List(n) {
-            launch(CommonPool) {
-                action()
+            launch(context) {
+                repeat(k) { action() }
             }
         }
         jobs.forEach { it.join() }
     }
-    println("Completed in $time ms")    
+    println("Completed ${n * k} actions in $time ms")    
 }
 
-var counter = AtomicInteger()
+@Volatile // in Kotlin `volatile` is an annotation 
+var counter = 0
 
 fun main(args: Array<String>) = runBlocking<Unit> {
-    massiveRun {
-        counter.incrementAndGet()
+    massiveRun(CommonPool) {
+        counter++
     }
-    println("Counter = ${counter.get()}")
+    println("Counter = $counter")
 }