Replace all volatiles with atomics in common code

In preparation to native multithreading.

Author: elizarov

kotlinx-coroutines-core/common/src/Await.kt

@@ -1,12 +1,11 @@
 /*
- * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines
 
 import kotlinx.atomicfu.*
 import kotlin.coroutines.*
-import kotlin.jvm.*
 
 /**
  * Awaits for completion of given deferred values without blocking a thread and resumes normally with the list of values
@@ -62,7 +61,7 @@ private class AwaitAll<T>(private val deferreds: Array<out Deferred<T>>) {
     suspend fun await(): List<T> = suspendCancellableCoroutine { cont ->
         // Intricate dance here
         // Step 1: Create nodes and install them as completion handlers, they may fire!
-        val nodes = Array<AwaitAllNode>(deferreds.size) { i ->
+        val nodes = Array(deferreds.size) { i ->
             val deferred = deferreds[i]
             deferred.start() // To properly await lazily started deferreds
             AwaitAllNode(cont, deferred).apply {
@@ -72,7 +71,7 @@ private class AwaitAll<T>(private val deferreds: Array<out Deferred<T>>) {
         val disposer = DisposeHandlersOnCancel(nodes)
         // Step 2: Set disposer to each node
         nodes.forEach { it.disposer = disposer }
-        // Here we know that if any code the nodes complete, it will dipsose the rest
+        // Here we know that if any code the nodes complete, it will dispose the rest
         // Step 3: Now we can check if continuation is complete
         if (cont.isCompleted) {
             // it is already complete while handlers were being installed -- dispose them all
@@ -94,18 +93,19 @@ private class AwaitAll<T>(private val deferreds: Array<out Deferred<T>>) {
     private inner class AwaitAllNode(private val continuation: CancellableContinuation<List<T>>, job: Job) : JobNode<Job>(job) {
         lateinit var handle: DisposableHandle
 
-        @Volatile
-        var disposer: DisposeHandlersOnCancel? = null
+        private val _disposer = atomic<DisposeHandlersOnCancel?>(null)
+        var disposer: DisposeHandlersOnCancel?
+            get() = _disposer.value
+            set(value) { _disposer.value = value }
 
         override fun invoke(cause: Throwable?) {
             if (cause != null) {
                 val token = continuation.tryResumeWithException(cause)
                 if (token != null) {
                     continuation.completeResume(token)
                     // volatile read of disposer AFTER continuation is complete
-                    val disposer = this.disposer
                     // and if disposer was already set (all handlers where already installed, then dispose them all)
-                    if (disposer != null) disposer.disposeAll()
+                    disposer?.disposeAll()
                 }
             } else if (notCompletedCount.decrementAndGet() == 0) {
                 continuation.resume(deferreds.map { it.getCompleted() })

kotlinx-coroutines-core/common/src/CancellableContinuationImpl.kt

@@ -63,8 +63,10 @@ internal open class CancellableContinuationImpl<in T>(
      */
     private val _state = atomic<Any?>(Active)
 
-    @Volatile
-    private var parentHandle: DisposableHandle? = null
+    private val _parentHandle = atomic<DisposableHandle?>(null)
+    private var parentHandle: DisposableHandle?
+        get() = _parentHandle.value
+        set(value) { _parentHandle.value = value }
 
     internal val state: Any? get() = _state.value
 

kotlinx-coroutines-core/common/src/EventLoop.common.kt

