Describe the feature

Extend existing async/await support to executeBlocking / blockingHandler threads

Use cases

Performance uplift (in theory) for any existing executeBlocking / blockingHandler code implementation in microthreads

Contribution

Who should implement this feature ? are you volunteering for implementing this feature or do you know that is able and willing implement this feature ?

I dunno who should, or is willing to implement this feature

Motivation:

Already completed futures should not park the thread. Currently, they do. This can improve performance.

This also resolves an ordering issue in the default event loop context.

Version

947f181

Context

I observed the following while load testing a complex application. The core part of this load test runs many SQL transactions. These transactions are implemented as awaiting Vertx JOOQ reactive client executing a SQL query against a PgPool.

java.lang.NullPointerException: Cannot invoke "java.lang.Runnable.run()" because the return value of "java.util.LinkedList.poll()" is null
	at io.vertx.await.impl.EventLoopScheduler.lambda$new$1(EventLoopScheduler.java:75)
	at io.vertx.await.impl.EventLoopScheduler.lambda$new$0(EventLoopScheduler.java:60)
	at io.vertx.await.impl.EventLoopScheduler.lambda$new$2(EventLoopScheduler.java:74)
	at java.base/java.lang.VirtualThread.submitRunContinuation(VirtualThread.java:234)
	at java.base/java.lang.VirtualThread.submitRunContinuation(VirtualThread.java:253)
	at java.base/java.lang.VirtualThread.start(VirtualThread.java:462)
	at java.base/java.lang.VirtualThread.start(VirtualThread.java:475)
	at io.vertx.await.impl.EventLoopScheduler.execute(EventLoopScheduler.java:89)
	at io.vertx.await.impl.VirtualThreadContext.execute2(VirtualThreadContext.java:118)
	at io.vertx.await.impl.VirtualThreadContext.execute(VirtualThreadContext.java:67)
	at io.vertx.await.impl.VirtualThreadContext.execute(VirtualThreadContext.java:84)
	at io.vertx.core.impl.ContextBase.execute(ContextBase.java:225)
	at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:51)
	at io.vertx.core.impl.future.FutureImpl.tryComplete(FutureImpl.java:211)
	at io.vertx.core.impl.future.PromiseImpl.tryComplete(PromiseImpl.java:23)
	at io.vertx.core.impl.future.PromiseImpl.onSuccess(PromiseImpl.java:49)
	at io.vertx.core.impl.future.PromiseImpl.handle(PromiseImpl.java:41)
	at io.vertx.core.impl.future.PromiseImpl.handle(PromiseImpl.java:23)
	at io.vertx.sqlclient.impl.command.CommandResponse.fire(CommandResponse.java:46)
	at io.vertx.sqlclient.impl.SocketConnectionBase.handleMessage(SocketConnectionBase.java:292)
	at io.vertx.pgclient.impl.PgSocketConnection.handleMessage(PgSocketConnection.java:97)
	at io.vertx.sqlclient.impl.SocketConnectionBase.lambda$init$0(SocketConnectionBase.java:105)
	at io.vertx.core.impl.EventLoopContext.emit(EventLoopContext.java:55)
	at io.vertx.core.impl.ContextBase.emit(ContextBase.java:239)
	at io.vertx.core.net.impl.NetSocketImpl.handleMessage(NetSocketImpl.java:390)
	at io.vertx.core.net.impl.ConnectionBase.read(ConnectionBase.java:157)
	at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:153)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
	at io.vertx.pgclient.impl.codec.PgEncoder.lambda$write$0(PgEncoder.java:98)
	at io.vertx.pgclient.impl.codec.PgCommandCodec.handleReadyForQuery(PgCommandCodec.java:139)
	at io.vertx.pgclient.impl.codec.PgDecoder.decodeReadyForQuery(PgDecoder.java:237)
	at io.vertx.pgclient.impl.codec.PgDecoder.channelRead(PgDecoder.java:96)
	at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
	at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
	at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:1589)

For your convenience, the context of this code is: https://github.com/vert-x3/vertx-virtual-threads-incubator/blob/947f1812290f0c3cc357d834f62ab24e620b4c1f/vertx-async-await-incubator/src/main/java/io/vertx/await/impl/EventLoopScheduler.java#L75

I observed this issue again in load testing, now within a gRPC stub implementation:

