Merge pull request #13 from aaronriekenberg/020623_jmh

JMH Benchmarks: Avoid byte array copy for reads in NMA valueSerializer

Author: aaronriekenberg

benchmarks/build.gradle.kts

@@ -16,10 +16,10 @@ dependencies {
 }
 
 jmh {
-    warmupIterations.set(3)
-    iterations.set(3)
+    warmupIterations.set(5)
+    iterations.set(5)
     fork.set(1)
     forceGC.set(true)
     failOnError.set(true)
-    jvmArgs.set(listOf("-Xmx2G"))
+    jvmArgs.set(listOf("-Xmx4G"))
 }

benchmarks/src/jmh/kotlin/com/target/nativememoryallocator/benchmarks/OffHeapGetPutBenchmark.kt

@@ -5,6 +5,7 @@ import com.target.nativememoryallocator.benchmarks.impl.RocksDBOffHeapCache
 import com.target.nativememoryallocator.benchmarks.impl.UnimplementedOffHeapCache
 import mu.KotlinLogging
 import org.openjdk.jmh.annotations.*
+import java.nio.ByteBuffer
 
 private val logger = KotlinLogging.logger {}
 
@@ -26,7 +27,7 @@ open class OffHeapGetPutBenchmark {
     )
     var cacheType: String = ""
 