@@ -182,15 +182,16 @@ internal abstract class EventLoopImplBase: EventLoopImplPlatform(), Delay {
     // Allocated only only once
     private val _delayed = atomic<DelayedTaskQueue?>(null)
 
-    @Volatile
-    private var isCompleted = false
+    private val _isCompleted = atomic(false)
+    private var isCompleted
+        get() = _isCompleted.value
+        set(value) { _isCompleted.value = value }
 
     override val isEmpty: Boolean get() {
         if (!isUnconfinedQueueEmpty) return false
         val delayed = _delayed.value
         if (delayed != null && !delayed.isEmpty) return false
-        val queue = _queue.value
-        return when (queue) {
+        return when (val queue = _queue.value) {
             null -> true
             is Queue<*> -> queue.isEmpty
             else -> queue === CLOSED_EMPTY

kotlinx-coroutines-core/common/src/JobSupport.kt

@@ -127,9 +127,10 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     // Note: use shared objects while we have no listeners
     private val _state = atomic<Any?>(if (active) EMPTY_ACTIVE else EMPTY_NEW)
 
-    @Volatile
-    @JvmField
-    internal var parentHandle: ChildHandle? = null
+    private val _parentHandle = atomic<ChildHandle?>(null)
+    internal var parentHandle: ChildHandle?
+        get() = _parentHandle.value
+        set(value) { _parentHandle.value = value }
 
     // ------------ initialization ------------
 
@@ -1019,23 +1020,33 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     @Suppress("UNCHECKED_CAST")
     private class Finishing(
         override val list: NodeList,
-        @Volatile
-        @JvmField var isCompleting: Boolean,
-        @Volatile
-        @JvmField var rootCause: Throwable? // NOTE: rootCause is kept even when SEALED
+        isCompleting: Boolean,
+        rootCause: Throwable?
     ) : SynchronizedObject(), Incomplete {
-        @Volatile
-        private var _exceptionsHolder: Any? = null // Contains null | Throwable | ArrayList | SEALED
+        private val _isCompleting = atomic(isCompleting)
+        var isCompleting: Boolean
+            get() = _isCompleting.value
+            set(value) { _isCompleting.value = value }
+
+        private val _rootCause = atomic(rootCause)
+        var rootCause: Throwable? // NOTE: rootCause is kept even when SEALED
+            get() = _rootCause.value
+            set(value) { _rootCause.value = value }
+
+        private val _exceptionsHolder = atomic<Any?>(null)
+        private var exceptionsHolder: Any? // Contains null | Throwable | ArrayList | SEALED
+            get() = _exceptionsHolder.value
+            set(value) { _exceptionsHolder.value = value }
 
         // NotE: cannot be modified when sealed
-        val isSealed: Boolean get() = _exceptionsHolder === SEALED
+        val isSealed: Boolean get() = exceptionsHolder === SEALED
         val isCancelling: Boolean get() = rootCause != null
         override val isActive: Boolean get() = rootCause == null // !isCancelling
 
         // Seals current state and returns list of exceptions
         // guarded by `synchronized(this)`
         fun sealLocked(proposedException: Throwable?): List<Throwable> {
-            val list = when(val eh = _exceptionsHolder) { // volatile read
+            val list = when(val eh = exceptionsHolder) { // volatile read
                 null -> allocateList()
                 is Throwable -> allocateList().also { it.add(eh) }
                 is ArrayList<*> -> eh as ArrayList<Throwable>
@@ -1044,7 +1055,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
             val rootCause = this.rootCause // volatile read
             rootCause?.let { list.add(0, it) } // note -- rootCause goes to the beginning
             if (proposedException != null && proposedException != rootCause) list.add(proposedException)
-            _exceptionsHolder = SEALED
+            exceptionsHolder = SEALED
             return list
         }
 
@@ -1056,11 +1067,11 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
                 return
             }
             if (exception === rootCause) return // nothing to do
-            when (val eh = _exceptionsHolder) { // volatile read
-                null -> _exceptionsHolder = exception
+            when (val eh = exceptionsHolder) { // volatile read
+                null -> exceptionsHolder = exception
                 is Throwable -> {
                     if (exception === eh) return // nothing to do
-                    _exceptionsHolder = allocateList().apply {
+                    exceptionsHolder = allocateList().apply {
                         add(eh)
                         add(exception)
 
@@ -1074,7 +1085,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
         private fun allocateList() = ArrayList<Throwable>(4)
 
         override fun toString(): String =
-            "Finishing[cancelling=$isCancelling, completing=$isCompleting, rootCause=$rootCause, exceptions=$_exceptionsHolder, list=$list]"
+            "Finishing[cancelling=$isCancelling, completing=$isCompleting, rootCause=$rootCause, exceptions=$exceptionsHolder, list=$list]"
     }
 
     private val Incomplete.isCancelling: Boolean

kotlinx-coroutines-core/common/src/channels/ArrayBroadcastChannel.kt

@@ -4,10 +4,10 @@
 
 package kotlinx.coroutines.channels
 
+import kotlinx.atomicfu.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.internal.*
 import kotlinx.coroutines.selects.*
-import kotlin.jvm.*
 
 /**
  * Broadcast channel with array buffer of a fixed [capacity].
@@ -44,12 +44,21 @@ internal class ArrayBroadcastChannel<E>(
 
     // head & tail are Long (64 bits) and we assume that they never wrap around
     // head, tail, and size are guarded by bufferLock
-    @Volatile
-    private var head: Long = 0 // do modulo on use of head
-    @Volatile
-    private var tail: Long = 0 // do modulo on use of tail
-    @Volatile
-    private var size: Int = 0
+
+    private val _head = atomic(0L)
+    private var head: Long // do modulo on use of head
+        get() = _head.value
+        set(value) { _head.value = value }
+
+    private val _tail = atomic(0L)
+    private var tail: Long // do modulo on use of tail
+        get() = _tail.value
+        set(value) { _tail.value = value }
+    
+    private val _size = atomic(0)
+    private var size: Int
+        get() = _size.value
+        set(value) { _size.value = value }
 
     private val subscribers = subscriberList<Subscriber<E>>()
 
@@ -199,9 +208,10 @@ internal class ArrayBroadcastChannel<E>(
     ) : AbstractChannel<E>(), ReceiveChannel<E> {
         private val subLock = ReentrantLock()
 
-        @Volatile
-        @JvmField
-        var subHead: Long = 0 // guarded by subLock
+        private val _subHead = atomic(0L)
+        var subHead: Long // guarded by subLock
+            get() = _subHead.value
+            set(value) { _subHead.value = value }
 
         override val isBufferAlwaysEmpty: Boolean get() = false
         override val isBufferEmpty: Boolean get() = subHead >= broadcastChannel.tail

kotlinx-coroutines-core/common/src/channels/ArrayChannel.kt

@@ -4,10 +4,10 @@
 
 package kotlinx.coroutines.channels
 
+import kotlinx.atomicfu.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.internal.*
 import kotlinx.coroutines.selects.*
-import kotlin.jvm.*
 import kotlin.math.*
 
 /**
@@ -36,8 +36,11 @@ internal open class ArrayChannel<E>(
      */
     private var buffer: Array<Any?> = arrayOfNulls<Any?>(min(capacity, 8))
     private var head: Int = 0
-    @Volatile
-    private var size: Int = 0 // Invariant: size <= capacity
+
+    private val _size = atomic(0)
+    private var size: Int // Invariant: size <= capacity
+        get() = _size.value
+        set(value) { _size.value = value }
 
     protected final override val isBufferAlwaysEmpty: Boolean get() = false
     protected final override val isBufferEmpty: Boolean get() = size == 0

kotlinx-coroutines-core/common/src/selects/Select.kt

@@ -232,8 +232,10 @@ internal class SelectBuilderImpl<in R>(
     private val _result = atomic<Any?>(UNDECIDED)
 
     // cancellability support
-    @Volatile
-    private var parentHandle: DisposableHandle? = null
+    private val _parentHandle = atomic<DisposableHandle?>(null)
+    private var parentHandle: DisposableHandle?
+        get() = _parentHandle.value
+        set(value) { _parentHandle.value = value }
 
     /* Result state machine