fix(model-server): do not execute long-running repository operation o…

…n request threads Execute them on Dispatchers.IO because they are mostly blocked by waiting for data from store. Long-running operations are `computeDelta` and `mergeChanges`.
modelix · Mar 7, 2024 · 85297b9 · 85297b9
1 parent 2c8c792
commit 85297b9
Show file tree

Hide file tree

Showing 4 changed files with 36 additions and 13 deletions.
diff --git a/model-server/src/main/kotlin/org/modelix/model/server/handlers/HistoryHandler.kt b/model-server/src/main/kotlin/org/modelix/model/server/handlers/HistoryHandler.kt
@@ -8,6 +8,8 @@ import io.ktor.server.resources.get
 import io.ktor.server.resources.post
 import io.ktor.server.response.respondRedirect
 import io.ktor.server.routing.routing
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
 import kotlinx.datetime.Instant
 import kotlinx.datetime.TimeZone
 import kotlinx.datetime.toJavaLocalDateTime
@@ -92,7 +94,9 @@ class HistoryHandler(val client: IModelClient, private val repositoriesManager:
                     val fromVersion = params["from"]!!
                     val toVersion = params["to"]!!
                     val user = getUserName()
-                    revert(branch, fromVersion, toVersion, user)
+                    withContext(Dispatchers.IO) {
+                        revert(branch, fromVersion, toVersion, user)
+                    }
                     call.respondRedirect(".")
                 }
 //                post("/history/{repoId}/{branch}/delete") {

diff --git a/model-server/src/main/kotlin/org/modelix/model/server/handlers/KeyValueLikeModelServer.kt b/model-server/src/main/kotlin/org/modelix/model/server/handlers/KeyValueLikeModelServer.kt
@@ -27,6 +27,8 @@ import io.ktor.server.resources.put
 import io.ktor.server.response.respondText
 import io.ktor.server.routing.routing
 import io.ktor.util.pipeline.PipelineContext
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
 import kotlinx.html.br
 import kotlinx.html.div
 import kotlinx.html.h1
@@ -164,7 +166,9 @@ class KeyValueLikeModelServer(val repositoriesManager: RepositoriesManager) {
                     val key = call.parameters["key"]!!
                     val value = call.receiveText()
                     try {
-                        putEntries(mapOf(key to value))
+                        withContext(Dispatchers.IO) {
+                            putEntries(mapOf(key to value))
+                        }
                         call.respondText("OK")
                     } catch (e: NotFoundException) {
                         call.respondText(e.message ?: "Not found", status = HttpStatusCode.NotFound)
@@ -283,7 +287,7 @@ class KeyValueLikeModelServer(val repositoriesManager: RepositoriesManager) {
         return result
     }
 
-    protected fun CallContext.putEntries(newEntries: Map<String, String?>) {
+    protected suspend fun CallContext.putEntries(newEntries: Map<String, String?>) = withContext(Dispatchers.IO) {
         val referencedKeys: MutableSet<String> = HashSet()
         for ((key, value) in newEntries) {
             checkKeyPermission(key, EPermissionType.WRITE)

diff --git a/model-server/src/main/kotlin/org/modelix/model/server/handlers/ModelReplicationServer.kt b/model-server/src/main/kotlin/org/modelix/model/server/handlers/ModelReplicationServer.kt
@@ -189,7 +189,9 @@ class ModelReplicationServer(val repositoriesManager: RepositoriesManager) {
             val deltaFromClient = call.receive<VersionDelta>()
             deltaFromClient.checkObjectHashes()
             storeClient.putAll(deltaFromClient.getAllObjects())
-            val mergedHash = repositoriesManager.mergeChanges(branchRef(), deltaFromClient.versionHash)
+            val mergedHash = withContext(Dispatchers.IO) {
+                repositoriesManager.mergeChanges(branchRef(), deltaFromClient.versionHash)
+            }
             call.respondDelta(mergedHash, deltaFromClient.versionHash)
         }
 
@@ -226,14 +228,15 @@ class ModelReplicationServer(val repositoriesManager: RepositoriesManager) {
             webSocket("listen") {
                 var lastVersionHash = call.request.queryParameters["lastKnown"]
                 while (coroutineContext[Job]?.isCancelled == false) {
-                    val newVersionHash =
-                        repositoriesManager.pollVersionHash(call.branchRef(), lastVersionHash)
+                    val newVersionHash = repositoriesManager.pollVersionHash(call.branchRef(), lastVersionHash)
+                    val objectMap = withContext(Dispatchers.IO) {
+                        repositoriesManager.computeDelta(newVersionHash, lastVersionHash).checkObjectHashes().toMap()
+                    }
                     val delta = VersionDelta(
                         newVersionHash,
                         lastVersionHash,
-                        objectsMap = repositoriesManager.computeDelta(newVersionHash, lastVersionHash).toMap(),
+                        objectsMap = objectMap,
                     )
-                    delta.checkObjectHashes()
                     send(Json.encodeToString(delta))
                     lastVersionHash = newVersionHash
                 }
@@ -295,7 +298,9 @@ class ModelReplicationServer(val repositoriesManager: RepositoriesManager) {
                         baseVersion = version,
                         operations = ops.map { it.getOriginalOp() }.toTypedArray(),
                     )
-                    repositoriesManager.mergeChanges(branchRef, newVersion.getContentHash())
+                    withContext(Dispatchers.IO) {
+                        repositoriesManager.mergeChanges(branchRef, newVersion.getContentHash())
+                    }
                 }
             })
         }
@@ -367,16 +372,22 @@ class ModelReplicationServer(val repositoriesManager: RepositoriesManager) {
     }
 
     private suspend fun ApplicationCall.respondDeltaAsJson(versionHash: String, baseVersionHash: String?) {
+        val objectsMap = withContext(Dispatchers.IO) {
+            repositoriesManager.computeDelta(versionHash, baseVersionHash).checkObjectHashes().toMap()
+        }
         val delta = VersionDelta(
             versionHash,
             baseVersionHash,
-            objectsMap = repositoriesManager.computeDelta(versionHash, baseVersionHash).toMap(),
+            objectsMap = objectsMap,
         )
-        delta.checkObjectHashes()
         respond(delta)
     }
 
-    private suspend fun ApplicationCall.respondDeltaAsObjectStream(versionHash: String, baseVersionHash: String?, plainText: Boolean) {
+    private suspend fun ApplicationCall.respondDeltaAsObjectStream(
+        versionHash: String,
+        baseVersionHash: String?,
+        plainText: Boolean,
+    ) {
         respondTextWriter(contentType = if (plainText) ContentType.Text.Plain else VersionDeltaStream.CONTENT_TYPE) {
             repositoriesManager.computeDelta(versionHash, baseVersionHash)
                 .checkObjectHashes()

diff --git a/modelql-server/src/main/kotlin/org/modelix/modelql/server/ModelQLServer.kt b/modelql-server/src/main/kotlin/org/modelix/modelql/server/ModelQLServer.kt
@@ -80,7 +80,11 @@ class ModelQLServer private constructor(val rootNodeProvider: () -> INode?, val
             handleCall(call, { rootNode to area }, {})
         }
 
-        suspend fun handleCall(call: ApplicationCall, input: (write: Boolean) -> Pair<INode, IArea>, afterQueryExecution: () -> Unit = {}) {
+        suspend fun handleCall(
+            call: ApplicationCall,
+            input: (write: Boolean) -> Pair<INode, IArea>,
+            afterQueryExecution: suspend () -> Unit = {},
+        ) {
             try {
                 val serializedQuery = call.receiveText()
                 val json = UntypedModelQL.json