gRPC Client / Server Code gen and api

[marcoferrer]

This issue is for tracking any discussions related to #16 PR.

The PR introduces new apis with subtle differences to those introduced in the stub ext generator.
Please reference the client / server example included. It contains a sample of the expected output for discussion.

Here are some key points and examples from the PR

1. All around better integration with Structured Concurrency
2. Backpressure Support has been implemented.
   • Related Links
     • Understanding Reactive gRPC Flow Control
     • Manual Flow Control
3. Generation of Client Stubs which implement the CoroutineScope interface
   • Allows client stubs to work well with Structured Concurrency
   • Cancellations can now be propagated across usages of a specific stub instance.
4. Generation of abstract Service Base Impl's
   • Allows services to still support common grpc-java patterns, while still fully embracing coroutine idioms and features.
5. Convenience exts for sending requests and responses in both client, server code.
   • Full support for familiar Kroto-plus convenience lambda exts on SendChannel.send { } and CompletableDeferred.complete { }
6. Support for annotation processor assistance via RpcMethod annotation from grpc-java. RpcMethod.java

suspend fun performUnaryCall(stub: GreeterCoroutineGrpc.GreeterCoroutineStub){

    val unaryResponse = stub.sayHello { name = "John" }

    println("Unary Response: ${unaryResponse.message}")
}

suspend fun performServerStreamingCall(stub: GreeterCoroutineGrpc.GreeterCoroutineStub){

    val responseChannel = stub.sayHelloServerStreaming { name = "John" }

    responseChannel.consumeEach {
        println("Server Streaming Response: ${it.message}")
    }
}

suspend fun CoroutineScope.performClientStreamingCall(stub: GreeterCoroutineGrpc.GreeterCoroutineStub){

    // Client Streaming RPC
    val (requestChannel, response) = stub.sayHelloClientStreaming()

    launch {
        repeat(5){
            requestChannel.send { name = "person #$it" }
        }
        requestChannel.close()
    }

    println("Client Streaming Response: ${response.await().toString().trim()}")
}

suspend fun CoroutineScope.performBidiCall(stub: GreeterCoroutineGrpc.GreeterCoroutineStub){

    val (requestChannel, responseChannel) = stub.sayHelloStreaming()

    launch {
        repeat(5){
            requestChannel.send { name = "person #$it" }
        }
        requestChannel.close()
    }

    launch {
        responseChannel.consumeEach {
            println("Bidi Response: ${it.message}")
        }
    }
}

[marcoferrer]

@brandmooffin tagging for visibility.

[lifk]

I see in your example that for the server implementation you a CompletableDeferred or a ResponseChannel as an argument in the method to override from gRPC. Taking a look at https://github.com/rouzwawi/grpc-kotlin which seems to be a project with similar objectives I think their method signature is easier to understand by people less familiar with gRPC.

override suspend fun greet(request: GreetRequest): GreetReply

I think shifting the response from an argument to the return of the function makes it more understandable.

[marcoferrer]

@lifk Thanks for the feedback

Thats a similar approach to the reactive-grpc project. I think it worked well for them, because the reactive paradigm maps everything into their respective primitives (Observable / Flowable). These primitives have error propagation baked into their semantics.

I was reluctant to go down that route for a couple of reason. Using the functions return as a result inverts gRPCs error propagation semantics. Instead of letting users build their error responses and return them explicitly, that method requires them to "throw" an error response.

Couple of problems I see with that.

• Throwing exceptions in coroutines can cause cooperative cancellation to trigger, which may lead to unintended behavior. Especially if your service attempts to send some data in the background, unrelated to the result of your method.
• Throwing exceptions can become expensive since it is accompanied by a stacktrace. Sed stacktrace doesnt provide any value to the caller of the service.
• The responsibility of propagating the service's intended response (error or result) shifts from the users to the library.
• Although gRPCs error mechanism is built around exceptions, I dont think they intended service level errors to be propagated as cross boundary application exceptions.

The current design was built around the goal of accurately mapping gRPC semantics and primitives to their coroutine counter parts. It was intended to lower the complexity while still allowing grpc-java users an easy way to convert existing services.

