Fix various bugs (#99)

* Fix wrong parent for buffer pool

* Add toString debug information

* Add default length to readUtf8String

* Fix readAvailableTo block return value

* Prevent redundant buffer borrow

* Add tests
* Remove obsolete branch in utf8 decoding

Author: e5l

core/commonMain/src/kotlinx/io/Input.kt

@@ -136,9 +136,10 @@ public abstract class Input : Closeable {
 
             return readBufferRange { buffer, startOffset, endOffset ->
                 destination.writeBuffer(buffer, startOffset, endOffset)
-                endOffset
+                endOffset - startOffset
             }
         }
+
         // No bytes in cache: fill [destination] buffer direct.
         return destination.writeBuffer { buffer, startIndex, endIndex ->
             startIndex + fill(buffer, startIndex, endIndex)
@@ -278,8 +279,7 @@ public abstract class Input : Closeable {
         var remaining = count
         while (remaining > 0) {
             val skipCount = readBufferRange { _, startOffset, endOffset ->
-                val skipCount = min(remaining, endOffset - startOffset)
-                startOffset + skipCount
+                min(remaining, endOffset - startOffset)
             }
 
             if (skipCount == 0) {
@@ -410,7 +410,7 @@ public abstract class Input : Closeable {
     }
 
     /**
-     * Allows direct read from a buffer, operates on startOffset + endOffset (exclusive), returns new position.
+     * Allows direct read from a buffer, operates on startOffset + endOffset (exclusive), returns consumed bytes count.
      * NOTE: Dangerous to use, if non-local return then position will not be updated.
      *
      * @return consumed bytes count
@@ -420,10 +420,9 @@ public abstract class Input : Closeable {
             return 0
         }
 
-        val newPosition = reader(buffer, position, limit)
-        val result = newPosition - position
-        position = newPosition
-        return result
+        val consumed = reader(buffer, position, limit)
+        position += consumed
+        return consumed
     }
 
     /**
@@ -473,7 +472,11 @@ public abstract class Input : Closeable {
         }
 
         val bytes = preview ?: startPreview()
-        return readThroughPreview(bytes)
+        return fillAndStoreInPreview(bytes)
+    }
+
+    private fun prepareNewBuffer() {
+        buffer = bufferPool.borrow()
     }
 
     /**
@@ -493,14 +496,15 @@ public abstract class Input : Closeable {
      * The [bytes] shouldn't be empty.
      */
     private fun fetchFromPreviewAndDiscard(bytes: Bytes): Int {
-        bufferPool.recycle(buffer)
         bytes.discardFirst()
 
         if (bytes.isEmpty()) {
             previewBytes = null
             return fillFromSource()
         }
 
+        bufferPool.recycle(buffer)
+
         bytes.pointed(0) { newBuffer, newLimit ->
             position = 0
             buffer = newBuffer
@@ -510,12 +514,13 @@ public abstract class Input : Closeable {
         return limit
     }
 
-    private fun readThroughPreview(bytes: Bytes): Int {
+    private fun fillAndStoreInPreview(bytes: Bytes): Int {
         val nextIndex = previewIndex + 1
 
         if (bytes.isAfterLast(nextIndex)) {
+            prepareNewBuffer()
             val fetched = fillFromSource() // received data can be empty
-            bytes.append(buffer, limit) // buffer can be empty
+            bytes.append(buffer, fetched) // buffer can be empty
             previewIndex = nextIndex
             return fetched
         }
@@ -532,11 +537,9 @@ public abstract class Input : Closeable {
     }
 
     private fun fillFromSource(): Int {
-        val source = bufferPool.borrow()
-        val fetched = fill(source)
-        limit = fetched
+        val fetched = fill(buffer, 0, buffer.size)
         position = 0
-        buffer = source
+        limit = fetched
         return fetched
     }
 

core/commonMain/src/kotlinx/io/buffer/PrimitiveArraysOperations.commom.kt

@@ -28,4 +28,4 @@ public inline fun Buffer.storeByteArray(
     source.useBuffer(sourceOffset, count) { buffer ->
         buffer.copyTo(this, 0, count, offset)
     }
-}
+}

core/commonMain/src/kotlinx/io/buffer/UnmanagedBufferPool.kt

@@ -5,9 +5,14 @@ import kotlin.native.concurrent.*
 
 internal class UnmanagedBufferPool(
     private val bufferSize: Int = DEFAULT_BUFFER_SIZE
-) : DefaultPool<Buffer>(DEFAULT_POOL_CAPACITY) {
+) : ObjectPool<Buffer> {
+    override val capacity: Int = Int.MAX_VALUE
 
-    override fun produceInstance(): Buffer = bufferOf(ByteArray(bufferSize))
+    override fun borrow(): Buffer = bufferOf(ByteArray(bufferSize))
+
+    override fun recycle(instance: Buffer) {}
+
+    override fun close() {}
 
     @ThreadLocal
     companion object {

core/commonMain/src/kotlinx/io/text/TextInputObsolete.kt

@@ -1,7 +1,8 @@
 package kotlinx.io.text
 
-import kotlinx.io.*
+import kotlinx.io.Input
 
+private const val DEFAULT_CAPACITY: Int = 32
 
 /**
  * Read UTF-8 string until [delimiter] to [output].
@@ -34,7 +35,7 @@ public fun Input.readUtf8StringUntilDelimitersTo(output: Appendable, delimiters:
  *
  * @throws MalformedInputException if decoder fail to recognize charset.
  */
-public fun Input.readUtf8StringTo(output: Appendable, length: Int): Int {
+public fun Input.readUtf8StringTo(output: Appendable, length: Int = Int.MAX_VALUE): Int {
     var remaining = length
     return decodeUtf8Chars {
         output.append(it)
@@ -74,8 +75,11 @@ public fun Input.readUtf8LineTo(output: Appendable, limit: Int = Int.MAX_VALUE)
  *
  * @throws MalformedInputException if decoder fail to recognize charset.
  */
-public fun Input.readUtf8String(length: Int): String = buildString(length) {
-    readUtf8StringTo(this, length)
+public fun Input.readUtf8String(length: Int = Int.MAX_VALUE): String {
+    val capacity = if (length == Int.MAX_VALUE) DEFAULT_CAPACITY else length
+    return buildString(capacity) {
+        readUtf8StringTo(this, length)
+    }
 }
 
 /**
@@ -117,20 +121,13 @@ private inline fun Input.decodeUtf8Chars(consumer: (Char) -> Boolean): Int {
                             malformedInput(value)
                         if (!consumer(byte.toChar())) {
                             state = STATE_FINISH
-                            return@readBufferRange offset + 1
+                            return@readBufferRange offset - startOffset + 1
                         }
                         count++
                     }
                     byteCount == 0 -> {
                         // first unicode byte
                         when {
-                            byte < 0x80 -> {
-                                if (!consumer(byte.toChar())) {
-                                    state = STATE_FINISH
-                                    return@readBufferRange offset + 1
-                                }
-                                count++
-                            }
                             byte < 0xC0 -> {
                                 byteCount = 0
                                 value = byte and 0x7F
@@ -188,15 +185,15 @@ private inline fun Input.decodeUtf8Chars(consumer: (Char) -> Boolean): Int {
                             }
                             if (!more) {
                                 state = STATE_FINISH
-                                return@readBufferRange offset + 1
+                                return@readBufferRange offset - startOffset + 1
                             }
 
                             value = 0
                         }
                     }
                 }
             }
-            endOffset
+            endOffset - startOffset
         }
     }
     return count
@@ -254,6 +251,7 @@ private val Utf8StateMachine = intArrayOf(
 )
 
 private const val STATE_FINISH = -2
+
 //private const val Utf8_STATE_ASCII = -1
 internal const val STATE_UTF_8 = 0
 internal const val STATE_REJECT = 1

core/commonTest/src/kotlinx/io/InputOutputTest.common.kt

@@ -1,8 +1,9 @@
 @file:Suppress("FORBIDDEN_IDENTITY_EQUALS")
+
 package kotlinx.io
 
 import kotlinx.io.buffer.*
-import kotlinx.io.pool.*
+import kotlinx.io.pool.DefaultPool
 import kotlin.test.*
 
 class InputOutputTest {
@@ -49,7 +50,6 @@ class InputOutputTest {
         assertTrue(instance === result)
     }
 
-
     @Test
     fun testFillDirect() {
         val myBuffer = bufferOf(ByteArray(1024))
@@ -164,6 +164,106 @@ class InputOutputTest {
         assertEquals(1023, end)
     }
 
+    @Test
+    fun testReadAvailableToReturnValue() {
+        var readIndex = 0
+        var writeIndex = 0
+
+        val input = object : Input() {
+            override fun fill(buffer: Buffer, startIndex: Int, endIndex: Int): Int {
+                readIndex++
+
+                buffer.storeByteAt(startIndex, 42)
+                return 1
+            }
+
+            override fun closeSource() {
+                return
+            }
+        }
+
+        val output = object : Output() {
+            override fun flush(source: Buffer, startIndex: Int, endIndex: Int) {
+                writeIndex++
+
+                assertEquals(startIndex + 1, endIndex)
+                assertEquals(42, source.loadByteAt(startIndex))
+            }
+
+            override fun closeSource() {
+            }
+        }
+
+        repeat(DEFAULT_BUFFER_SIZE * 2) {
+            assertEquals(1, input.readAvailableTo(output))
+            output.flush()
+            assertEquals(it + 1, readIndex)
+            assertEquals(it + 1, writeIndex)
+        }
+
+        repeat(DEFAULT_BUFFER_SIZE * 2) {
+            input.prefetch(1)
+            assertEquals(1, input.readAvailableTo(output))
+
+            assertEquals(DEFAULT_BUFFER_SIZE * 2 + it + 1, readIndex)
+            assertEquals(DEFAULT_BUFFER_SIZE * 2 + it + 1, writeIndex)
+        }
+    }
+
+    @Test
+    fun testReadAvailableToReturnValueAfterSkipByte() {
+        var readIndex = 0
+        var writeIndex = 0
+
+        val input = object : Input() {
+            override fun fill(buffer: Buffer, startIndex: Int, endIndex: Int): Int {
+                check(startIndex + 10 <= endIndex)
+                readIndex += 10
+                return 10
+            }
+
+            override fun closeSource() {
+                return
+            }
+        }
+
+        val output = object : Output() {
+            override fun flush(source: Buffer, startIndex: Int, endIndex: Int) {
+                val count = endIndex - startIndex
+                writeIndex += count
+                assertEquals(9, count)
+            }
+
+            override fun closeSource() {
+            }
+        }
+
+
+        repeat(DEFAULT_BUFFER_SIZE * 2) {
+            input.readByte()
+            assertEquals(9, input.readAvailableTo(output))
+            output.flush()
+
+            assertEquals((it + 1) * 10, readIndex)
+            assertEquals((it + 1) * 9, writeIndex)
+        }
+
+        var previousRead = readIndex
+        var previousWrite = writeIndex
+        repeat(DEFAULT_BUFFER_SIZE * 2) {
+            input.prefetch(20)
+            input.readByte()
+            assertEquals(9, input.readAvailableTo(output))
+            input.readByte()
+            assertEquals(9, input.readAvailableTo(output))
+
+            assertEquals(20, readIndex - previousRead)
+            assertEquals(18, writeIndex - previousWrite)
+            previousRead = readIndex
+            previousWrite = writeIndex
+        }
+    }
+
     private fun checkException(block: () -> Unit) {
         var fail = false
         try {

core/commonTest/src/kotlinx/io/text/InputStringTest.kt

@@ -18,16 +18,23 @@ class InputStringTest {
     )
 
     @Test
-    fun decodeUtf8FromInputUntil() = bufferSizes.forEach { size ->
+    fun testDecodeUtf8FromInputUntil() = bufferSizes.forEach { size ->
         val input = buildBytes(size) { writeByteArray(content) }.input()
         val result = input.readUtf8StringUntilDelimiter('.')
         assertEquals(expected.dropLast(1), result)
     }
 
     @Test
-    fun decodeUtf8FromInput() = bufferSizes.forEach { size ->
+    fun testDecodeUtf8FromInput() = bufferSizes.forEach { size ->
         val input = buildBytes(size) { writeByteArray(content) }.input()
         val result = input.readUtf8String(expected.length)
         assertEquals(expected, result)
     }
+
+    @Test
+    fun testDecodeUtf8FromInputWithoutLength() = bufferSizes.forEach { size ->
+        val input = buildBytes(size) { writeByteArray(content) }.input()
+        val result = input.readUtf8String()
+        assertEquals(expected, result)
+    }
 }