grpc-native: Address PR comments

Signed-off-by: Johannes Zottele <[email protected]>

Author: Jozott00

grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/internal/ClientCall.kt

@@ -22,6 +22,10 @@ import kotlinx.rpc.internal.utils.InternalRpcApi
  *   Internally, it buffers the messages.
  * - On Native, you can only call [sendMessage] when [isReady] returns true. There is no buffering; therefore,
  *   only one message can be sent at a time.
+ *
+ * Request message number:
+ * - On JVM, there is no limit on the number of messages you can [request].
+ * - On Native, you can only call [request] with up to `16`.
  */
 @InternalRpcApi
 public expect abstract class ClientCall<Request, Response> {

grpc/grpc-core/src/commonTest/kotlin/kotlinx/rpc/grpc/test/RawClientTest.kt

@@ -4,7 +4,6 @@
 
 package kotlinx.rpc.grpc.test
 
-import grpc.examples.echo.*
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.test.runTest
@@ -17,6 +16,7 @@ import kotlin.test.assertEquals
 /**
  * Tests for JVM and Native clients.
  */
+// TODO: Start echo service server automatically
 class RawClientTest {
 
     @Test
@@ -102,9 +102,5 @@ class RawClientTest {
             channel.shutdown()
             channel.awaitTermination()
         }
-
-
     }
-
-
 }

grpc/grpc-core/src/nativeInterop/cinterop/libkgrpc.def

@@ -1,9 +1,9 @@
 headers = kgrpc.h grpc/grpc.h grpc/credentials.h grpc/byte_buffer_reader.h \
-    grpc/support/alloc.h
+    grpc/support/alloc.h grpc/impl/propagation_bits.h
 
 headerFilter= kgrpc.h grpc/slice.h grpc/byte_buffer.h grpc/grpc.h \
     grpc/impl/grpc_types.h grpc/credentials.h grpc/support/time.h grpc/byte_buffer_reader.h \
-    grpc/support/alloc.h
+    grpc/support/alloc.h grpc/impl/propagation_bits.h
 
 noStringConversion = grpc_slice_from_copied_buffer my_grpc_slice_from_copied_buffer
 strictEnums = grpc_status_code grpc_connectivity_state grpc_call_error

grpc/grpc-core/src/nativeMain/kotlin/kotlinx/rpc/grpc/ManagedChannel.native.kt

@@ -21,7 +21,7 @@ public actual abstract class ManagedChannelPlatform : GrpcChannel()
  */
 public actual abstract class ManagedChannelBuilder<T : ManagedChannelBuilder<T>> {
     public actual open fun usePlaintext(): T {
-        error("Builder does not override usePlaintext()")
+        error("Builder does not support usePlaintext()")
     }
 }
 

grpc/grpc-core/src/nativeMain/kotlin/kotlinx/rpc/grpc/internal/CallbackFuture.kt

