Pool byte arrays for charset decoding

formats/json/jvmMain/src/kotlinx/serialization/json/JvmStreams.kt

@@ -56,7 +56,12 @@ public fun <T> Json.decodeFromStream(
     deserializer: DeserializationStrategy<T>,
     stream: InputStream
 ): T {
-    return decodeByReader(deserializer, JavaStreamSerialReader(stream))
+    val reader = JavaStreamSerialReader(stream)
+    try {
+        return decodeByReader(deserializer, reader)
+    } finally {
+        reader.release()
+    }
 }
 
 /**

formats/json/jvmMain/src/kotlinx/serialization/json/internal/CharArrayPool.kt → formats/json/jvmMain/src/kotlinx/serialization/json/internal/ArrayPools.kt

@@ -3,20 +3,18 @@
  */
 package kotlinx.serialization.json.internal
 
-import java.util.concurrent.*
+/*
+ * Not really documented kill switch as a workaround for potential
+ * (unlikely) problems with memory consumptions.
+ */
+private val MAX_CHARS_IN_POOL = runCatching {
+    System.getProperty("kotlinx.serialization.json.pool.size").toIntOrNull()
+}.getOrNull() ?: 1024 * 1024
 
 internal open class CharArrayPoolBase {
     private val arrays = ArrayDeque<CharArray>()
     private var charsTotal = 0
 
-    /*
-     * Not really documented kill switch as a workaround for potential
-     * (unlikely) problems with memory consumptions.
-     */
-    private val MAX_CHARS_IN_POOL = runCatching {
-        System.getProperty("kotlinx.serialization.json.pool.size").toIntOrNull()
-    }.getOrNull() ?: 1024 * 1024 // 2 MB seems to be a reasonable constraint, (1M of chars)
-
     protected fun take(size: Int): CharArray {
         /*
          * Initially the pool is empty, so an instance will be allocated
@@ -52,3 +50,40 @@ internal actual object CharArrayPoolBatchSize : CharArrayPoolBase() {
         releaseImpl(array)
     }
 }
+
+// Byte array pool
+
+internal open class ByteArrayPoolBase {
+    private val arrays = ArrayDeque<kotlin.ByteArray>()
+    private var bytesTotal = 0
+
+    protected fun take(size: Int): ByteArray {
+        /*
+         * Initially the pool is empty, so an instance will be allocated
+         * and the pool will be populated in the 'release'
+         */
+        val candidate = synchronized(this) {
+            arrays.removeLastOrNull()?.also { bytesTotal -= it.size / 2 }
+        }
+        return candidate ?: ByteArray(size)
+    }
+
+    protected fun releaseImpl(array: ByteArray): Unit = synchronized(this) {
+        if (bytesTotal + array.size >= MAX_CHARS_IN_POOL) return@synchronized
+        bytesTotal += array.size / 2
+        arrays.addLast(array)
+    }
+}
+
+internal object ByteArrayPool8k : ByteArrayPoolBase() {
+    fun take(): ByteArray = super.take(8196)
+
+    fun release(array: ByteArray) = releaseImpl(array)
+}
+
+
+internal object ByteArrayPool : ByteArrayPoolBase() {
+    fun take(): ByteArray = super.take(128)
+
+    fun release(array: ByteArray) = releaseImpl(array)
+}

formats/json/jvmMain/src/kotlinx/serialization/json/internal/CharsetReader.kt

@@ -19,7 +19,7 @@ internal class CharsetReader(
         decoder = charset.newDecoder()
             .onMalformedInput(CodingErrorAction.REPLACE)
             .onUnmappableCharacter(CodingErrorAction.REPLACE)
-        byteBuffer = ByteBuffer.allocate(32)
+        byteBuffer = ByteBuffer.wrap(ByteArrayPool8k.take())
         byteBuffer.flip() // Make empty
     }
 
@@ -117,4 +117,8 @@ internal class CharsetReader(
             else -> error("Unreachable state: $bytesRead")
         }
     }
+
+    public fun release() {
+        ByteArrayPool8k.release(byteBuffer.array())
+    }
 }

formats/json/jvmMain/src/kotlinx/serialization/json/internal/JvmJsonStreams.kt

@@ -254,10 +254,14 @@ internal class JsonToJavaStreamWriter(private val stream: OutputStream) : JsonWr
 }
 
 internal class JavaStreamSerialReader(stream: InputStream) : SerialReader {
-    // NB: not closed on purpose, it is responsibility of the caller
+    // NB: not closed on purpose, it is the responsibility of the caller
     private val reader = CharsetReader(stream, Charsets.UTF_8)
 
     override fun read(buffer: CharArray, bufferOffset: Int, count: Int): Int {
         return reader.read(buffer, bufferOffset, count)
     }
+
+    fun release() {
+        reader.release()
+    }
 }