I also wanted to maintain the testability of service methods. Being able to mock the request and response channels allows users to test their business logic exclusively

Another small benefit to the current approach is message builders being implicitly available for building response vs having to import them explicitly.

response.complete { name = "John" } 

//vs 

return HelloWorldProtoBuilders.GreetReply {
    name = "John"    
}

[lifk]

I see that you have good points on why to follow that approach and I think that either way it's a big improvement over using the grpc-java implementation directly.

One of my reasons to suggest return types is that the compiler is gonna enforce you to have that return
in the code, most of our code at a grpc level is a try catch that then maps to a StatusRuntimeException. Sometimes we had problems with calls that didn't close their channel because the onError was not called and that is something that I think could be handled better.

[marcoferrer]

You bring up a good point. If I understand you correctly, I think you might be covered anyways by the safety net that surrounds service method invocation. I wanted to leave a fallback to clean up resources for cases such as yours.

   launch(GrpcContextElement()) {
        try {
            block(requestChannel, completableResponse)
        }catch (e: Throwable){
            completableResponse.completeExceptionally(e)
        }
   }

https://github.com/marcoferrer/kroto-plus/blob/grpc-server-prototype/kroto-plus-coroutines/src/main/kotlin/com/github/marcoferrer/krotoplus/coroutines/server/ServerCalls.kt

[lifk]

Having a fallback is a really good idea but I didn't mean exactly that, let's consider the following example

override fun sayHello(request: HelloRequest, completableResponse: CompletableDeferred<HelloReply>) {
        try {
            // do something that crashes
            throw Exception()
        } catch (exception: Exception) {
            // log the error and do some logic
            // application is dead here because I forgot to propagate the exception or to finish the completable response
        }
    }
override fun sayHello(request: HelloRequest): HelloReply {
        try {
            // do something that crashes
            throw Exception()
        } catch (exception: Exception) {
            // log the error and do some logic
            // this doesn't compile because I need to throw or return
        }
    }

In the first implementation I can make the mistake of not handling anything in the catch and the application stays there waiting, in the second one the compiler will enforce a return and that will make me notice that I'm missing something.

Maybe if you see too much issues into handling the exceptions maybe you could introduce a Result type to handle the response, something like:

fun sayHello(request: HelloRequest): Result<HelloReply, StatusRuntimeException>

[marcoferrer]

Sorry for the delayed reply. I was taking time off for the holiday.

I did some further digging and have a couple of things I wanted to bring up.

Your question actually helped me find some gaps that needed to be addressed in the service code.

While I think you bring up valid argument for wanting compile time checking for service methods, I think a small ext function might provide the functionality your looking for.

    suspend fun <T> CompletableDeferred<T>.respondWith(block: suspend CoroutineScope.() -> T) {
        coroutineScope {
            try {
                complete(block())
            } catch (e: Throwable) {
                completeExceptionally(e)
            }
        }
    }

Using this method your service method implementation can be checked at compile time.

    override suspend fun sayHello(request: HelloRequest, completableResponse: CompletableDeferred<HelloReply>) =
        completableResponse.respondWith {

            if (request.name.matches(validNameRegex)) {
                HelloWorldProtoBuilders.HelloReply {
                    message = "Hello there, ${request.name}!"
                }
            } else {
                throw Status.INVALID_ARGUMENT.asException()
            }
        }

What are your thoughts on this? @lifk

On a seperate note, while digging into this I realized that the current service examples have some bugs in regards to structured concurrency.

[marcoferrer]

Other Concerns

In order to have proper parallel decomposition (Outlined here) the services need to wrap their rpc implementations in a coroutineScope{ } block and not rely on the scope on the service.

    override suspend fun sayHello(request: HelloRequest, completableResponse: CompletableDeferred<HelloReply>) {
        coroutineScope {
            if (request.name.matches(validNameRegex)) {
                completableResponse
                    .complete { message = "Hello there, ${request.name}!" }
            } else {
                completableResponse
                    .completeExceptionally(Status.INVALID_ARGUMENT.asRuntimeException())
            }
        }
    }

