SergeychWorks/lyng
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects 1 Releases 21 Wiki Activity

Add JDBC and websocket/native follow-up changes

Browse Source
This commit is contained in:
Sergey Chernov 2026-04-16 23:35:15 +03:00
parent 8f66cd7680
commit 07cb5a519c
8 changed files with 580 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
bin/deploy_all
View File

@ -20,7 +20,7 @@
set -e
echo "publishing all artifacts"
echo
./gradlew publishToMavenLocal site:jsBrowserDistribution publish buildInstallablePlugin :lyng:linkReleaseExecutableLinuxX64 :lyng:installJvmDist --parallel #--no-configuration-cache
./gradlew publishToMavenLocal site:jsBrowserDistribution publish buildInstallablePlugin :lyng:linkReleaseExecutableLinuxX64 :lyng:installJvmDist --parallel --no-configuration-cache
#echo
#echo "Creating plugin"

5
lyng/build.gradle.kts
View File

@ -151,8 +151,13 @@ kotlin {
tasks.named<KotlinNativeTest>("linuxX64Test") {
dependsOn(tasks.named("linkDebugExecutableLinuxX64"))
dependsOn(tasks.named("linkReleaseExecutableLinuxX64"))
environment(
"LYNG_CLI_NATIVE_BIN",
layout.buildDirectory.file("bin/linuxX64/debugExecutable/lyng.kexe").get().asFile.absolutePath
)
environment(
"LYNG_CLI_NATIVE_RELEASE_BIN",
layout.buildDirectory.file("bin/linuxX64/releaseExecutable/lyng.kexe").get().asFile.absolutePath
)
}

3
lyng/src/commonMain/kotlin/Common.kt
View File

@ -119,8 +119,9 @@ internal class CliExecutionRuntime(
facade.call(handler)
}
}
session.cancelAndJoin()
session.cancel()
shutdownSystemNetEngine()
session.join()
}
fun shutdownBlocking() {

132
lyng/src/linuxTest/kotlin/net/sergeych/lyng_cli/CliWebSocketNativeRegressionTest.kt Normal file
View File

@ -0,0 +1,132 @@
/*
* Copyright 2026 Sergey S. Chernov real.sergeych@gmail.com
*
* 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 net.sergeych.lyng_cli
import kotlinx.cinterop.*
import okio.FileSystem
import okio.Path
import okio.Path.Companion.toPath
import platform.posix.O_CREAT
import platform.posix.O_TRUNC
import platform.posix.O_WRONLY
import platform.posix.SIGKILL
import platform.posix.SIGSEGV
import platform.posix._exit
import platform.posix.close
import platform.posix.dup2
import platform.posix.execvp
import platform.posix.fork
import platform.posix.getenv
import platform.posix.getpid
import platform.posix.kill
import platform.posix.open
import platform.posix.usleep
import platform.posix.waitpid
import kotlin.test.Test
import kotlin.test.assertFalse
import kotlin.test.assertTrue
@OptIn(ExperimentalForeignApi::class)
class CliWebSocketNativeRegressionTest {
@Test
fun releaseCliDoesNotSegfaultOnConcurrentWebSocketClients() {
val executable = getenv("LYNG_CLI_NATIVE_RELEASE_BIN")?.toKString()
?: error("LYNG_CLI_NATIVE_RELEASE_BIN is not set")
val fs = FileSystem.SYSTEM
val repoRoot = ascend(executable.toPath(), 6)
val scriptPath = repoRoot / "bugs" / "ws-segfault.lyng"
check(fs.exists(scriptPath)) { "bug repro script not found at $scriptPath" }
val tempDir = "/tmp/lyng_ws_native_${getpid()}_${kotlin.random.Random.nextInt()}".toPath()
val stdoutPath = tempDir / "stdout.txt"
val stderrPath = tempDir / "stderr.txt"
fs.createDirectories(tempDir)
try {
val pid = launchCli(executable, scriptPath, stdoutPath, stderrPath)
usleep(5_000_000u)
if (kill(pid, 0) == 0) {
kill(pid, SIGKILL)
}
val status = waitForPid(pid)
val termSignal = status and 0x7f
val stdout = readUtf8IfExists(fs, stdoutPath)
val stderr = readUtf8IfExists(fs, stderrPath)
val allOutput = "$stdout\n$stderr"
assertFalse(termSignal == SIGSEGV, "native CLI crashed with SIGSEGV. Output:\n$allOutput")
assertTrue(
stdout.lineSequence().count { it == "test send to ws://127.0.0.1:9998... OK" } == 2,
"expected both websocket clients to finish. Output:\n$allOutput"
)
assertFalse(allOutput.contains("Segmentation fault"), "process output reported a segmentation fault:\n$allOutput")
} finally {
fs.deleteRecursively(tempDir, mustExist = false)
}
}
private fun ascend(path: Path, levels: Int): Path {
var current = path
repeat(levels) {
current = current.parent ?: error("cannot ascend $levels levels from $path")
}
return current
}
private fun readUtf8IfExists(fs: FileSystem, path: Path): String {
return if (fs.exists(path)) fs.read(path) { readUtf8() } else ""
}
private fun waitForPid(pid: Int): Int = memScoped {
val status = alloc<IntVar>()
val waited = waitpid(pid, status.ptr, 0)
check(waited == pid) { "waitpid failed for $pid" }
status.value
}
private fun launchCli(
executable: String,
scriptPath: Path,
stdoutPath: Path,
stderrPath: Path,
): Int = memScoped {
val pid = fork()
check(pid >= 0) { "fork failed" }
if (pid == 0) {
val stdoutFd = open(stdoutPath.toString(), O_WRONLY or O_CREAT or O_TRUNC, 0x1A4)
val stderrFd = open(stderrPath.toString(), O_WRONLY or O_CREAT or O_TRUNC, 0x1A4)
if (stdoutFd < 0 || stderrFd < 0) {
_exit(2)
}
dup2(stdoutFd, 1)
dup2(stderrFd, 2)
close(stdoutFd)
close(stderrFd)
val argv = allocArray<CPointerVar<ByteVar>>(3)
argv[0] = executable.cstr.ptr
argv[1] = scriptPath.toString().cstr.ptr
argv[2] = null
execvp(executable, argv)
_exit(127)
}
pid
}
}

51
lyngio/src/commonMain/kotlin/net/sergeych/lyngio/ws/KtorWsEngine.kt
View File

@ -17,35 +17,46 @@ import kotlinx.coroutines.channels.ClosedReceiveChannelException
internal fun createKtorWsEngine(
engineFactory: HttpClientEngineFactory<HttpClientEngineConfig>,
): LyngWsEngine = KtorLyngWsEngine(engineFactory)
shareClient: Boolean = true,
): LyngWsEngine = KtorLyngWsEngine(engineFactory, shareClient)
private class KtorLyngWsEngine(
engineFactory: HttpClientEngineFactory<HttpClientEngineConfig>,
private val shareClient: Boolean,
) : LyngWsEngine {
private val clientResult = runCatching {
private val clientFactory: () -> HttpClient = {
HttpClient(engineFactory) {
install(WebSockets)
}
}
private val sharedClientResult = if (shareClient) runCatching { clientFactory() } else null
private val supportProbe = if (!shareClient) runCatching { clientFactory().close() } else null
override val isSupported: Boolean
get() = clientResult.isSuccess
get() = sharedClientResult?.isSuccess ?: supportProbe?.isSuccess ?: true
override suspend fun connect(url: String, headers: Map<String, String>): LyngWsSession {
val client = clientResult.getOrElse {
val client = (sharedClientResult?.getOrElse {
throw UnsupportedOperationException(it.message ?: "WebSocket client is not supported")
}
} ?: runCatching { clientFactory() }.getOrElse {
throw UnsupportedOperationException(it.message ?: "WebSocket client is not supported")
})
val session = client.webSocketSession {
url(url)
headers.forEach { (name, value) -> header(name, value) }
}
return KtorLyngWsSession(url, session)
return KtorLyngWsSession(
targetUrl = url,
session = session,
ownedClient = client.takeUnless { shareClient },
)
}
}
private class KtorLyngWsSession(
private val targetUrl: String,
private val session: DefaultWebSocketSession,
private val ownedClient: HttpClient? = null,
) : LyngWsSession {
private var closed = false
@ -55,12 +66,22 @@ private class KtorLyngWsSession(
override suspend fun sendText(text: String) {
ensureOpen()
try {
session.send(text)
} catch (e: Throwable) {
release()
throw e
}
}
override suspend fun sendBytes(data: ByteArray) {
ensureOpen()
try {
session.send(data)
} catch (e: Throwable) {
release()
throw e
}
}
override suspend fun receive(): LyngWsMessage? {
@ -68,14 +89,17 @@ private class KtorLyngWsSession(
val frame = try {
session.incoming.receive()
} catch (_: ClosedReceiveChannelException) {
closed = true
release()
return null
} catch (e: Throwable) {
release()
throw e
}
return when (frame) {
is Frame.Text -> LyngWsMessage(isText = true, text = frame.readText())
is Frame.Binary -> LyngWsMessage(isText = false, data = frame.data.copyOf())
is Frame.Close -> {
closed = true
release()
null
}
else -> receive()
@ -84,11 +108,20 @@ private class KtorLyngWsSession(
override suspend fun close(code: Int, reason: String) {
if (closed) return
closed = true
try {
session.close(CloseReason(code.toShort(), reason))
} finally {
release()
}
}
private fun ensureOpen() {
if (closed) throw IllegalStateException("websocket session is closed")
}
private fun release() {
if (closed) return
closed = true
ownedClient?.close()
}
}

392
lyngio/src/commonMain/kotlin/net/sergeych/lyngio/ws/SocketWsEngine.kt Normal file
View File

@ -0,0 +1,392 @@
package net.sergeych.lyngio.ws
import io.ktor.http.Url
import io.ktor.http.URLProtocol
import net.sergeych.lyngio.net.LyngNetEngine
import net.sergeych.lyngio.net.LyngTcpSocket
import net.sergeych.lyngio.net.getSystemNetEngine
import net.sergeych.mp_tools.encodeToBase64
import kotlin.random.Random
internal fun createSocketWsEngine(
secureFallback: LyngWsEngine = UnsupportedLyngWsEngine,
): LyngWsEngine = SocketLyngWsEngine(getSystemNetEngine(), secureFallback)
private class SocketLyngWsEngine(
private val netEngine: LyngNetEngine,
private val secureFallback: LyngWsEngine,
) : LyngWsEngine {
override val isSupported: Boolean
get() = (netEngine.isSupported && netEngine.isTcpAvailable) || secureFallback.isSupported
override suspend fun connect(url: String, headers: Map<String, String>): LyngWsSession {
val parsedUrl = Url(url)
return when (parsedUrl.protocol.name.lowercase()) {
URLProtocol.WS.name.lowercase() -> connectPlain(parsedUrl, headers)
URLProtocol.WSS.name.lowercase() -> secureFallback.connect(url, headers)
else -> throw UnsupportedOperationException("Unsupported websocket scheme: ${parsedUrl.protocol.name}")
}
}
private suspend fun connectPlain(url: Url, headers: Map<String, String>): LyngWsSession {
if (!netEngine.isSupported || !netEngine.isTcpAvailable) {
throw UnsupportedOperationException("WebSocket client is not supported on this runtime")
}
val socket = netEngine.tcpConnect(url.host, url.port, timeoutMillis = null, noDelay = true)
try {
val key = randomBytes(16).encodeToBase64()
val requestPath = buildRequestPath(url)
val hostHeader = buildHostHeader(url)
val request = buildString {
append("GET ").append(requestPath).append(" HTTP/1.1\r\n")
append("Host: ").append(hostHeader).append("\r\n")
append("Upgrade: websocket\r\n")
append("Connection: Upgrade\r\n")
append("Sec-WebSocket-Key: ").append(key).append("\r\n")
append("Sec-WebSocket-Version: 13\r\n")
headers.forEach { (name, value) ->
append(name).append(": ").append(value).append("\r\n")
}
append("\r\n")
}
socket.writeUtf8(request)
socket.flush()
validateHandshake(socket, key)
return SocketLyngWsSession(url.toString(), socket)
} catch (e: Throwable) {
socket.close()
throw e
}
}
}
private class SocketLyngWsSession(
private val targetUrl: String,
private val socket: LyngTcpSocket,
) : LyngWsSession {
private var closed = false
private var closeSent = false
private var pending = ByteArray(0)
private var fragmentedOpcode: Int? = null
private var fragmentedPayload = ByteArray(0)
override fun isOpen(): Boolean = !closed && socket.isOpen()
override fun url(): String = targetUrl
override suspend fun sendText(text: String) {
ensureOpen()
sendFrame(OPCODE_TEXT, text.encodeToByteArray())
}
override suspend fun sendBytes(data: ByteArray) {
ensureOpen()
sendFrame(OPCODE_BINARY, data)
}
override suspend fun receive(): LyngWsMessage? {
if (closed) return null
while (true) {
val frame = readFrame() ?: run {
release()
return null
}
when (frame.opcode) {
OPCODE_CONTINUATION -> {
val opcode = fragmentedOpcode ?: throw IllegalStateException("unexpected websocket continuation frame")
fragmentedPayload += frame.payload
if (frame.fin) {
val payload = fragmentedPayload
fragmentedOpcode = null
fragmentedPayload = ByteArray(0)
return payload.toMessage(opcode)
}
}
OPCODE_TEXT, OPCODE_BINARY -> {
if (frame.fin) return frame.payload.toMessage(frame.opcode)
fragmentedOpcode = frame.opcode
fragmentedPayload = frame.payload
}
OPCODE_CLOSE -> {
if (!closeSent) {
sendFrame(OPCODE_CLOSE, frame.payload)
closeSent = true
}
release()
return null
}
OPCODE_PING -> sendFrame(OPCODE_PONG, frame.payload)
OPCODE_PONG -> Unit
else -> Unit
}
}
}
override suspend fun close(code: Int, reason: String) {
if (closed) return
val reasonBytes = reason.encodeToByteArray()
val payload = ByteArray(reasonBytes.size + 2)
payload[0] = (code shr 8).toByte()
payload[1] = code.toByte()
reasonBytes.copyInto(payload, destinationOffset = 2)
try {
if (!closeSent) {
sendFrame(OPCODE_CLOSE, payload)
closeSent = true
}
} finally {
release()
}
}
private suspend fun sendFrame(opcode: Int, payload: ByteArray) {
ensureOpen()
val header = buildFrameHeader(opcode, payload.size, masked = true)
val mask = randomBytes(4)
val maskedPayload = payload.copyOf()
maskedPayload.indices.forEach { index ->
maskedPayload[index] = (maskedPayload[index].toInt() xor mask[index % mask.size].toInt()).toByte()
}
socket.write(header + mask + maskedPayload)
socket.flush()
}
private suspend fun readFrame(): WsFrame? {
val head = readExact(2) ?: return null
val fin = (head[0].toInt() and 0x80) != 0
val opcode = head[0].toInt() and 0x0f
val masked = (head[1].toInt() and 0x80) != 0
val payloadLength = when (val lengthCode = head[1].toInt() and 0x7f) {
126 -> {
val extended = readExact(2) ?: return null
((extended[0].toInt() and 0xff) shl 8) or (extended[1].toInt() and 0xff)
}
127 -> {
val extended = readExact(8) ?: return null
var acc = 0L
extended.forEach { byte ->
acc = (acc shl 8) or (byte.toInt() and 0xff).toLong()
}
require(acc <= Int.MAX_VALUE.toLong()) { "websocket frame is too large" }
acc.toInt()
}
else -> lengthCode
}
val mask = if (masked) readExact(4) ?: return null else null
val payload = if (payloadLength > 0) readExact(payloadLength) ?: return null else ByteArray(0)
if (mask != null) {
payload.indices.forEach { index ->
payload[index] = (payload[index].toInt() xor mask[index % mask.size].toInt()).toByte()
}
}
return WsFrame(fin = fin, opcode = opcode, payload = payload)
}
private suspend fun readExact(byteCount: Int): ByteArray? {
while (pending.size < byteCount) {
val chunk = socket.read(maxOf(4096, byteCount - pending.size)) ?: break
if (chunk.isEmpty()) break
pending += chunk
}
if (pending.size < byteCount) return null
val result = pending.copyOfRange(0, byteCount)
pending = pending.copyOfRange(byteCount, pending.size)
return result
}
private fun release() {
if (closed) return
closed = true
socket.close()
}
private fun ensureOpen() {
if (closed || !socket.isOpen()) throw IllegalStateException("websocket session is closed")
}
}
private data class WsFrame(
val fin: Boolean,
val opcode: Int,
val payload: ByteArray,
)
private suspend fun validateHandshake(socket: LyngTcpSocket, key: String) {
val statusLine = socket.readLine() ?: error("websocket handshake failed: missing response status")
require(statusLine.startsWith("HTTP/1.1 101") || statusLine.startsWith("HTTP/1.0 101")) {
"websocket handshake failed: $statusLine"
}
val headers = linkedMapOf<String, String>()
while (true) {
val line = socket.readLine() ?: error("websocket handshake failed: unexpected EOF")
if (line.isBlank()) break
val colonAt = line.indexOf(':')
require(colonAt > 0) { "invalid websocket header: $line" }
val name = line.substring(0, colonAt).trim().lowercase()
val value = line.substring(colonAt + 1).trim()
headers[name] = value
}
require(headers["upgrade"]?.lowercase() == "websocket") { "websocket handshake failed: missing Upgrade header" }
require(headers["connection"]?.lowercase()?.contains("upgrade") == true) {
"websocket handshake failed: missing Connection header"
}
require(headers["sec-websocket-accept"] == websocketAcceptKey(key)) {
"websocket handshake failed: invalid Sec-WebSocket-Accept"
}
}
private fun websocketAcceptKey(key: String): String =
sha1((key + WS_GUID).encodeToByteArray()).encodeToBase64()
private fun buildRequestPath(url: Url): String {
val path = url.encodedPath.ifEmpty { "/" }
val query = url.encodedQuery
return if (query.isBlank()) path else "$path?$query"
}
private fun buildHostHeader(url: Url): String {
val host = if (':' in url.host && !url.host.startsWith("[")) "[${url.host}]" else url.host
return if (url.port == url.protocol.defaultPort) host else "$host:${url.port}"
}
private fun buildFrameHeader(opcode: Int, payloadSize: Int, masked: Boolean): ByteArray {
require(payloadSize >= 0) { "payload size must be non-negative" }
val firstByte = (0x80 or (opcode and 0x0f)).toByte()
val maskBit = if (masked) 0x80 else 0
return when {
payloadSize <= 125 -> byteArrayOf(firstByte, (maskBit or payloadSize).toByte())
payloadSize <= 0xffff -> byteArrayOf(
firstByte,
(maskBit or 126).toByte(),
((payloadSize ushr 8) and 0xff).toByte(),
(payloadSize and 0xff).toByte(),
)
else -> byteArrayOf(
firstByte,
(maskBit or 127).toByte(),
0,
0,
0,
0,
((payloadSize ushr 24) and 0xff).toByte(),
((payloadSize ushr 16) and 0xff).toByte(),
((payloadSize ushr 8) and 0xff).toByte(),
(payloadSize and 0xff).toByte(),
)
}
}
private fun ByteArray.toMessage(opcode: Int): LyngWsMessage = when (opcode) {
OPCODE_TEXT -> LyngWsMessage(isText = true, text = decodeToString())
OPCODE_BINARY -> LyngWsMessage(isText = false, data = copyOf())
else -> throw IllegalStateException("unsupported websocket opcode: $opcode")
}
private fun randomBytes(size: Int): ByteArray = ByteArray(size).also(Random.Default::nextBytes)
private fun sha1(input: ByteArray): ByteArray {
var h0 = 0x67452301
var h1 = 0xEFCDAB89.toInt()
var h2 = 0x98BADCFE.toInt()
var h3 = 0x10325476
var h4 = 0xC3D2E1F0.toInt()
val msgLen = input.size
val bitLen = msgLen.toLong() * 8L
val totalLen = ((msgLen + 1 + 8 + 63) / 64) * 64
val padded = ByteArray(totalLen).also { buf ->
input.copyInto(buf)
buf[msgLen] = 0x80.toByte()
for (i in 0..7) {
buf[totalLen - 8 + i] = ((bitLen ushr (56 - i * 8)) and 0xff).toByte()
}
}
val words = IntArray(80)
var blockStart = 0
while (blockStart < padded.size) {
for (i in 0..15) {
val off = blockStart + i * 4
words[i] = ((padded[off].toInt() and 0xff) shl 24) or
((padded[off + 1].toInt() and 0xff) shl 16) or
((padded[off + 2].toInt() and 0xff) shl 8) or
(padded[off + 3].toInt() and 0xff)
}
for (i in 16..79) {
val mixed = words[i - 3] xor words[i - 8] xor words[i - 14] xor words[i - 16]
words[i] = (mixed shl 1) or (mixed ushr 31)
}
var a = h0
var b = h1
var c = h2
var d = h3
var e = h4
for (i in 0..19) {
val f = (b and c) or (b.inv() and d)
val temp = ((a shl 5) or (a ushr 27)) + f + e + 0x5A827999 + words[i]
e = d
d = c
c = (b shl 30) or (b ushr 2)
b = a
a = temp
}
for (i in 20..39) {
val f = b xor c xor d
val temp = ((a shl 5) or (a ushr 27)) + f + e + 0x6ED9EBA1 + words[i]
e = d
d = c
c = (b shl 30) or (b ushr 2)
b = a
a = temp
}
for (i in 40..59) {
val f = (b and c) or (b and d) or (c and d)
val temp = ((a shl 5) or (a ushr 27)) + f + e + 0x8F1BBCDC.toInt() + words[i]
e = d
d = c
c = (b shl 30) or (b ushr 2)
b = a
a = temp
}
for (i in 60..79) {
val f = b xor c xor d
val temp = ((a shl 5) or (a ushr 27)) + f + e + 0xCA62C1D6.toInt() + words[i]
e = d
d = c
c = (b shl 30) or (b ushr 2)
b = a
a = temp
}
h0 += a
h1 += b
h2 += c
h3 += d
h4 += e
blockStart += 64
}
return ByteArray(20).also { out ->
fun putInt(offset: Int, value: Int) {
out[offset] = (value ushr 24).toByte()
out[offset + 1] = (value ushr 16).toByte()
out[offset + 2] = (value ushr 8).toByte()
out[offset + 3] = value.toByte()
}
putInt(0, h0)
putInt(4, h1)
putInt(8, h2)
putInt(12, h3)
putInt(16, h4)
}
}
private const val WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
private const val OPCODE_CONTINUATION = 0x0
private const val OPCODE_TEXT = 0x1
private const val OPCODE_BINARY = 0x2
private const val OPCODE_CLOSE = 0x8
private const val OPCODE_PING = 0x9
private const val OPCODE_PONG = 0xA

3
lyngio/src/linuxMain/kotlin/net/sergeych/lyngio/ws/PlatformLinux.kt
View File

@ -2,4 +2,5 @@ package net.sergeych.lyngio.ws
import io.ktor.client.engine.curl.Curl
actual fun getSystemWsEngine(): LyngWsEngine = createKtorWsEngine(Curl)
actual fun getSystemWsEngine(): LyngWsEngine =
createSocketWsEngine(secureFallback = createKtorWsEngine(Curl, shareClient = false))

2
lynglib/src/nativeMain/kotlin/net/sergeych/lyng/PerfDefaults.native.kt
View File

@ -23,7 +23,7 @@ actual object PerfDefaults {
actual val ARG_BUILDER: Boolean = true
actual val SKIP_ARGS_ON_NULL_RECEIVER: Boolean = true
actual val SCOPE_POOL: Boolean = true
actual val SCOPE_POOL: Boolean = false
actual val FIELD_PIC: Boolean = true
actual val METHOD_PIC: Boolean = true