Reconnectable hangs on iOS platform

[webfrea-k]

Reconnectable starts correctly, but then hangs on the delay() call.

Expected Behavior

It should stay in delay for as many milliseconds as specified and then continue

Actual Behavior

It never continues after delay()

Steps to Reproduce

Configure reconnectable and specify delay(1000) in it, then restart the server :)

reconnectable(5) {
    delay(1000)
    true
}

Possible Solution

I believe the problem is in the coroutine scope inside internal suspend fun ReconnectableRSocket().
For delay() to work on native-mt, coroutine should start from the main scope.

.stateIn(CoroutineScope(Dispatchers.Main + job))  // only for native platform

Your Environment

Latest

• RSocket version(s) used: 1.12.0
• Other relevant libraries versions (eg. netty, ...): /
• Platform (eg. JVM version (javar -version) or Node version (node --version)): native
• OS and version (eg uname -a): /

[whyoleg]

hey, first question, is do you see any errors thrown, like InvalidMutabilityException ?
Can you try to change TCP transport from default implementation to smth like this

//instead of
//val transport = aSocket(SelectorManager()).tcp().clientTransport("0.0.0.0", 8000)

//do this
val selector = SelectorManager()
val transport = ClientTransport {
    aSocket(selector).tcp().clientTransport("0.0.0.0", 8000).connect()
}

[webfrea-k]

Thanks for the suggestion. I've tried it and it appears there is no difference. I get InvalidMutabilityException if I subscribe for the job completion (everything you try to do in here throws mutability exception).

 RSocket.job.invokeOnCompletion {
    //attempt to re-connect here throws invalid mutability exception
 }

iOS part gets into reconnectable once, but then stops. Let me know if you need more info.

[whyoleg]

Can you provide stacktrace for InvalidMutabilityException. Is it happened in your code or internally in rsocket-kotlin?
For now, in some places, there is still thread switch inside rsocket-kotlin, can be related to it.

[webfrea-k]

Sure, here is the stacktrace:

Mon, 19 Apr 2021 08:25:38 UTC ❌ APP: kotlin.native.concurrent.InvalidMutabilityException: mutation attempt of frozen my.namespace.sdk.APP@5040d6d8
    at 0   APP Demo                          0x000000010d4af1ff kfun:kotlin.Throwable#<init>(kotlin.String?){} + 95
    at 1   APP Demo                          0x000000010d4a777d kfun:kotlin.Exception#<init>(kotlin.String?){} + 93
    at 2   APP Demo                          0x000000010d4a79ed kfun:kotlin.RuntimeException#<init>(kotlin.String?){} + 93
    at 3   APP Demo                          0x000000010d4e111d kfun:kotlin.native.concurrent.InvalidMutabilityException#<init>(kotlin.String){} + 93
    at 4   APP Demo                          0x000000010d4e291f ThrowInvalidMutabilityException + 431
    at 5   APP Demo                          0x000000010d5d9f70 MutationCheck + 128
    at 6   APP Demo                          0x000000010c0c01a9 kfun:my.namespace.sdk.APP.<set-isInitialiseInProgress>#internal + 89
    at 7   APP Demo                          0x000000010c0c035e kfun:my.namespace.sdk.APP#finalizeInit(){} + 62
    at 8   APP Demo                          0x000000010c0b8b2b kfun:my.namespace.sdk.APP.$initializeCOROUTINE$108#invokeSuspend(kotlin.Result<kotlin.Any?>){}kotlin.Any? + 6187
    at 9   APP Demo                          0x000000010d4d2d40 kfun:kotlin.coroutines.native.internal.BaseContinuationImpl#resumeWith(kotlin.Result<kotlin.Any?>){} + 800
    at 10  APP Demo                          0x000000010d82decc kfun:io.ktor.util.pipeline.SuspendFunctionGun.resumeRootWith#internal + 972
    at 11  APP Demo                          0x000000010d82da6e kfun:io.ktor.util.pipeline.SuspendFunctionGun.loop#internal + 1854
    at 12  APP Demo                          0x000000010d830572 kfun:io.ktor.util.pipeline.SuspendFunctionGun.object-1.resumeWith#internal + 418
    at 13  APP Demo                          0x000000010d4d3031 kfun:kotlin.coroutines.native.internal.BaseContinuationImpl#resumeWith(kotlin.Result<kotlin.Any?>){} + 1553
    at 14  APP Demo                          0x000000010d82decc kfun:io.ktor.util.pipeline.SuspendFunctionGun.resumeRootWith#internal + 972
    at 15  APP Demo                          0x000000010d82da6e kfun:io.ktor.util.pipeline.SuspendFunctionGun.loop#internal + 1854
    at 16  APP Demo                          0x000000010d830572 kfun:io.ktor.util.pipeline.SuspendFunctionGun.object-1.resumeWith#internal + 418
    at 17  APP Demo                          0x000000010d4d3031 kfun:kotlin.coroutines.native.internal.BaseContinuationImpl#resumeWith(kotlin.Result<kotlin.Any?>){} + 1553
    at 18  APP Demo                          0x000000010d82decc kfun:io.ktor.util.pipeline.SuspendFunctionGun.resumeRootWith#internal + 972
    at 19  APP Demo                          0x000000010d82da6e kfun:io.ktor.util.pipeline.SuspendFunctionGun.loop#internal + 1854
    at 20  APP Demo                          0x000000010d830572 kfun:io.ktor.util.pipeline.SuspendFunctionGun.object-1.resumeWith#internal + 418
    at 21  APP Demo                          0x000000010d4d3031 kfun:kotlin.coroutines.native.internal.BaseContinuationImpl#resumeWith(kotlin.Result<kotlin.Any?>){} + 1553
    at 22  APP Demo                          0x000000010d82decc kfun:io.ktor.util.pipeline.SuspendFunctionGun.resumeRootWith#internal + 972
    at 23  APP Demo                          0x000000010d82da6e kfun:io.ktor.util.pipeline.SuspendFunctionGun.loop#internal + 1854
    at 24  APP Demo                          0x000000010d830572 kfun:io.ktor.util.pipeline.SuspendFunctionGun.object-1.resumeWith#internal + 418
    at 25  APP Demo                          0x000000010d4d3031 kfun:kotlin.coroutines.native.internal.BaseContinuationImpl#resumeWith(kotlin.Result<kotlin.Any?>){} + 1553
    at 26  APP Demo                          0x000000010d6b2262 kfun:kotlinx.coroutines.DispatchedTask#run(){} + 2866
    at 27  APP Demo                          0x000000010d6f4caa kfun:kotlinx.coroutines.DarwinMainDispatcher.dispatch$lambda-0#internal + 90
    at 28  APP Demo                          0x000000010d6f50b0 kfun:kotlinx.coroutines.DarwinMainDispatcher.$dispatch$lambda-0$FUNCTION_REFERENCE$44.invoke#internal + 64
    at 29  APP Demo                          0x000000010d6f5110 kfun:kotlinx.coroutines.DarwinMainDispatcher.$dispatch$lambda-0$FUNCTION_REFERENCE$44.$<bridge-UNN>invoke(){}#internal + 64

