Even with useBlockingTaskExecutor enabled, gRPC's ServerInterceptor runs in a non-blocking thread

[be-hase]

Issue

First of all, gRPC's ServerInterceptor is a synchronous API.

Therefore, when implementing an authentication process that requires network IO on the ServerInterceptor, blocking the thread is the only way.

In the case of gRPC-java, this is not a major problem since it runs on a thread pool. (ref)
However, in Armeria, it is a problem because it runs on the event loop.
Unfortunately, even if I specify useBlockingTaskExecutor, the ServerInterceptor is executed on the event loop. (ref)

Suggestion

Unfortunately, the ServerInterceptor is synchronous at this time, so it would be better to allow the ServerInterceptor to run on a thread pool, as in gRPC-java.
(At least, if useBlockingTaskExecutor is enabled, it would be better to do so.)

For reference, in the case of gRPC-Kotlin, the ServerInterceptor runs in a thread pool and the Service runs in coroutines.

Ideally, of course, the ServerInterceptor should support async.

Or we could consider another method like this.
#2607 (comment)

[be-hase]

Or we may prepare an AsyncServerInterceptor using the technique introduced here.
https://stackoverflow.com/questions/53651024/grpc-java-async-call-in-serverinterceptor

But, it would be nice if it was prepared on the gRPC-Java side...

[ikhoon]

Unfortunately, even if I specify useBlockingTaskExecutor, the ServerInterceptor is executed on the event loop. (ref)

😱 It should be a bug. Let us fix it soon.

[ikhoon]

I was working on refactoring Armeria's ServerCall implementations. #4192
The change to fix this issue won't be difficult but it may heavily conflict with #4192. I am going to fix this problem once the refactoring PR is merged.

[be-hase]

Thank you.

In my opinion, ServerInterceptor should be runned at BlockingTaskExecutor when the user has specified useBlockingTaskExecutor=true.

By the way, I am using gRPC-Kotlin. I wrote the following code using the technique introduced in this article and it seems to work fine.
(However, it's still a rough code, and the test isn't enough)

It would be very convenient if armeria had a ServerInterceptor that could be written asynchronously.
(This code uses Coroutines, but armeria needs to be usable in java, so it's better to use CompletableFuture.)

/**
 * ref: https://stackoverflow.com/questions/53651024/grpc-java-async-call-in-serverinterceptor
 */
abstract class SuspendableServerInterceptor(
    private val context: CoroutineContext = EmptyCoroutineContext
) : ServerInterceptor {
    override fun <ReqT : Any, RespT : Any> interceptCall(
        call: ServerCall<ReqT, RespT>,
        headers: Metadata,
        next: ServerCallHandler<ReqT, RespT>
    ): ServerCall.Listener<ReqT> {
        val armeriaCtx = ServiceRequestContext.current()
        val delayedListener = DelayedListener<ReqT>()
        delayedListener.job = CoroutineScope(
            ArmeriaRequestCoroutineContext(armeriaCtx)
                    + GrpcContextElement.current()
                    + COROUTINE_CONTEXT_KEY.get()
                    + context
        ).launch {
            try {
                delayedListener.realListener = suspendableInterceptCall(call, headers, next)
                delayedListener.drainPendingCallbacks()
            } catch (e: CancellationException) {
                log.debug { "Caught CancellationException. $e" }
                call.close(Status.CANCELLED, Metadata())
            } catch (e: Exception) {
                log.error(e) { "Unhandled exception. $e" }
                call.close(Status.UNKNOWN, Metadata())
            }
        }
        return delayedListener
    }

    abstract suspend fun <ReqT : Any, RespT : Any> suspendableInterceptCall(
        call: ServerCall<ReqT, RespT>,
        headers: Metadata,
        next: ServerCallHandler<ReqT, RespT>
    ): ServerCall.Listener<ReqT>

    /**
     * ref: https://github.com/grpc/grpc-java/blob/84edc332397ed01fae2400c25196fc90d8c1a6dd/core/src/main/java/io/grpc/internal/DelayedClientCall.java#L415
     */
    private class DelayedListener<ReqT> : ServerCall.Listener<ReqT>() {
        var realListener: ServerCall.Listener<ReqT>? = null

        @Volatile
        private var passThrough = false

        @GuardedBy("this")
        private var pendingCallbacks: MutableList<Runnable> = mutableListOf()

        var job: Job? = null

        override fun onMessage(message: ReqT) {
            if (passThrough) {
                checkNotNull(realListener).onMessage(message)
            } else {
                delayOrExecute { checkNotNull(realListener).onMessage(message) }
            }
        }

        override fun onHalfClose() {
            if (passThrough) {
                checkNotNull(realListener).onHalfClose()
            } else {
                delayOrExecute { checkNotNull(realListener).onHalfClose() }
            }
        }

        override fun onCancel() {
            job?.cancel()
            if (passThrough) {
                checkNotNull(realListener).onCancel()
            } else {
                delayOrExecute { checkNotNull(realListener).onCancel() }
            }
        }

        override fun onComplete() {
            if (passThrough) {
                checkNotNull(realListener).onComplete()
            } else {
                delayOrExecute { checkNotNull(realListener).onComplete() }
            }
        }

        override fun onReady() {
            if (passThrough) {
                checkNotNull(realListener).onReady()
            } else {
                delayOrExecute { checkNotNull(realListener).onReady() }
            }
        }

        private fun delayOrExecute(runnable: Runnable) {
            synchronized(this) {
                if (!passThrough) {
                    pendingCallbacks.add(runnable)
                    return
                }
            }
            runnable.run()
        }

        fun drainPendingCallbacks() {
            check(!passThrough)
            var toRun: MutableList<Runnable> = mutableListOf()
            while (true) {
                synchronized(this) {
                    if (pendingCallbacks.isEmpty()) {
                        pendingCallbacks = mutableListOf()
                        passThrough = true
                        return
                    }
                    // Since there were pendingCallbacks, we need to process them. To maintain ordering we
                    // can't set passThrough=true until we run all pendingCallbacks, but new Runnables may be
                    // added after we drop the lock. So we will have to re-check pendingCallbacks.
                    val tmp: MutableList<Runnable> = toRun
                    toRun = pendingCallbacks
                    pendingCallbacks = tmp
                }
                for (runnable in toRun) {
                    // Avoid calling listener while lock is held to prevent deadlocks.
                    runnable.run()
                }
                toRun.clear()
            }
        }
    }

    companion object {
        private val log = KotlinLogging.logger {}

        @Suppress("UNCHECKED_CAST")
        // Get by using reflection
        internal val COROUTINE_CONTEXT_KEY: Context.Key<CoroutineContext> =
            CoroutineContextServerInterceptor::class.let { kclass ->
                val companionObject = kclass.companionObject!!
                val property = companionObject.memberProperties.single { it.name == "COROUTINE_CONTEXT_KEY" }
                checkNotNull(property.getter.call(kclass.companionObjectInstance!!)) as Context.Key<CoroutineContext>
            }
    }
}