perf(model-server): make version delta computation suspendable

Execute non suspendable part outside request threads.

Author: Oleksandr Dzhychko

model-server/src/main/kotlin/org/modelix/model/server/handlers/RepositoriesManager.kt

@@ -13,11 +13,15 @@
  */
 package org.modelix.model.server.handlers
 
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.async
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.asFlow
 import kotlinx.coroutines.flow.channelFlow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.datetime.Clock
 import org.apache.commons.collections4.map.LRUMap
@@ -29,8 +33,6 @@ import org.modelix.model.api.IReadTransaction
 import org.modelix.model.api.ITree
 import org.modelix.model.api.IdGeneratorDummy
 import org.modelix.model.api.PBranch
-import org.modelix.model.api.runSynchronized
-import org.modelix.model.client2.checkObjectHashes
 import org.modelix.model.lazy.BranchReference
 import org.modelix.model.lazy.CLTree
 import org.modelix.model.lazy.CLVersion
@@ -46,6 +48,7 @@ import org.modelix.model.server.store.IStoreClient
 import org.modelix.model.server.store.LocalModelClient
 import org.modelix.model.server.store.pollEntry
 import org.modelix.model.server.store.runTransactionSuspendable
+import org.slf4j.LoggerFactory
 import java.lang.ref.SoftReference
 import java.util.UUID
 
@@ -258,58 +261,52 @@ class RepositoriesManager(val client: LocalModelClient) {
             ?: throw IllegalStateException("No version found for branch '${branch.branchName}' in repository '${branch.repositoryId}'")
     }
 
