Allocate underlying buffer in ArrayChannel in on-demand manner

Rationale:
Such change will allow us to use huge buffers in various flow operators without having a serious footprint in suspension-free scenarios

Author: qwwdfsad

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

@@ -8,6 +8,7 @@ import kotlinx.coroutines.*
 import kotlinx.coroutines.internal.*
 import kotlinx.coroutines.selects.*
 import kotlin.jvm.*
+import kotlin.math.*
 
 /**
  * Channel with array buffer of a fixed [capacity].
@@ -29,10 +30,14 @@ internal open class ArrayChannel<E>(
     }
 
     private val lock = ReentrantLock()
-    private val buffer: Array<Any?> = arrayOfNulls<Any?>(capacity)
+    /*
+     * Guarded by lock.
+     * Allocate minimum of capacity and 16 to avoid excess memory pressure for large channels when it's not necessary.
+     */
+    private var buffer: Array<Any?> = arrayOfNulls<Any?>(min(capacity, 16))
     private var head: Int = 0
     @Volatile
-    private var size: Int = 0
+    private var size: Int = 0 // Invariant: size <= capacity
 
     protected final override val isBufferAlwaysEmpty: Boolean get() = false
     protected final override val isBufferEmpty: Boolean get() = size == 0
@@ -64,6 +69,7 @@ internal open class ArrayChannel<E>(
                         }
                     }
                 }
+                ensureCapacity(size)
                 buffer[(head + size) % capacity] = element // actually queue element
                 return OFFER_SUCCESS
             }
@@ -112,6 +118,7 @@ internal open class ArrayChannel<E>(
                     this.size = size // restore size
                     return ALREADY_SELECTED
                 }
+                ensureCapacity(size)
                 buffer[(head + size) % capacity] = element // actually queue element
                 return OFFER_SUCCESS
             }
@@ -123,6 +130,16 @@ internal open class ArrayChannel<E>(
         return receive!!.offerResult
     }
 
+    // Guarded by lock
+    private fun ensureCapacity(currentSize: Int) {
+        if (currentSize == buffer.size ) {
+            val newSize = min(buffer.size * 2, capacity)
+            val newBuffer = arrayOfNulls<Any?>(newSize)
+            buffer.copyInto(newBuffer)
+            buffer = newBuffer
+        }
+    }
+
     // result is `E | POLL_FAILED | Closed`
     protected override fun pollInternal(): Any? {
         var send: Send? = null

kotlinx-coroutines-core/common/test/channels/ArrayChannelTest.kt

@@ -144,4 +144,31 @@ class ArrayChannelTest : TestBase() {
         channel.cancel(TestCancellationException())
         channel.receiveOrNull()
     }
+
+    @Test
+    fun testBufferSize() = runTest {
+        val capacity = 42
+        val channel = Channel<Int>(capacity)
+        launch {
+            expect(2)
+            repeat(42) {
+                channel.send(it)
+            }
+            expect(3)
+            channel.send(42)
+            expect(5)
+            channel.close()
+        }
+
+        expect(1)
+        yield()
+
+        expect(4)
+        val result = ArrayList<Int>(42)
+        channel.consumeEach {
+            result.add(it)
+        }
+        assertEquals((0..capacity).toList(), result)
+        finish(6)
+    }
 }