Skip to content

Commit

Permalink
perf(model-client): prefetch data to reduce the number of small requests
Browse files Browse the repository at this point in the history
  • Loading branch information
slisson committed Jun 20, 2024
1 parent ff4b503 commit 65ca207
Show file tree
Hide file tree
Showing 26 changed files with 937 additions and 220 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -86,8 +86,4 @@ interface IModelClientV2 {
suspend fun <R> query(branch: BranchReference, body: (IMonoStep<INode>) -> IMonoStep<R>): R

suspend fun <R> query(repositoryId: RepositoryId, versionHash: String, body: (IMonoStep<INode>) -> IMonoStep<R>): R

suspend fun getObjects(repository: RepositoryId, keys: Sequence<String>): Map<String, String>

suspend fun pushObjects(repository: RepositoryId, objects: Sequence<Pair<String, String>>)
}
Original file line number Diff line number Diff line change
@@ -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>)
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -265,27 +269,16 @@ class ModelClientV2(
}
}

override suspend fun getObjects(repository: RepositoryId, keys: Sequence<String>): Map<String, String> {
val response = httpClient.post {
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())
}

val content = response.bodyAsChannel()
val objects = HashMap<String, String>()
while (true) {
val key = checkNotNull(content.readUTF8Line()) { "Empty line expected at the end of the stream" }
if (key == "") {
check(content.readUTF8Line() == null) { "Empty line is only allowed at the end of the stream" }
break
}
val value = checkNotNull(content.readUTF8Line()) { "Object missing for hash $key" }
objects[key] = value
}
return objects
}

override suspend fun push(branch: BranchReference, version: IVersion, baseVersion: IVersion?): IVersion {
Expand Down Expand Up @@ -315,7 +308,7 @@ class ModelClientV2(
}
}

override suspend fun pushObjects(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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
173 changes: 102 additions & 71 deletions model-client/src/jvmMain/kotlin/org/modelix/model/client2/LazyLoading.kt
Original file line number Diff line number Diff line change
@@ -1,71 +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.ObjectStoreCache
import org.modelix.model.lazy.RepositoryId

fun IModelClientV2.lazyLoadVersion(repositoryId: RepositoryId, versionHash: String, cacheSize: Int = 100_000): IVersion {
val store = ObjectStoreCache(ModelClientAsStore(this, repositoryId), cacheSize)
return CLVersion.loadFromHash(versionHash, store)
}

suspend fun IModelClientV2.lazyLoadVersion(branchRef: BranchReference, cacheSize: Int = 100_000): IVersion {
return lazyLoadVersion(branchRef.repositoryId, pullHash(branchRef), cacheSize)
}

class ModelClientAsStore(val client: IModelClientV2, val repositoryId: RepositoryId) : IKeyValueStore {
override fun get(key: String): String? {
return getAll(listOf(key))[key]
}

override fun put(key: String, value: String?) {
TODO("Not yet implemented")
}

override fun getAll(keys: Iterable<String>): Map<String, String?> {
return runBlocking {
client.getObjects(repositoryId, keys.asSequence())
}
}

override fun putAll(entries: Map<String, String?>) {
TODO("Not yet implemented")
}

override fun prefetch(key: String) {
TODO("Not yet implemented")
}

override fun listen(key: String, listener: IKeyListener) {
TODO("Not yet implemented")
}

override fun removeListener(key: String, listener: IKeyListener) {
TODO("Not yet implemented")
}

override fun getPendingSize(): Int {
TODO("Not yet implemented")
}
}
/*
* 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
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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?>
Expand Down
Loading

0 comments on commit 65ca207

Please sign in to comment.