Outbound flow control bugfix

CHANGELOG.md

@@ -1,9 +1,9 @@
-## Version 0.4.1
+## Version 0.5.0-RC
 _\*\*-\*\*_
-* New: Update to Kotlin `1.3.40`
-* New: Update to Kotlin Coroutines `1.2.2`
-* New: Update to gRPC `1.21.0`
-* New: Update to protobuf `3.7.1`
+* New: Update to Kotlin `1.3.41`
+* New: Update to Kotlin Coroutines `1.3.0-RC`
+* New: Update to gRPC `1.22.1`
+* New: Update to protobuf `3.9.0`
 
 
 

buildSrc/src/main/kotlin/Versions.kt

@@ -1,7 +1,7 @@
 object Versions {
-    const val protobuf = "3.7.1"
-    const val grpc = "1.21.0"
-    const val kotlin = "1.3.40"
-    const val coroutines = "1.2.2"
+    const val protobuf = "3.9.0"
+    const val grpc = "1.22.1"
+    const val kotlin = "1.3.41"
+    const val coroutines = "1.3.0-RC"
     const val mockk = "1.9.1"
 }

kroto-plus-coroutines/build.gradle

@@ -8,6 +8,7 @@ def experimentalFlags = [
     "-Xuse-experimental=kotlin.Experimental",
     "-Xuse-experimental=kotlinx.coroutines.ExperimentalCoroutinesApi",
     "-Xuse-experimental=kotlinx.coroutines.ObsoleteCoroutinesApi",
+    "-Xuse-experimental=kotlinx.coroutines.FlowPreview",
     "-Xuse-experimental=com.github.marcoferrer.krotoplus.coroutines.KrotoPlusInternalApi"
 ]
 
@@ -35,6 +36,7 @@ dependencies {
     testImplementation project(':test-api:java')
     testImplementation "io.mockk:mockk:${Versions.mockk}"
     testImplementation "org.jetbrains.kotlinx:kotlinx-coroutines-test:${Versions.coroutines}"
+    testImplementation "org.jetbrains.kotlinx:kotlinx-coroutines-debug:${Versions.coroutines}"
 }
 
 tasks.withType(JavaCompile) {

kroto-plus-coroutines/src/main/kotlin/com/github/marcoferrer/krotoplus/coroutines/call/FlowControl.kt

@@ -20,9 +20,10 @@ import io.grpc.stub.CallStreamObserver
 import kotlinx.coroutines.CoroutineExceptionHandler
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.channels.ActorScope
 import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.launch
-import java.util.concurrent.atomic.AtomicBoolean
+import kotlinx.coroutines.channels.actor
+import kotlinx.coroutines.channels.consumeEach
 import java.util.concurrent.atomic.AtomicInteger
 
 
@@ -42,40 +43,51 @@ internal fun <T> CallStreamObserver<*>.applyInboundFlowControl(
     }
 }
 
