IO: fix verifying pool usage

buffer view doesn't hold original pool reference to make
pool chaining working

Author: Sergey Mashkov

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/ByteBufferChannel.kt

@@ -955,7 +955,7 @@ internal class ByteBufferChannel(
                         }
                     }
                 } catch (t: Throwable) {
-                    node.release()
+                    node.release(packet.pool)
                     packet.release()
                     throw t
                 }
@@ -965,7 +965,7 @@ internal class ByteBufferChannel(
                 }
             }
 
-            node.release()
+            node.release(packet.pool)
         }
     }
 
@@ -984,13 +984,13 @@ internal class ByteBufferChannel(
                     }
 
                     if (node.readRemaining == 0) {
-                        node.release()
+                        node.release(packet.pool)
                         node = packet.steal()
                     }
                 }
             }
         } catch (t: Throwable) {
-            node?.release()
+            node?.release(packet.pool)
             packet.release()
             throw t
         }

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/buffers/BufferView.kt

@@ -6,12 +6,10 @@ import kotlinx.coroutines.experimental.io.internal.*
 import kotlinx.coroutines.experimental.io.packet.*
 
 internal class BufferView private constructor(private var content: ByteBuffer,
-                                                 private val origin: BufferView?,
-                                                 private val pool: ObjectPool<BufferView>?) {
+                                                 private val origin: BufferView?) {
 
-    constructor(pool: ObjectPool<ByteBuffer>) : this(pool.borrow(), null, null)
-    constructor(buffer: ByteBuffer, pool: ObjectPool<BufferView>) : this(buffer, null, pool)
-    constructor(buffer: ByteBuffer) : this(buffer, null, null)
+    constructor(pool: ObjectPool<ByteBuffer>) : this(pool.borrow(), null)
+    constructor(buffer: ByteBuffer) : this(buffer, null)
 
     private var duplicated: ByteBuffer? = null
 
@@ -88,7 +86,7 @@ internal class BufferView private constructor(private var content: ByteBuffer,
         require(offset < dst.size)
         require(offset + length <= dst.size)
 
-        val dup = readDuplicated()
+        val dup = readDuplicated(length)
         dup.get(dst, offset, length)
         readPosition = rp + length
     }
@@ -202,7 +200,7 @@ internal class BufferView private constructor(private var content: ByteBuffer,
         val to = duplicated()
         to.limit(rp)
         to.position(rp - size)
-        val from = other.readDuplicated()
+        val from = other.readDuplicated(Int.MAX_VALUE)
 
         to.put(from)
         readPosition = rp - size
@@ -250,9 +248,10 @@ internal class BufferView private constructor(private var content: ByteBuffer,
     fun isExclusivelyOwned(): Boolean = refCount.value == 1L
 
     fun makeView(): BufferView {
-        (origin ?: this).acquire()
+        val newOrigin = origin ?: this
+        newOrigin.acquire()
 
-        val view = BufferView(content, origin, null)
+        val view = BufferView(content, newOrigin)
         view.attachment = attachment
         view.readPosition = readPosition
         view.writePosition = writePosition
@@ -261,15 +260,15 @@ internal class BufferView private constructor(private var content: ByteBuffer,
         return view
     }
 
-    fun release() {
+    fun release(pool: ObjectPool<BufferView>) {
         if (releaseRefCount()) {
             resetForWrite()
 
             if (origin != null) {
                 unlink()
-                origin.release()
+                origin.release(pool)
             } else {
-                pool?.recycle(this) ?: unlink()
+                pool.recycle(this)
             }
         }
     }
@@ -298,7 +297,7 @@ internal class BufferView private constructor(private var content: ByteBuffer,
         writePosition += delta
     }
 
-    private fun readDuplicated(limit: Int = Int.MAX_VALUE): ByteBuffer {
+    private fun readDuplicated(limit: Int): ByteBuffer {
         val position = readPosition
         val upLimit = (position.toLong() + limit).coerceAtMost(Int.MAX_VALUE.toLong()).toInt()
 
@@ -308,7 +307,7 @@ internal class BufferView private constructor(private var content: ByteBuffer,
         return buffer
     }
 
-    private fun writeDuplicated(limit: Int = Int.MAX_VALUE): ByteBuffer {
+    private fun writeDuplicated(limit: Int): ByteBuffer {
         val position = writePosition
         val upLimit = (position.toLong() + limit).coerceAtMost(Int.MAX_VALUE.toLong()).toInt()
 
@@ -349,7 +348,7 @@ internal class BufferView private constructor(private var content: ByteBuffer,
     }
 
     internal object Pool : ObjectPoolImpl<BufferView>(PACKET_BUFFER_POOL_SIZE) {
-        override fun produceInstance() = BufferView(ByteBuffer.allocateDirect(PACKET_BUFFER_SIZE), this)
+        override fun produceInstance() = BufferView(ByteBuffer.allocateDirect(PACKET_BUFFER_SIZE))
         override fun clearInstance(instance: BufferView): BufferView {
             return instance.apply {
                 next = null

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/buffers/Utils.kt

@@ -1,9 +1,11 @@
 package kotlinx.coroutines.experimental.io.buffers
 
-internal tailrec fun BufferView?.releaseAll() {
+import kotlinx.coroutines.experimental.io.internal.*
+
+internal tailrec fun BufferView?.releaseAll(pool: ObjectPool<BufferView>) {
     if (this == null) return
-    release()
-    next.releaseAll()
+    release(pool)
+    next.releaseAll(pool)
 }
 
 internal fun BufferView.copyAll(): BufferView {

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteReadPacketEmpty.kt

@@ -1,5 +1,3 @@
 package kotlinx.coroutines.experimental.io.packet
 
-import kotlinx.coroutines.experimental.io.buffers.*
-
-val ByteReadPacketEmpty: ByteReadPacket = ByteReadPacketViewBased(BufferView.Empty)
+val ByteReadPacketEmpty: ByteReadPacket = ByteReadPacketViewBased.Empty

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteReadPacketViewBased.kt

@@ -6,19 +6,20 @@ import kotlinx.coroutines.experimental.io.internal.*
 import java.io.*
 import java.nio.charset.*
 
-internal class ByteReadPacketViewBased(private var head: BufferView) : ByteReadPacket {
+internal class ByteReadPacketViewBased(private var head: BufferView,
+                                       internal val pool: ObjectPool<BufferView>) : ByteReadPacket {
 
     override val remaining: Int
         get() = head.remainingAll().toInt() // TODO Long or Int?
 
     override val isEmpty: Boolean
         get() = head.isEmpty()
 
-    override fun copy(): ByteReadPacket = ByteReadPacketViewBased(head.copyAll())
+    override fun copy(): ByteReadPacket = ByteReadPacketViewBased(head.copyAll(), pool)
 
     override fun release() {
         if (head !== BufferView.Empty) {
-            head.releaseAll()
+            head.releaseAll(pool)
             head = BufferView.Empty
         }
     }
@@ -86,6 +87,7 @@ internal class ByteReadPacketViewBased(private var head: BufferView) : ByteReadP
                         true
                     }
                     '\n' -> {
+                        afterRead()
                         return true
                     }
                     else -> {
@@ -94,15 +96,18 @@ internal class ByteReadPacketViewBased(private var head: BufferView) : ByteReadP
                             return@decodeUTF8 false
                         }
 
-                        if (decoded == limit) throw BufferLimitExceededException("Too many characters in line: limit $limit exceeded")
+                        if (decoded == limit) {
+                            afterRead()
+                            throw BufferLimitExceededException("Too many characters in line: limit $limit exceeded")
+                        }
                         decoded++
                         out.append(ch)
                         true
                     }
                 }
             }
 
-            if (size == 0) {
+            if (size == 0 || end) {
                 afterRead()
                 size = 1
             }
@@ -306,7 +311,7 @@ internal class ByteReadPacketViewBased(private var head: BufferView) : ByteReadP
         head.writeBufferAppend(next, minSize)
         if (next.readRemaining == 0) {
             head.next = next.next
-            next.release()
+            next.release(pool)
         }
 
         if (head.readRemaining >= minSize) return head
@@ -325,11 +330,13 @@ internal class ByteReadPacketViewBased(private var head: BufferView) : ByteReadP
     private fun releaseHead(head: BufferView) {
         val next = head.next
         this.head = next ?: BufferView.Empty
-        head.release()
+        head.release(pool)
     }
 
     companion object {
-        val Empty = ByteReadPacketViewBased(BufferView.Empty)
+        val Empty = ByteReadPacketViewBased(BufferView.Empty, object: NoPoolImpl<BufferView>() {
+            override fun borrow() = BufferView.Empty
+        })
         val ReservedSize = 8
         private val SkipBuffer = CharArray(8192)
     }

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/ByteWritePacketImpl.kt

@@ -144,7 +144,7 @@ internal class ByteWritePacketImpl(private var headerSizeHint: Int, private val
             tail.writeBufferAppend(foreignStolen)
             tail.next = foreignStolen.next
             this.tail = foreignStolen.tail().takeUnless { it === foreignStolen } ?: tail
-            foreignStolen.release()
+            foreignStolen.release(p.pool)
             size = head.remainingAll().toInt()
         } else if (appendSize == -1 || prependSize < appendSize) {
             // do prepend
@@ -162,7 +162,7 @@ internal class ByteWritePacketImpl(private var headerSizeHint: Int, private val
 
                 pre.next = foreignStolen
             }
-            tail.release()
+            tail.release(pool)
 
             this.tail = foreignStolen.tail()
             size = head.remainingAll().toInt()
@@ -326,7 +326,7 @@ internal class ByteWritePacketImpl(private var headerSizeHint: Int, private val
         this.size = 0
 
         if (head === BufferView.Empty) return ByteReadPacketViewBased.Empty
-        return ByteReadPacketViewBased(head)
+        return ByteReadPacketViewBased(head, pool)
     }
 
     override fun release() {
@@ -336,7 +336,7 @@ internal class ByteWritePacketImpl(private var headerSizeHint: Int, private val
         if (head !== empty) {
             this.head = empty
             this.tail = empty
-            head.releaseAll()
+            head.releaseAll(pool)
             size = 0
         }
     }

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/Channels.kt

@@ -23,7 +23,7 @@ fun WritableByteChannel.writePacket(p: ByteReadPacket) {
                 }
             }
         } finally {
-            b?.release()
+            b?.release(p.pool)
             p.release()
         }
     } else {
@@ -59,6 +59,7 @@ private fun ReadableByteChannel.readPacketImpl(min: Long, max: Long): ByteReadPa
 
     if (max == 0L) return ByteReadPacketEmpty
 
+    val pool = BufferView.Pool
     val empty = BufferView.Empty
     var head: BufferView = empty
     var tail: BufferView = empty
@@ -69,7 +70,7 @@ private fun ReadableByteChannel.readPacketImpl(min: Long, max: Long): ByteReadPa
         while (read < min || (read == min && min == 0L)) {
             val remInt = (max - read).coerceAtMost(Int.MAX_VALUE.toLong()).toInt()
 
-            val part = tail.takeIf { it.writeRemaining.let { it > 200 || it >= remInt } } ?: BufferView.Pool.borrow().also {
+            val part = tail.takeIf { it.writeRemaining.let { it > 200 || it >= remInt } } ?: pool.borrow().also {
                 if (head === empty) {
                     head = it; tail = it
                 }
@@ -93,10 +94,10 @@ private fun ReadableByteChannel.readPacketImpl(min: Long, max: Long): ByteReadPa
             }
         }
     } catch (t: Throwable) {
-        head.releaseAll()
+        head.releaseAll(pool)
         throw t
     }
 
-    return ByteReadPacketViewBased(head)
+    return ByteReadPacketViewBased(head, pool)
 }
 

core/kotlinx-coroutines-io/src/main/kotlin/kotlinx/coroutines/experimental/io/packet/Packets.kt

@@ -32,6 +32,7 @@ inline fun buildPacket(headerSizeHint: Int = 0, block: ByteWritePacket.() -> Uni
 fun WritePacket(headerSizeHint: Int = 0): ByteWritePacket = ByteWritePacketImpl(headerSizeHint, BufferView.Pool)
 
 fun ByteReadPacket.readUTF8Line(estimate: Int = 16, limit: Int = Int.MAX_VALUE): String? {
+    if (isEmpty) return null
     val sb = StringBuilder(estimate)
     return if (readUTF8LineTo(sb, limit)) sb.toString() else null
 }

core/kotlinx-coroutines-io/src/test/kotlin/kotlinx/coroutines/experimental/io/ByteBufferChannelTest.kt

@@ -31,10 +31,10 @@ class ByteBufferChannelTest {
     private val failures = ErrorCollector()
 
     @get:Rule
-    private val pool = VerifyingObjectPool(BufferObjectNoPool)
+    internal val pool = VerifyingObjectPool(BufferObjectNoPool)
 
     @get:Rule
-    private val pktPool = VerifyingObjectPool(BufferView.Pool)
+    internal val pktPool = VerifyingObjectPool(BufferView.Pool)
 
     private val Size = BUFFER_SIZE - RESERVED_SIZE
     private val ch = ByteBufferChannel(autoFlush = false, pool = pool)
@@ -567,7 +567,7 @@ class ByteBufferChannelTest {
             writeStringUtf8("Hello")
         } as ByteReadPacketViewBased
 
-        val packet = ByteReadPacketViewBased(packet0.steal()!!)
+        val packet = ByteReadPacketViewBased(packet0.steal()!!, pktPool)
 
         ch.writeInt(packet.remaining)
         ch.writePacket(packet)

core/kotlinx-coroutines-io/src/test/kotlin/kotlinx/coroutines/experimental/io/BytePacketBuildTest.kt

@@ -9,7 +9,7 @@ import kotlin.test.*
 
 class BytePacketBuildTest {
     @get:Rule
-    private val pool = VerifyingObjectPool(BufferView.Pool)
+    internal val pool = VerifyingObjectPool(BufferView.Pool)
 
     @Test
     fun smokeSingleBufferTest() {
@@ -160,6 +160,8 @@ class BytePacketBuildTest {
             p.readInt()
             fail()
         } catch (expected: EOFException) {
+        } finally {
+            p.release()
         }
     }