How can I fire off a mainloop to a separate thread? I've been using oldschool `Thread` so far, but t...

# coroutines

[reactormonk]

How can I fire off a mainloop to a separate thread? I've been using oldschool Thread so far, but the Exception handling is atrocious

[reactormonk]

Ideally, I'd rewrite the underlying library to use proper coroutines, but alas

[Sam]

Can you clarify a bit? What do you mean by mainloop? Are we talking about Android?

[reactormonk]

Kinda, running the mainloop for a rabbitmq service

[reactormonk]

But yeah, the whole system runs on android

[Sam]

Can you share an example of what you tried with threads? It shouldn't be too hard to coroutinify 🤞

[reactormonk]

suspend fun createRPCServer(block: suspend com.rabbitmq.client.Channel.() -> RpcServer) {
            val channel = connection.createChannel()
            val rpcServer = block(channel)

            val thread = Thread {
                rpcServer.mainloop()
            }
            try {
                thread.start()
            } catch (e: Exception) {
                Timber.e(e)
            }
            connection.addShutdownListener {
                rpcServer.close()
                thread.interrupt()
            }
        }

[reactormonk]

The try/catch I added recently, after the Exception from the Thread ended up in Timbuktu

[uli]

Did you try launching a coroutine on Dispatchers.IO for the rpcServer.mainloop()?

[reactormonk]

Is that Dispatcher made for hugging threads?

[Sam]

Dispatchers.IO could work well 👍. Or, if you want a dedicated thread, you can use newSingleThreadContext(), which will make a dispatcher with its own thread.

[reactormonk]

Ok, I have no issues shoving it over to <http://Dispatchers.IO|Dispatchers.IO>, I was initially reluctant due to thread hugging

[uli]

it is made for things mostly waiting on IO. As rpcServer.mainloop will not supsend. It also keeps running on the same thread for it’s life time

[Sam]

👍 try this:

runInterruptible(<http://Dispatchers.IO|Dispatchers.IO>) {
  rpcServer.mainLoop()
}

[Sam]

It'll be cancellable, and it'll propagate exceptions correctly

[reactormonk]

suspend fun createRPCServer(block: suspend com.rabbitmq.client.Channel.() -> RpcServer) {
            val channel = connection.createChannel()
            val rpcServer = block(channel)

            val serverJob = Job()
            runInterruptible(<http://Dispatchers.IO|Dispatchers.IO> + serverJob) {
                rpcServer.mainloop()
            }
            connection.addShutdownListener {
                rpcServer.close()
                serverJob.cancel()
            }
        }

Something like this?

[Sam]

Passing a custom Job like that is generally not a good idea. In this case it means the mainLoop() won't react to cancellation any more.

[reactormonk]

How can I create a job that's still connected to the parent context, aka doesn't override it?

[Sam]

coroutineScope {
  val serverJob = launch {
    runInterruptible(<http://Dispatchers.IO|Dispatchers.IO>) {
      rpcServer.mainloop()
    }
  }

  connection.addShutdownListener {
    rpcServer.close()
    serverJob.cancel()
  }
}

[reactormonk]

Ah, yeah, forgot about coroutineScope to acquire the context

[reactormonk]

Thanks for the help

[uli]

Well, the original fun createRPCServer would return after starting the service. coroutineScope {} will suspend until the launched coroutine terminates. You might want to create a new scope to launch the service in

[Sam]

That's a good spot, the original code with the Thread { … } didn't actually include a way to wait for the mainLoop() to finish.

[Sam]

I had not noticed that.

[uli]

If you launch on a dedicated scope, createRPCServer would also not need to be suspend.
So either provide your own scope or take a scope as receiver:

fun CoroutineScope.createRPCServer(block: suspend com.rabbitmq.client.Channel.() -> RpcServer): Job {
  val serverJob = launch {
    runInterruptible(<http://Dispatchers.IO|Dispatchers.IO>) {
      rpcServer.mainloop()
    }
  }

  connection.addShutdownListener {
    rpcServer.close()
    serverJob.cancel()
  }

  return serverJob
}

[Zach Klippenstein (he/him) [MOD]]

IO dispatcher is probably fine for running this sort of thing on Android, but it’s probably not a good idea in general. The IO dispatcher is tuned for tasks that hold threads while doing IO operations for some time, but not for the entire life of the process. It has a maximum number of threads it will allocate, so theoretically holding one or more of these forever could limit normal users of the dispatcher. But on Android, I think you’re extremely unlikely to get close to that limit anyway, so it probably doesn’t matter.

[uli]

Yes if this is a concern go with newSingleThreadContext

[uli]

Be aware though that this has to be cleaned up if no longer used

[Oliver.O]

The IO dispatcher [...] has a maximum number of threads it will allocate, so theoretically holding one or more of these forever could limit normal users of the dispatcher.

Using Dispatchers.IO.limitedParallelism(...) makes these problems disappear. For details and references, see KTOR-6462.

[reactormonk]

Probably would be best to rewrite everything using coroutines and no more mainloop, but that'll take some time 😭