Skip to content

Commit dd4559f

Browse files
twisti-devtwisti-dev
authored
Merge pull request #51 from SLNE-Development/perf/optimize-performance
Optimize Redis message parsing and argument handling to reduce allocations
2 parents 11eeeec + 2b03b21 commit dd4559f

10 files changed

Lines changed: 308 additions & 285 deletions

File tree

‎gradle.properties‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,4 +3,4 @@ kotlin.stdlib.default.dependency=false
33
org.gradle.parallel=true
44
#org.gradle.caching=true
55
#org.gradle.configureondemand=true
6-
version=1.5.1
6+
version=1.6.0

‎surf-redis-api/api/surf-redis-api.api‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -66,10 +66,8 @@ public final class dev/slne/surf/redis/RedisComponentProvider$Companion : dev/sl
6666
}
6767

6868
public final class dev/slne/surf/redis/cache/RedisSetIndex {
69-
public final fun extractStrings (Ljava/lang/Object;)Ljava/util/Set;
7069
public final fun getName ()Ljava/lang/String;
7170
public fun toString ()Ljava/lang/String;
72-
public final fun valueString (Ljava/lang/Object;)Ljava/lang/String;
7371
}
7472

7573
public abstract class dev/slne/surf/redis/cache/RedisSetIndexes {

‎surf-redis-api/src/main/kotlin/dev/slne/surf/redis/cache/RedisSetIndexes.kt‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,12 +10,16 @@ class RedisSetIndex<T : Any, V : Any> internal constructor(
1010
private val valueToString: (V) -> String,
1111
private val normalize: (String) -> String
1212
) {
13-
fun extractStrings(element: T): Set<String> = valuesOf(element).asSequence()
13+
@InternalRedisAPI
14+
fun extractStringsSequence(element: T): Sequence<String> = valuesOf(element).asSequence()
1415
.map(valueToString)
1516
.map(normalize)
1617
.filter { it.isNotEmpty() }
17-
.toSet()
1818

19+
@InternalRedisAPI
20+
fun extractStrings(element: T): Set<String> = extractStringsSequence(element).toSet()
21+
22+
@InternalRedisAPI
1923
fun valueString(value: V): String =
2024
normalize(valueToString(value)).also {
2125
require(it.isNotEmpty()) { "Index '$name' produced blank key for value '$value'" }

‎surf-redis-core/src/main/kotlin/dev/slne/surf/redis/cache/SimpleRedisCacheImpl.kt‎

Lines changed: 71 additions & 75 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ import com.sksamuel.aedile.core.expireAfterWrite
66
import dev.slne.surf.api.core.util.logger
77
import dev.slne.surf.redis.RedisApi
88
import dev.slne.surf.redis.util.*
9-
import it.unimi.dsi.fastutil.objects.ObjectArrayList
109
import kotlinx.coroutines.reactor.awaitSingle
1110
import kotlinx.coroutines.reactor.awaitSingleOrNull
1211
import kotlinx.serialization.KSerializer
@@ -16,7 +15,6 @@ import org.redisson.api.RStreamReactive
1615
import org.redisson.api.stream.StreamMessageId
1716
import reactor.core.Disposable
1817
import reactor.core.publisher.Mono
19-
import java.util.concurrent.atomic.AtomicBoolean
2018
import java.util.concurrent.atomic.AtomicLong
2119
import java.util.concurrent.atomic.AtomicReference
2220
import kotlin.time.Duration
@@ -103,7 +101,15 @@ class SimpleRedisCacheImpl<K : Any, V : Any>(
103101

104102

105103
private val lastVersion = AtomicLong(0L)
106-
private val cursorId = AtomicReference<StreamMessageId>(StreamMessageId(0, 0))
104+
private val cursorId = AtomicReference(StreamMessageId(0, 0))
105+
106+
// Pre-computed per-instance constants reused on every Lua script invocation.
107+
// These avoid reallocating identical Strings / List<Any> wrappers on the hot path.
108+
private val messageDelimiterStr: String = MESSAGE_DELIMITER.toString()
109+
private val maxStreamLengthStr: String = MAX_STREAM_LENGTH.toString()
110+
private val ttlMillisStr: String = ttl.inWholeMilliseconds.toString()
111+
private val scriptKeys: List<Any> = listOf(idsKey, streamKey, versionKey)
112+
private val touchScriptKeys: List<Any> = listOf(idsKey)
107113

108114
private fun redisKey(key: K): String = "$keyPrefix$VALUE_KEY_INFIX${keyToString(key)}"
109115
private fun localKey(key: K): String = keyToString(key)
@@ -158,15 +164,24 @@ class SimpleRedisCacheImpl<K : Any, V : Any>(
158164
}
159165

160166
private fun processStreamMessage(type: String, msg: String) {
161-
// version<NUL>origin<NUL>payload
162-
val parts = msg.split(MESSAGE_DELIMITER, limit = 3)
163-
if (parts.size < 2) {
167+
// version<NUL>origin<NUL>payload — parse without allocating an ArrayList per message.
168+
val firstDelim = msg.indexOf(MESSAGE_DELIMITER)
169+
if (firstDelim < 0) {
164170
log.atWarning().log("Malformed stream message in cache '$namespace': $msg")
165171
return
166172
}
167-
val versionStr = parts[0]
168-
val originId = parts[1]
169-
val payload = if (parts.size >= 3) parts[2] else ""
173+
val secondDelim = msg.indexOf(MESSAGE_DELIMITER, firstDelim + 1)
174+
val versionStr = msg.substring(0, firstDelim)
175+
val originId: String
176+
val payload: String
177+
178+
if (secondDelim < 0) { // payload omitted in message
179+
originId = msg.substring(firstDelim + 1)
180+
payload = ""
181+
} else {
182+
originId = msg.substring(firstDelim + 1, secondDelim)
183+
payload = msg.substring(secondDelim + 1)
184+
}
170185

171186
val version = versionStr.toLongOrNull()
172187
if (version == null) {
@@ -175,20 +190,16 @@ class SimpleRedisCacheImpl<K : Any, V : Any>(
175190
return
176191
}
177192

178-
val shouldInvalidate = AtomicBoolean(false)
179193
val oldVersion = lastVersion.getAndUpdate { currentVer ->
180194
when {
181195
currentVer == 0L -> version
182196
version <= currentVer -> currentVer
183-
version == currentVer + 1 -> version
184-
else -> {
185-
shouldInvalidate.set(true)
186-
version
187-
}
197+
else -> version
188198
}
189199
}
190200

191-
if (shouldInvalidate.get()) {
201+
val shouldInvalidate = oldVersion != 0L && version > oldVersion + 1
202+
if (shouldInvalidate) {
192203
log.atWarning()
193204
.log("Version gap detected in cache '$namespace': last=$oldVersion, new=$version. Clearing near-cache.")
194205
clearNearCacheOnly()
@@ -254,24 +265,21 @@ class SimpleRedisCacheImpl<K : Any, V : Any>(
254265
requireNoNul(localKey, "key")
255266
val raw = api.json.encodeToString(serializer, value)
256267

257-
val argv = ObjectArrayList<String>(10)
258-
argv += instanceId
259-
argv += MESSAGE_DELIMITER.toString()
260-
argv += MAX_STREAM_LENGTH.toString()
261-
argv += ttl.inWholeMilliseconds.toString()
262-
argv += STREAM_FIELD_TYPE
263-
argv += STREAM_FIELD_MSG
264-
argv += OP_VAL
265-
argv += localKey
266-
argv += raw
267-
argv += keyPrefix
268-
269268
scriptExecutor.execute<Long>(
270269
PUT_SCRIPT,
271270
RScript.Mode.READ_WRITE,
272271
RScript.ReturnType.LONG,
273-
listOf(idsKey, streamKey, versionKey),
274-
*argv.toTypedArray()
272+
scriptKeys,
273+
/* argv */ instanceId,
274+
messageDelimiterStr,
275+
maxStreamLengthStr,
276+
ttlMillisStr,
277+
STREAM_FIELD_TYPE,
278+
STREAM_FIELD_MSG,
279+
OP_VAL,
280+
localKey,
281+
raw,
282+
keyPrefix
275283
).awaitSingle()
276284

277285
nearCache.put(localKey, CacheEntry.Value(value))
@@ -328,23 +336,20 @@ class SimpleRedisCacheImpl<K : Any, V : Any>(
328336
val localKey = localKey(key)
329337
requireNoNul(localKey, "key")
330338

331-
val argv = ObjectArrayList<String>(9)
332-
argv += instanceId
333-
argv += MESSAGE_DELIMITER.toString()
334-
argv += MAX_STREAM_LENGTH.toString()
335-
argv += ttl.inWholeMilliseconds.toString()
336-
argv += STREAM_FIELD_TYPE
337-
argv += STREAM_FIELD_MSG
338-
argv += OP_VAL
339-
argv += localKey
340-
argv += keyPrefix
341-
342339
scriptExecutor.execute<Long>(
343340
PUT_NULL_SCRIPT,
344341
RScript.Mode.READ_WRITE,
345342
RScript.ReturnType.LONG,
346-
listOf(idsKey, streamKey, versionKey),
347-
*argv.toTypedArray()
343+
scriptKeys,
344+
/* argv */instanceId,
345+
messageDelimiterStr,
346+
maxStreamLengthStr,
347+
ttlMillisStr,
348+
STREAM_FIELD_TYPE,
349+
STREAM_FIELD_MSG,
350+
OP_VAL,
351+
localKey,
352+
keyPrefix
348353
).awaitSingle()
349354
nearCache.put(localKey, CacheEntry.Null)
350355
}
@@ -353,23 +358,20 @@ class SimpleRedisCacheImpl<K : Any, V : Any>(
353358
val localKey = localKey(key)
354359
requireNoNul(localKey, "key")
355360

356-
val argv = ObjectArrayList<String>(9)
357-
argv += instanceId
358-
argv += MESSAGE_DELIMITER.toString()
359-
argv += MAX_STREAM_LENGTH.toString()
360-
argv += ttl.inWholeMilliseconds.toString()
361-
argv += STREAM_FIELD_TYPE
362-
argv += STREAM_FIELD_MSG
363-
argv += OP_VAL
364-
argv += localKey
365-
argv += keyPrefix
366-
367361
val removed: Long = scriptExecutor.execute<Long>(
368362
REMOVE_SCRIPT,
369363
RScript.Mode.READ_WRITE,
370364
RScript.ReturnType.LONG,
371-
listOf(idsKey, streamKey, versionKey),
372-
*argv.toTypedArray()
365+
scriptKeys,
366+
/* argv */ instanceId,
367+
messageDelimiterStr,
368+
maxStreamLengthStr,
369+
ttlMillisStr,
370+
STREAM_FIELD_TYPE,
371+
STREAM_FIELD_MSG,
372+
OP_VAL,
373+
localKey,
374+
keyPrefix
373375
).awaitSingle()
374376

375377
if (removed > 0) {
@@ -381,22 +383,19 @@ class SimpleRedisCacheImpl<K : Any, V : Any>(
381383
}
382384

383385
override suspend fun invalidateAll(): Long {
384-
val argv = ObjectArrayList<String>(8)
385-
argv += instanceId
386-
argv += MESSAGE_DELIMITER.toString()
387-
argv += MAX_STREAM_LENGTH.toString()
388-
argv += ttl.inWholeMilliseconds.toString()
389-
argv += STREAM_FIELD_TYPE
390-
argv += STREAM_FIELD_MSG
391-
argv += OP_ALL
392-
argv += keyPrefix
393-
394386
val deleted: Long = scriptExecutor.execute<Long>(
395387
CLEAR_SCRIPT,
396388
RScript.Mode.READ_WRITE,
397389
RScript.ReturnType.LONG,
398-
listOf(idsKey, streamKey, versionKey),
399-
*argv.toTypedArray()
390+
scriptKeys,
391+
/* argv */ instanceId,
392+
messageDelimiterStr,
393+
maxStreamLengthStr,
394+
ttlMillisStr,
395+
STREAM_FIELD_TYPE,
396+
STREAM_FIELD_MSG,
397+
OP_ALL,
398+
keyPrefix
400399
).awaitSingle()
401400
clearNearCacheOnly()
402401

@@ -407,17 +406,14 @@ class SimpleRedisCacheImpl<K : Any, V : Any>(
407406
val shouldRefresh = refreshGate.asMap().putIfAbsent(localKey, Unit) == null
408407
if (!shouldRefresh) return
409408

410-
val argv = ObjectArrayList<String>(3)
411-
argv += ttl.inWholeMilliseconds.toString()
412-
argv += localKey
413-
argv += keyPrefix
414-
415409
scriptExecutor.execute<Long>(
416410
TOUCH_SCRIPT,
417411
RScript.Mode.READ_WRITE,
418412
RScript.ReturnType.LONG,
419-
keys = listOf(idsKey),
420-
*argv.toTypedArray()
413+
keys = touchScriptKeys,
414+
/* argv */ ttlMillisStr,
415+
localKey,
416+
keyPrefix
421417
).subscribe(
422418
{ /* result isn't used */ },
423419
{ e ->

0 commit comments

Comments
 (0)