Use coroutines #277

JcMinarro · 2023-02-28T00:05:20Z

Convert Handler.handleUpdate() method into a suspend method. It allows calling coroutines method from dispatcher block code.
The ExecutorLooper use coroutines as well instead of a ThreadPoolExecutor that only runs a new thread.
Fix #251

JcMinarro · 2023-03-02T16:38:08Z

@seik @vjgarciag96 are you keeping maintaining this repo?
Could I have a review of this PR?

vjgarciag96 · 2023-03-04T11:49:05Z

@seik @vjgarciag96 are you keeping maintaining this repo?
Could I have a review of this PR?

@JcMinarro It's still maintained, but as you can see it has not been very active lately. I'll try and have a look at the code through the weekend. Thank you so much for the contribution, excited to finally see coroutines support in the library!

JcMinarro · 2023-03-04T12:32:21Z

Cool!!!
Thank you for the project, it is handy.
If you feel something needs to be changed, let me know 🙃

vjgarciag96

This is looking great! Thank you so much for the work. I added a few small comments, but nothing major. I think this version is very close to be ready to merge and ship 🚀

vjgarciag96 · 2023-03-05T17:37:14Z

telegram/src/main/kotlin/com/github/kotlintelegrambot/dispatcher/Dispatcher.kt

 ) {

    internal lateinit var bot: Bot

    private val commandHandlers = linkedSetOf<Handler>()
    private val errorHandlers = arrayListOf<ErrorHandler>()

-    @Volatile private var stopped = false
+    private val scope: CoroutineScope = CoroutineScope(ioDispatcher)
+    private var job: Job? = null


I think this one may need to be volatile to guarantee memory consistency. There's nothing preventing startCheckingUpdates and stopCheckingUpdates from being called from different threads, and both are accessing job.

Sure, let me update it

vjgarciag96 · 2023-03-05T17:42:44Z

telegram/src/main/kotlin/com/github/kotlintelegrambot/dispatcher/Dispatcher.kt


    internal fun startCheckingUpdates() {
-        stopped = false
-        updatesExecutor.execute { checkQueueUpdates() }
+        job = scope.launch { checkQueueUpdates() }


I wonder if we should protect here against multiple calls to startCheckingUpdates (without a call to stopCheckingUpdates) to avoid clients from using the library APIs incorrectly without noticing. But we weren't protecting against that with the previous implementation anyways, so that's out of scope for this PR.

I think the case of running twice the startCheckingUpdates wasn't handled previously, but now with the current implementation should be as easy as canceling the previous job if it isn't null.
Let me update it

vjgarciag96 · 2023-03-05T18:04:37Z

telegram/src/main/kotlin/com/github/kotlintelegrambot/dispatcher/Dispatcher.kt

    private val logLevel: LogLevel,
+    ioDispatcher: CoroutineDispatcher = Dispatchers.IO,


I wonder if we should be using Dispatchers.IO here. The dispatcher here is used to launch a single coroutine that will loop until stopped, checking for updates on a queue, and running functions provided by consumers of the library for each of those (in those functions consumers can decide whether they need to run in a different dispatcher or not). But we don't know whether those functions will be blocking IO operations so it doesn't feel right to use this Dispatcher here. The main downside I could see of this is reducing the potential for parallelism of Dispatchers.IO, which is global to the app, because we have one of its threads always busy. In practice, I'm not sure whether this will actually make a difference but it may for some use cases? Any thoughts on this? One alternative could be to just have a dispatcher backed by a singleThreadExecutor that we create only for this.

A bit unrelated. This also makes me wonder about next steps for better coroutine support, and feels like replacing BlockingQueue for a Channel could be a nice impovement, so we suspend while waiting for new updates rather than blocking. Also out of scope of this PR, but just bringing it up to get your thoughts about it.

Dispatcher is a bit different than Thread, A Dispatcher is more like a ThreadPool where coroutines are executed (Except the Dispatcher.Main on Android implementation that use the UI Thread), so it shouldnt be blocking IO Operations on the rest of the app. Even the coroutine we launch on this Dispatcher will be running indefinitely until the loop is stopped, on every iteration of our infinite loop we are "suspending" our logic and delegating to the dispatcher we can continue (our job has not been canceled) or if there are others coroutines waiting to be executed and need to be run before us. We achieve it by calling yield() method.

Apart from that, I had a doubt about using Dispatchers.Default or Dispatchers.IO. Reading the docs First one is used to run "hard jobs" that consume machine recurses and the second one to operations that will be interacting with IO Resources. In our case, I think the IO definition is closer to our implementation (We run IO calls on every iteration). Maybe, something we could do is run updates on Dispatchers.Default but switch to Dispatchers.IO when API calls are executed. For that would be better to improve the "Retrofit integration" to use coroutines internally, but I think it can be achieved in a separate PR if we decide it is needed.

Lastly, regarding the BlockingQueue, of course, we could use Channel instead, I didn't think about that, but it is a really good point

I will try to do separate PR using Channel instead of BlockingQueue, because If I change it here, lot of code will need to be updated (Tests are using fake blockingQueue that will need to be updated as well..)

Yes, right. I see the point of this code being "cooperative" by doing yield every time an item is received from the queue, so it's not going to be taking a thread forever. For now, it will only be blocking in queue.take but that will become suspending when we migrate to use Channel so I think this class is doing good.

But I don't think we run any IO operation in Dispatcher (we run them on Updater) so that's why I was thinking that maybe it would be nicer to use our own dispatcher backed by a single thread executor so that we don't take any of the resources from Dispatchers.IO here. There's this nice extension function to make our work easy https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/as-coroutine-dispatcher.html.

I agree it could make sense to use Dispatchers.IO on the Updater class, which is where we actually run the getUpdates API operation.

Re. Retrofit it might be nice to use the suspend version of the functions but as far as I understand OkHttp doesn't support non-blocking IO operations yet (square/okio#531). One alternative would be to use Ktor instead https://github.com/ktorio/ktor, but that would be a big task, so I guess we can do with executing blocking IO operations on Dispatchers.IO for now.

Ok, I see. Let me create our own Dispatcher and inject it, instead of injecting the Dispatcher.IO

Re Network Layer, I think it is out of the scope of this PR, it can be covered later

Yes, completely agree.

vjgarciag96 · 2023-03-05T18:07:05Z

telegram/src/main/kotlin/com/github/kotlintelegrambot/updater/Looper.kt

-internal class ExecutorLooper(
-    private val loopExecutor: Executor,
-) : Looper {
+internal class ExecutorLooper(ioDispatcher: CoroutineDispatcher) : Looper {


Not important. The name of this class and the comment seems a bit outdated now that we are using a CoroutineDispatcher rather than an Executor.

vjgarciag96 · 2023-03-05T18:09:58Z

telegram/src/main/kotlin/com/github/kotlintelegrambot/updater/Looper.kt

+    private val scope: CoroutineScope = CoroutineScope(ioDispatcher)
+    private var job: Job? = null


Similar comments to the ones in the Dispatcher class:

We may want to make job volatile for memory consistency because we can't guarantee loop and quit are going to be called from the same thead, and both are accessing job.

I also wonder about the Dispatcher we should be using here. In this case, this is actually used to run the API operations so it might make sense for it to be Dispatchers.IO.

I am thinking and, as far as we cancel the job just before a new job is started, we shouldn't have memory problems. A job can be called to cancel() multiple times.

The memory consistency point was about memory visibility, sorry I may have made it confusing.

As we don't know or control what thread a consumer is calling loop and quit from, we could have thread A calling loop and thread B calling quit. loop writes job and quit reads it, and without additional synchronization, it's not guaranteed that thread B will be able to see the value that thread A wrote to job. That would be solved by making job volatile. Does this make sense?

Ok, make sense. Let me update it.

vjgarciag96 · 2023-03-05T18:13:24Z

telegram/src/test/kotlin/com/github/kotlintelegrambot/updater/BoundLooper.kt

-        repeat(loopIterations) {
-            loopBody()
+    override fun loop(loopBody: suspend () -> Unit) {
+        scope.launch {


Is there a reason why we need to launch a coroutine here for some of the tests or could we keep this being executed in the caller context?

We can't because loopBody is a suspend function, but loop isn't, and suspend function can only be executed inside of a coroutine or from another suspend function.
For now, as far as it wasn't needed, I wanted to keep the Looper method signature but change its internal implementation. If it is needed, we can change it later.

Ah true, that makes sense. Thanks for the explanation.

vjgarciag96 · 2023-03-05T18:15:24Z

telegram/src/test/kotlin/com/github/kotlintelegrambot/updater/ExecutorLooperTest.kt

-    @Test
-    fun `loops until thread is interrupted`() {
-        var count = 0
+        println("JcLog: Start test $expectedCount")


I imagine this test was giving some problems with the new changes

Switching from Thread to Coroutines always gives problems in tests.... 🙃
BTW, this log had to be removed before I committed it, but I forget, let me remove it now.

vjgarciag96 · 2023-03-05T18:15:53Z

telegram/src/test/kotlin/com/github/kotlintelegrambot/updater/ExecutorLooperTest.kt

-    }
-
-    @Test
-    fun `loops until an exception is thrown`() {


Is this one not needed anymore? What happens when a Exception is thrown now?

When an exception is thrown, the job is stopped, so maybe we can keep this test as well, but I don't think it gives us a lot of value.
Let me see if I can keep it as well

Ahh I see, but the exception is still propagated because we are using launch?

JcMinarro

Thanks for the review.
Let me update the PR

JcMinarro · 2023-03-05T22:23:06Z

telegram/src/test/kotlin/com/github/kotlintelegrambot/updater/ExecutorLooperTest.kt

-    @Test
-    fun `loops until thread is interrupted`() {
-        var count = 0
+        println("JcLog: Start test $expectedCount")


Switching from Thread to Coroutines always gives problems in tests.... 🙃
BTW, this log had to be removed before I committed it, but I forget, let me remove it now.

JcMinarro · 2023-03-05T22:29:52Z

telegram/src/test/kotlin/com/github/kotlintelegrambot/updater/BoundLooper.kt

-        repeat(loopIterations) {
-            loopBody()
+    override fun loop(loopBody: suspend () -> Unit) {
+        scope.launch {


We can't because loopBody is a suspend function, but loop isn't, and suspend function can only be executed inside of a coroutine or from another suspend function.
For now, as far as it wasn't needed, I wanted to keep the Looper method signature but change its internal implementation. If it is needed, we can change it later.

JcMinarro · 2023-03-05T22:32:27Z

telegram/src/main/kotlin/com/github/kotlintelegrambot/dispatcher/Dispatcher.kt


    internal fun startCheckingUpdates() {
-        stopped = false
-        updatesExecutor.execute { checkQueueUpdates() }
+        job = scope.launch { checkQueueUpdates() }


I think the case of running twice the startCheckingUpdates wasn't handled previously, but now with the current implementation should be as easy as canceling the previous job if it isn't null.
Let me update it

JcMinarro · 2023-03-05T22:32:57Z

telegram/src/main/kotlin/com/github/kotlintelegrambot/dispatcher/Dispatcher.kt

 ) {

    internal lateinit var bot: Bot

    private val commandHandlers = linkedSetOf<Handler>()
    private val errorHandlers = arrayListOf<ErrorHandler>()

-    @Volatile private var stopped = false
+    private val scope: CoroutineScope = CoroutineScope(ioDispatcher)
+    private var job: Job? = null


Sure, let me update it

JcMinarro · 2023-03-05T23:08:03Z

telegram/src/main/kotlin/com/github/kotlintelegrambot/dispatcher/Dispatcher.kt

    private val logLevel: LogLevel,
+    ioDispatcher: CoroutineDispatcher = Dispatchers.IO,


Dispatcher is a bit different than Thread, A Dispatcher is more like a ThreadPool where coroutines are executed (Except the Dispatcher.Main on Android implementation that use the UI Thread), so it shouldnt be blocking IO Operations on the rest of the app. Even the coroutine we launch on this Dispatcher will be running indefinitely until the loop is stopped, on every iteration of our infinite loop we are "suspending" our logic and delegating to the dispatcher we can continue (our job has not been canceled) or if there are others coroutines waiting to be executed and need to be run before us. We achieve it by calling yield() method.

Apart from that, I had a doubt about using Dispatchers.Default or Dispatchers.IO. Reading the docs First one is used to run "hard jobs" that consume machine recurses and the second one to operations that will be interacting with IO Resources. In our case, I think the IO definition is closer to our implementation (We run IO calls on every iteration). Maybe, something we could do is run updates on Dispatchers.Default but switch to Dispatchers.IO when API calls are executed. For that would be better to improve the "Retrofit integration" to use coroutines internally, but I think it can be achieved in a separate PR if we decide it is needed.

Lastly, regarding the BlockingQueue, of course, we could use Channel instead, I didn't think about that, but it is a really good point

JcMinarro · 2023-03-05T23:18:00Z

telegram/src/main/kotlin/com/github/kotlintelegrambot/updater/Looper.kt

+    private val scope: CoroutineScope = CoroutineScope(ioDispatcher)
+    private var job: Job? = null


I am thinking and, as far as we cancel the job just before a new job is started, we shouldn't have memory problems. A job can be called to cancel() multiple times.

JcMinarro · 2023-03-05T23:22:59Z

telegram/src/test/kotlin/com/github/kotlintelegrambot/updater/ExecutorLooperTest.kt

-    }
-
-    @Test
-    fun `loops until an exception is thrown`() {


When an exception is thrown, the job is stopped, so maybe we can keep this test as well, but I don't think it gives us a lot of value.
Let me see if I can keep it as well

JcMinarro · 2023-03-05T23:31:18Z

telegram/src/main/kotlin/com/github/kotlintelegrambot/dispatcher/Dispatcher.kt

    private val logLevel: LogLevel,
+    ioDispatcher: CoroutineDispatcher = Dispatchers.IO,


I will try to do separate PR using Channel instead of BlockingQueue, because If I change it here, lot of code will need to be updated (Tests are using fake blockingQueue that will need to be updated as well..)

vjgarciag96

Great stuff, thanks for actioning the feedback! Once we resolve the discussion around the Dispatcher and the volatile we are good to go.

JcMinarro · 2023-03-07T09:18:14Z

Great stuff, thanks for actioning the feedback! Once we resolve the discussion around the Dispatcher and the volatile we are good to go.

Ok, I think it should be fine now ;)

vjgarciag96

LGTM! This is wonderful, thank you so much for the contribution again!

JinKoro · 2023-05-15T21:00:00Z

@vjgarciag96 When is the release of this task planned?

JcMinarro · 2023-05-15T21:16:30Z

@vjgarciag96 When is the release of this task planned?

I am not sure if there is any new release planned yet.
As far as this is on Jitpack, you can use a "commit SHA1" as a dependency version in your project.

dependencies {
  implementation 'com.github.kotlin-telegram-bot.kotlin-telegram-bot:dispatcher:f2c594d16c'
}

JcMinarro added 4 commits February 27, 2023 22:34

Add coroutines dependency

0f0aa97

Use coroutines to run Dispatcher updates

e422007

Make Handler.handleUpdate() a suspend method

656dc4c

Use coroutines within ExecutorLooper

4521b12

vjgarciag96 self-requested a review March 4, 2023 11:49

vjgarciag96 self-assigned this Mar 4, 2023

vjgarciag96 reviewed Mar 5, 2023

View reviewed changes

JcMinarro commented Mar 5, 2023

View reviewed changes

Solve PR's comments

87605f8

JcMinarro mentioned this pull request Mar 6, 2023

Use Channel instead of BlockingQueue #278

Merged

vjgarciag96 reviewed Mar 6, 2023

View reviewed changes

JcMinarro added 2 commits March 7, 2023 10:02

Use @Volatile annotation to avoid concurrency issues

d01c65a

Inject our own CoroutineDispatcher

ca8e4e4

vjgarciag96 approved these changes Mar 7, 2023

View reviewed changes

vjgarciag96 merged commit c41efdf into kotlin-telegram-bot:main Mar 7, 2023

JcMinarro deleted the use-coroutines branch March 7, 2023 09:30

		private val logLevel: LogLevel,
		ioDispatcher: CoroutineDispatcher = Dispatchers.IO,

		private val scope: CoroutineScope = CoroutineScope(ioDispatcher)
		private var job: Job? = null

Use coroutines #277

Use coroutines #277

Conversation

JcMinarro commented Feb 28, 2023 • edited

JcMinarro commented Mar 2, 2023

vjgarciag96 commented Mar 4, 2023 • edited

JcMinarro commented Mar 4, 2023

vjgarciag96 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vjgarciag96 Mar 5, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vjgarciag96 Mar 6, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JcMinarro left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vjgarciag96 left a comment

Choose a reason for hiding this comment

JcMinarro commented Mar 7, 2023

vjgarciag96 left a comment

Choose a reason for hiding this comment

JinKoro commented May 15, 2023

JcMinarro commented May 15, 2023

JcMinarro commented Feb 28, 2023 •

edited

vjgarciag96 commented Mar 4, 2023 •

edited

vjgarciag96 Mar 5, 2023 •

edited

vjgarciag96 Mar 6, 2023 •

edited