- remove coroutines (most packets will only have one group, so it doesn't help much)

- don't materialize to named stores (unneeded)
This commit is contained in:
Adam 2022-03-14 00:00:04 +01:00
parent 0ebf492b42
commit 057eee6a68

View file

@ -18,7 +18,6 @@ import dev.adamko.kafkatorio.processor.serdes.jsonMapper
import dev.adamko.kafkatorio.processor.serdes.kxsBinary import dev.adamko.kafkatorio.processor.serdes.kxsBinary
import dev.adamko.kotka.extensions.consumedAs import dev.adamko.kotka.extensions.consumedAs
import dev.adamko.kotka.extensions.groupedAs import dev.adamko.kotka.extensions.groupedAs
import dev.adamko.kotka.extensions.materializedAs
import dev.adamko.kotka.extensions.materializedWith import dev.adamko.kotka.extensions.materializedWith
import dev.adamko.kotka.extensions.producedAs import dev.adamko.kotka.extensions.producedAs
import dev.adamko.kotka.extensions.repartitionedAs import dev.adamko.kotka.extensions.repartitionedAs
@ -34,13 +33,20 @@ import dev.adamko.kotka.extensions.tables.join
import dev.adamko.kotka.extensions.tables.mapValues import dev.adamko.kotka.extensions.tables.mapValues
import dev.adamko.kotka.extensions.tables.toStream import dev.adamko.kotka.extensions.tables.toStream
import dev.adamko.kotka.kxs.serde import dev.adamko.kotka.kxs.serde
import kotlin.coroutines.CoroutineContext import kotlin.collections.List
import kotlinx.coroutines.CoroutineName import kotlin.collections.Map
import kotlinx.coroutines.Dispatchers import kotlin.collections.associate
import kotlinx.coroutines.SupervisorJob import kotlin.collections.component1
import kotlinx.coroutines.async import kotlin.collections.component2
import kotlinx.coroutines.awaitAll import kotlin.collections.getOrElse
import kotlinx.coroutines.runBlocking import kotlin.collections.groupBy
import kotlin.collections.isNotEmpty
import kotlin.collections.isNullOrEmpty
import kotlin.collections.joinToString
import kotlin.collections.map
import kotlin.collections.mapValues
import kotlin.collections.mutableSetOf
import kotlin.collections.plus
import kotlinx.serialization.Serializable import kotlinx.serialization.Serializable
import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.StreamsBuilder
import org.apache.kafka.streams.Topology import org.apache.kafka.streams.Topology
@ -120,10 +126,6 @@ fun groupMapChunks(
} }
private val coroutineContext: CoroutineContext =
Dispatchers.Default + SupervisorJob() + CoroutineName("groupTilesIntoChunksWithColours")
private fun groupTilesIntoChunksWithColours( private fun groupTilesIntoChunksWithColours(
mapChunkUpdatePacketsStream: KStream<FactorioServerId, FactorioEventUpdatePacket>, mapChunkUpdatePacketsStream: KStream<FactorioServerId, FactorioEventUpdatePacket>,
tileProtoColourDict: KTable<FactorioServerId, TileColourDict>, tileProtoColourDict: KTable<FactorioServerId, TileColourDict>,
@ -163,35 +165,30 @@ private fun groupTilesIntoChunksWithColours(
val flatMappedChunkTiles: List<Pair<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>> = val flatMappedChunkTiles: List<Pair<ServerMapChunkId, ServerMapChunkTiles<TileProtoHashCode>>> =
// println("[${System.currentTimeMillis()}] Flat-mapped ${flatMappedChunkTiles.size} chunk, with ${flatMappedChunkTiles.joinToString { it.second.map.size.toString() }} tiles") // println("[${System.currentTimeMillis()}] Flat-mapped ${flatMappedChunkTiles.size} chunk, with ${flatMappedChunkTiles.joinToString { it.second.map.size.toString() }} tiles")
runBlocking(coroutineContext) {
mapTiles mapTiles
.tiles .tiles
.groupBy { tile -> .groupBy { tile ->
tile.position.toMapChunkPosition(standardChunkSize.tilesPerChunk) tile.position.toMapChunkPosition(standardChunkSize.tilesPerChunk)
} }
.map { (chunkPos, tiles) -> .map { (chunkPos, tiles) ->
async { val chunkId = ServerMapChunkId(
serverId = serverId,
chunkPosition = chunkPos,
surfaceIndex = mapTiles.surfaceIndex,
chunkSize = standardChunkSize,
)
val chunkId = ServerMapChunkId( val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) }
serverId = serverId,
chunkPosition = chunkPos,
surfaceIndex = mapTiles.surfaceIndex,
chunkSize = standardChunkSize,
)
val tilesMap = tiles.associate { it.position to TileProtoHashCode(it) } val chunkTiles = ServerMapChunkTiles(chunkId, tilesMap)
val chunkTiles = ServerMapChunkTiles(chunkId, tilesMap) chunkId to chunkTiles
}
chunkId to chunkTiles
}
}.awaitAll()
// println("[${System.currentTimeMillis()}] Flat-mapped ${flatMappedChunkTiles.size} chunk, with ${flatMappedChunkTiles.joinToString { it.second.map.size.toString() }} tiles") // println("[${System.currentTimeMillis()}] Flat-mapped ${flatMappedChunkTiles.size} chunk, with ${flatMappedChunkTiles.joinToString { it.second.map.size.toString() }} tiles")
}
flatMappedChunkTiles flatMappedChunkTiles
} }
.filter("server-map-data.tiles.filter-not-empty") { _, value -> .filter("server-map-data.tiles.filter-not-empty") { _, value ->
@ -206,8 +203,7 @@ private fun groupTilesIntoChunksWithColours(
) )
.reduce( .reduce(
"server-map-data.tiles.reduce", "server-map-data.tiles.reduce",
materializedAs( materializedWith(
"server-map-data.tiles.store",
kxsBinary.serde<ServerMapChunkId>(), kxsBinary.serde<ServerMapChunkId>(),
kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(), kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>>(),
) )
@ -222,8 +218,7 @@ private fun groupTilesIntoChunksWithColours(
) )
).toTable( ).toTable(
"server-map-data.tiles.repartition.to-table", "server-map-data.tiles.repartition.to-table",
materializedAs( materializedWith(
"server-map-data.tiles.repartition.to-table.store",
kxsBinary.serde<ServerMapChunkId>(), kxsBinary.serde<ServerMapChunkId>(),
kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>?>(), kxsBinary.serde<ServerMapChunkTiles<TileProtoHashCode>?>(),
) )
@ -255,8 +250,7 @@ private fun groupTilesIntoChunksWithColours(
) )
).toTable( ).toTable(
"tileProtoColourDict.repartition.to-table", "tileProtoColourDict.repartition.to-table",
materializedAs( materializedWith(
"tileProtoColourDict.repartition.to-table.store",
kxsBinary.serde<FactorioServerId>(), kxsBinary.serde<FactorioServerId>(),
kxsBinary.serde<TileColourDict?>(), kxsBinary.serde<TileColourDict?>(),
) )
@ -280,8 +274,8 @@ private fun groupTilesIntoChunksWithColours(
.join( .join(
other = tileProtoColourDictRepartitioned, other = tileProtoColourDictRepartitioned,
tableJoined = tableJoined("server-map-data.join-tiles-with-prototypes"), tableJoined = tableJoined("server-map-data.join-tiles-with-prototypes"),
materialized = materializedAs( materialized = materializedWith(
"server-map-data.join-tiles-with-prototypes.store", // "server-map-data.join-tiles-with-prototypes.store",
kxsBinary.serde<ServerMapChunkId>(), kxsBinary.serde<ServerMapChunkId>(),
kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(), kxsBinary.serde<ServerMapChunkTiles<ColourHex>>(),
), ),