Further improve stream decoding performance

[qwwdfsad]

Current dev:

JacksonComparisonBenchmark.kotlinFromStream          thrpt   10  247.197 ± 19.929  ops/ms

This branch:

JacksonComparisonBenchmark.kotlinFromStream          thrpt   10  564.784 ± 4.125  ops/ms

[qwwdfsad]

Rebased & ready for review

[slavonnet]

    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)
    }

Logic have hidden issue. If you take 128b buffer then release (buffer added to pool), then call take (512) you get last 128 buffer.
On children we have static size of take and now all ok, but if some one create another child?

And i think buffer sizes very small. Most systems are configured to use block sizes of 4096 or 8192(up to 64k). Ideal look to system memory page size.

[slavonnet]

Maybe Rewrite this code to remove many synchronized calls and speed up code?

internal class ByteArrayPoolBase(val size: Int) {

// new pool instance in every kotlin object. Not need super.take and protected overload or change to glocal static array / kotlin object
      private val arrays = ArrayDeque<kotlin.ByteArray>()
      private val bufferSize =size / 2
// bytesCount is private and not used in code 

 fun take(): ByteArray =
        if (array.isEmpty) { // try check for empty without synchronized
               ByteArray(size)
        } else {
            synchronized(this) {
                arrays.removeLastOrNull() 
            } ?: ByteArray(size)
        }        

}
    fun release(array: ByteArray): Unit {
             // arrays.size atomic. But If will be data race we temporary add +1 element to poll. synchronized not needed.  MAX_CHARS_IN_POOL pool size for chars. is for bytebuffer is same?
             if (arrays.size * bufferSize  < MAX_CHARS_IN_POOL) 
                 synchronized(this) {
                       arrays.addLast(array)
                }
             
    }


}

internal object ByteArrayPool8k : ByteArrayPoolBase(8196)
internal object ByteArrayPool : ByteArrayPoolBase(512)

[slavonnet]

also ByteArray, CharArray can replace with template T :)