-    private val deltaCache = LRUMap<Pair<String, String?>, SoftReference<Lazy<ObjectDataMap>>>(10)
-    fun computeDelta(versionHash: String, baseVersionHash: String?): ObjectData {
+    private val versionDeltaCache = VersionDeltaCache(client.storeCache)
+    suspend fun computeDelta(versionHash: String, baseVersionHash: String?): ObjectData {
         if (versionHash == baseVersionHash) return ObjectData.empty
         if (baseVersionHash == null) {
             // no need to cache anything if there is no delta computation happening
             return allObjectDataAsFlow(versionHash)
         }
 
-        return runSynchronized(deltaCache) {
-            val key = versionHash to baseVersionHash
-            deltaCache.get(key)?.get() ?: lazy {
-                // lazy { ... } allows to run the computation without locking deltaCache
-                // SoftReference because deltas can be very large
-                val version = CLVersion(versionHash, client.storeCache)
-                val baseVersion = CLVersion(baseVersionHash, client.storeCache)
-                val objectsMap = version.computeDelta(baseVersion)
-                ObjectDataMap(objectsMap)
-            }.also { deltaCache[key] = SoftReference(it) }
-        }.value
+        return versionDeltaCache.getOrComputeDelta(versionHash, baseVersionHash)
     }
 
     private fun allObjectDataAsFlow(versionHash: String): ObjectDataFlow {
         val hashObjectFlow = channelFlow {
-            val version = CLVersion(versionHash, objectStore)
-            // Use a bulk query to make as few request to the underlying store as possible.
-            val bulkQuery = objectStore.newBulkQuery()
-            // It is unsatisfactory that we have to keep already emitted hashes in memory.
-            // But without changing the underlying model,
-            // we have to do this to not emit objects more than once.
-            val seenHashes = mutableSetOf<String>()
-            fun emitObjects(entry: KVEntryReference<*>) {
-                if (seenHashes.contains(entry.getHash())) return
-                seenHashes.add(entry.getHash())
-                bulkQuery.get(entry).onSuccess {
-                    val value = checkNotNull(it) { "No value received for ${entry.getHash()}" }
-                    // Use `send` instead of `trySend`,
-                    // because `trySend` fails if the channel capacity is full.
-                    // This might happen if the data is produced faster than consumed.
-                    // A better solution would be to have bulk queries which itself are asynchronous
-                    // but doing that needs more consideration.
-                    runBlocking {
-                        // Maybe we should avoid Flow<Pair<String, String>> and use Flow<String>.
-                        // This needs profiling.
-                        channel.send(entry.getHash() to value.serialize())
-                    }
-                    for (referencedEntry in value.getReferencedEntries()) {
-                        emitObjects(referencedEntry)
+            // Our bulk query is blocking, therefor we explicitly launch it on one of the Dispatchers.IO.
+            // Without it, the consumer could accidentally start the flow on this thread and block it.
+            launch(Dispatchers.IO) {
+                // Use a bulk query to make as few request to the underlying store as possible.
+                val bulkQuery = objectStore.newBulkQuery()
+                // It is unsatisfactory that we have to keep already emitted hashes in memory.
+                // But without changing the underlying model,
+                // we have to do this to not emit objects more than once.
+                val seenHashes = mutableSetOf<String>()
+                fun emitObjects(entry: KVEntryReference<*>) {
+                    if (seenHashes.contains(entry.getHash())) return
+                    seenHashes.add(entry.getHash())
+                    bulkQuery.get(entry).onSuccess {
+                        val value = checkNotNull(it) { "No value received for ${entry.getHash()}" }
+                        // Use `send` instead of `trySend`,
+                        // because `trySend` fails if the channel capacity is full.
+                        // This might happen if the data is produced faster than consumed.
+                        // A better solution would be to have bulk queries which itself are asynchronous
+                        // but doing that needs more consideration.
+                        runBlocking {
+                            // Maybe we should avoid Flow<Pair<String, String>> and use Flow<String>.
+                            // This needs profiling
+                            channel.send(entry.getHash() to value.serialize())
+                        }
+                        for (referencedEntry in value.getReferencedEntries()) {
+                            emitObjects(referencedEntry)
+                        }
                     }
                 }
+                emitObjects(KVEntryReference(versionHash, CPVersion.DESERIALIZER))
+                LOG.debug("Starting to bulk query all objects.")
+                bulkQuery.process()
             }
-            emitObjects(KVEntryReference(versionHash, CPVersion.DESERIALIZER))
-            bulkQuery.process()
         }
         val checkedHashObjectFlow = hashObjectFlow.checkObjectHashes()
         val objectData = ObjectDataFlow(checkedHashObjectFlow)
@@ -347,6 +344,7 @@ class RepositoriesManager(val client: LocalModelClient) {
     }
 
     companion object {
+        private val LOG = LoggerFactory.getLogger(RepositoriesManager::class.java)
         const val KEY_PREFIX = ":v2"
         private const val REPOSITORIES_LIST_KEY = "$KEY_PREFIX:repositories"
         const val LEGACY_SERVER_ID_KEY = "repositoryId"
@@ -382,3 +380,38 @@ class ObjectDataFlow(private val hashObjectFlow: Flow<Pair<String, String>>) : O
 private fun Flow<Pair<String, String>>.checkObjectHashes(): Flow<Pair<String, String>> {
     return onEach { HashUtil.checkObjectHash(it.first, it.second) }
 }
+
+class VersionDeltaCache(val store: IDeserializingKeyValueStore) {
+
+    companion object {
+        private val LOG = LoggerFactory.getLogger(VersionDeltaCache::class.java)
+    }
+
+    private val cacheMap = LRUMap<Pair<String, String?>, SoftReference<Deferred<ObjectDataMap>>>(10)
+
+    suspend fun getOrComputeDelta(versionHash: String, baseVersionHash: String): ObjectDataMap {
+        val deferredDelta = synchronized(cacheMap) {
+            val key = versionHash to baseVersionHash
+            val existingDeferredDelta = cacheMap[key]?.get()
+            if (existingDeferredDelta != null) {
+                LOG.debug("Version delta found in cache for {}.", key)
+                existingDeferredDelta
+            } else {
+                LOG.debug("Version delta not found in cache for {}.", key)
+                val version = CLVersion(versionHash, store)
+                val baseVersion = CLVersion(baseVersionHash, store)
+                val newDeferredDelta = runBlocking(Dispatchers.IO) {
+                    async {
+                        LOG.debug("Computing for delta for {}.", key)
+                        val result = ObjectDataMap(version.computeDelta(baseVersion))
+                        LOG.debug("Computed version delta for {}.", key)
+                        result
+                    }
+                }
+                cacheMap[key] = SoftReference(newDeferredDelta)
+                newDeferredDelta
+            }
+        }
+        return deferredDelta.await()
+    }
+}