Provide a way to pass parent CoroutineScope to socket builders

[altavir]

Currently, raw socket builders like

aSocket(ActorSelectorManager(Dispatchers.IO)).tcp().serverTransport(port = rawSocketPort)

are breaking structured concurrency since each socket creates a new stand-alone Job, not connected to a parent coroutine. It causes problems when we need to automatically close a server socket on parent context failure or manual cancelation.

The solution is to add an optional parent scope parameter in one of the builders and possibly add an additional builder like CoroutineScope.aSocketInScope(), which automatically uses parent socket.

[darvld]

I've also run into this problem. However AFAIK the aSocket, the ActorSelectorManager and the tcp() calls belong to Ktor. So I guess this issue should be moved to their project.
My solution to this was to create a custom TcpConnection class, which takes a parent Job as an optional constructor parameter, and a custom implementation of ServerTransport, which passes its supervisor job into every new connection.

if to look on current master branch, it has a little different TCP transport creation:

rsocket-kotlin/rsocket-transport-ktor/src/commonMain/kotlin/io/rsocket/kotlin/transport/ktor/TcpServerTransport.kt

Lines 36 to 52 in ccfbd91

	public fun TcpServerTransport(
	selector: SelectorManager,
	localAddress: NetworkAddress? = null,
	configure: SocketOptions.AcceptorOptions.() -> Unit = {},
	): ServerTransport<Job> = ServerTransport { accept ->
	val serverSocket = aSocket(selector).tcp().bind(localAddress, configure)
	GlobalScope.launch(serverSocket.socketContext + Dispatchers.Unconfined + ignoreExceptionHandler, CoroutineStart.UNDISPATCHED) {
	supervisorScope {
	while (isActive) {
	val clientSocket = serverSocket.accept()
	val connection = TcpConnection(clientSocket)
	launch(start = CoroutineStart.UNDISPATCHED) { accept(connection) }
	}
	}
	}
	serverSocket.socketContext
	}

KTOR aSocket when starting server will create ServerSocket with Job in it, and closing it should also happen only by cancelling it. So providing CoroutineScope during TCP server creation makes no sense, you will anyway need to cancel parent serverSocket Job manually, as it's out of standard Job hierarchy.

So, to solve this issue, it's better to create an issue in https://youtrack.jetbrains.com/issues/KTOR about TCP implementation to support parent scope/job for aSocket

[altavir]

@olme04 The code explicitly uses GloablaScope thus violating structured concurrency. It must take a scope as a parameter. If it does, it is quite easy to add coroutineContext[Job]?.invokeOnCompletion{ serverSocket.close()} to close the socket. So I do not see how it is a Ktor problem.

@altavir yes, it uses GlobalScope, but it takes socket Job - socketContext type is Job, if you will provide another scope, it will not change anything, as Job will be overridden anyway.
On current moment ktor create ServerSocket object with fresh new Job:
https://github.com/ktorio/ktor/blob/2a836a54ef3293b5265cce58992d5cbc7cf9cc23/ktor-network/jvm/src/io/ktor/network/sockets/ServerSocketImpl.kt#L21
And as so, it's socketContext: Job should be root job for hierarchy.

Anyway, not sure, that here CoroutineScope should be provided from user. Starting server is an entry point of an application, and it's lifetime is represented as Job, which is returned after call of RSocketServer.bind.
But, some tweaking is possible.
Can you share a use case on why do you need, that starting server happen in some specific CoroutineScope?

[altavir]

The problem could be still mitigated on RSocket side, but I've created an issue: https://youtrack.jetbrains.com/issue/KTOR-2864.

The use case is rather simple. I want to control a lifetime of a socket by the external scope and close it automatically when the external scope is canceled or failed. In my case, I have a service that runs multiple protocols on multiple ports simultaneously and I want a unified way to close them all.

You can do it now using simple extension function like this (instead of RSocketServer.bind):

fun RSocketServer.bindInScope(scope: CoroutineScope, transport: ServerTransport<Job>): Job = bind(transport).also { serverJob ->
  scope.job.invokeOnCompletion { serverJob.cancel("some message", it) }
  //and if to cancel scope on server fail
  serverJob.invokeOnCompletion { scope.cancel("server closed", it) }
}

[yschimke]

@olme04 is that something we should make a part of the core code?

@yschimke IMO - no, I think, that it's more a specific use case then a general feature. such function is only needed for TCP Server transport now, as WS server transport is configured differently. Im still thinking, that without changes in KTOR we couldn't provide some general API here.

@altavir can you check if PR #178 will be enough for your use case?
TCP Server creation will look like this:

val transport = TcpServerTransport("0.0.0.0", 4444)
val server = RSocketServer()
val scope: CoroutineScope = TODO() //some scope in which to start server
//make sure not to run tcp server on Default dispatcher, as `SelectorManager` will block on thread to manage interactions, still TODO how to improve this place
server.bindIn(scope + Dispatchers.IO, transport) { 
  RSocketRequestHandler {
    //handlers here
  }
}.handlerJob.join() //awaiting server completion if needed

[altavir]

Yes. looks nice. With the corrections that it is probably context, no scope, but it is fine in any case.

it's CoroutineScope -

rsocket-kotlin/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/core/RSocketServer.kt

Lines 40 to 44 in 381d464

	public fun <T> bindIn(
	scope: CoroutineScope,
	transport: ServerTransport<T>,
	acceptor: ConnectionAcceptor,
	): T = with(transport) {

plus is shortcut from kx.coroutines - https://github.com/Kotlin/kotlinx.coroutines/blob/8c98180f177bbe4b26f1ed9685a9280fea648b9c/kotlinx-coroutines-core/common/src/CoroutineScope.kt#L95-L96

[altavir]

OK, for some reason I've missed that one can do the same for scopes.