ReactiveSubscriber seems not following reactive rules

[whojes-toss]

hi everyone,
i read the issue here, and found that ReactiveSubscriber code is not following the rule 2.7.

I think this can cause problem (which i'm currently strugggling with), but i'm not sure there is another protection code.

(My problem is too hard to reproduce, so i'm investigating every single point..)

Describe the bug

What happened? What should have happened instead?

Provide a Reproducer

• If possible, please provide a small self-contained project (or even just a single file) where the issue reproduces.
• If you can't pinpoint the issue, please provide at least some project where this reproduces, for example, your production one. If you are not ready to show the project publicly, we are open to discussing the details privately.
• If you really can't provide any code, please do still open an issue. This may prompt other people to chime in with their reproducers.

[dkhalanskyjb]

Sorry, but after skimming through the code, I see no violations of 2.7. Could you elaborate? Under what specific circumstances can Publisher.asFlow behave incorrectly?

[whojes-toss]

Rule 2.7 implies that the subscription's request and close operations should not be executed concurrently, and in that respect, I understood that the above issue was prevented by using @Synchronized. I also thought that since the makeRequest/cancel function calls the subscription's request/cancel method and can be invoked externally, some sort of safeguard should be in place.

@Synchronized
private fun withSubscriptionLock(block: () -> Unit) {
    block()
}

fun makeRequest() {
    withSubscriptionLock { 
        subscription.request(requestSize)
    }
}

fun cancel() {
    withSubscriptionLock { 
        subscription.cancel()
    }
}

We handle tens of thousands of coroutines in a batch process, and very occasionally, we experience an issue where a coroutine gets stuck.

[dkhalanskyjb]

The class is private, and each instance is only used inside a single function call, which doesn't do any multithreaded processing.

[whojes-toss]

i'm sorry but i don't get it...

this code is it's only usage, and it's a suspend function, but how doesn't do multithreaded?

i wrote down this simple code, and running thread changes. ("pppp" has 10,000 entries.)
please let me know what i have missed, thank you in advance

[whojes-toss]

oh i think i get it..

[dkhalanskyjb]

Here's a simple example: https://pl.kotl.in/uoNXi1sTy

• Many threads are used.
• It's a suspend function.
• However, because execution is linear, no synchronization is needed to access the shared variable, access to it will always be serialized anyway.

but how doesn't do multithreaded

Let's look at it.

publisher.injectCoroutineContext(injectContext).subscribe(subscriber)
try {
    var consumed = 0L
    while (true) {
        val value = subscriber.takeNextOrNull() ?: break
        coroutineContext.ensureActive()
        collector.emit(value)
        if (++consumed == requestSize) {
            consumed = 0L
            subscriber.makeRequest()
        }
    }
} finally {
    subscriber.cancel()
}

First, the call to subscribe finishes. Then, the existing elements get processed until they are no more, and then, subscriber.makeRequest() gets called. When the channel is closed, the same function calls subscriber.cancel(), and it can't be calling makeRequest() at the same time. I don't see how any of these operations can happen in parallel.

[whojes-toss]

execution is linear and i was confused a little
thanks for your kindness

[dkhalanskyjb]

"Execution is linear" means that there are no operations happening in parallel. The code may switch between running on different threads, but two pieces of the function can not run at the same time.

Do you still believe that our implementation violates the rule 2.7, or can we close this issue?

[whojes-toss]

we can close this issue. it's quite clear to me now.