This happens when the function tries to call itself (when connection is disrupted).

In essence (I tried to implement my own reconnectable while default is not working on iOS):

private suspend fun connectDataSource() {
    val transport = aSocket(SelectorManager()).tcp().clientTransport2(remoteAddress = NetworkAddress(host, 8000))
    api.connection.value = api.connector(uniqueDeviceId).connect(transport)
    api.connection.value?.job?.invokeOnCompletion {
        CoroutineScope(dispatcher).launch {
            delay(1000)
            connectDataSource() //recursive call
         }
    }
}

[whyoleg]

Looks like it's just an issue on your side with K/N memory model. as you use API class somewhere where context switches. On current moment it can happen also using rsocket-kotlin (will be changed in future).
Do you have the same issue/error by using reconnectable ?

[webfrea-k]

No, if I use reconnectable builder, it does not throw any errors. It just never continues executing code after the delay. Should I prepare a reproducer branch?

[whyoleg]

@webfrea-k yes please, reproducer would be nice

[webfrea-k]

@whyoleg here is the reproducer: https://github.com/webfrea-k/rsocket-kotlin/tree/ios-reconnectable-reproducer

1. Run the TcpServerAppKt
2. Run the iosApp & trigger request stream
3. Stop TcpServerAppKt
4. Restart TcpServerAppKt after a couple of seconds

To match the behaviour with JVM, it should loop over "Reconnecting in 1 sec...." until the server is back online or max reconnect retries were reached.
Actual behaviour (iOS console):

[1, 1, 1]
[]
[]
Error was thrown, caught & ignored
Optional(kotlin.Unit)
nil
Reconnecting in 1 sec....

[whyoleg]

thx, will look on it later today

[whyoleg]

Hey, I've not found yet the root cause, but think, that it's because somehow/somewhere Main dispatcher of coroutines (which is used in your case on UI I think) is blocked.
workaround that works on my side is to wrap delay inside reconnectable in withContext(Dispatchers.Default) :)
So connector will be:

RSocketConnector {
  reconnectable(10) {
    withContext(Dispatchers.Default) { delay(1000) }
    true
  }
}

Can you try, if it works for you?
Or, another thing, may be it will be better to use rSocket from separate dispatcher? or Default, or created with newSingleThreadContext, IMO, it will be better, than use network from UI thread.

[webfrea-k]

You are right. I've been doing all network calls from Dispatchers.Main on iOS, because KTOR does the switching under the hood. I just assumed it should be the same for rSocket.
I've tried to pass default dispatcher like this:

val transport = tcpClientTransport(SelectorManager(Dispatchers.Default), host, 8000)

but this does not have any effect on reconnectable for me. The workaround with withContext(Dispatchers.Default) works perfectly. Big thanks! :) Should I close this issue or do you plan to change something in the reconnectable?

P.S. Do you plan to build any snapshot builds in the near future? It would be great to have latest changes and latest coroutine and ktor dependencies in the project :) Tnx!

[whyoleg]

Several points:

1. I just assumed it should be the same for rSocket. - we rely on ktor here for now, and Im just not sure, how ktor TCP on K/N works here, it will just work with buffers in current thread, but not sure about network part. But, reconnectable now starts in current thread with Dispatchers.Unconfined. Not sure about, will it be better to change it or leave it like this, will check.
2. I've tried to pass default dispatcher like this - SelectorManager on K/N just ignores provided coroutine context now. It uses one worker under the hood, so anyway it will work in another thread. Comparing to K/JVM where this parameter is using (yeah, some inconsistencies), it's again, how ktor works here. So, it's expected, that it has no effect
3. Let's leave an issue opened for now, I will investigate a little more about how we can improve user experience here
4. snapshot can be retrieved with repository https://oss.jfrog.org/oss-snapshot-local and version 0.13.0-SNAPSHOT, but for now master branch is broken because of rebase, will fix, and ping here, when latest snapshot of master will be published

[whyoleg]

Hey @webfrea-k, sorry for delay, CI was a little broken and unstable. Latest 0.13.0-SNAPSHOT is now available

[webfrea-k]

Hey @whyoleg, thanks for the snapshot version. I have tested it just now and reconnectable now works perfectly on both platforms. Closing the issue.