Merge pull request #766 from modelix/MODELIX-911-Efficient-lazy-loading

MODELIX-911 Efficient lazy loading (JVM)

Author: slisson

model-api/src/commonMain/kotlin/org/modelix/model/api/INode.kt

@@ -314,6 +314,10 @@ interface IReplaceableNode : INode {
 @Deprecated("Use .key(INode), .key(IBranch), .key(ITransaction) or .key(ITree)")
 fun IRole.key(): String = RoleAccessContext.getKey(this)
 fun IRole.key(node: INode): String = if (node.usesRoleIds()) getUID() else getSimpleName()
+fun IChildLink.key(node: INode): String? = when (this) {
+    is NullChildLink -> null
+    else -> (this as IRole).key(node)
+}
 fun INode.usesRoleIds(): Boolean = if (this is INodeEx) this.usesRoleIds() else false
 fun INode.getChildren(link: IChildLink): Iterable<INode> = if (this is INodeEx) getChildren(link) else getChildren(link.key(this))
 fun INode.moveChild(role: IChildLink, index: Int, child: INode): Unit = if (this is INodeEx) moveChild(role, index, child) else moveChild(role.key(this), index, child)
@@ -433,3 +437,5 @@ fun INode.getContainmentLink() = if (this is INodeEx) {
 fun INode.getRoot(): INode = parent?.getRoot() ?: this
 fun INode.isInstanceOf(superConcept: IConcept?): Boolean = concept.let { it != null && it.isSubConceptOf(superConcept) }
 fun INode.isInstanceOfSafe(superConcept: IConcept): Boolean = tryGetConcept()?.isSubConceptOf(superConcept) ?: false
+
+fun INode.addNewChild(role: IChildLink, index: Int) = addNewChild(role, index, null as IConceptReference?)

model-client/build.gradle.kts

@@ -21,6 +21,7 @@ val ktorVersion: String by rootProject
 val kotlinxSerializationVersion: String by rootProject
 
 kotlin {
+    jvmToolchain(11)
     jvm()
     js(IR) {
         browser {

model-client/src/commonMain/kotlin/org/modelix/model/client/GarbageFilteringStore.kt

@@ -29,6 +29,10 @@ class GarbageFilteringStore(private val store: IKeyValueStore) : IKeyValueStoreW
         return if (pendingEntries.containsKey(key)) pendingEntries[key] else store[key]
     }
 
+    override fun getIfCached(key: String): String? {
+        return if (pendingEntries.containsKey(key)) pendingEntries[key] else store.getIfCached(key)
+    }
+
     override fun getPendingSize(): Int = store.getPendingSize() + pendingEntries.size
 
     override fun getWrapped(): IKeyValueStore = store

model-client/src/commonMain/kotlin/org/modelix/model/client2/IModelClientV2Internal.kt

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2024.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.modelix.model.client2
+
+import org.modelix.model.lazy.RepositoryId
+import org.modelix.model.server.api.v2.ObjectHash
+import org.modelix.model.server.api.v2.ObjectHashAndSerializedObject
+import org.modelix.model.server.api.v2.SerializedObject
+
+/**
+ * Should only be used by Modelix components.
+ */
+interface IModelClientV2Internal : IModelClientV2 {
+    /**
+     * Required for lazy loading.
+     * Use [IModelClientV2.lazyLoadVersion]
+     */
+    suspend fun getObjects(repository: RepositoryId, keys: Sequence<ObjectHash>): Map<ObjectHash, SerializedObject>
+
+    /**
+     * Required for lazy loading.
+     * Use [IModelClientV2.lazyLoadVersion]
+     */
+    suspend fun pushObjects(repository: RepositoryId, objects: Sequence<ObjectHashAndSerializedObject>)
+}

model-client/src/commonMain/kotlin/org/modelix/model/client2/ModelClientV2.kt

@@ -68,6 +68,10 @@ import org.modelix.model.lazy.computeDelta
 import org.modelix.model.operations.OTBranch
 import org.modelix.model.persistent.HashUtil
 import org.modelix.model.persistent.MapBasedStore
+import org.modelix.model.server.api.v2.ImmutableObjectsStream
+import org.modelix.model.server.api.v2.ObjectHash
+import org.modelix.model.server.api.v2.ObjectHashAndSerializedObject
+import org.modelix.model.server.api.v2.SerializedObject
 import org.modelix.model.server.api.v2.VersionDelta
 import org.modelix.model.server.api.v2.VersionDeltaStream
 import org.modelix.model.server.api.v2.VersionDeltaStreamV2
@@ -83,7 +87,7 @@ class ModelClientV2(
     private val httpClient: HttpClient,
     val baseUrl: String,
     private var clientProvidedUserId: String?,
-) : IModelClientV2, Closable {
+) : IModelClientV2, IModelClientV2Internal, Closable {
     private var clientId: Int = 0
     private var idGenerator: IIdGenerator = IdGeneratorDummy()
     private var serverProvidedUserId: String? = null
@@ -265,6 +269,18 @@ class ModelClientV2(
         }
     }
 
+    override suspend fun getObjects(repository: RepositoryId, keys: Sequence<ObjectHash>): Map<ObjectHash, SerializedObject> {
+        return httpClient.preparePost {
+            url {
+                takeFrom(baseUrl)
+                appendPathSegments("repositories", repository.id, "objects", "getAll")
+            }
+            setBody(keys.joinToString("\n"))
+        }.execute { response ->
+            ImmutableObjectsStream.decode(response.bodyAsChannel())
+        }
+    }
+
     override suspend fun push(branch: BranchReference, version: IVersion, baseVersion: IVersion?): IVersion {
         LOG.debug { "${clientId.toString(16)}.push($branch, $version, $baseVersion)" }
         require(version is CLVersion)
@@ -274,7 +290,7 @@ class ModelClientV2(
         HashUtil.checkObjectHashes(objects)
         val delta = if (objects.size > 1000) {
             // large HTTP requests and large Json objects don't scale well
-            uploadObjects(branch.repositoryId, objects.asSequence().map { it.key to it.value })
+            pushObjects(branch.repositoryId, objects.asSequence().map { it.key to it.value })
             VersionDelta(version.getContentHash(), null)
         } else {
             VersionDelta(version.getContentHash(), null, objectsMap = objects)
@@ -292,7 +308,7 @@ class ModelClientV2(
         }
     }
 
-    private suspend fun uploadObjects(repository: RepositoryId, objects: Sequence<Pair<String, String>>) {
+    override suspend fun pushObjects(repository: RepositoryId, objects: Sequence<ObjectHashAndSerializedObject>) {
         LOG.debug { "${clientId.toString(16)}.pushObjects($repository)" }
         objects.chunked(100_000).forEach { unsortedChunk ->
             // Entries are sorted to avoid deadlocks on the server side between transactions.

model-client/src/jvmMain/kotlin/org/modelix/model/KeyValueStoreCache.kt

@@ -61,6 +61,10 @@ class KeyValueStoreCache(private val store: IKeyValueStore) : IKeyValueStoreWrap
         return getAll(setOf(key))[key]
     }
 
+    override fun getIfCached(key: String): String? {
+        return cache[key] ?: store.getIfCached(key)
+    }
+
     override fun getAll(keys: Iterable<String>): Map<String, String?> {
         val remainingKeys = toStream(keys).collect(Collectors.toList())
         val result: MutableMap<String, String?> = LinkedHashMap(16, 0.75.toFloat(), false)

model-client/src/jvmMain/kotlin/org/modelix/model/client/AsyncStore.kt

@@ -33,6 +33,12 @@ class AsyncStore(private val store: IKeyValueStore) : IKeyValueStoreWrapper {
         return store[key]
     }
 
+    override fun getIfCached(key: String): String? {
+        return synchronized(pendingWrites) {
+            pendingWrites[key]
+        } ?: store.getIfCached(key)
+    }
+
     override fun getWrapped(): IKeyValueStore = store
 
     override fun getPendingSize(): Int = store.getPendingSize() + pendingWrites.size

model-client/src/jvmMain/kotlin/org/modelix/model/client/RestWebModelClient.kt

@@ -371,6 +371,10 @@ class RestWebModelClient @JvmOverloads constructor(
         return runBlocking { getA(key) }
     }
 
+    override fun getIfCached(key: String): String? {
+        return null // doesn't contain any caches
+    }
+
     override suspend fun getA(key: String): String? {
         val isHash = HashUtil.isSha256(key)
         if (isHash) {

model-client/src/jvmMain/kotlin/org/modelix/model/client2/LazyLoading.kt

@@ -0,0 +1,102 @@
+/*
+ * Copyright (c) 2024.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.modelix.model.client2
+
+import kotlinx.coroutines.runBlocking
+import org.modelix.model.IKeyListener
+import org.modelix.model.IKeyValueStore
+import org.modelix.model.IVersion
+import org.modelix.model.lazy.BranchReference
+import org.modelix.model.lazy.CLVersion
+import org.modelix.model.lazy.CacheConfiguration
+import org.modelix.model.lazy.ObjectStoreCache
+import org.modelix.model.lazy.RepositoryId
+import org.modelix.model.persistent.HashUtil
+
+/**
+ * This function loads parts of the model lazily while it is iterated and limits the amount of data that is cached on
+ * the client side.
+ *
+ * IModelClientV2#loadVersion eagerly loads the whole model. For large models this can be slow and requires lots of
+ * memory.
+ * To reduce the relative overhead of requests to the server, the lazy loading algorithm tries to predict which nodes
+ * are required next and fill a "prefetch cache" by using "free capacity" of the regular requests. That means,
+ * the number of requests doesn't change by this prefetching, but small requests are filled to up to their limit with
+ * additional prefetch requests.
+ */
+fun IModelClientV2.lazyLoadVersion(repositoryId: RepositoryId, versionHash: String, config: CacheConfiguration = CacheConfiguration()): IVersion {
+    val store = ObjectStoreCache(ModelClientAsStore(this, repositoryId), config)
+    return CLVersion.loadFromHash(versionHash, store)
+}
+
+/**
+ * An overload of [IModelClientV2.lazyLoadVersion] that reads the current version hash of the branch from the server and
+ * then loads that version with lazy loading support.
+ */
+suspend fun IModelClientV2.lazyLoadVersion(branchRef: BranchReference, config: CacheConfiguration = CacheConfiguration()): IVersion {
+    return lazyLoadVersion(branchRef.repositoryId, pullHash(branchRef), config)
+}
+
+private class ModelClientAsStore(client: IModelClientV2, val repositoryId: RepositoryId) : IKeyValueStore {
+    private val client: IModelClientV2Internal = client as IModelClientV2Internal
+
+    override fun get(key: String): String? {
+        return getAll(listOf(key))[key]
+    }
+
+    override fun getIfCached(key: String): String? {
+        return null
+    }
+
+    override fun put(key: String, value: String?) {
+        putAll(mapOf(key to value))
+    }
+
+    override fun getAll(keys: Iterable<String>): Map<String, String?> {
+        return runBlocking {
+            client.getObjects(repositoryId, keys.asSequence())
+        }
+    }
+
+    override fun putAll(entries: Map<String, String?>) {
+        runBlocking {
+            client.pushObjects(
+                repositoryId,
+                entries.asSequence().map { (key, value) ->
+                    require(HashUtil.isSha256(key) && value != null) { "Only immutable objects are allowed: $key -> $value" }
+                    key to value
+                },
+            )
+        }
+    }
+
+    override fun prefetch(key: String) {
+        throw UnsupportedOperationException()
+    }
+
+    override fun listen(key: String, listener: IKeyListener) {
+        throw UnsupportedOperationException()
+    }
+
+    override fun removeListener(key: String, listener: IKeyListener) {
+        throw UnsupportedOperationException()
+    }
+
+    override fun getPendingSize(): Int {
+        return 0
+    }
+}

model-datastructure/src/commonMain/kotlin/org/modelix/model/IKeyValueStore.kt

@@ -16,12 +16,14 @@
 package org.modelix.model
 
 import org.modelix.model.lazy.BulkQuery
+import org.modelix.model.lazy.BulkQueryConfiguration
 import org.modelix.model.lazy.IBulkQuery
 import org.modelix.model.lazy.IDeserializingKeyValueStore
 
 interface IKeyValueStore {
-    fun newBulkQuery(deserializingCache: IDeserializingKeyValueStore): IBulkQuery = BulkQuery(deserializingCache)
+    fun newBulkQuery(deserializingCache: IDeserializingKeyValueStore, config: BulkQueryConfiguration): IBulkQuery = BulkQuery(deserializingCache, config)
     operator fun get(key: String): String?
+    fun getIfCached(key: String): String?
     suspend fun getA(key: String): String? = get(key)
     fun put(key: String, value: String?)
     fun getAll(keys: Iterable<String>): Map<String, String?>