@@ -11,7 +11,7 @@ import kotlinx.atomicfu.atomic
  */
 internal class CallbackFuture<T : Any> {
     private sealed interface State<out T> {
-        data class Pending<T>(val cbs: List<(T) -> Unit> = emptyList()) : State<T>
+        data class Pending<T>(val callbacks: List<(T) -> Unit> = emptyList()) : State<T>
         data class Done<T>(val value: T) : State<T>
     }
 
@@ -22,7 +22,7 @@ internal class CallbackFuture<T : Any> {
         while (true) {
             when (val s = state.value) {
                 is State.Pending -> if (state.compareAndSet(s, State.Done(result))) {
-                    toInvoke = s.cbs
+                    toInvoke = s.callbacks
                     break
                 }
 
@@ -40,7 +40,7 @@ internal class CallbackFuture<T : Any> {
                 }
 
                 is State.Pending -> {
-                    val next = State.Pending(s.cbs + callback) // copy-on-write append
+                    val next = State.Pending(s.callbacks + callback) // copy-on-write append
                     if (state.compareAndSet(s, next)) return
                 }
             }

grpc/grpc-core/src/nativeMain/kotlin/kotlinx/rpc/grpc/internal/CompletionQueue.kt

@@ -30,18 +30,18 @@ internal sealed interface BatchResult {
     /**
      * Happens when the batch couldn't be submitted for some reason.
      */
-    data class CallError(val error: grpc_call_error) : BatchResult
+    data class SubmitError(val error: grpc_call_error) : BatchResult
 
     /**
      * Happens when the batch was successfully submitted.
      * The [future] will be completed with `true` if the batch was successful, `false` otherwise.
      * In the case of `false`, the status of the `RECV_STATUS_ON_CLIENT` batch will provide the error details.
      */
-    data class Called(val future: CallbackFuture<Boolean>) : BatchResult
+    data class Submitted(val future: CallbackFuture<Boolean>) : BatchResult
 }
 
 /**
- * The Kotlin wrapper for the native grpc_completion_queue.
+ * A thread-safe Kotlin wrapper for the native grpc_completion_queue.
  * It is based on the "new" callback API; therefore, there are no kotlin-side threads required to poll
  * the queue.
  * Users can attach to the returned [CallbackFuture] if the batch was successfully submitted (see [BatchResult]).
@@ -50,7 +50,7 @@ internal class CompletionQueue {
 
     internal enum class State { OPEN, SHUTTING_DOWN, CLOSED }
 
-    // if the queue was called with forceShutdown = true,
+    // if the shutdown() was called with forceShutdown = true,
     // it will reject all new batches and wait for all current ones to finish.
     private var forceShutdown = false
 
@@ -122,16 +122,20 @@ internal class CompletionQueue {
         if (err != grpc_call_error.GRPC_CALL_OK) {
             // if the call was not successful, the callback will not be invoked.
             deleteCbTag(tag)
-            return BatchResult.CallError(err)
+            return BatchResult.SubmitError(err)
         }
 
-        return BatchResult.Called(completion)
+        return BatchResult.Submitted(completion)
     }
 
     /**
      * Shuts down the queue.
      * The method returns immediately, but the queue will be shut down asynchronously.
      * The returned [CallbackFuture] will be completed with `Unit` when the queue is shut down.
+     *
+     * @param force if `true`, the queue will reject all new batches with [BatchResult.CQShutdown].
+     *        Otherwise, the queue allows submitting new batches and shutdown only when there are no more
+     *        ongoing batches.
      */
     fun shutdown(force: Boolean = false): CallbackFuture<Unit> {
         if (force) {

grpc/grpc-core/src/nativeMain/kotlin/kotlinx/rpc/grpc/internal/NativeClientCall.kt

@@ -54,7 +54,7 @@ internal class NativeClientCall<Request, Response>(
     private var closed = atomic(false)
 
     // tracks how many operations are in flight (not yet completed by the listener).
-    // if 0, there are no more operations (except for the RECV_STATUS_ON_CLIENT op).
+    // if 0 and we got a closeInfo (containing the status), there are no more ongoing operations.
     // in this case, we can safely call onClose on the listener.
     // we need this mechanism to ensure that onClose is not called while any other callback is still running
     // on the listener.
@@ -64,7 +64,7 @@ internal class NativeClientCall<Request, Response>(
     // if null, the call is still in progress. otherwise, the call can be closed as soon as inFlight is 0.
     private val closeInfo = atomic<Pair<Status, GrpcTrailers>?>(null)
 
-    // we currently don't buffer messages, so after one `sendMessage` call, ready turns false.
+    // we currently don't buffer messages, so after one `sendMessage` call, ready turns false. (KRPC-192)
     private val ready = atomic(true)
 
     /**
@@ -81,11 +81,11 @@ internal class NativeClientCall<Request, Response>(
      * AND the corresponding listener callback returned.
      *
      * If the counter reaches 0, no more listener callbacks are executed, and the call can be closed by
-     * calling [tryDeliverClose].
+     * calling [tryToCloseCall].
      */
     private fun endOp() {
         if (inFlight.decrementAndGet() == 0) {
-            tryDeliverClose()
+            tryToCloseCall()
         }
     }
 
@@ -97,23 +97,23 @@ internal class NativeClientCall<Request, Response>(
      * - If the [inFlight] counter is not 0, this does nothing.
      * - Otherwise, the listener's onClose callback is invoked and the call is closed.
      */
-    private fun tryDeliverClose() {
-        val s = closeInfo.value ?: return
+    private fun tryToCloseCall() {
+        val info = closeInfo.value ?: return
         if (inFlight.value == 0 && closed.compareAndSet(expect = false, update = true)) {
-            val lst = checkNotNull(listener) { "Not yet started" }
+            val lst = checkNotNull(listener) { internalError("Not yet started") }
             // allows the managed channel to join for the call to finish.
             callJob.complete()
-            lst.onClose(s.first, s.second)
+            lst.onClose(info.first, info.second)
         }
     }
 
     /**
-     * Sets the [closeInfo] and calls [tryDeliverClose].
-     * This is called as soon as the RECV_STATUS_ON_CLIENT batch is finished.
+     * Sets the [closeInfo] and calls [tryToCloseCall].
+     * This is called as soon as the RECV_STATUS_ON_CLIENT batch (started with [startRecvStatus]) finished.
      */
     private fun markClosePending(status: Status, trailers: GrpcTrailers) {
         if (closeInfo.compareAndSet(null, Pair(status, trailers))) {
-            tryDeliverClose()
+            tryToCloseCall()
         }
     }
 
@@ -132,8 +132,8 @@ internal class NativeClientCall<Request, Response>(
         responseListener: Listener<Response>,
         headers: GrpcTrailers,
     ) {
-        check(listener == null) { "Already started" }
-        check(!cancelled) { "Already cancelled." }
+        check(listener == null) { internalError("Already started") }
+        check(!cancelled) { internalError("Already cancelled.") }
 
         listener = responseListener
 
@@ -164,7 +164,7 @@ internal class NativeClientCall<Request, Response>(
         beginOp()
 
         when (val callResult = cq.runBatch(this@NativeClientCall, ops, nOps)) {
-            is BatchResult.Called -> {
+            is BatchResult.Submitted -> {
                 callResult.future.onComplete { success ->
                     try {
                         if (success) {
@@ -184,7 +184,7 @@ internal class NativeClientCall<Request, Response>(
                 cancelInternal(grpc_status_code.GRPC_STATUS_UNAVAILABLE, "Channel shutdown")
             }
 
-            is BatchResult.CallError -> {
+            is BatchResult.SubmitError -> {
                 cleanup()
                 endOp()
                 cancelInternal(
@@ -196,7 +196,7 @@ internal class NativeClientCall<Request, Response>(
     }
 
     /**
-     * Starts a batch operation to receive the status from the completion queue.
+     * Starts a batch operation to receive the status from the completion queue (RECV_STATUS_ON_CLIENT).
      * This operation is bound to the lifetime of the call, so it will finish once all other operations are done.
      * If this operation fails, it will call [markClosePending] with the corresponding error, as the entire call
      * si considered failed.
@@ -206,7 +206,7 @@ internal class NativeClientCall<Request, Response>(
      */
     @OptIn(ExperimentalStdlibApi::class)
     private fun startRecvStatus(): Boolean {
-        checkNotNull(listener) { "Not yet started" }
+        checkNotNull(listener) { internalError("Not yet started") }
         val arena = Arena()
         val statusCode = arena.alloc<grpc_status_code.Var>()
         val statusDetails = arena.alloc<grpc_slice>()
@@ -221,7 +221,7 @@ internal class NativeClientCall<Request, Response>(
         }
 
         when (val callResult = cq.runBatch(this@NativeClientCall, op.ptr, 1u)) {
-            is BatchResult.Called -> {
+            is BatchResult.Submitted -> {
                 callResult.future.onComplete {
                     val details = statusDetails.toByteArray().toKString()
                     val status = Status(statusCode.value.toKotlin(), details, null)
@@ -244,7 +244,7 @@ internal class NativeClientCall<Request, Response>(
                 return false
             }
 
-            is BatchResult.CallError -> {
+            is BatchResult.SubmitError -> {
                 arena.clear()
                 markClosePending(
                     Status(StatusCode.INTERNAL, "Failed to start call: ${callResult.error}"),
@@ -285,9 +285,11 @@ internal class NativeClientCall<Request, Response>(
      * This must only be called again after [numMessages] were received in the [Listener.onMessage] callback.
      */
     override fun request(numMessages: Int) {
-        check(numMessages > 0) { "numMessages must be > 0" }
-        val listener = checkNotNull(listener) { "Not yet started" }
-        check(!cancelled) { "Already cancelled" }
+        check(numMessages > 0) { internalError("numMessages must be > 0") }
+        // limit numMessages to prevent potential stack overflows
+        check(numMessages <= 16) { internalError("numMessages must be <= 16") }
+        val listener = checkNotNull(listener) { internalError("Not yet started") }
+        check(!cancelled) { internalError("Already cancelled") }
 
         var remainingMessages = numMessages
 
@@ -333,8 +335,8 @@ internal class NativeClientCall<Request, Response>(
     }
 
     override fun halfClose() {
-        check(!halfClosed) { "Already half closed." }
-        check(!cancelled) { "Already cancelled." }
+        check(!halfClosed) { internalError("Already half closed.") }
+        check(!cancelled) { internalError("Already cancelled.") }
         halfClosed = true
 
         val arena = Arena()
@@ -350,10 +352,10 @@ internal class NativeClientCall<Request, Response>(
     override fun isReady(): Boolean = ready.value
 
     override fun sendMessage(message: Request) {
-        checkNotNull(listener) { "Not yet started" }
-        check(!halfClosed) { "Already half closed." }
-        check(!cancelled) { "Already cancelled." }
-        check(isReady()) { "Not yet ready." }
+        checkNotNull(listener) { internalError("Not yet started") }
+        check(!halfClosed) { internalError("Already half closed.") }
+        check(!cancelled) { internalError("Already cancelled.") }
+        check(isReady()) { internalError("Not yet ready.") }
 
         // set ready false, as only one message can be sent at a time.
         ready.value = false
@@ -368,14 +370,12 @@ internal class NativeClientCall<Request, Response>(
         }
 
         runBatch(op.ptr, 1u, cleanup = {
-            // no mater what happens, we need to set ready to true again.
-            turnReady()
-
             // actual cleanup
             grpc_byte_buffer_destroy(byteBuffer)
             arena.clear()
         }) {
-            // Nothing to do here
+            // set ready true, as we can now send another message.
+            turnReady()
         }
     }
 }

grpc/grpc-core/src/nativeMain/kotlin/kotlinx/rpc/grpc/internal/NativeGrpcLibrary.kt

@@ -21,17 +21,16 @@ internal object GrpcRuntime {
     /** Acquire a runtime reference. Must be closed exactly once. */
     fun acquire(): AutoCloseable {
         refLock.withLock {
-            val prev = 0
-            refs++
+            val prev = refs++
             if (prev == 0) grpc_init()
         }
         return object : AutoCloseable {
-            private var done = atomic(false)
+            private val done = atomic(false)
             override fun close() {
                 if (!done.compareAndSet(expect = false, update = true)) return
                 refLock.withLock {
                     val now = --refs
-                    require(now >= 0) { "release() without matching acquire()" }
+                    require(now >= 0) { internalError("release() without matching acquire()") }
                     if (now == 0) {
                         grpc_shutdown()
                     }