-    private var cache: OffHeapCache<String, ByteArray> = UnimplementedOffHeapCache
+    private var cache: OffHeapCache<String, ByteBuffer> = UnimplementedOffHeapCache
 
     @State(Scope.Thread)
     open class ThreadState {
@@ -60,7 +61,7 @@ open class OffHeapGetPutBenchmark {
         }
 
         for (i in 0 until NUM_ENTRIES) {
-            cache.put(key = i.toString(), value = ByteArray(VALUE_SIZE))
+            cache.put(key = i.toString(), value = ByteBuffer.wrap(ByteArray(VALUE_SIZE)))
         }
 
         cache.logMetadata()
@@ -78,7 +79,7 @@ open class OffHeapGetPutBenchmark {
     @Benchmark
     @Group("read_only")
     @GroupThreads(8)
-    fun readOnly(threadState: ThreadState): ByteArray? {
+    fun readOnly(threadState: ThreadState): ByteBuffer? {
         val key = threadState.nextIndex().toString()
         val value = cache.get(key = key)
         if (value == null) {
@@ -92,13 +93,13 @@ open class OffHeapGetPutBenchmark {
     @GroupThreads(8)
     fun writeOnly(threadState: ThreadState) {
         val key = threadState.nextIndex().toString()
-        cache.put(key = key, value = ByteArray(VALUE_SIZE))
+        cache.put(key = key, value = ByteBuffer.wrap(ByteArray(VALUE_SIZE)))
     }
 
     @Benchmark
     @Group("readwrite")
     @GroupThreads(6)
-    fun readwrite_get(threadState: ThreadState): ByteArray? {
+    fun readwrite_get(threadState: ThreadState): ByteBuffer? {
         val key = threadState.nextIndex().toString()
         val value = cache.get(key = key)
         if (value == null) {
@@ -112,7 +113,7 @@ open class OffHeapGetPutBenchmark {
     @GroupThreads(2)
     fun readwrite_put(threadState: ThreadState) {
         val key = threadState.nextIndex().toString()
-        cache.put(key = key, value = ByteArray(VALUE_SIZE))
+        cache.put(key = key, value = ByteBuffer.wrap(ByteArray(VALUE_SIZE)))
     }
 
 }

benchmarks/src/jmh/kotlin/com/target/nativememoryallocator/benchmarks/impl/NMAOffHeapCache.kt

@@ -7,14 +7,15 @@ import com.target.nativememoryallocator.map.NativeMemoryMapBackend
 import com.target.nativememoryallocator.map.NativeMemoryMapBuilder
 import com.target.nativememoryallocator.map.NativeMemoryMapSerializer
 import mu.KotlinLogging
+import java.nio.ByteBuffer
 
 private val logger = KotlinLogging.logger {}
 
 
 /**
  * NMA implementation of OffHeapCache.
  */
-class NMAOffHeapCache : OffHeapCache<String, ByteArray> {
+class NMAOffHeapCache : OffHeapCache<String, ByteBuffer> {
 
     init {
         logger.info { "initializing NMAOffHeapCache" }
@@ -25,27 +26,27 @@ class NMAOffHeapCache : OffHeapCache<String, ByteArray> {
         nativeMemorySizeBytes = (10L * 1024 * 1024 * 1024), //10gb,
     ).build()
 
-    private val valueSerializer = object : NativeMemoryMapSerializer<ByteArray> {
-        override fun deserializeFromOnHeapMemoryBuffer(onHeapMemoryBuffer: OnHeapMemoryBuffer): ByteArray {
-            return onHeapMemoryBuffer.toTrimmedArray()
+    private val valueSerializer = object : NativeMemoryMapSerializer<ByteBuffer> {
+        override fun deserializeFromOnHeapMemoryBuffer(onHeapMemoryBuffer: OnHeapMemoryBuffer): ByteBuffer {
+            return onHeapMemoryBuffer.asByteBuffer()
         }
 
-        override fun serializeToByteArray(value: ByteArray): ByteArray {
-            return value
+        override fun serializeToByteArray(value: ByteBuffer): ByteArray {
+            return value.array()
         }
     };
 
-    private val nativeMemoryMap = NativeMemoryMapBuilder<String, ByteArray>(
+    private val nativeMemoryMap = NativeMemoryMapBuilder<String, ByteBuffer>(
         valueSerializer = valueSerializer,
         nativeMemoryAllocator = nativeMemoryAllocator,
         backend = NativeMemoryMapBackend.CAFFEINE,
     ).build()
 
-    override fun get(key: String): ByteArray? {
+    override fun get(key: String): ByteBuffer? {
         return nativeMemoryMap.get(key = key)
     }
 
-    override fun put(key: String, value: ByteArray) {
+    override fun put(key: String, value: ByteBuffer) {
         nativeMemoryMap.put(key = key, value = value)
     }
 

benchmarks/src/jmh/kotlin/com/target/nativememoryallocator/benchmarks/impl/RocksDBOffHeapCache.kt

@@ -4,13 +4,14 @@ import com.target.nativememoryallocator.benchmarks.OffHeapCache
 import mu.KotlinLogging
 import org.rocksdb.RocksDB
 import java.io.File
+import java.nio.ByteBuffer
 
 private val logger = KotlinLogging.logger {}
 
 /**
  * RocksDB implementation of OffHeapCache.
  */
-class RocksDBOffHeapCache : OffHeapCache<String, ByteArray> {
+class RocksDBOffHeapCache : OffHeapCache<String, ByteBuffer> {
 
     private val rocksDbDir: String
 
@@ -28,12 +29,12 @@ class RocksDBOffHeapCache : OffHeapCache<String, ByteArray> {
         rocksDB = RocksDB.open(rocksDbDir)
     }
 
-    override fun get(key: String): ByteArray? {
-        return rocksDB.get(key.toByteArray())
+    override fun get(key: String): ByteBuffer? {
+        return rocksDB.get(key.toByteArray())?.let { ByteBuffer.wrap(it) }
     }
 
-    override fun put(key: String, value: ByteArray) {
-        rocksDB.put(key.toByteArray(), value)
+    override fun put(key: String, value: ByteBuffer) {
+        rocksDB.put(key.toByteArray(), value.array())
     }
 
     override fun size(): Int {

benchmarks/src/jmh/kotlin/com/target/nativememoryallocator/benchmarks/impl/UnimplementedOffHeapCache.kt

@@ -1,18 +1,19 @@
 package com.target.nativememoryallocator.benchmarks.impl
 
 import com.target.nativememoryallocator.benchmarks.OffHeapCache
+import java.nio.ByteBuffer
 
 /**
  * Unimplemented OffHeapCache.
  * All methods throw NotImplementedError.
  */
-object UnimplementedOffHeapCache : OffHeapCache<String, ByteArray> {
+object UnimplementedOffHeapCache : OffHeapCache<String, ByteBuffer> {
 
-    override fun get(key: String): ByteArray? {
+    override fun get(key: String): ByteBuffer? {
         throw NotImplementedError()
     }
 
-    override fun put(key: String, value: ByteArray) {
+    override fun put(key: String, value: ByteBuffer) {
         throw NotImplementedError()
     }