This raises the following questions.

• Should wrapping the rpc method in a coroutineScope be enforced at runtime, compile time, or left up to the library consumer to follow best practices?
  • The structured concurrency blog post states that this should be done for reliable behavior from backend services
  • It can be enforced if need be by making service method require a return value of type RpcJob a special coroutine type with coresponding builder.
• Should the generated service base continue to implement CoroutineScope?
  • The structured concurrency blog post states that objects implementing CoroutineScope are meant to be short lived and the lifetime of grpc service instances span the servers lifetime.
  • Implementing CoroutineScope has the potential of confusing users as to which scope they are invoking their rpc method on.
• When service methods are invoked, should we mark a request as abandoned if the method body completes without returning a response to the client? This would only work if consumers follow proper best practices for structure concurrency.

It looks like these issues are present in other grpc coroutine implementations as well, my aim is to provided an implementation that balances all the best practices from both technologies.

[lifk]

I think the extension function can be a good idea to allow optionally for the consumers of the library to call service methods in a more secure way.

About your concerns about coroutineScope, I think implementing the scope per service is not useful, as you say the grpc service is gonna have a lifespan as big as the server lifetime so we wouldn't get any benefit from being able to cancel the scope together with all the children coroutines because the only case where the grpc service is stopped is when the server shuts down.

I didn't work a lot yet with newer versions of coroutines so I don't have a lot of knowledge on best practices about this problem but my intuition as the consumer of the library would be that the method of the grpc service is wrap in a coroutine scope that represents a single call, that way if the call is cancelled all the children coroutines will die. I think if this behavior is handled by the generated code it should be properly documented to avoid confusion about in which scope is the code being executed.

[marcoferrer]

So after a bit of feedback from a few other parties I think Im going to move forward with changing the service method signature to have an explicit return type. There were some valid points brought up.

• Marking a method as suspending AND accepting a completable deferred as an parameter can confuse users. The norm is to mark a method as suspending so that you can compute and return a result without blocking. My reason for marking the methods as suspending was to be able to support cross boundary cooperative cancellation, but this can still be done with explicit returns.
• Grpcs stream observer api exists as a way of passing what is essentially a callback to the service method. Since once of the benefits of coroutines is being able to use callback based libraries in a sequential manner, it looks we would benefit from abandoning certain grpc semantics in order to support better integration.

In regards to your comment on implementing CoroutineScope in the base service. As long as the base scope doesnt have a Job set, it wont cancel the children. The true reasoning behind implementing CoroutineScope in the base service is to allow users to configure the initial context that their service method will execute on. More importantly which dispatcher should be used. A root job is created on a per rpc level.

Apparently the confusion regarding ambiguous scope resolution in classes implementing coroutine scope has been brought up before. There is currently an issue open for adding a warning to intellij for such cases. KT-27493 It seems like worrying about consumers following best practices for this specific case wont be much of an issue in the near future.

Im close to having everything updated for the new signatures. Once Im done benchmarking and testing the manual control flow with the new implementation, Ill update this issue for visibility

[marcoferrer]

So the latest release 0.2.2-RC1 contains the updated code generator as well as coroutines lib. Its current state reflects the discussions outlined here as well as through other channels. Im going to leave this issue open while continuing to gather community feedback.

https://github.com/marcoferrer/kroto-plus/releases/tag/v0.2.2-RC1

[marcoferrer]

Just published the latest snapshot with the near final version of the proposed APIs. One key change was that services no longer implement CoroutineScope. A new interface was introduced to allow service impls to provide an initial coroutine context. PR related to this change is located here

Release: v0.2.2-RC2

This will be the last RC before the gRPC Coroutines API is finalized. Any feedback or questions is encouraged. Support for the legacy coroutine stub extension APIs will be removed in the next release. This release refactors generated stub extensions to use the new gRPC Coroutines API.

[marcoferrer]

Im going to close out this design issue now that coroutine support has been fully release. Any new suggestions or api changes can take place in their own issue