toki
71e709d30d
- Kotlin: TCP parallel test, WebSocket latency fix, stress test updates - TypeScript: TCP client, WebSocket client/server fixes, stress test, WS test updates - Archive completed subtasks (03_kotlin_tcp_parallel, 04_ws_fixed_latency) - Update roadmap milestone
796 lines
33 KiB
Kotlin
796 lines
33 KiB
Kotlin
@file:JvmName("StressKt")
|
|
|
|
// Same-language Kotlin stress / benchmark harness.
|
|
//
|
|
// 고성능 병렬 운용 기준선 Milestone의 lang-baseline Task에서, TypeScript stress.ts와 같은 출력
|
|
// 계약(ROW|/SKIP|/SUMMARY|)으로 Kotlin same-language TCP/WS baseline을 남긴다. 절대 성능 합격선은
|
|
// 고정하지 않고 local 환경 baseline(throughput, p50/p95/p99 latency, used heap)을 기록하며, 안정성
|
|
// 합격선만 hard fail로 둔다: timeout 0, nonce mismatch 0, response type mismatch 0, connection별
|
|
// FIFO 위반 0, 종료 후 pending leak 0.
|
|
//
|
|
// 실행: ./gradlew run -PmainClass=com.tokilabs.proto_socket.crosstest.StressKt --args="--quick --transport=tcp,ws"
|
|
|
|
package com.tokilabs.proto_socket.crosstest
|
|
|
|
import com.tokilabs.proto_socket.Communicator
|
|
import com.tokilabs.proto_socket.ParserMap
|
|
import com.tokilabs.proto_socket.TcpClient
|
|
import com.tokilabs.proto_socket.TcpServer
|
|
import com.tokilabs.proto_socket.WsClient
|
|
import com.tokilabs.proto_socket.WsServer
|
|
import com.tokilabs.proto_socket.DialTcp
|
|
import com.tokilabs.proto_socket.DialWs
|
|
import com.tokilabs.proto_socket.addListenerTyped
|
|
import com.tokilabs.proto_socket.addRequestListenerTyped
|
|
import com.tokilabs.proto_socket.packets.TestData
|
|
import com.tokilabs.proto_socket.typeNameOf
|
|
import kotlinx.coroutines.CompletableDeferred
|
|
import kotlinx.coroutines.Dispatchers
|
|
import kotlinx.coroutines.channels.Channel
|
|
import kotlinx.coroutines.coroutineScope
|
|
import kotlinx.coroutines.delay
|
|
import kotlinx.coroutines.launch
|
|
import kotlinx.coroutines.runBlocking
|
|
import kotlinx.coroutines.withContext
|
|
import kotlinx.coroutines.withTimeoutOrNull
|
|
import java.net.InetAddress
|
|
import java.net.ServerSocket
|
|
import java.util.concurrent.ConcurrentLinkedQueue
|
|
import java.util.concurrent.atomic.AtomicInteger
|
|
import java.util.concurrent.atomic.AtomicLong
|
|
import kotlin.system.exitProcess
|
|
|
|
private const val HOST = "127.0.0.1"
|
|
private const val LANGUAGE = "Kotlin"
|
|
private const val WS_PATH = "/"
|
|
private val ALL_PROFILES = listOf("roundtrip", "burst", "sustained", "parallel", "slow-mix", "payload")
|
|
|
|
// payload size matrix 축. quick은 작은 샘플(1KB/64KB), full은 Milestone 기준(1KB/64KB/1MB)을 측정한다.
|
|
private const val PAYLOAD_SEED = "0123456789abcdef"
|
|
private val PAYLOAD_SIZES_QUICK = listOf(1024, 65536)
|
|
private val PAYLOAD_SIZES_FULL = listOf(1024, 65536, 1048576)
|
|
|
|
private val rowsEmitted = AtomicInteger(0)
|
|
private val totalViolations = AtomicLong(0)
|
|
|
|
/** 안정성 위반 카운터. 0이 아니면 해당 row는 FAIL이고 프로세스도 non-zero exit한다. */
|
|
private class StressStability {
|
|
val timeouts = AtomicLong(0)
|
|
val nonceMismatch = AtomicLong(0)
|
|
val typeMismatch = AtomicLong(0)
|
|
val fifoViolations = AtomicLong(0)
|
|
val pendingLeak = AtomicLong(0)
|
|
|
|
fun violations(): Long =
|
|
timeouts.get() + nonceMismatch.get() + typeMismatch.get() + fifoViolations.get() + pendingLeak.get()
|
|
}
|
|
|
|
private class StressServerHandle(
|
|
val port: Int,
|
|
val stop: () -> Unit,
|
|
private val connected: Channel<Unit>,
|
|
) {
|
|
suspend fun waitForConnections(count: Int = 1) {
|
|
repeat(count) {
|
|
if (withTimeoutOrNull(3_000L) { connected.receive() } == null) {
|
|
throw IllegalStateException("server did not finish client setup")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private class StressClientHandle(
|
|
val comm: Communicator,
|
|
val close: () -> Unit,
|
|
// [0]=maxStaged, [1]=droppedCount, [2]=totalCopyNanos, [3]=receivedCount
|
|
val wsMetrics: (() -> LongArray)? = null,
|
|
)
|
|
|
|
private suspend fun stopServer(handle: StressServerHandle) {
|
|
handle.stop()
|
|
delay(50)
|
|
}
|
|
|
|
private fun parserMap(): ParserMap = mapOf(typeNameOf<TestData>() to { bytes: ByteArray -> TestData.parseFrom(bytes) })
|
|
|
|
private fun log(line: String) = System.err.println(line)
|
|
|
|
private fun memMb(): Double {
|
|
val rt = Runtime.getRuntime()
|
|
return (rt.totalMemory() - rt.freeMemory()).toDouble() / (1024 * 1024)
|
|
}
|
|
|
|
private fun freePort(): Int = ServerSocket(0, 50, InetAddress.getByName(HOST)).use { it.localPort }
|
|
|
|
private fun testDataPayloadBytes(message: String): Int =
|
|
TestData.newBuilder().setIndex(0).setMessage(message).build().toByteArray().size
|
|
|
|
// payloadFiller는 size 바이트 길이의 deterministic filler string을 만든다.
|
|
private fun payloadFiller(size: Int): String {
|
|
if (size <= 0) return ""
|
|
val reps = (size + PAYLOAD_SEED.length - 1) / PAYLOAD_SEED.length
|
|
return PAYLOAD_SEED.repeat(reps).substring(0, size)
|
|
}
|
|
|
|
// payloadLabel은 payload 크기를 사람이 읽는 축 이름으로 바꾼다(1024 -> 1KB, 1048576 -> 1MB).
|
|
private fun payloadLabel(bytes: Int): String = when {
|
|
bytes >= 1048576 && bytes % 1048576 == 0 -> "${bytes / 1048576}MB"
|
|
bytes >= 1024 && bytes % 1024 == 0 -> "${bytes / 1024}KB"
|
|
else -> "${bytes}B"
|
|
}
|
|
|
|
private fun percentile(sorted: List<Double>, p: Double): Double {
|
|
if (sorted.isEmpty()) return 0.0
|
|
var rank = Math.ceil(p / 100.0 * sorted.size).toInt() - 1
|
|
if (rank < 0) rank = 0
|
|
if (rank > sorted.size - 1) rank = sorted.size - 1
|
|
return sorted[rank]
|
|
}
|
|
|
|
private fun classifyRequestError(error: Throwable, s: StressStability) {
|
|
val msg = (error.message ?: error.toString()).lowercase()
|
|
when {
|
|
msg.contains("timeout") -> s.timeouts.incrementAndGet()
|
|
msg.contains("expected") || msg.contains("mismatch") -> s.typeMismatch.incrementAndGet()
|
|
else -> s.nonceMismatch.incrementAndGet()
|
|
}
|
|
}
|
|
|
|
private fun emitRow(
|
|
profile: String,
|
|
axis: String,
|
|
transport: String,
|
|
payloadBytes: Int,
|
|
clientCount: Int,
|
|
requests: Int,
|
|
throughput: Double,
|
|
p50: Double,
|
|
p95: Double,
|
|
p99: Double,
|
|
mem: Double,
|
|
s: StressStability,
|
|
queueBacklog: Long = 0L,
|
|
gatewayBacklog: Long = 0L,
|
|
) {
|
|
val violations = s.violations()
|
|
val status = if (violations == 0L) "PASS" else "FAIL"
|
|
totalViolations.addAndGet(violations)
|
|
rowsEmitted.incrementAndGet()
|
|
val fields = listOf(
|
|
"ROW",
|
|
profile,
|
|
axis,
|
|
LANGUAGE,
|
|
transport,
|
|
payloadBytes.toString(),
|
|
clientCount.toString(),
|
|
requests.toString(),
|
|
String.format("%.1f", throughput),
|
|
String.format("%.3f", p50),
|
|
String.format("%.3f", p95),
|
|
String.format("%.3f", p99),
|
|
s.timeouts.get().toString(),
|
|
s.nonceMismatch.get().toString(),
|
|
s.typeMismatch.get().toString(),
|
|
s.fifoViolations.get().toString(),
|
|
s.pendingLeak.get().toString(),
|
|
queueBacklog.toString(), // queueBacklog: WS=maxStaged, TCP=0 (same-language baseline에는 inbound gateway가 없다)
|
|
gatewayBacklog.toString(), // gatewayBacklog: WS=droppedCount, TCP=0
|
|
"off",
|
|
String.format("%.1f", mem),
|
|
status,
|
|
)
|
|
println(fields.joinToString("|"))
|
|
// fixed-latency 진단용 stderr diagnostic. stdout ROW 계약은 그대로 두고, transport별 p50/p99를
|
|
// grep 가능한 형태로 남겨 ws fixed latency 분석에서 Kotlin WS row를 쉽게 추출한다.
|
|
log(
|
|
"DIAG fixed-latency profile=$profile transport=$transport " +
|
|
"p50=${String.format("%.3f", p50)} p99=${String.format("%.3f", p99)}",
|
|
)
|
|
}
|
|
|
|
private fun emitSkip(profile: String, axis: String, transport: String, reason: String) {
|
|
println("SKIP|$profile|$axis|$LANGUAGE|$transport|$reason")
|
|
}
|
|
|
|
@Suppress("UNUSED_PARAMETER")
|
|
private fun deferredBottleneck(transport: String, profile: String, concurrency: Int?): String? {
|
|
if (profile == "slow-mix") {
|
|
return "deferred: optimize Epic — Kotlin slow-mix fires same-client requests concurrently, so response completion order is coroutine-scheduling dependent"
|
|
}
|
|
return null
|
|
}
|
|
|
|
// TCP parallel clients>=128 deferred skip은 제거됐다. TcpClient가 read 전용 dispatcher와 soTimeout
|
|
// 기반 read poll을 쓰도록 고쳐(read/write 스레드 풀 분리) thread starvation deadlock이 사라졌고,
|
|
// clients=128/512/1024 row를 실제로 측정한다. 남는 deferred는 deferredBottleneck(slow-mix)뿐이다.
|
|
@Suppress("UNUSED_PARAMETER")
|
|
private fun parallelDeadlockBottleneck(transport: String, clientCount: Int): String? = null
|
|
|
|
private fun summarize(
|
|
profile: String,
|
|
axis: String,
|
|
transport: String,
|
|
latencies: List<Double>,
|
|
elapsedMs: Double,
|
|
clientCount: Int,
|
|
s: StressStability,
|
|
mem: Double,
|
|
) {
|
|
val sorted = latencies.sorted()
|
|
val throughput = if (elapsedMs > 0) latencies.size / elapsedMs * 1000.0 else 0.0
|
|
emitRow(
|
|
profile, axis, transport, testDataPayloadBytes("req-0"), clientCount, latencies.size,
|
|
throughput, percentile(sorted, 50.0), percentile(sorted, 95.0), percentile(sorted, 99.0), mem, s,
|
|
)
|
|
}
|
|
|
|
private fun emitThroughput(
|
|
profile: String,
|
|
axis: String,
|
|
transport: String,
|
|
count: Int,
|
|
elapsedMs: Double,
|
|
clientCount: Int,
|
|
payloadBytes: Int,
|
|
s: StressStability,
|
|
mem: Double,
|
|
) {
|
|
val throughput = if (elapsedMs > 0) count / elapsedMs * 1000.0 else 0.0
|
|
emitRow(profile, axis, transport, payloadBytes, clientCount, count, throughput, 0.0, 0.0, 0.0, mem, s)
|
|
}
|
|
|
|
private fun startServer(transport: String, port: Int, onConnected: (Communicator) -> Unit): StressServerHandle {
|
|
val connected = Channel<Unit>(Channel.UNLIMITED)
|
|
when (transport) {
|
|
"tcp" -> {
|
|
val srv = TcpServer(HOST, port) { socket -> TcpClient.fromSocket(socket, 0, 0, parserMap()) }
|
|
srv.onClientConnected = { client ->
|
|
onConnected(client.communicator)
|
|
connected.trySend(Unit)
|
|
}
|
|
srv.start()
|
|
return StressServerHandle(srv.port(), { srv.stop() }, connected)
|
|
}
|
|
"ws" -> {
|
|
val srv = WsServer(HOST, port, WS_PATH) { conn -> WsClient.forServer(conn, 0, 0, parserMap()) }
|
|
srv.onClientConnected = { client ->
|
|
onConnected(client.communicator)
|
|
connected.trySend(Unit)
|
|
}
|
|
srv.start()
|
|
return StressServerHandle(srv.port(), { srv.stop() }, connected)
|
|
}
|
|
else -> throw IllegalArgumentException("unknown transport $transport")
|
|
}
|
|
}
|
|
|
|
// request-response 축에서는 client가 concurrency C로 동시에 send하므로 frame 도착 순서가 결정적이지
|
|
// 않다. 따라서 여기서는 FIFO를 측정하지 않고 응답 정확성/timeout만 본다. per-connection FIFO는 단일
|
|
// connection 순차 send인 burst 축에서 검증한다.
|
|
private fun startEchoServer(transport: String, port: Int): StressServerHandle =
|
|
startServer(transport, port) { comm ->
|
|
addRequestListenerTyped<TestData, TestData>(comm) { req ->
|
|
TestData.newBuilder().setIndex(req.index * 2).setMessage("echo:" + req.message).build()
|
|
}
|
|
}
|
|
|
|
private fun startSlowMixServer(transport: String, port: Int, slowEvery: Int, delayMs: Long): StressServerHandle =
|
|
startServer(transport, port) { comm ->
|
|
addRequestListenerTyped<TestData, TestData>(comm) { req ->
|
|
if ((req.index + 1) % slowEvery == 0) delay(delayMs)
|
|
TestData.newBuilder().setIndex(req.index * 2).setMessage("echo:" + req.message).build()
|
|
}
|
|
}
|
|
|
|
private suspend fun dial(transport: String, port: Int): StressClientHandle = when (transport) {
|
|
"tcp" -> {
|
|
val c = DialTcp(HOST, port, 0, 0, parserMap())
|
|
StressClientHandle(comm = c.communicator, close = { c.close() })
|
|
}
|
|
"ws" -> {
|
|
val c = DialWs(HOST, port, WS_PATH, 0, 0, parserMap())
|
|
StressClientHandle(comm = c.communicator, close = { c.close() }, wsMetrics = { c.getMetricsDiagnostics() })
|
|
}
|
|
else -> throw IllegalArgumentException("unknown transport $transport")
|
|
}
|
|
|
|
private suspend fun dialWithRetry(transport: String, port: Int): StressClientHandle {
|
|
val deadline = System.nanoTime() + 3_000_000_000L
|
|
var lastError: Throwable? = null
|
|
while (System.nanoTime() < deadline) {
|
|
try {
|
|
return dial(transport, port)
|
|
} catch (error: Throwable) {
|
|
lastError = error
|
|
withContext(Dispatchers.IO) { Thread.sleep(50) }
|
|
}
|
|
}
|
|
throw IllegalStateException("connect $transport:$port timed out: $lastError")
|
|
}
|
|
|
|
private suspend fun sendOne(comm: Communicator, index: Int, prefix: String, timeoutMs: Long, latencies: ConcurrentLinkedQueue<Double>, s: StressStability) {
|
|
val started = System.nanoTime()
|
|
try {
|
|
val res: TestData = com.tokilabs.proto_socket.sendRequestTyped(
|
|
comm, TestData.newBuilder().setIndex(index).setMessage("$prefix-$index").build(), timeoutMs,
|
|
)
|
|
latencies.add((System.nanoTime() - started) / 1e6)
|
|
if (res.index != index * 2 || res.message != "echo:$prefix-$index") {
|
|
s.nonceMismatch.incrementAndGet()
|
|
}
|
|
} catch (error: Throwable) {
|
|
classifyRequestError(error, s)
|
|
}
|
|
}
|
|
|
|
private suspend fun sendPayloadOne(
|
|
comm: Communicator,
|
|
index: Int,
|
|
filler: String,
|
|
expectedEcho: String,
|
|
timeoutMs: Long,
|
|
latencies: ConcurrentLinkedQueue<Double>,
|
|
s: StressStability,
|
|
) {
|
|
val started = System.nanoTime()
|
|
try {
|
|
val res: TestData = com.tokilabs.proto_socket.sendRequestTyped(
|
|
comm, TestData.newBuilder().setIndex(index).setMessage(filler).build(), timeoutMs,
|
|
)
|
|
latencies.add((System.nanoTime() - started) / 1e6)
|
|
if (res.index != index * 2 || res.message != expectedEcho) {
|
|
s.nonceMismatch.incrementAndGet()
|
|
}
|
|
} catch (error: Throwable) {
|
|
classifyRequestError(error, s)
|
|
}
|
|
}
|
|
|
|
private suspend fun sendSlowMixOne(
|
|
comm: Communicator,
|
|
clientId: Int,
|
|
index: Int,
|
|
timeoutMs: Long,
|
|
completedByClient: Array<AtomicInteger>,
|
|
latencies: ConcurrentLinkedQueue<Double>,
|
|
s: StressStability,
|
|
) {
|
|
val message = "sm-$clientId-$index"
|
|
val started = System.nanoTime()
|
|
try {
|
|
val res: TestData = com.tokilabs.proto_socket.sendRequestTyped(
|
|
comm, TestData.newBuilder().setIndex(index).setMessage(message).build(), timeoutMs,
|
|
)
|
|
latencies.add((System.nanoTime() - started) / 1e6)
|
|
if (res.index != index * 2 || res.message != "echo:$message") {
|
|
s.nonceMismatch.incrementAndGet()
|
|
}
|
|
if (index <= completedByClient[clientId].get()) {
|
|
s.fifoViolations.incrementAndGet()
|
|
}
|
|
completedByClient[clientId].set(index)
|
|
} catch (error: Throwable) {
|
|
classifyRequestError(error, s)
|
|
}
|
|
}
|
|
|
|
private suspend fun runRequestLoad(comm: Communicator, total: Int, concurrency: Int, timeoutMs: Long, s: StressStability): List<Double> {
|
|
val latencies = ConcurrentLinkedQueue<Double>()
|
|
var issued = 0
|
|
var nextIndex = 0
|
|
while (issued < total) {
|
|
val batchSize = minOf(concurrency, total - issued)
|
|
coroutineScope {
|
|
for (i in 0 until batchSize) {
|
|
val index = nextIndex++
|
|
launch(Dispatchers.Default) { sendOne(comm, index, "req", timeoutMs, latencies, s) }
|
|
}
|
|
}
|
|
issued += batchSize
|
|
}
|
|
return latencies.toList()
|
|
}
|
|
|
|
private suspend fun profileRoundtrip(transport: String, mode: String) {
|
|
val concurrencies = listOf(1, 16, 64, 256)
|
|
val batches = if (mode == "quick") 2 else 20
|
|
val timeoutMs = if (mode == "quick") 5_000L else 15_000L
|
|
log("[roundtrip] transport=$transport mode=$mode concurrencies=1,16,64,256 batches/level=$batches")
|
|
for (concurrency in concurrencies) {
|
|
val deferred = deferredBottleneck(transport, "roundtrip", concurrency)
|
|
if (deferred != null) {
|
|
emitSkip("roundtrip", "concurrency=$concurrency", transport, deferred)
|
|
continue
|
|
}
|
|
val s = StressStability()
|
|
val total = concurrency * batches
|
|
val srv = startEchoServer(transport, freePort())
|
|
try {
|
|
val handle = dialWithRetry(transport, srv.port)
|
|
srv.waitForConnections()
|
|
try {
|
|
val started = System.nanoTime()
|
|
val latencies = runRequestLoad(handle.comm, total, concurrency, timeoutMs, s)
|
|
val snap = handle.wsMetrics?.invoke()
|
|
val queueBacklog = snap?.get(0) ?: 0L
|
|
val gatewayBacklog = snap?.get(1) ?: 0L
|
|
if (snap != null) {
|
|
val copyMs = snap[2] / 1_000_000.0
|
|
log(
|
|
"[roundtrip] transport=$transport concurrency=$concurrency " +
|
|
"wsCopyMs=${String.format("%.3f", copyMs)} " +
|
|
"wsMaxStaged=${snap[0]} wsDropped=${snap[1]} " +
|
|
"wsReceived=${snap[3]}"
|
|
)
|
|
}
|
|
val sorted = latencies.sorted()
|
|
val elapsedMs = (System.nanoTime() - started) / 1e6
|
|
val throughput = if (elapsedMs > 0) latencies.size / elapsedMs * 1000.0 else 0.0
|
|
emitRow(
|
|
"roundtrip", "concurrency=$concurrency", transport,
|
|
testDataPayloadBytes("req-0"), 1, latencies.size,
|
|
throughput, percentile(sorted, 50.0), percentile(sorted, 95.0), percentile(sorted, 99.0),
|
|
memMb(), s, queueBacklog, gatewayBacklog,
|
|
)
|
|
} finally {
|
|
handle.close()
|
|
if (handle.comm.isAlive()) s.pendingLeak.incrementAndGet()
|
|
}
|
|
} catch (error: Throwable) {
|
|
s.timeouts.incrementAndGet()
|
|
log("[roundtrip] concurrency=$concurrency error=$error")
|
|
emitRow("roundtrip", "concurrency=$concurrency", transport, testDataPayloadBytes("req-0"), 1, 0, 0.0, 0.0, 0.0, 0.0, memMb(), s)
|
|
} finally {
|
|
stopServer(srv)
|
|
}
|
|
log("[roundtrip] transport=$transport concurrency=$concurrency done violations=${s.violations()}")
|
|
}
|
|
}
|
|
|
|
private suspend fun profileBurst(transport: String, mode: String) {
|
|
val counts = if (mode == "quick") listOf(200) else listOf(1000, 10000, 100000)
|
|
log("[burst] transport=$transport mode=$mode counts=$counts")
|
|
val deferred = deferredBottleneck(transport, "burst", null)
|
|
if (deferred != null) {
|
|
for (count in counts) emitSkip("burst", "count=$count", transport, deferred)
|
|
return
|
|
}
|
|
for (count in counts) {
|
|
val s = StressStability()
|
|
val received = AtomicInteger(0)
|
|
val lastIndex = AtomicInteger(-1)
|
|
val done = CompletableDeferred<Unit>()
|
|
val srv = startServer(transport, freePort()) { comm ->
|
|
addListenerTyped<TestData>(comm) { data ->
|
|
if (data.index <= lastIndex.get()) s.fifoViolations.incrementAndGet()
|
|
lastIndex.set(data.index)
|
|
if (received.incrementAndGet() >= count) done.complete(Unit)
|
|
}
|
|
}
|
|
var handle: StressClientHandle? = null
|
|
try {
|
|
handle = dialWithRetry(transport, srv.port)
|
|
srv.waitForConnections()
|
|
val started = System.nanoTime()
|
|
for (i in 0 until count) {
|
|
handle.comm.send(TestData.newBuilder().setIndex(i).setMessage("b-$i").build())
|
|
}
|
|
val timeoutMs = if (mode == "quick") 10_000L else 60_000L
|
|
if (withTimeoutOrNull(timeoutMs) { done.await() } == null) {
|
|
s.timeouts.incrementAndGet()
|
|
log("[burst] count=$count dispatch timeout received=${received.get()}")
|
|
}
|
|
val elapsed = (System.nanoTime() - started) / 1e6
|
|
if (received.get() != count) s.pendingLeak.incrementAndGet()
|
|
emitThroughput("burst", "count=$count", transport, count, elapsed, 1, testDataPayloadBytes("b-0"), s, memMb())
|
|
} catch (error: Throwable) {
|
|
s.timeouts.incrementAndGet()
|
|
log("[burst] count=$count error=$error")
|
|
emitThroughput("burst", "count=$count", transport, received.get(), 0.0, 1, testDataPayloadBytes("b-0"), s, memMb())
|
|
} finally {
|
|
handle?.let {
|
|
it.close()
|
|
if (it.comm.isAlive()) s.pendingLeak.incrementAndGet()
|
|
}
|
|
stopServer(srv)
|
|
}
|
|
log("[burst] transport=$transport count=$count received=${received.get()} violations=${s.violations()}")
|
|
}
|
|
}
|
|
|
|
private suspend fun profileSustained(transport: String, mode: String) {
|
|
val durationsMs = if (mode == "quick") listOf(2000L) else listOf(30000L, 300000L, 1800000L)
|
|
val concurrency = 16
|
|
val timeoutMs = 15_000L
|
|
log("[sustained] transport=$transport mode=$mode durations=$durationsMs ms concurrency=$concurrency")
|
|
val deferred = deferredBottleneck(transport, "sustained", concurrency)
|
|
if (deferred != null) {
|
|
for (durationMs in durationsMs) emitSkip("sustained", "duration=${durationMs}ms", transport, deferred)
|
|
return
|
|
}
|
|
for (durationMs in durationsMs) {
|
|
val s = StressStability()
|
|
var peak = memMb()
|
|
val srv = startEchoServer(transport, freePort())
|
|
try {
|
|
val handle = dialWithRetry(transport, srv.port)
|
|
srv.waitForConnections()
|
|
try {
|
|
val latencies = ConcurrentLinkedQueue<Double>()
|
|
var nextIndex = 0
|
|
val deadline = System.nanoTime() + durationMs * 1_000_000L
|
|
while (System.nanoTime() < deadline) {
|
|
coroutineScope {
|
|
for (i in 0 until concurrency) {
|
|
val index = nextIndex++
|
|
launch(Dispatchers.Default) { sendOne(handle.comm, index, "s", timeoutMs, latencies, s) }
|
|
}
|
|
}
|
|
val cur = memMb()
|
|
if (cur > peak) peak = cur
|
|
}
|
|
summarize("sustained", "duration=${durationMs}ms", transport, latencies.toList(), durationMs.toDouble(), 1, s, peak)
|
|
} finally {
|
|
handle.close()
|
|
if (handle.comm.isAlive()) s.pendingLeak.incrementAndGet()
|
|
}
|
|
} catch (error: Throwable) {
|
|
s.timeouts.incrementAndGet()
|
|
log("[sustained] duration=${durationMs}ms error=$error")
|
|
summarize("sustained", "duration=${durationMs}ms", transport, emptyList(), 0.0, 1, s, peak)
|
|
} finally {
|
|
stopServer(srv)
|
|
}
|
|
log("[sustained] transport=$transport duration=${durationMs}ms done peakMemMb=${String.format("%.1f", peak)} violations=${s.violations()}")
|
|
}
|
|
}
|
|
|
|
private suspend fun profileParallel(transport: String, mode: String) {
|
|
val clientCounts = if (mode == "quick") listOf(4) else listOf(16, 128, 512, 1024)
|
|
val perClient = if (mode == "quick") 50 else 500
|
|
val concurrency = 8
|
|
val timeoutMs = 15_000L
|
|
log("[parallel] transport=$transport mode=$mode clients=$clientCounts perClient=$perClient")
|
|
val deferred = deferredBottleneck(transport, "parallel", concurrency)
|
|
if (deferred != null) {
|
|
for (clientCount in clientCounts) emitSkip("parallel", "clients=$clientCount", transport, deferred)
|
|
return
|
|
}
|
|
for (clientCount in clientCounts) {
|
|
val parallelDeferred = parallelDeadlockBottleneck(transport, clientCount)
|
|
if (parallelDeferred != null) {
|
|
emitSkip("parallel", "clients=$clientCount", transport, parallelDeferred)
|
|
continue
|
|
}
|
|
val s = StressStability()
|
|
val srv = startEchoServer(transport, freePort())
|
|
val handles = mutableListOf<StressClientHandle>()
|
|
try {
|
|
for (i in 0 until clientCount) {
|
|
try {
|
|
handles.add(dialWithRetry(transport, srv.port))
|
|
} catch (error: Throwable) {
|
|
s.timeouts.incrementAndGet()
|
|
log("[parallel] clients=$clientCount dial error=$error")
|
|
}
|
|
}
|
|
srv.waitForConnections(handles.size)
|
|
val allLatencies = ConcurrentLinkedQueue<Double>()
|
|
val started = System.nanoTime()
|
|
coroutineScope {
|
|
for (handle in handles) {
|
|
launch(Dispatchers.Default) {
|
|
allLatencies.addAll(runRequestLoad(handle.comm, perClient, concurrency, timeoutMs, s))
|
|
}
|
|
}
|
|
}
|
|
summarize("parallel", "clients=$clientCount", transport, allLatencies.toList(), (System.nanoTime() - started) / 1e6, clientCount, s, memMb())
|
|
} finally {
|
|
for (handle in handles) {
|
|
handle.close()
|
|
if (handle.comm.isAlive()) s.pendingLeak.incrementAndGet()
|
|
}
|
|
stopServer(srv)
|
|
}
|
|
log("[parallel] transport=$transport clients=$clientCount done violations=${s.violations()}")
|
|
}
|
|
}
|
|
|
|
private suspend fun profileSlowMix(transport: String, mode: String) {
|
|
val clientCount = if (mode == "quick") 4 else 16
|
|
val perClient = if (mode == "quick") 20 else 200
|
|
val slowEvery = if (mode == "quick") 5 else 10
|
|
val delayMs = if (mode == "quick") 50L else 100L
|
|
val timeoutMs = if (mode == "quick") 10_000L else 30_000L
|
|
val axis = "clients=$clientCount,slowEvery=$slowEvery,delayMs=$delayMs"
|
|
log("[slow-mix] transport=$transport mode=$mode $axis perClient=$perClient")
|
|
val deferred = deferredBottleneck(transport, "slow-mix", perClient)
|
|
if (deferred != null) {
|
|
emitSkip("slow-mix", axis, transport, deferred)
|
|
return
|
|
}
|
|
|
|
val s = StressStability()
|
|
var peak = memMb()
|
|
val srv = startSlowMixServer(transport, freePort(), slowEvery, delayMs)
|
|
val handles = mutableListOf<Pair<Int, StressClientHandle>>()
|
|
val latencies = ConcurrentLinkedQueue<Double>()
|
|
var elapsedMs = 0.0
|
|
try {
|
|
for (clientId in 0 until clientCount) {
|
|
try {
|
|
handles.add(clientId to dialWithRetry(transport, srv.port))
|
|
} catch (error: Throwable) {
|
|
s.timeouts.incrementAndGet()
|
|
log("[slow-mix] client=$clientId dial error=$error")
|
|
}
|
|
}
|
|
srv.waitForConnections(handles.size)
|
|
val completedByClient = Array(clientCount) { AtomicInteger(-1) }
|
|
val started = System.nanoTime()
|
|
coroutineScope {
|
|
for ((clientId, handle) in handles) {
|
|
for (index in 0 until perClient) {
|
|
launch(Dispatchers.Default) {
|
|
sendSlowMixOne(handle.comm, clientId, index, timeoutMs, completedByClient, latencies, s)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
val cur = memMb()
|
|
if (cur > peak) peak = cur
|
|
elapsedMs = (System.nanoTime() - started) / 1e6
|
|
} finally {
|
|
for ((_, handle) in handles) {
|
|
handle.close()
|
|
if (handle.comm.isAlive()) s.pendingLeak.incrementAndGet()
|
|
}
|
|
stopServer(srv)
|
|
}
|
|
val sorted = latencies.toList().sorted()
|
|
val throughput = if (elapsedMs > 0) latencies.size / elapsedMs * 1000.0 else 0.0
|
|
emitRow(
|
|
"slow-mix", axis, transport, testDataPayloadBytes("sm-0-0"), clientCount, latencies.size,
|
|
throughput, percentile(sorted, 50.0), percentile(sorted, 95.0), percentile(sorted, 99.0), peak, s,
|
|
)
|
|
log("[slow-mix] transport=$transport clients=$clientCount done peakMemMb=${String.format("%.1f", peak)} violations=${s.violations()}")
|
|
}
|
|
|
|
// profilePayload는 payload size matrix를 측정한다. 동일 connection request-response를 1KB/64KB/1MB
|
|
// payload에서 돌리고, message field를 deterministic filler로 채워 실제 serialized payload bytes,
|
|
// payload별 p50/p95/p99 latency, throughput, used heap, stability counters를 결과 row에 남긴다.
|
|
private suspend fun profilePayload(transport: String, mode: String) {
|
|
val sizes = if (mode == "quick") PAYLOAD_SIZES_QUICK else PAYLOAD_SIZES_FULL
|
|
val concurrency = if (mode == "quick") 4 else 8
|
|
val requests = if (mode == "quick") 20 else 100
|
|
val timeoutMs = if (mode == "quick") 10_000L else 30_000L
|
|
log("[payload] transport=$transport mode=$mode sizes=${sizes.map(::payloadLabel)} concurrency=$concurrency requests/size=$requests")
|
|
val deferred = deferredBottleneck(transport, "payload", concurrency)
|
|
if (deferred != null) {
|
|
for (size in sizes) emitSkip("payload", "payload=${payloadLabel(size)}", transport, deferred)
|
|
return
|
|
}
|
|
for (size in sizes) {
|
|
val s = StressStability()
|
|
val filler = payloadFiller(size)
|
|
val expectedEcho = "echo:$filler"
|
|
val payloadBytes = testDataPayloadBytes(filler)
|
|
val axis = "payload=${payloadLabel(size)}"
|
|
var peak = memMb()
|
|
val srv = startEchoServer(transport, freePort())
|
|
try {
|
|
val handle = dialWithRetry(transport, srv.port)
|
|
srv.waitForConnections()
|
|
try {
|
|
val latencies = ConcurrentLinkedQueue<Double>()
|
|
var issued = 0
|
|
var nextIndex = 0
|
|
val started = System.nanoTime()
|
|
while (issued < requests) {
|
|
val batchSize = minOf(concurrency, requests - issued)
|
|
coroutineScope {
|
|
for (i in 0 until batchSize) {
|
|
val index = nextIndex++
|
|
launch(Dispatchers.Default) {
|
|
sendPayloadOne(handle.comm, index, filler, expectedEcho, timeoutMs, latencies, s)
|
|
}
|
|
}
|
|
}
|
|
issued += batchSize
|
|
val cur = memMb()
|
|
if (cur > peak) peak = cur
|
|
}
|
|
val elapsedMs = (System.nanoTime() - started) / 1e6
|
|
val sorted = latencies.toList().sorted()
|
|
val throughput = if (elapsedMs > 0) latencies.size / elapsedMs * 1000.0 else 0.0
|
|
val snap = handle.wsMetrics?.invoke()
|
|
val queueBacklog = snap?.get(0) ?: 0L
|
|
val gatewayBacklog = snap?.get(1) ?: 0L
|
|
if (snap != null) {
|
|
val copyMs = snap[2] / 1_000_000.0
|
|
log(
|
|
"[payload] transport=$transport size=${payloadLabel(size)} " +
|
|
"wsCopyMs=${String.format("%.3f", copyMs)} " +
|
|
"wsMaxStaged=${snap[0]} wsDropped=${snap[1]} " +
|
|
"wsReceived=${snap[3]} peakMemMb=${String.format("%.1f", peak)}"
|
|
)
|
|
}
|
|
emitRow(
|
|
"payload", axis, transport, payloadBytes, 1, latencies.size, throughput,
|
|
percentile(sorted, 50.0), percentile(sorted, 95.0), percentile(sorted, 99.0),
|
|
peak, s, queueBacklog, gatewayBacklog,
|
|
)
|
|
} finally {
|
|
handle.close()
|
|
if (handle.comm.isAlive()) s.pendingLeak.incrementAndGet()
|
|
}
|
|
} catch (error: Throwable) {
|
|
s.timeouts.incrementAndGet()
|
|
log("[payload] size=${payloadLabel(size)} error=$error")
|
|
emitRow("payload", axis, transport, payloadBytes, 1, 0, 0.0, 0.0, 0.0, 0.0, peak, s)
|
|
} finally {
|
|
stopServer(srv)
|
|
}
|
|
log("[payload] transport=$transport size=${payloadLabel(size)} bytes=$payloadBytes peakMemMb=${String.format("%.1f", peak)} violations=${s.violations()}")
|
|
}
|
|
}
|
|
|
|
private fun parseList(args: Array<String>, prefix: String, def: List<String>): List<String> {
|
|
for (arg in args) {
|
|
if (arg.startsWith(prefix)) {
|
|
val items = arg.substring(prefix.length).split(",").map { it.trim() }.filter { it.isNotEmpty() }
|
|
if (items.isNotEmpty()) return items
|
|
}
|
|
}
|
|
return def
|
|
}
|
|
|
|
fun main(args: Array<String>) = runBlocking {
|
|
var mode = "quick"
|
|
for (arg in args) {
|
|
if (arg == "--full" || arg == "--mode=full") mode = "full"
|
|
if (arg == "--quick" || arg == "--mode=quick") mode = "quick"
|
|
}
|
|
var transports = parseList(args, "--transport=", listOf("tcp", "ws"))
|
|
transports = parseList(args, "--transports=", transports)
|
|
var profiles = parseList(args, "--profiles=", ALL_PROFILES)
|
|
profiles = parseList(args, "--profile=", profiles)
|
|
|
|
log(
|
|
"INFO stress harness language=$LANGUAGE mode=$mode transports=${transports.joinToString(",")} " +
|
|
"profiles=${profiles.joinToString(",")} typeName=${typeNameOf<TestData>()}",
|
|
)
|
|
|
|
val runners: Map<String, suspend (String, String) -> Unit> = mapOf(
|
|
"roundtrip" to ::profileRoundtrip,
|
|
"burst" to ::profileBurst,
|
|
"sustained" to ::profileSustained,
|
|
"parallel" to ::profileParallel,
|
|
"slow-mix" to ::profileSlowMix,
|
|
"payload" to ::profilePayload,
|
|
)
|
|
|
|
for (transport in transports) {
|
|
if (transport != "tcp" && transport != "ws") {
|
|
emitSkip("all", "transport", transport, "unsupported transport for Kotlin same-language baseline")
|
|
continue
|
|
}
|
|
for (profile in profiles) {
|
|
val runner = runners[profile]
|
|
if (runner == null) {
|
|
emitSkip(profile, "profile", transport, "profile not part of Kotlin same-language baseline (gateway is TypeScript-specific)")
|
|
continue
|
|
}
|
|
runner(transport, mode)
|
|
}
|
|
}
|
|
|
|
val status = if (totalViolations.get() == 0L) "PASS" else "FAIL"
|
|
println(
|
|
"SUMMARY|status=$status|language=$LANGUAGE|mode=$mode|transports=${transports.joinToString(",")}|" +
|
|
"profiles=${profiles.joinToString(",")}|rows=${rowsEmitted.get()}|stability_violations=${totalViolations.get()}",
|
|
)
|
|
System.out.flush()
|
|
if (status != "PASS") exitProcess(1)
|
|
}
|