+private typealias MessageHandler = suspend ActorScope<*>.() -> Unit
+
 internal fun <T> CoroutineScope.applyOutboundFlowControl(
     streamObserver: CallStreamObserver<T>,
     targetChannel: Channel<T>
 ){
-    val isOutboundJobRunning = AtomicBoolean()
+
     val channelIterator = targetChannel.iterator()
-    streamObserver.setOnReadyHandler {
-        if(targetChannel.isClosedForReceive){
-            streamObserver.completeSafely()
-        }else if(
+    val messageHandlerBlock: MessageHandler = {
+        while(
             streamObserver.isReady &&
-            !targetChannel.isClosedForReceive &&
-            isOutboundJobRunning.compareAndSet(false, true)
+            channelIterator.hasNext()
         ){
-            launch(Dispatchers.Unconfined + CoroutineExceptionHandler { _, e ->
-                streamObserver.completeSafely(e)
-                targetChannel.close(e)
-            }) {
-                try{
-                    while(
-                        streamObserver.isReady &&
-                        !targetChannel.isClosedForReceive &&
-                        channelIterator.hasNext()
-                    ){
-                        val value = channelIterator.next()
-                        streamObserver.onNext(value)
-                    }
-                    if(targetChannel.isClosedForReceive){
-                        streamObserver.onCompleted()
-                    }
-                } finally {
-                    isOutboundJobRunning.set(false)
-                }
+            val value = channelIterator.next()
+            streamObserver.onNext(value)
+        }
+        if(targetChannel.isClosedForReceive){
+            channel.close()
+        }
+    }
+
+    val messageHandlerActor = actor<MessageHandler>(
+        capacity = 1,
+        context = Dispatchers.Unconfined + CoroutineExceptionHandler { _, e ->
+            streamObserver.completeSafely(e)
+            targetChannel.close(e)
+        }
+    ) {
+        consumeEach { it.invoke(this) }
+        if(targetChannel.isClosedForReceive){
+            streamObserver.onCompleted()
+        }
+    }
+
+    streamObserver.setOnReadyHandler {
+        try {
+            if(targetChannel.isClosedForReceive){
+                messageHandlerActor.close()
+            }else{
+                messageHandlerActor.offer(messageHandlerBlock)
             }
+        }catch (e: Throwable){
+            // If offer throws an exception then it is
+            // either already closed or there was a failure
+            // which has already cleaned up call resources
         }
     }
 }

kroto-plus-coroutines/src/test/kotlin/com/github/marcoferrer/krotoplus/coroutines/CallOptionsTest.kt

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+
+
 package com.github.marcoferrer.krotoplus.coroutines
 
 import io.grpc.CallOptions
@@ -28,6 +30,7 @@ import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertNotEquals
 
+@Suppress("DEPRECATION")
 class CallOptionsTest {
 
     @Test

kroto-plus-coroutines/src/test/kotlin/com/github/marcoferrer/krotoplus/coroutines/StubSupportTests.kt

@@ -36,7 +36,7 @@ class NewGrpcStubTests {
         val stub = scope
             .newGrpcStub(GreeterCoroutineGrpc.GreeterCoroutineStub, channel)
 
-        assertEquals(nameElement.name, stub.coroutineContext[CoroutineName]?.name)
+        assertEquals(nameElement.name, stub.context[CoroutineName]?.name)
     }
 
 
@@ -50,7 +50,7 @@ class NewGrpcStubTests {
         val stub = CoroutineScope(scopeNameElement + scopeJob)
             .newGrpcStub(GreeterCoroutineGrpc.GreeterCoroutineStub, channel,expectedNameElement)
 
-        assertEquals(expectedNameElement.name, stub.coroutineContext[CoroutineName]?.name)
-        assertEquals(scopeJob, stub.coroutineContext[Job])
+        assertEquals(expectedNameElement.name, stub.context[CoroutineName]?.name)
+        assertEquals(scopeJob, stub.context[Job])
     }
 }

kroto-plus-coroutines/src/test/kotlin/com/github/marcoferrer/krotoplus/coroutines/client/ClientCallBidiStreamingTests.kt

@@ -23,6 +23,7 @@ import com.github.marcoferrer.krotoplus.coroutines.withCoroutineContext
 import io.grpc.CallOptions
 import io.grpc.ClientCall
 import io.grpc.Status
+import io.grpc.examples.helloworld.GreeterCoroutineGrpc
 import io.grpc.examples.helloworld.GreeterGrpc
 import io.grpc.examples.helloworld.HelloReply
 import io.grpc.examples.helloworld.HelloRequest
@@ -38,13 +39,19 @@ import kotlinx.coroutines.Job
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.SendChannel
-import kotlinx.coroutines.channels.map
-import kotlinx.coroutines.channels.toList
+import kotlinx.coroutines.debug.DebugProbes
+import kotlinx.coroutines.debug.junit4.CoroutinesTimeout
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.collectIndexed
+import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import org.junit.Rule
 import org.junit.Test
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.coroutines.CoroutineContext
 import kotlin.test.BeforeTest
 import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
@@ -55,6 +62,10 @@ class ClientCallBidiStreamingTests {
     @[Rule JvmField]
     var grpcServerRule = GrpcServerRule().directExecutor()
 
+    @[Rule JvmField]
+    public val timeout = CoroutinesTimeout.seconds(30)
+
+
     private val methodDescriptor = GreeterGrpc.getSayHelloStreamingMethod()
     private val service = spyk(object : GreeterGrpc.GreeterImplBase() {})
 
@@ -149,7 +160,7 @@ class ClientCallBidiStreamingTests {
                 requestChannel.close()
             }
 
-            responseChannel.map { it.message }.toList()
+            responseChannel.consumeAsFlow().map { it.message }.toList()
         }
 
         assertEquals(3,result.size)
@@ -332,4 +343,58 @@ class ClientCallBidiStreamingTests {
         assert(responseChannel.isClosedForReceive) { "Response channel should be closed for receive" }
     }
 
+    @Test
+    fun `High throughput call succeeds`() {
+        grpcServerRule.serviceRegistry.addService(object : GreeterCoroutineGrpc.GreeterImplBase() {
+            override val initialContext: CoroutineContext = Dispatchers.Default
+            override suspend fun sayHelloStreaming(
+                requestChannel: ReceiveChannel<HelloRequest>,
+                responseChannel: SendChannel<HelloReply>
+            ) {
+                requestChannel.consumeAsFlow().collectIndexed { index, value ->
+                    if (index % 1000 == 0) {
+//                          println("Server received $index")
+                    }
+
+                    responseChannel.send(HelloReply.newBuilder().setMessage(value.name).build())
+                }
+                responseChannel.close()
+            }
+        })
+        val stub = GreeterCoroutineGrpc.newStub(grpcServerRule.channel)
+
+        val (requestChannel, responseChannel) = stub
+            .clientCallBidiStreaming(methodDescriptor)
+
+        val numMessages = 100000
+        val receivedCount = AtomicInteger()
+        runBlocking(Dispatchers.Default) {
+            DebugProbes.install()
+            val req = HelloRequest.newBuilder()
+                .setName("test").build()
+
+            launch {
+                repeat(numMessages) {
+//                     if (it % 1000 == 0) println("Client sent $it")
+                    requestChannel.send(req)
+                }
+                requestChannel.close()
+            }
+
+            launch {
+                repeat(numMessages) {
+//                     if (it % 1000 == 0) println("Client received $it")
+                    responseChannel.receive()
+                    receivedCount.incrementAndGet()
+                }
+            }
+        }
+        // Sleep so that we can ensure the response channel
+        // has had enough time to close before being asserted on
+        Thread.sleep(50)
+        assert(requestChannel.isClosedForSend) { "Request channel should be closed for send" }
+        assert(responseChannel.isClosedForReceive) { "Response channel should be closed for receive" }
+        assertEquals(numMessages, receivedCount.get(), "Must response count must equal request count")
+    }
+
 }

kroto-plus-coroutines/src/test/kotlin/com/github/marcoferrer/krotoplus/coroutines/integration/ClientStreamingBackPressureTests.kt

@@ -30,6 +30,7 @@ import io.mockk.every
 import io.mockk.spyk
 import io.mockk.verify
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.cancel
@@ -74,10 +75,10 @@ class ClientStreamingBackPressureTests {
 
     @Test
     fun `Client send suspends until server invokes receive`() {
-        lateinit var serverRequestChannel: ReceiveChannel<HelloRequest>
+        val deferredServerChannel = CompletableDeferred<ReceiveChannel<HelloRequest>>()
         grpcServerRule.serviceRegistry.addService(object : GreeterCoroutineGrpc.GreeterImplBase(){
             override suspend fun sayHelloClientStreaming(requestChannel: ReceiveChannel<HelloRequest>): HelloReply {
-                serverRequestChannel = spyk(requestChannel)
+                deferredServerChannel.complete(spyk(requestChannel))
                 delay(Long.MAX_VALUE)
                 return HelloReply.getDefaultInstance()
             }
@@ -105,6 +106,7 @@ class ClientStreamingBackPressureTests {
                     }
                 }
 
+                val serverRequestChannel = deferredServerChannel.await()
                 repeat(3){
                     delay(10L)
                     assertEquals(it + 1, requestCount.get())

kroto-plus-coroutines/src/test/kotlin/com/github/marcoferrer/krotoplus/coroutines/server/ServerCallBidiStreamingTests.kt

@@ -42,9 +42,10 @@ import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.channels.consumeEach
-import kotlinx.coroutines.channels.mapTo
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.consumeAsFlow
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.yield
@@ -63,7 +64,9 @@ class ServerCallBidiStreamingTests {
     private val expectedResponse = HelloReply.newBuilder().setMessage("reply").build()
     private val responseObserver = spyk<StreamObserver<HelloReply>>(object : StreamObserver<HelloReply> {
         override fun onNext(value: HelloReply?) {}
-        override fun onError(t: Throwable?) {}
+        override fun onError(t: Throwable?) {
+            print(t?.message)
+        }
         override fun onCompleted() {}
     })
 
@@ -93,9 +96,11 @@ class ServerCallBidiStreamingTests {
             ) {
                 reqChannel = requestChannel
                 respChannel = responseChannel
-                requestChannel.mapTo(responseChannel) {
-                    HelloReply.newBuilder().setMessage("Reply: ${it.name}").build()
-                }
+                requestChannel
+                    .consumeAsFlow()
+                    .collect {
+                        responseChannel.send(HelloReply.newBuilder().setMessage("Reply: ${it.name}").build())
+                    }
             }
         })
 
@@ -128,10 +133,11 @@ class ServerCallBidiStreamingTests {
                 respChannel = responseChannel
                 var groupCount = 0
                 val requestIter = requestChannel.iterator()
+
                 while (requestIter.hasNext()) {
                     val requestValues = listOf(
-                        requestIter.next(),
-                        requestIter.next(),
+                        requestIter.next().also { requestIter.hasNext() },
+                        requestIter.next().also { requestIter.hasNext() },
                         requestIter.next()
                     )
                     responseChannel.send {

protoc-gen-kroto-plus/generator-tests/src/test/kotlin/com/github/marcoferrer/krotoplus/generators/GrpcCoroutinesGeneratorTests.kt

@@ -144,7 +144,6 @@ class GrpcCoroutinesGeneratorTests {
 
             val (requestChannel, responseChannel) = stub.sayHelloStreaming()
 
-//            launchProducerJob(requestChannel) {
             launch(Dispatchers.Default) {
                 repeat(3) {
                     requestChannel.send { name = "name $it" }
@@ -153,7 +152,6 @@ class GrpcCoroutinesGeneratorTests {
             }
 
             val results = responseChannel.toList()
-            println(results)
             assertEquals(9, results.size)
 
             val expected = "name 0|name 0|name 0" +