Merge pull request #13 from normanmaurer/allochandle_read
Correctly respect RecvByteBufAllocator.Handle when reading
This commit is contained in:
commit
70682b238f
@ -56,45 +56,53 @@ abstract class AbstractIOUringServerChannel extends AbstractIOUringChannel imple
|
||||
final class UringServerChannelUnsafe extends AbstractIOUringChannel.AbstractUringUnsafe {
|
||||
@Override
|
||||
protected void scheduleRead0() {
|
||||
final IOUringRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
|
||||
allocHandle.reset(config());
|
||||
allocHandle.attemptedBytesRead(1);
|
||||
|
||||
IOUringSubmissionQueue submissionQueue = submissionQueue();
|
||||
//Todo get network addresses
|
||||
submissionQueue.addAccept(fd().intValue());
|
||||
submissionQueue.submit();
|
||||
}
|
||||
|
||||
// TODO: Respect MAX_MESSAGES_READ
|
||||
protected void readComplete0(int res) {
|
||||
final IOUringRecvByteAllocatorHandle allocHandle =
|
||||
(IOUringRecvByteAllocatorHandle) unsafe()
|
||||
.recvBufAllocHandle();
|
||||
final ChannelPipeline pipeline = pipeline();
|
||||
if (res >= 0) {
|
||||
allocHandle.incMessagesRead(1);
|
||||
try {
|
||||
Channel channel = newChildChannel(res);
|
||||
// Register accepted channel for POLLRDHUP
|
||||
IOUringSubmissionQueue submissionQueue = submissionQueue();
|
||||
submissionQueue.addPollRdHup(res);
|
||||
submissionQueue.submit();
|
||||
allocHandle.lastBytesRead(res);
|
||||
|
||||
pipeline.fireChannelRead(channel);
|
||||
} catch (Throwable cause) {
|
||||
if (res >= 0) {
|
||||
allocHandle.incMessagesRead(1);
|
||||
try {
|
||||
Channel channel = newChildChannel(res);
|
||||
// Register accepted channel for POLLRDHUP
|
||||
IOUringSubmissionQueue submissionQueue = submissionQueue();
|
||||
submissionQueue.addPollRdHup(res);
|
||||
submissionQueue.submit();
|
||||
|
||||
pipeline.fireChannelRead(channel);
|
||||
if (allocHandle.continueReading()) {
|
||||
scheduleRead();
|
||||
} else {
|
||||
allocHandle.readComplete();
|
||||
pipeline.fireExceptionCaught(cause);
|
||||
pipeline.fireChannelReadComplete();
|
||||
}
|
||||
if (config().isAutoRead()) {
|
||||
scheduleRead();
|
||||
}
|
||||
} else {
|
||||
} catch (Throwable cause) {
|
||||
allocHandle.readComplete();
|
||||
// Check if we did fail because there was nothing to accept atm.
|
||||
if (res != ERRNO_EAGAIN_NEGATIVE && res != ERRNO_EWOULDBLOCK_NEGATIVE) {
|
||||
// Something bad happened. Convert to an exception.
|
||||
pipeline.fireExceptionCaught(Errors.newIOException("io_uring accept", res));
|
||||
}
|
||||
pipeline.fireChannelReadComplete();
|
||||
pipeline.fireExceptionCaught(cause);
|
||||
}
|
||||
} else {
|
||||
allocHandle.readComplete();
|
||||
pipeline.fireChannelReadComplete();
|
||||
// Check if we did fail because there was nothing to accept atm.
|
||||
if (res != ERRNO_EAGAIN_NEGATIVE && res != ERRNO_EWOULDBLOCK_NEGATIVE) {
|
||||
// Something bad happened. Convert to an exception.
|
||||
pipeline.fireExceptionCaught(Errors.newIOException("io_uring accept", res));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
@ -202,12 +202,12 @@ abstract class AbstractIOUringStreamChannel extends AbstractIOUringChannel imple
|
||||
final ChannelConfig config = config();
|
||||
|
||||
final ByteBufAllocator allocator = config.getAllocator();
|
||||
final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
|
||||
final IOUringRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
|
||||
allocHandle.reset(config);
|
||||
|
||||
ByteBuf byteBuf = allocHandle.allocate(allocator);
|
||||
IOUringSubmissionQueue submissionQueue = submissionQueue();
|
||||
unsafe().recvBufAllocHandle().attemptedBytesRead(byteBuf.writableBytes());
|
||||
allocHandle.attemptedBytesRead(byteBuf.writableBytes());
|
||||
|
||||
assert readBuffer == null;
|
||||
readBuffer = byteBuf;
|
||||
@ -217,13 +217,11 @@ abstract class AbstractIOUringStreamChannel extends AbstractIOUringChannel imple
|
||||
submissionQueue.submit();
|
||||
}
|
||||
|
||||
// TODO: Respect MAX_MESSAGE_READ.
|
||||
@Override
|
||||
protected void readComplete0(int res) {
|
||||
boolean close = false;
|
||||
|
||||
final IOUringRecvByteAllocatorHandle allocHandle =
|
||||
(IOUringRecvByteAllocatorHandle) unsafe()
|
||||
.recvBufAllocHandle();
|
||||
final IOUringRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
|
||||
final ChannelPipeline pipeline = pipeline();
|
||||
ByteBuf byteBuf = this.readBuffer;
|
||||
this.readBuffer = null;
|
||||
@ -256,10 +254,9 @@ abstract class AbstractIOUringStreamChannel extends AbstractIOUringChannel imple
|
||||
}
|
||||
|
||||
allocHandle.incMessagesRead(1);
|
||||
boolean writable = byteBuf.isWritable();
|
||||
pipeline.fireChannelRead(byteBuf);
|
||||
byteBuf = null;
|
||||
if (!writable && config().isAutoRead()) {
|
||||
if (allocHandle.continueReading()) {
|
||||
// Let's schedule another read.
|
||||
scheduleRead();
|
||||
} else {
|
||||
@ -269,7 +266,6 @@ abstract class AbstractIOUringStreamChannel extends AbstractIOUringChannel imple
|
||||
} catch (Throwable t) {
|
||||
handleReadException(pipeline, byteBuf, t, close, allocHandle);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf,
|
||||
|
@ -34,11 +34,8 @@ final class IOUringEventLoop extends SingleThreadEventLoop implements
|
||||
private static final InternalLogger logger = InternalLoggerFactory.getInstance(IOUringEventLoop.class);
|
||||
|
||||
//Todo set config ring buffer size
|
||||
private final int ringSize = 32;
|
||||
private final int ENOENT = -2;
|
||||
|
||||
//just temporary -> Todo use ErrorsStaticallyReferencedJniMethods like in Epoll
|
||||
private static final int SOCKET_ERROR_EPIPE = -32;
|
||||
private static final int ringSize = 32;
|
||||
private static final int ENOENT = -2;
|
||||
private static final long ETIME = -62;
|
||||
static final long ECANCELED = -125;
|
||||
|
||||
@ -263,7 +260,7 @@ final class IOUringEventLoop extends SingleThreadEventLoop implements
|
||||
try {
|
||||
eventfd.close();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
logger.warn("Failed to close the event fd.", e);
|
||||
}
|
||||
ringBuffer.close();
|
||||
iovecArrayPool.release();
|
||||
|
@ -41,9 +41,4 @@ final class IOUringRecvByteAllocatorHandle extends RecvByteBufAllocator.Delegati
|
||||
public boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) {
|
||||
return ((RecvByteBufAllocator.ExtendedHandle) delegate()).continueReading(maybeMoreDataSupplier);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean continueReading() {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user