Dec 22, 2022 4:32:39 PM io.grpc.internal.SerializingExecutor run
SEVERE: Exception while executing runnable io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1MessagesAvailable@3600a37f
java.lang.NullPointerException: Cannot invoke "java.lang.Runnable.run()" because the return value of "java.util.LinkedList.poll()" is null
	at io.vertx.await.impl.EventLoopScheduler.lambda$new$1(EventLoopScheduler.java:75)
	at io.vertx.await.impl.EventLoopScheduler.lambda$new$0(EventLoopScheduler.java:60)
	at io.vertx.await.impl.EventLoopScheduler.lambda$new$2(EventLoopScheduler.java:74)
	at java.base/java.lang.VirtualThread.submitRunContinuation(VirtualThread.java:234)
	at java.base/java.lang.VirtualThread.submitRunContinuation(VirtualThread.java:253)
	at java.base/java.lang.VirtualThread.start(VirtualThread.java:462)
	at java.base/java.lang.VirtualThread.start(VirtualThread.java:475)
	at io.vertx.await.impl.EventLoopScheduler.execute(EventLoopScheduler.java:89)
	at io.vertx.core.eventbus.impl.HandlerRegistration.receive(HandlerRegistration.java:46)
	at io.vertx.core.eventbus.impl.EventBusImpl.deliverMessageLocally(EventBusImpl.java:375)
	at io.vertx.core.eventbus.impl.EventBusImpl.sendLocally(EventBusImpl.java:341)
	at io.vertx.core.eventbus.impl.EventBusImpl.sendOrPub(EventBusImpl.java:329)
	at io.vertx.core.eventbus.impl.OutboundDeliveryContext.execute(OutboundDeliveryContext.java:109)
	at io.vertx.core.eventbus.impl.DeliveryContextBase.next(DeliveryContextBase.java:72)
	at io.vertx.core.eventbus.impl.OutboundDeliveryContext.next(OutboundDeliveryContext.java:28)
	at io.vertx.core.eventbus.impl.EventBusImpl.sendOrPubInternal(EventBusImpl.java:422)
	at io.vertx.core.eventbus.impl.EventBusImpl.sendOrPubInternal(EventBusImpl.java:428)
	at io.vertx.core.eventbus.impl.EventBusImpl.publish(EventBusImpl.java:164)
// context: this is forwarding grpc messages to the event bus
	at com.hiddenswitch.framework.impl.ServerGameContext.clientToServer(ServerGameContext.java:281)
	at com.hiddenswitch.framework.impl.ServerGameContext.lambda$subscribeGame$3(ServerGameContext.java:271)
// end context
	at io.vertx.grpc.stub.StreamObserverReadStream.onNext(StreamObserverReadStream.java:37)
	at io.grpc.stub.ServerCalls$StreamingServerCallHandler$StreamingServerCallListener.onMessage(ServerCalls.java:262)
	at io.grpc.ForwardingServerCallListener.onMessage(ForwardingServerCallListener.java:33)
	at io.grpc.Contexts$ContextualizedServerCallListener.onMessage(Contexts.java:76)
	at com.avast.grpc.jwt.server.DelayedServerCallListener.onMessage(DelayedServerCallListener.java:17)
	at io.grpc.ForwardingServerCallListener.onMessage(ForwardingServerCallListener.java:33)
	at com.be_hase.grpc.micrometer.MicrometerServerCallListener.onMessage(MicrometerServerCallListener.java:22)
	at io.grpc.ForwardingServerCallListener.onMessage(ForwardingServerCallListener.java:33)
	at io.grpc.Contexts$ContextualizedServerCallListener.onMessage(Contexts.java:76)
	at io.grpc.ForwardingServerCallListener.onMessage(ForwardingServerCallListener.java:33)
	at io.opentracing.contrib.grpc.TracingServerInterceptor$2.onMessage(TracingServerInterceptor.java:221)
	at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.messagesAvailableInternal(ServerCallImpl.java:318)
	at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.messagesAvailable(ServerCallImpl.java:301)
	at io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1MessagesAvailable.runInContext(ServerImpl.java:834)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at io.vertx.grpc.VertxServer$ActualServer.lambda$null$0(VertxServer.java:96)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:264)
	at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:246)
	at io.vertx.core.impl.EventLoopContext.lambda$runOnContext$0(EventLoopContext.java:43)
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:566)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:1589)

Do you have a reproducer?

(Investigating)

Steps to reproduce

(Investigating)

Extra

macOS 13, Java 19

These tests show that Thread.currentThread().isInterrupted() doesn't get set to true, as expected, when a virtual thread is interrupted in Vertx.

Additionally, the DefaultVirtualThreadContextTest cannot interrupt Thread.Sleep(), but it's not clear to me if the reason is expected.

Signed-off-by: Paulo Lopes [email protected]

Motivation:

This is a first step to allow mutating the internals of the router, which will later allow add the concept of filters