Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix scalability issue due to checkcast on context's invoke operations #12806

Merged
merged 5 commits into from Sep 27, 2022
Merged

Fix scalability issue due to checkcast on context's invoke operations #12806

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b0ddc0a
Fix scalability issue due to checkcast on context's invoke operations
franz1981 Sep 15, 2022
747e470
Improved style and comments
franz1981 Sep 16, 2022
8e0185c
Micro-benchmark
franz1981 Sep 20, 2022
e67fb4b
Adding duplex/non-duplex
franz1981 Sep 26, 2022
3edcb9d
Addressing scalability of HeadContext
franz1981 Sep 26, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
273 changes: 209 additions & 64 deletions transport/src/main/java/io/netty/channel/AbstractChannelHandlerContext.java
View file
Open in desktop
Expand Up @@ -16,6 +16,7 @@
package io.netty.channel;

import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.socket.DuplexChannel;
franz1981 marked this conversation as resolved.
Show resolved Hide resolved
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import io.netty.util.ReferenceCountUtil;
Expand Down Expand Up @@ -162,10 +163,19 @@ public void run() {

private void invokeChannelRegistered() {
if (invokeHandler()) {
try {
((ChannelInboundHandler) handler()).channelRegistered(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).channelRegistered(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
} else {
try {
((ChannelInboundHandler) handler).channelRegistered(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
}
franz1981 marked this conversation as resolved.
Show resolved Hide resolved
} else {
fireChannelRegistered();
Expand Down Expand Up @@ -194,10 +204,19 @@ public void run() {

private void invokeChannelUnregistered() {
if (invokeHandler()) {
try {
((ChannelInboundHandler) handler()).channelUnregistered(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).channelUnregistered(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
} else {
try {
((ChannelInboundHandler) handler).channelUnregistered(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
}
} else {
fireChannelUnregistered();
Expand Down Expand Up @@ -226,10 +245,19 @@ public void run() {

private void invokeChannelActive() {
if (invokeHandler()) {
try {
((ChannelInboundHandler) handler()).channelActive(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).channelActive(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
} else {
try {
((ChannelInboundHandler) handler).channelActive(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
}
} else {
fireChannelActive();
Expand Down Expand Up @@ -258,10 +286,19 @@ public void run() {

private void invokeChannelInactive() {
if (invokeHandler()) {
try {
((ChannelInboundHandler) handler()).channelInactive(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).channelInactive(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
} else {
try {
((ChannelInboundHandler) handler).channelInactive(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
}
} else {
fireChannelInactive();
Expand Down Expand Up @@ -342,10 +379,19 @@ public void run() {

private void invokeUserEventTriggered(Object event) {
if (invokeHandler()) {
try {
((ChannelInboundHandler) handler()).userEventTriggered(this, event);
} catch (Throwable t) {
invokeExceptionCaught(t);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).userEventTriggered(this, event);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
} else {
try {
((ChannelInboundHandler) handler).userEventTriggered(this, event);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
}
} else {
fireUserEventTriggered(event);
Expand Down Expand Up @@ -375,10 +421,19 @@ public void run() {

private void invokeChannelRead(Object msg) {
if (invokeHandler()) {
try {
((ChannelInboundHandler) handler()).channelRead(this, msg);
} catch (Throwable t) {
invokeExceptionCaught(t);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).channelRead(this, msg);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
} else {
try {
((ChannelInboundHandler) handler).channelRead(this, msg);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
}
} else {
fireChannelRead(msg);
Expand Down Expand Up @@ -406,10 +461,19 @@ static void invokeChannelReadComplete(final AbstractChannelHandlerContext next)

private void invokeChannelReadComplete() {
if (invokeHandler()) {
try {
((ChannelInboundHandler) handler()).channelReadComplete(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).channelReadComplete(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
} else {
try {
((ChannelInboundHandler) handler).channelReadComplete(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
}
} else {
fireChannelReadComplete();
Expand Down Expand Up @@ -437,10 +501,19 @@ static void invokeChannelWritabilityChanged(final AbstractChannelHandlerContext

private void invokeChannelWritabilityChanged() {
if (invokeHandler()) {
try {
((ChannelInboundHandler) handler()).channelWritabilityChanged(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).channelWritabilityChanged(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
} else {
try {
((ChannelInboundHandler) handler).channelWritabilityChanged(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
}
} else {
fireChannelWritabilityChanged();
Expand Down Expand Up @@ -502,10 +575,19 @@ public void run() {

private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {
if (invokeHandler()) {
try {
((ChannelOutboundHandler) handler()).bind(this, localAddress, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).bind(this, localAddress, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
} else {
try {
((ChannelOutboundHandler) handler).bind(this, localAddress, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
}
} else {
bind(localAddress, promise);
Expand Down Expand Up @@ -544,10 +626,19 @@ public void run() {

private void invokeConnect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
if (invokeHandler()) {
try {
((ChannelOutboundHandler) handler()).connect(this, remoteAddress, localAddress, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).connect(this, remoteAddress, localAddress, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
} else {
try {
((ChannelOutboundHandler) handler).connect(this, remoteAddress, localAddress, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
}
} else {
connect(remoteAddress, localAddress, promise);
Expand Down Expand Up @@ -583,10 +674,19 @@ public void run() {

private void invokeDisconnect(ChannelPromise promise) {
if (invokeHandler()) {
try {
((ChannelOutboundHandler) handler()).disconnect(this, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).disconnect(this, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
} else {
try {
((ChannelOutboundHandler) handler).disconnect(this, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
}
} else {
disconnect(promise);
Expand Down Expand Up @@ -618,10 +718,19 @@ public void run() {

private void invokeClose(ChannelPromise promise) {
if (invokeHandler()) {
try {
((ChannelOutboundHandler) handler()).close(this, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).close(this, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
} else {
try {
((ChannelOutboundHandler) handler).close(this, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
}
} else {
close(promise);
Expand Down Expand Up @@ -653,10 +762,19 @@ public void run() {

private void invokeDeregister(ChannelPromise promise) {
if (invokeHandler()) {
try {
((ChannelOutboundHandler) handler()).deregister(this, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).deregister(this, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
} else {
try {
((ChannelOutboundHandler) handler).deregister(this, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
}
} else {
deregister(promise);
Expand All @@ -682,10 +800,19 @@ public ChannelHandlerContext read() {

private void invokeRead() {
if (invokeHandler()) {
try {
((ChannelOutboundHandler) handler()).read(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).read(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
} else {
try {
((ChannelOutboundHandler) handler).read(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
}
} else {
read();
Expand Down Expand Up @@ -713,10 +840,19 @@ void invokeWrite(Object msg, ChannelPromise promise) {
}

private void invokeWrite0(Object msg, ChannelPromise promise) {
try {
((ChannelOutboundHandler) handler()).write(this, msg, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).write(this, msg, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
} else {
try {
((ChannelOutboundHandler) handler).write(this, msg, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
}
}

Expand Down Expand Up @@ -746,10 +882,19 @@ private void invokeFlush() {
}

private void invokeFlush0() {
try {
((ChannelOutboundHandler) handler()).flush(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
final ChannelHandler handler = handler();
if (handler instanceof ChannelDuplexHandler) {
try {
((ChannelDuplexHandler) handler).flush(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
} else {
try {
((ChannelOutboundHandler) handler).flush(this);
} catch (Throwable t) {
invokeExceptionCaught(t);
}
}
}

Expand Down