grpc-native: Working version of NativeClientCall

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

Author: Jozott00

cinterop-c/include/grpcpp_c.h

@@ -47,7 +47,6 @@ typedef struct {
 } grpc_cb_tag;
 
 
-
 grpc_client_t *grpc_client_create_insecure(const char *target);
 void grpc_client_delete(const grpc_client_t *client);
 
@@ -77,6 +76,10 @@ typedef struct grpc_channel_credentials grpc_channel_credentials_t;
 
 bool kgrpc_iomgr_run_in_background();
 
+
+/////// UTILS ///////
+
+
 #ifdef __cplusplus
     }
 #endif

cinterop-c/src/grpcpp_c.cpp

@@ -208,6 +208,7 @@ extern "C" {
         return grpc_iomgr_run_in_background();
     }
 
+
 }
 
 

grpc/grpc-core/src/commonTest/proto/helloworld.proto

@@ -0,0 +1,14 @@
+syntax = "proto3";
+
+
+// The request message containing the user's name.
+message HelloRequest {
+  string name = 1;
+  optional uint32 timeout = 2;
+}
+
+
+// The response message containing the greetings
+message HelloReply {
+  string message = 1;
+}

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

@@ -1,7 +1,8 @@
-headers = grpcpp_c.h grpc/grpc.h grpc/credentials.h
-headerFilter= grpcpp_c.h grpc/slice.h grpc/byte_buffer.h grpc/grpc.h grpc/impl/grpc_types.h grpc/credentials.h grpc/support/time.h
+headers = grpcpp_c.h grpc/grpc.h grpc/credentials.h grpc/byte_buffer_reader.h
+headerFilter= grpcpp_c.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
 
 noStringConversion = grpc_slice_from_copied_buffer my_grpc_slice_from_copied_buffer
-strictEnums = grpc_status_code, grpc_connectivity_state grpc_call_error
+strictEnums = grpc_status_code grpc_connectivity_state grpc_call_error
 
 staticLibraries = libgrpcpp_c_static.a

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

@@ -10,10 +10,8 @@ package kotlinx.rpc.grpc
 import cnames.structs.grpc_channel
 import kotlinx.cinterop.CPointer
 import kotlinx.cinterop.ExperimentalForeignApi
-import kotlinx.rpc.grpc.internal.ClientCall
-import kotlinx.rpc.grpc.internal.GrpcCallOptions
-import kotlinx.rpc.grpc.internal.GrpcChannel
-import kotlinx.rpc.grpc.internal.MethodDescriptor
+import kotlinx.coroutines.*
+import kotlinx.rpc.grpc.internal.*
 import libgrpcpp_c.*
 import kotlin.experimental.ExperimentalNativeApi
 import kotlin.native.ref.createCleaner
@@ -34,13 +32,15 @@ public actual abstract class ManagedChannelBuilder<T : ManagedChannelBuilder<T>>
 }
 
 internal actual fun ManagedChannelBuilder<*>.buildChannel(): ManagedChannel {
-    return NativeManagedChannel(
-        target = "localhost:50051",
-        credentials = GrpcCredentials(
-            grpc_insecure_credentials_create()
-                ?: error("Failed to create credentials")
-        )
-    )
+    TODO("Not yet implemented")
+//    return NativeManagedChannel(
+//        target = "localhost:50051",
+//        credentials = GrpcCredentials(
+//            grpc_insecure_credentials_create()
+//                ?: error("Failed to create credentials")
+//        ),
+//
+//    )
 }
 
 internal actual fun ManagedChannelBuilder(hostname: String, port: Int): ManagedChannelBuilder<*> {
@@ -52,43 +52,93 @@ internal actual fun ManagedChannelBuilder(target: String): ManagedChannelBuilder
 }
 
 
