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

Execute non suspendable part outside request threads.
modelix · Mar 8, 2024 · f2f6c1b · f2f6c1b
1 parent 4800396
commit f2f6c1b
Showing 1 changed file with 35 additions and 15 deletions.
diff --git a/model-server/src/main/kotlin/org/modelix/model/server/handlers/RepositoriesManager.kt b/model-server/src/main/kotlin/org/modelix/model/server/handlers/RepositoriesManager.kt
@@ -13,12 +13,18 @@
  */
 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.runBlocking
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
 import kotlinx.datetime.Clock
 import org.apache.commons.collections4.map.LRUMap
 import org.modelix.model.IKeyValueStore
@@ -29,8 +35,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
@@ -258,25 +262,15 @@ 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 {
@@ -382,3 +376,29 @@ 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) {
+    private val cacheMap = LRUMap<Pair<String, String?>, SoftReference<Deferred<ObjectDataMap>>>(10)
+    private val cacheMapMutex = Mutex()
+
+    suspend fun getOrComputeDelta(versionHash: String, baseVersionHash: String): ObjectDataMap {
+        val deferredDelta = cacheMapMutex.withLock {
+            val key = versionHash to baseVersionHash
+            val existingDeferredDelta = cacheMap[key]?.get()
+            if (existingDeferredDelta != null) {
+                existingDeferredDelta
+            } else {
+                val version = CLVersion(versionHash, store)
+                val baseVersion = CLVersion(baseVersionHash, store)
+                val newDeferredDelta = withContext(Dispatchers.IO) {
+                    async {
+                        ObjectDataMap(version.computeDelta(baseVersion))
+                    }
+                }
+                cacheMap[key] = SoftReference(newDeferredDelta)
+                newDeferredDelta
+            }
+        }
+        return deferredDelta.await()
+    }
+}