andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Continue reading when the number of bytes is less then the configured… (#11089)

Browse Source 
... number of bytes when using DatagramChannels

Motivation:

In our FixedRecvByteBufAllocator we dont continue to read if the number of bytes is less then what was configured. This is correct when using it for TCP but not when using it for UDP. When using UDP the number of bytes is the maximum of what we want to support but we often end up processing smaller datagrams in general. Because of this we should use contineReading(UncheckedBooleanSupplier) to determite if we should continue reading

Modifications:

- use contineReading(UncheckedBooleanSupplier) for DatagramChannels

Result:

Read more then once in the general case for DatagramChannels with the default config
This commit is contained in:
Norman Maurer 2021-03-17 13:10:57 +01:00 committed by GitHub
parent 95652fef12
commit a82af64047
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 26 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java
View File

@ -35,6 +35,7 @@ import io.netty.channel.unix.Errors.NativeIoException;
import io.netty.channel.unix.Socket;
import io.netty.channel.unix.UnixChannelUtil;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.UncheckedBooleanSupplier;
import io.netty.util.internal.ObjectUtil;
import io.netty.util.internal.RecyclableArrayList;
import io.netty.util.internal.StringUtil;
@ -498,7 +499,9 @@ public final class EpollDatagramChannel extends AbstractEpollChannel implements
} else {
break;
}
} while (allocHandle.continueReading());
// We use the TRUE_SUPPLIER as it is also ok to read less then what we did try to read (as long
// as we read anything).
} while (allocHandle.continueReading(UncheckedBooleanSupplier.TRUE_SUPPLIER));
} catch (Throwable t) {
exception = t;
}

6
transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDatagramChannel.java
View File

@ -31,6 +31,7 @@ import io.netty.channel.unix.DatagramSocketAddress;
import io.netty.channel.unix.Errors;
import io.netty.channel.unix.IovArray;
import io.netty.channel.unix.UnixChannelUtil;
import io.netty.util.UncheckedBooleanSupplier;
import io.netty.util.internal.ObjectUtil;
import io.netty.util.internal.StringUtil;
import io.netty.util.internal.UnstableApi;
@ -471,7 +472,10 @@ public final class KQueueDatagramChannel extends AbstractKQueueChannel implement
pipeline.fireChannelRead(packet);
byteBuf = null;
} while (allocHandle.continueReading());
// We use the TRUE_SUPPLIER as it is also ok to read less then what we did try to read (as long
// as we read anything).
} while (allocHandle.continueReading(UncheckedBooleanSupplier.TRUE_SUPPLIER));
} catch (Throwable t) {
if (byteBuf != null) {
byteBuf.release();

6
transport/src/main/java/io/netty/channel/nio/AbstractNioMessageChannel.java
View File

@ -55,6 +55,10 @@ public abstract class AbstractNioMessageChannel extends AbstractNioChannel {
super.doBeginRead();
}
protected boolean continueReading(RecvByteBufAllocator.Handle allocHandle) {
return allocHandle.continueReading();
}
private final class NioMessageUnsafe extends AbstractNioUnsafe {
private final List<Object> readBuf = new ArrayList<Object>();
@ -82,7 +86,7 @@ public abstract class AbstractNioMessageChannel extends AbstractNioChannel {
}
allocHandle.incMessagesRead(localRead);
} while (allocHandle.continueReading());
} while (continueReading(allocHandle));
} catch (Throwable t) {
exception = t;
}

12
transport/src/main/java/io/netty/channel/socket/nio/NioDatagramChannel.java
View File

@ -30,6 +30,7 @@ import io.netty.channel.nio.AbstractNioMessageChannel;
import io.netty.channel.socket.DatagramChannelConfig;
import io.netty.channel.socket.DatagramPacket;
import io.netty.channel.socket.InternetProtocolFamily;
import io.netty.util.UncheckedBooleanSupplier;
import io.netty.util.internal.ObjectUtil;
import io.netty.util.internal.SocketUtils;
import io.netty.util.internal.PlatformDependent;
@ -598,4 +599,15 @@ public final class NioDatagramChannel
}
return super.closeOnReadError(cause);
}
@Override
protected boolean continueReading(RecvByteBufAllocator.Handle allocHandle) {
if (allocHandle instanceof RecvByteBufAllocator.ExtendedHandle) {
// We use the TRUE_SUPPLIER as it is also ok to read less then what we did try to read (as long
// as we read anything).
return ((RecvByteBufAllocator.ExtendedHandle) allocHandle)
.continueReading(UncheckedBooleanSupplier.TRUE_SUPPLIER);
}
return allocHandle.continueReading();
}
}