perf(quartz): streaming filter deserializer to remove tree allocation
The relay-inbound path (REQ / COUNT / NEG-OPEN) was the only place ManualFilterDeserializer still went through `jp.codec.readTree(jp)`, materializing a full ObjectNode per filter before walking it. The Command/Message/Event deserializers next door already stream straight off the JsonParser; this brings filter parsing onto the same shape. - Add ManualFilterDeserializer.fromJson(JsonParser): mirrors EventDeserializer's hand-rolled token loop. Dispatches on field name, reads the seven fixed keys + dynamic #x / &x tag arrays via nextToken / nextTextValue / longValue / intValue, and drops invalid entries silently (same tolerance as the tree-based mapNotNull path). - Wire the four internal call sites — three in CommandDeserializer (REQ, COUNT, NEG-OPEN) and the standalone FilterDeserializer — to the streaming overload. - Keep the existing fromJson(ObjectNode) overload as-is for any external/cross-format adapter that already has a tree in hand. For a single REQ with N filters this drops N ObjectNode trees per inbound frame, which at 10k connections × ~5 filters × 1 msg/s is ~50k tree allocations/sec we don't have to do. No behavioral change: all quartz JVM tests pass (including the cross-mapper round-trip suite that compares Jackson and KotlinSerialization output) and geode's relay tests pass.
This commit is contained in:
+6
-10
@@ -24,7 +24,6 @@ import com.fasterxml.jackson.core.JsonParser
|
|||||||
import com.fasterxml.jackson.core.JsonToken
|
import com.fasterxml.jackson.core.JsonToken
|
||||||
import com.fasterxml.jackson.databind.DeserializationContext
|
import com.fasterxml.jackson.databind.DeserializationContext
|
||||||
import com.fasterxml.jackson.databind.deser.std.StdDeserializer
|
import com.fasterxml.jackson.databind.deser.std.StdDeserializer
|
||||||
import com.fasterxml.jackson.databind.node.ObjectNode
|
|
||||||
import com.vitorpamplona.quartz.nip01Core.jackson.EventDeserializer
|
import com.vitorpamplona.quartz.nip01Core.jackson.EventDeserializer
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
|
||||||
import com.vitorpamplona.quartz.nip01Core.relay.filters.ManualFilterDeserializer
|
import com.vitorpamplona.quartz.nip01Core.relay.filters.ManualFilterDeserializer
|
||||||
@@ -52,9 +51,9 @@ class CommandDeserializer : StdDeserializer<Command>(Command::class.java) {
|
|||||||
val filters = mutableListOf<Filter>()
|
val filters = mutableListOf<Filter>()
|
||||||
|
|
||||||
while (jp.nextToken() != JsonToken.END_ARRAY) {
|
while (jp.nextToken() != JsonToken.END_ARRAY) {
|
||||||
val filterObj: ObjectNode = jp.codec.readTree(jp)
|
// currentToken is now START_OBJECT for each filter;
|
||||||
val filter = ManualFilterDeserializer.fromJson(filterObj)
|
// the streaming parser consumes through END_OBJECT.
|
||||||
filters.add(filter)
|
filters.add(ManualFilterDeserializer.fromJson(jp))
|
||||||
}
|
}
|
||||||
|
|
||||||
ReqCmd(
|
ReqCmd(
|
||||||
@@ -68,9 +67,7 @@ class CommandDeserializer : StdDeserializer<Command>(Command::class.java) {
|
|||||||
val filters = mutableListOf<Filter>()
|
val filters = mutableListOf<Filter>()
|
||||||
|
|
||||||
while (jp.nextToken() != JsonToken.END_ARRAY) {
|
while (jp.nextToken() != JsonToken.END_ARRAY) {
|
||||||
val filterObj: ObjectNode = jp.codec.readTree(jp)
|
filters.add(ManualFilterDeserializer.fromJson(jp))
|
||||||
val filter = ManualFilterDeserializer.fromJson(filterObj)
|
|
||||||
filters.add(filter)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
CountCmd(
|
CountCmd(
|
||||||
@@ -102,9 +99,8 @@ class CommandDeserializer : StdDeserializer<Command>(Command::class.java) {
|
|||||||
|
|
||||||
NegOpenCmd.LABEL -> {
|
NegOpenCmd.LABEL -> {
|
||||||
val subId = jp.nextTextValue()
|
val subId = jp.nextTextValue()
|
||||||
jp.nextToken()
|
jp.nextToken() // advance to filter's START_OBJECT
|
||||||
val filterObj: ObjectNode = jp.codec.readTree(jp)
|
val filter = ManualFilterDeserializer.fromJson(jp)
|
||||||
val filter = ManualFilterDeserializer.fromJson(filterObj)
|
|
||||||
val initialMessage = jp.nextTextValue()
|
val initialMessage = jp.nextTextValue()
|
||||||
|
|
||||||
NegOpenCmd(
|
NegOpenCmd(
|
||||||
|
|||||||
+140
-1
@@ -21,6 +21,7 @@
|
|||||||
package com.vitorpamplona.quartz.nip01Core.relay.filters
|
package com.vitorpamplona.quartz.nip01Core.relay.filters
|
||||||
|
|
||||||
import com.fasterxml.jackson.core.JsonParser
|
import com.fasterxml.jackson.core.JsonParser
|
||||||
|
import com.fasterxml.jackson.core.JsonToken
|
||||||
import com.fasterxml.jackson.databind.DeserializationContext
|
import com.fasterxml.jackson.databind.DeserializationContext
|
||||||
import com.fasterxml.jackson.databind.deser.std.StdDeserializer
|
import com.fasterxml.jackson.databind.deser.std.StdDeserializer
|
||||||
import com.fasterxml.jackson.databind.node.ObjectNode
|
import com.fasterxml.jackson.databind.node.ObjectNode
|
||||||
@@ -32,11 +33,149 @@ class FilterDeserializer : StdDeserializer<Filter>(Filter::class.java) {
|
|||||||
override fun deserialize(
|
override fun deserialize(
|
||||||
jp: JsonParser,
|
jp: JsonParser,
|
||||||
ctxt: DeserializationContext,
|
ctxt: DeserializationContext,
|
||||||
): Filter = ManualFilterDeserializer.fromJson(jp.codec.readTree(jp))
|
): Filter = ManualFilterDeserializer.fromJson(jp)
|
||||||
}
|
}
|
||||||
|
|
||||||
class ManualFilterDeserializer {
|
class ManualFilterDeserializer {
|
||||||
companion object {
|
companion object {
|
||||||
|
/**
|
||||||
|
* Streaming filter parser. Reads field-by-field off [jp] without
|
||||||
|
* materializing an intermediate `JsonNode` tree — same shape as
|
||||||
|
* [com.vitorpamplona.quartz.nip01Core.jackson.EventDeserializer],
|
||||||
|
* which is what makes high-fan-out REQ traffic cheap on the
|
||||||
|
* relay-inbound path.
|
||||||
|
*
|
||||||
|
* Caller must position the parser so [jp.currentToken] is the
|
||||||
|
* `START_OBJECT` opening the filter. On return, [jp.currentToken]
|
||||||
|
* is the matching `END_OBJECT`.
|
||||||
|
*
|
||||||
|
* Tolerant by design — invalid array entries (wrong type, JSON
|
||||||
|
* null) are silently dropped, mirroring the
|
||||||
|
* `mapNotNull { it.asTextOrNull() }` behavior of the tree-based
|
||||||
|
* overload below. Unknown top-level fields are skipped via
|
||||||
|
* [JsonParser.skipChildren].
|
||||||
|
*/
|
||||||
|
fun fromJson(jp: JsonParser): Filter {
|
||||||
|
var ids: MutableList<String>? = null
|
||||||
|
var authors: MutableList<String>? = null
|
||||||
|
var kinds: MutableList<Int>? = null
|
||||||
|
var tags: MutableMap<String, List<String>>? = null
|
||||||
|
var tagsAll: MutableMap<String, List<String>>? = null
|
||||||
|
var since: Long? = null
|
||||||
|
var until: Long? = null
|
||||||
|
var limit: Int? = null
|
||||||
|
var search: String? = null
|
||||||
|
|
||||||
|
while (jp.nextToken() != JsonToken.END_OBJECT) {
|
||||||
|
val name = jp.currentName()
|
||||||
|
jp.nextToken() // advance to value
|
||||||
|
when {
|
||||||
|
name == "ids" -> {
|
||||||
|
ids = readStringArray(jp)
|
||||||
|
}
|
||||||
|
|
||||||
|
name == "authors" -> {
|
||||||
|
authors = readStringArray(jp)
|
||||||
|
}
|
||||||
|
|
||||||
|
name == "kinds" -> {
|
||||||
|
kinds = readIntArray(jp)
|
||||||
|
}
|
||||||
|
|
||||||
|
name == "since" -> {
|
||||||
|
if (jp.currentToken != JsonToken.VALUE_NULL) since = jp.longValue
|
||||||
|
}
|
||||||
|
|
||||||
|
name == "until" -> {
|
||||||
|
if (jp.currentToken != JsonToken.VALUE_NULL) until = jp.longValue
|
||||||
|
}
|
||||||
|
|
||||||
|
name == "limit" -> {
|
||||||
|
if (jp.currentToken != JsonToken.VALUE_NULL) limit = jp.intValue
|
||||||
|
}
|
||||||
|
|
||||||
|
name == "search" -> {
|
||||||
|
if (jp.currentToken != JsonToken.VALUE_NULL) search = jp.text
|
||||||
|
}
|
||||||
|
|
||||||
|
name.length > 1 && name[0] == '#' -> {
|
||||||
|
val map = tags ?: mutableMapOf<String, List<String>>().also { tags = it }
|
||||||
|
map[name.substring(1)] = readStringArray(jp)
|
||||||
|
}
|
||||||
|
|
||||||
|
name.length > 1 && name[0] == '&' -> {
|
||||||
|
val map = tagsAll ?: mutableMapOf<String, List<String>>().also { tagsAll = it }
|
||||||
|
map[name.substring(1)] = readStringArray(jp)
|
||||||
|
}
|
||||||
|
|
||||||
|
else -> {
|
||||||
|
jp.skipChildren()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return Filter(
|
||||||
|
ids = ids,
|
||||||
|
authors = authors,
|
||||||
|
kinds = kinds,
|
||||||
|
tags = tags,
|
||||||
|
tagsAll = tagsAll,
|
||||||
|
since = since,
|
||||||
|
until = until,
|
||||||
|
limit = limit,
|
||||||
|
search = search,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Reads a string array off [jp]. Drops non-string entries and
|
||||||
|
* JSON nulls — matches the `mapNotNull { it.asTextOrNull() }`
|
||||||
|
* tolerance of the tree-based path. Returns an empty list when
|
||||||
|
* the value is anything other than a `START_ARRAY` (incl.
|
||||||
|
* `null`), so callers don't have to special-case that.
|
||||||
|
*/
|
||||||
|
private fun readStringArray(jp: JsonParser): MutableList<String> {
|
||||||
|
val out = mutableListOf<String>()
|
||||||
|
if (jp.currentToken == JsonToken.START_ARRAY) {
|
||||||
|
while (jp.nextToken() != JsonToken.END_ARRAY) {
|
||||||
|
if (jp.currentToken == JsonToken.VALUE_STRING) {
|
||||||
|
out.add(jp.text)
|
||||||
|
} else if (jp.currentToken == JsonToken.START_OBJECT || jp.currentToken == JsonToken.START_ARRAY) {
|
||||||
|
jp.skipChildren()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else if (jp.currentToken == JsonToken.START_OBJECT) {
|
||||||
|
jp.skipChildren()
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Reads an int array off [jp]. Same tolerance rules as
|
||||||
|
* [readStringArray] — non-numeric entries are dropped.
|
||||||
|
*/
|
||||||
|
private fun readIntArray(jp: JsonParser): MutableList<Int> {
|
||||||
|
val out = mutableListOf<Int>()
|
||||||
|
if (jp.currentToken == JsonToken.START_ARRAY) {
|
||||||
|
while (jp.nextToken() != JsonToken.END_ARRAY) {
|
||||||
|
when (jp.currentToken) {
|
||||||
|
JsonToken.VALUE_NUMBER_INT, JsonToken.VALUE_NUMBER_FLOAT -> out.add(jp.intValue)
|
||||||
|
JsonToken.START_OBJECT, JsonToken.START_ARRAY -> jp.skipChildren()
|
||||||
|
else -> Unit
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else if (jp.currentToken == JsonToken.START_OBJECT) {
|
||||||
|
jp.skipChildren()
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Tree-based overload kept for callers that already have an
|
||||||
|
* `ObjectNode` in hand (e.g. cross-format adapters). New code on
|
||||||
|
* the relay-inbound path should use the streaming overload —
|
||||||
|
* this one materializes the full filter tree first.
|
||||||
|
*/
|
||||||
fun fromJson(jsonObject: ObjectNode): Filter {
|
fun fromJson(jsonObject: ObjectNode): Filter {
|
||||||
val tagsIn = mutableListOf<String>()
|
val tagsIn = mutableListOf<String>()
|
||||||
jsonObject.fieldNames().forEach {
|
jsonObject.fieldNames().forEach {
|
||||||
|
|||||||
Reference in New Issue
Block a user