grpc-native: Write docs

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

Author: Jozott00

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

@@ -16,6 +16,12 @@ import kotlinx.rpc.internal.utils.InternalRpcApi
  * Callback execution:
  * - On JVM it is guaranteed that callbacks aren't executed concurrently.
  * - On Native, it is only guaranteed that `onClose` is called after all other callbacks finished.
+ *
+ * Sending message readiness:
+ * - On JVM, it is possible to call [sendMessage] multiple times, without checking [isReady].
+ *   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.
  */
 @InternalRpcApi
 public expect abstract class ClientCall<Request, Response> {

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

@@ -14,7 +14,10 @@ import kotlinx.rpc.grpc.internal.*
 import kotlin.test.Test
 import kotlin.test.assertEquals
 
-class CancellationClientTest {
+/**
+ * Tests for JVM and Native clients.
+ */
+class RawClientTest {
 
     @Test
     fun unaryEchoTest() = runTest(
@@ -39,8 +42,8 @@ class CancellationClientTest {
     }
 
     @Test
-    fun clientStreamingTest() = runTest(
-        methodName = "ServerStreamingEcho",
+    fun clientStreamingEchoTest() = runTest(
+        methodName = "ClientStreamingEcho",
         type = MethodType.CLIENT_STREAMING,
     ) { channel, descriptor ->
         val response = clientStreamingRpc(channel, descriptor, flow {
@@ -50,10 +53,29 @@ class CancellationClientTest {
                 emit(EchoRequest { message = "Eccchhooo" })
             }
         })
-        val expected = "Eccchhooo,Eccchhooo,Eccchhooo,Eccchhooo,Eccchhooo"
+        val expected = "Eccchhooo, Eccchhooo, Eccchhooo, Eccchhooo, Eccchhooo"
         assertEquals(expected, response.message)
     }
 
+    @Test
+    fun bidirectionalStreamingEchoTest() = runTest(
+        methodName = "BidirectionalStreamingEcho",
+        type = MethodType.BIDI_STREAMING,
+    ) { channel, descriptor ->
+        val response = bidirectionalStreamingRpc(channel, descriptor, flow {
+            repeat(5) {
+                emit(EchoRequest { message = "Eccchhooo" })
+            }
+        })
+
+        var i = 0
+        response.collect {
+            i++
+            assertEquals("Eccchhooo", it.message)
+        }
+        assertEquals(5, i)
+    }
+
     fun runTest(
         methodName: String,
         type: MethodType,

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

@@ -15,16 +15,36 @@ import platform.posix.memset
 import kotlin.experimental.ExperimentalNativeApi
 import kotlin.native.ref.createCleaner
 
+/**
+ * The result of a batch operation (see [CompletionQueue.runBatch]).
+ */
 internal sealed interface BatchResult {
+    /**
+     * Happens when a batch was submitted and...
+     * - the queue is closed
+     * - the queue is in the process of a force shutdown
+     * - the queue is in the process of a normal shutdown, and the batch is a new `RECV_STATUS_ON_CLIENT` batch.
+     */
     object CQShutdown : BatchResult
+
+    /**
+     * Happens when the batch couldn't be submitted for some reason.
+     */
     data class CallError(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
 }
 
 /**
- * A coroutine wrapper around the grpc completion_queue, which manages message operations.
+ * The 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]).
  */
 internal class CompletionQueue {
 
@@ -44,11 +64,13 @@ internal class CompletionQueue {
     @Suppress("PropertyName")
     internal val _shutdownDone = CallbackFuture<Unit>()
 
-    // used for spinning lock. false means not used (available)
+    // used to synchronize the start of a new batch operation.
     private val batchStartGuard = SynchronizedObject()
 
+    // a stable reference of this used as user_data in the shutdown callback.
     private val thisStableRef = StableRef.create(this)
 
+    // the shutdown functor/tag called when the queue is shut down.
     private val shutdownFunctor = nativeHeap.alloc<kgrpc_cb_tag> {
         functor.functor_run = SHUTDOWN_CB
         user_data = thisStableRef.asCPointer()
@@ -69,6 +91,10 @@ internal class CompletionQueue {
         require(kgrpc_iomgr_run_in_background()) { "The gRPC iomgr is not running background threads, required for callback based APIs." }
     }
 
+    /**
+     * Submits a batch operation to the queue.
+     * See [BatchResult] for possible outcomes.
+     */
     fun runBatch(call: NativeClientCall<*, *>, ops: CPointer<grpc_op>, nOps: ULong): BatchResult {
         val completion = CallbackFuture<Boolean>()
         val tag = newCbTag(completion, OPS_COMPLETE_CB)
@@ -102,7 +128,11 @@ internal class CompletionQueue {
         return BatchResult.Called(completion)
     }
 
-    // must not be canceled as it cleans resources and sets the state to CLOSED
+    /**
+     * 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.
+     */
     fun shutdown(force: Boolean = false): CallbackFuture<Unit> {
         if (force) {
             forceShutdown = true

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

@@ -34,6 +34,7 @@ internal class NativeClientCall<Request, Response>(
     }
 
     init {
+        // cancel the call if the job is canceled.
         callJob.invokeOnCompletion {
             when (it) {
                 is CancellationException -> {
@@ -63,6 +64,9 @@ 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.
+    private val ready = atomic(true)
+
     /**
      * Increments the [inFlight] counter by one.
      * This should be called before starting a batch.
@@ -113,6 +117,16 @@ internal class NativeClientCall<Request, Response>(
         }
     }
 
+    /**
+     * Sets the [ready] flag to true and calls the listener's onReady callback.
+     * This is called as soon as the RECV_MESSAGE batch is finished (or failed).
+     */
+    private fun turnReady() {
+        if (ready.compareAndSet(expect = false, update = true)) {
+            listener?.onReady()
+        }
+    }
+
 
     override fun start(
         responseListener: Listener<Response>,
@@ -124,13 +138,19 @@ internal class NativeClientCall<Request, Response>(
         listener = responseListener
 
         // start receiving the status from the completion queue,
-        // which is bound to the lifecycle of the call.
-        val success = initializeCallOnCQ()
+        // which is bound to the lifetime of the call.
+        val success = startRecvStatus()
         if (!success) return
 
+        // send and receive initial headers to/from the server
         sendAndReceiveInitialMetadata()
     }
 
+    /**
+     * Submits a batch operation to the [CompletionQueue] and handle the returned [BatchResult].
+     * If the batch was successfully submitted, [onSuccess] is called.
+     * In any case, [cleanup] is called.
+     */
     private fun runBatch(
         ops: CPointer<grpc_op>,
         nOps: ULong,
@@ -175,12 +195,19 @@ internal class NativeClientCall<Request, Response>(
         }
     }
 
+    /**
+     * Starts a batch operation to receive the status from the completion queue.
+     * 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.
+     *
+     * @return true if the batch was successfully submitted, false otherwise.
+     * In this case, the call is considered failed.
+     */
     @OptIn(ExperimentalStdlibApi::class)
-    private fun initializeCallOnCQ(): Boolean {
+    private fun startRecvStatus(): Boolean {
         checkNotNull(listener) { "Not yet started" }
         val arena = Arena()
-        // this must not be canceled as it sets the call status.
-        // if the client itself got canceled, this will return fast.
         val statusCode = arena.alloc<grpc_status_code.Var>()
         val statusDetails = arena.alloc<grpc_slice>()
         val errorStr = arena.alloc<CPointerVar<ByteVar>>()
@@ -253,12 +280,19 @@ internal class NativeClientCall<Request, Response>(
         }
     }
 
+    /**
+     * Requests [numMessages] messages from the server.
+     * 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" }
 
         var remainingMessages = numMessages
+
+        // we need to request only one message at a time, so we use a recursive function that
+        // requests one message and then calls itself again.
         fun post() {
             if (remainingMessages-- <= 0) return
 
@@ -272,13 +306,16 @@ internal class NativeClientCall<Request, Response>(
                 if (recvPtr.value != null) grpc_byte_buffer_destroy(recvPtr.value)
                 arena.clear()
             }) {
-                val buf = recvPtr.value ?: return@runBatch // EOS
+                // if the call was successful, but no message was received, we reached the end-of-stream.
+                val buf = recvPtr.value ?: return@runBatch
                 val msg = methodDescriptor.getResponseMarshaller()
                     .parse(buf.toKotlin().asInputStream())
                 listener.onMessage(msg)
-                post() // post next only now
+                post()
             }
         }
+
+        // start requesting messages
         post()
     }
 
@@ -310,19 +347,31 @@ 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." }
+
+        // set ready false, as only one message can be sent at a time.
+        ready.value = false
 
         val arena = Arena()
         val inputStream = methodDescriptor.getRequestMarshaller().stream(message)
         val byteBuffer = inputStream.asSource().toGrpcByteBuffer()
+
         val op = arena.alloc<grpc_op> {
             op = GRPC_OP_SEND_MESSAGE
             data.send_message.send_message = byteBuffer
         }
+
         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()
         }) {

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

@@ -19,6 +19,9 @@ import kotlin.experimental.ExperimentalNativeApi
 import kotlin.native.ref.createCleaner
 import kotlin.time.Duration
 
+/**
+ * Wrapper for [grpc_channel_credentials].
+ */
 internal sealed class GrpcCredentials(
     internal val raw: CPointer<grpc_channel_credentials>,
 ) {
@@ -27,12 +30,21 @@ internal sealed class GrpcCredentials(
     }
 }
 
+/**
+ * Insecure credentials.
+ */
 internal class GrpcInsecureCredentials() :
     GrpcCredentials(grpc_insecure_credentials_create() ?: error("Failed to create credentials"))
 
-
+// default propagation mask for all calls.
 private const val GRPC_PROPAGATE_DEFAULTS = 0x0000FFFFu
 
+/**
+ * Native implementation of [ManagedChannel].
+ *
+ * @param target The target address to connect to.
+ * @param credentials The credentials to use for the connection.
+ */
 internal class NativeManagedChannel(
     target: String,
     // we must store them, otherwise the credentials are getting released
@@ -89,11 +101,14 @@ internal class NativeManagedChannel(
             return
         }
         if (force) {
+            // cancel all jobs, such that the shutdown is completing faster (not immediate).
             // TODO: replace jobs by custom pendingCallClass.
             callJobSupervisor.cancelChildren(CancellationException("Channel is shutting down"))
         }
 
         // wait for the completion queue to shut down.
+        // the completion queue will be shut down after all requests are completed.
+        // therefore, we don't have to wait for the callJobs to be completed.
         cq.shutdown(force).onComplete {
             if (isTerminatedInternal.complete(Unit)) {
                 // release the grpc runtime, so it might call grpc_shutdown()