+private const val GRPC_PROPAGATE_DEFAULTS = 0x0000FFFFu
+
 internal class NativeManagedChannel(
-    private val target: String,
+    target: String,
     // we must store them, otherwise the credentials are getting released
-    private val credentials: GrpcCredentials,
+    credentials: GrpcCredentials,
+    dispatcher: CoroutineDispatcher = Dispatchers.Default,
 ) : ManagedChannel, ManagedChannelPlatform() {
 
+    private val channelJob = SupervisorJob()
+    private val callJobSupervisor = SupervisorJob(channelJob)
+    private val channelScope = CoroutineScope(channelJob + dispatcher)
+
+    // the channel's completion queue, handling all request operations
+    private val cq = CompletionQueue()
+
     internal val raw: CPointer<grpc_channel> = grpc_channel_create(target, credentials.raw, null)
         ?: error("Failed to create channel")
+
+    @Suppress("unused")
     private val rawCleaner = createCleaner(raw) {
         grpc_channel_destroy(it)
     }
 
     override val platformApi: ManagedChannelPlatform = this
 
-    override val isShutdown: Boolean
-        get() = TODO("Not yet implemented")
+    private var isShutdownInternal: Boolean = false
+    override val isShutdown: Boolean = isShutdownInternal
+    private var isTerminatedInternal = CompletableDeferred(Unit)
     override val isTerminated: Boolean
-        get() = TODO("Not yet implemented")
+        get() = isTerminatedInternal.isCompleted
 
     override suspend fun awaitTermination(duration: Duration): Boolean {
-        TODO("Not yet implemented")
+        withTimeoutOrNull(duration) {
+            isTerminatedInternal.await()
+        } ?: return false
+        return true
     }
 
     override fun shutdown(): ManagedChannel {
-        TODO("Not yet implemented")
+        channelScope.launch {
+            shutdownInternal(false)
+        }
+        return this
     }
 
     override fun shutdownNow(): ManagedChannel {
-        TODO("Not yet implemented")
+        channelScope.launch {
+            shutdownInternal(true)
+        }
+        return this
+    }
+
+    private suspend fun shutdownInternal(force: Boolean) {
+        isShutdownInternal = true
+        if (force) {
+            callJobSupervisor.cancelChildren(CancellationException("Channel is shutting down"))
+        }
+        // prevent any start() calls on already created jobs
+        callJobSupervisor.complete()
+        cq.shutdown(force)
+        // wait for child jobs to complete.
+        // should be immediate, as the completion queue is shutdown.
+        callJobSupervisor.join()
+        isTerminatedInternal.complete(Unit)
     }
 
 
     override fun <RequestT, ResponseT> newCall(
         methodDescriptor: MethodDescriptor<RequestT, ResponseT>,
         callOptions: GrpcCallOptions,
     ): ClientCall<RequestT, ResponseT> {
-        TODO("Not yet implemented")
+        check(!isShutdown) { "Channel is shutdown" }
+
+        val parent = channelScope.coroutineContext[Job]!!
+        val callJob = Job(parent)
+        val callScope = CoroutineScope(callJob)
+
+        val methodNameSlice = methodDescriptor.getFullMethodName().toGrpcSlice()
+        val rawCall = grpc_channel_create_call(
+            channel = raw, parent_call = null, propagation_mask = GRPC_PROPAGATE_DEFAULTS, completion_queue = cq.raw,
+            method = methodNameSlice, host = null, deadline = gpr_inf_future(GPR_CLOCK_REALTIME), reserved = null
+        ) ?: error("Failed to create call")
+
+        return NativeClientCall(
+            cq, rawCall, methodDescriptor, callScope
+        )
     }
 
     override fun authority(): String {
@@ -98,10 +148,15 @@ internal class NativeManagedChannel(
 }
 
 
-internal class GrpcCredentials(
+internal sealed class GrpcCredentials(
     internal val raw: CPointer<grpc_channel_credentials_t>,
 ) {
     val rawCleaner = createCleaner(raw) {
         grpc_channel_credentials_release(it)
     }
 }
+
+internal class GrpcInsecureCredentials() :
+    GrpcCredentials(grpc_insecure_credentials_create() ?: error("Failed to create credentials"))
+
+

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

@@ -2,11 +2,15 @@
  * Copyright 2023-2025 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
  */
 
+@file:OptIn(ExperimentalForeignApi::class, ExperimentalNativeApi::class)
+
 package kotlinx.rpc.grpc.internal
 
+import kotlinx.cinterop.ExperimentalForeignApi
 import kotlinx.rpc.grpc.GrpcTrailers
 import kotlinx.rpc.grpc.Status
 import kotlinx.rpc.internal.utils.InternalRpcApi
+import kotlin.experimental.ExperimentalNativeApi
 
 @InternalRpcApi
 public actual abstract class ClientCall<Request, Response> {

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

@@ -9,15 +9,23 @@ package kotlinx.rpc.grpc.internal
 import cnames.structs.grpc_call
 import kotlinx.atomicfu.atomic
 import kotlinx.cinterop.*
+import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.suspendCancellableCoroutine
+import kotlinx.coroutines.withContext
 import libgrpcpp_c.*
 import platform.posix.memset
 import kotlin.coroutines.Continuation
 import kotlin.coroutines.resume
-import kotlin.coroutines.resumeWithException
 import kotlin.experimental.ExperimentalNativeApi
 import kotlin.native.ref.createCleaner
 
+internal sealed interface BatchResult {
+    object Success : BatchResult
+    object ResultError : BatchResult
+    object CQShutdown : BatchResult
+    data class CallError(val error: grpc_call_error) : BatchResult
+}
+
 /**
  * A coroutine wrapper around the grpc completion_queue, which manages message operations.
  * It is based on the "new" callback API; therefore, there are no kotlin-side threads required to poll
@@ -27,6 +35,9 @@ internal class CompletionQueue {
 
     private enum class State { OPEN, SHUTTING_DOWN, CLOSED }
 
+    // if the queue was called with forceShutdown = true,
+    // it will reject all new batches and wait for all current ones to finish.
+    private var forceShutdown = false
     private val state = atomic(State.OPEN)
 
     // internal as it must be accessible from the SHUTDOWN_CB,
@@ -47,7 +58,10 @@ internal class CompletionQueue {
 
     val raw = grpc_completion_queue_create_for_callback(shutdownFunctor.ptr, null)
 
+    @Suppress("unused")
     private val thisStableRefCleaner = createCleaner(thisStableRef) { it.dispose() }
+
+    @Suppress("unused")
     private val shutdownFunctorCleaner = createCleaner(shutdownFunctor) { nativeHeap.free(it) }
 
     init {
@@ -56,16 +70,22 @@ internal class CompletionQueue {
         require(kgrpc_iomgr_run_in_background()) { "The gRPC iomgr is not running background threads, required for callback based APIs." }
     }
 
-    suspend fun runBatch(call: CPointer<grpc_call>, ops: CPointer<grpc_op>, nOps: ULong) =
-        suspendCancellableCoroutine<grpc_call_error> { cont ->
+    // TODO: Remove this method
+    suspend fun runBatch(call: NativeClientCall<*, *>, ops: CPointer<grpc_op>, nOps: ULong) =
+        runBatch(call.raw, ops, nOps)
+
+    suspend fun runBatch(call: CPointer<grpc_call>, ops: CPointer<grpc_op>, nOps: ULong): BatchResult =
+        suspendCancellableCoroutine<BatchResult> { cont ->
             val tag = newCbTag(cont, OPS_COMPLETE_CB)
 
             var err = grpc_call_error.GRPC_CALL_ERROR
             // synchronizes access to grpc_call_start_batch
             withBatchStartLock {
-                if (state.value != State.OPEN) {
+                if (forceShutdown || state.value == State.CLOSED) {
+                    // if the queue is either closed or in the process of a FORCE shutdown,
+                    // new batches will instantly fail.
                     deleteCbTag(tag)
-                    cont.resume(grpc_call_error.GRPC_CALL_ERROR_COMPLETION_QUEUE_SHUTDOWN)
+                    cont.resume(BatchResult.CQShutdown)
                     return@suspendCancellableCoroutine
                 }
 
@@ -75,7 +95,7 @@ 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)
-                cont.resume(err)
+                cont.resume(BatchResult.CallError(grpc_call_error.GRPC_CALL_ERROR))
                 return@suspendCancellableCoroutine
             }
 
@@ -87,16 +107,29 @@ internal class CompletionQueue {
                 // cancel the call if one of its batches is canceled.
                 // grpc_call_cancel is thread-safe and can be called several times.
                 // the callback is invoked anyway, so the tag doesn't get deleted here.
-                grpc_call_cancel(call, null)
+                if (it != null) {
+                    grpc_call_cancel_with_status(
+                        call,
+                        grpc_status_code.GRPC_STATUS_CANCELLED,
+                        "Call got cancelled: ${it.message}",
+                        null
+                    )
+                } else {
+                    grpc_call_cancel(call, null)
+                }
             }
         }
 
-    suspend fun shutdown() {
+    // must not be canceled as it cleans resources and sets the state to CLOSED
+    suspend fun shutdown(force: Boolean = false) = withContext(NonCancellable) {
+        if (force) {
+            forceShutdown = true
+        }
         if (!state.compareAndSet(State.OPEN, State.SHUTTING_DOWN)) {
             // the first call to shutdown() makes transition and to SHUTTING_DOWN and
             // initiates shut down. all other invocations await the shutdown.
             _shutdownDone.await()
-            return
+            return@withContext
         }
 
         // wait until all batch operations since the state transitions were started.
@@ -126,10 +159,10 @@ internal class CompletionQueue {
 @CName("kq_ops_complete_cb")
 private fun opsCompleteCb(functor: CPointer<grpc_completion_queue_functor>?, ok: Int) {
     val tag = functor!!.reinterpret<grpc_cb_tag>()
-    val cont = tag.pointed.user_data!!.asStableRef<Continuation<grpc_call_error>>().get()
+    val cont = tag.pointed.user_data!!.asStableRef<Continuation<BatchResult>>().get()
     deleteCbTag(tag)
-    if (ok != 0) cont.resume(grpc_call_error.GRPC_CALL_OK)
-    else cont.resumeWithException(IllegalStateException("batch failed"))
+    if (ok != 0) cont.resume(BatchResult.Success)
+    else cont.resume(BatchResult.ResultError)
 }
 
 @CName("kq_shutdown_cb")