[#530] Allow using a bounded ByteBuf as the first inbound buffer
- Clean up - Do not stop reading when reached at maxCapacity. - Just let handler drain the buffer and try again quickly. - No more magic number in OIO buffer expansion
This commit is contained in:
parent
9e75a33d3d
commit
44daa99d3f
@ -141,31 +141,28 @@ public class AioSocketChannel extends AbstractAioChannel implements SocketChanne
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
private static boolean expandReadBuffer(ByteBuf byteBuf) {
|
private static void expandReadBuffer(ByteBuf byteBuf) {
|
||||||
final int maxCapacity = byteBuf.maxCapacity();
|
final int writerIndex = byteBuf.writerIndex();
|
||||||
final int capacity = byteBuf.capacity();
|
final int capacity = byteBuf.capacity();
|
||||||
|
if (capacity != writerIndex) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
final int maxCapacity = byteBuf.maxCapacity();
|
||||||
if (capacity == maxCapacity) {
|
if (capacity == maxCapacity) {
|
||||||
return false;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
// FIXME: Magic number
|
// FIXME: Magic number
|
||||||
final int increment = 4096;
|
final int increment = 4096;
|
||||||
|
|
||||||
final int writerIndex = byteBuf.writerIndex();
|
|
||||||
if (writerIndex != capacity) {
|
|
||||||
// No need to expand because there's a room in the buffer.
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Expand to maximum capacity.
|
|
||||||
if (writerIndex + increment > maxCapacity) {
|
if (writerIndex + increment > maxCapacity) {
|
||||||
|
// Expand to maximum capacity.
|
||||||
byteBuf.capacity(maxCapacity);
|
byteBuf.capacity(maxCapacity);
|
||||||
return true;
|
} else {
|
||||||
}
|
|
||||||
|
|
||||||
// Expand by the increment.
|
// Expand by the increment.
|
||||||
byteBuf.ensureWritableBytes(increment);
|
byteBuf.ensureWritableBytes(increment);
|
||||||
return true;
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -44,7 +44,7 @@ abstract class AbstractNioByteChannel extends AbstractNioChannel {
|
|||||||
boolean read = false;
|
boolean read = false;
|
||||||
try {
|
try {
|
||||||
expandReadBuffer(byteBuf);
|
expandReadBuffer(byteBuf);
|
||||||
for (;;) {
|
loop: for (;;) {
|
||||||
int localReadAmount = doReadBytes(byteBuf);
|
int localReadAmount = doReadBytes(byteBuf);
|
||||||
if (localReadAmount > 0) {
|
if (localReadAmount > 0) {
|
||||||
read = true;
|
read = true;
|
||||||
@ -52,8 +52,25 @@ abstract class AbstractNioByteChannel extends AbstractNioChannel {
|
|||||||
closed = true;
|
closed = true;
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
if (!expandReadBuffer(byteBuf)) {
|
|
||||||
|
switch (expandReadBuffer(byteBuf)) {
|
||||||
|
case 0:
|
||||||
|
// Read all - stop reading.
|
||||||
|
break loop;
|
||||||
|
case 1:
|
||||||
|
// Keep reading until everything is read.
|
||||||
break;
|
break;
|
||||||
|
case 2:
|
||||||
|
// Let the inbound handler drain the buffer and continue reading.
|
||||||
|
if (read) {
|
||||||
|
read = false;
|
||||||
|
pipeline.fireInboundBufferUpdated();
|
||||||
|
if (!byteBuf.writable()) {
|
||||||
|
throw new IllegalStateException(
|
||||||
|
"an inbound handler whose buffer is full must consume at " +
|
||||||
|
"least one byte.");
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
@ -100,30 +117,32 @@ abstract class AbstractNioByteChannel extends AbstractNioChannel {
|
|||||||
protected abstract int doReadBytes(ByteBuf buf) throws Exception;
|
protected abstract int doReadBytes(ByteBuf buf) throws Exception;
|
||||||
protected abstract int doWriteBytes(ByteBuf buf, boolean lastSpin) throws Exception;
|
protected abstract int doWriteBytes(ByteBuf buf, boolean lastSpin) throws Exception;
|
||||||
|
|
||||||
private static boolean expandReadBuffer(ByteBuf byteBuf) {
|
// 0 - not expanded because the buffer is writable
|
||||||
final int maxCapacity = byteBuf.maxCapacity();
|
// 1 - expanded because the buffer was not writable
|
||||||
|
// 2 - could not expand because the buffer was at its maximum although the buffer is not writable.
|
||||||
|
private static int expandReadBuffer(ByteBuf byteBuf) {
|
||||||
|
final int writerIndex = byteBuf.writerIndex();
|
||||||
final int capacity = byteBuf.capacity();
|
final int capacity = byteBuf.capacity();
|
||||||
|
if (capacity != writerIndex) {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
final int maxCapacity = byteBuf.maxCapacity();
|
||||||
if (capacity == maxCapacity) {
|
if (capacity == maxCapacity) {
|
||||||
return false;
|
return 2;
|
||||||
}
|
}
|
||||||
|
|
||||||
// FIXME: Magic number
|
// FIXME: Magic number
|
||||||
final int increment = 4096;
|
final int increment = 4096;
|
||||||
|
|
||||||
final int writerIndex = byteBuf.writerIndex();
|
|
||||||
if (writerIndex != capacity) {
|
|
||||||
// No need to expand because there's a room in the buffer.
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
// Expand to maximum capacity.
|
|
||||||
if (writerIndex + increment > maxCapacity) {
|
if (writerIndex + increment > maxCapacity) {
|
||||||
|
// Expand to maximum capacity.
|
||||||
byteBuf.capacity(maxCapacity);
|
byteBuf.capacity(maxCapacity);
|
||||||
return true;
|
} else {
|
||||||
}
|
|
||||||
|
|
||||||
// Expand by the increment.
|
// Expand by the increment.
|
||||||
byteBuf.ensureWritableBytes(increment);
|
byteBuf.ensureWritableBytes(increment);
|
||||||
return true;
|
}
|
||||||
|
|
||||||
|
return 1;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -40,13 +40,44 @@ abstract class AbstractOioByteChannel extends AbstractOioChannel {
|
|||||||
boolean closed = false;
|
boolean closed = false;
|
||||||
boolean read = false;
|
boolean read = false;
|
||||||
try {
|
try {
|
||||||
expandReadBuffer(byteBuf);
|
for (;;) {
|
||||||
int localReadAmount = doReadBytes(byteBuf);
|
int localReadAmount = doReadBytes(byteBuf);
|
||||||
if (localReadAmount > 0) {
|
if (localReadAmount > 0) {
|
||||||
read = true;
|
read = true;
|
||||||
} else if (localReadAmount < 0) {
|
} else if (localReadAmount < 0) {
|
||||||
closed = true;
|
closed = true;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
final int available = available();
|
||||||
|
if (available <= 0) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (byteBuf.writable()) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
final int capacity = byteBuf.capacity();
|
||||||
|
final int maxCapacity = byteBuf.maxCapacity();
|
||||||
|
if (capacity == maxCapacity) {
|
||||||
|
if (read) {
|
||||||
|
read = false;
|
||||||
|
pipeline.fireInboundBufferUpdated();
|
||||||
|
if (!byteBuf.writable()) {
|
||||||
|
throw new IllegalStateException(
|
||||||
|
"an inbound handler whose buffer is full must consume at " +
|
||||||
|
"least one byte.");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
final int writerIndex = byteBuf.writerIndex();
|
||||||
|
if (writerIndex + available > maxCapacity) {
|
||||||
|
byteBuf.capacity(maxCapacity);
|
||||||
|
} else {
|
||||||
|
byteBuf.ensureWritableBytes(available);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
} catch (Throwable t) {
|
} catch (Throwable t) {
|
||||||
if (read) {
|
if (read) {
|
||||||
read = false;
|
read = false;
|
||||||
@ -65,38 +96,6 @@ abstract class AbstractOioByteChannel extends AbstractOioChannel {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private boolean expandReadBuffer(ByteBuf byteBuf) {
|
|
||||||
final int maxCapacity = byteBuf.maxCapacity();
|
|
||||||
final int capacity = byteBuf.capacity();
|
|
||||||
if (capacity == maxCapacity) {
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
final int available = available();
|
|
||||||
final int writerIndex = byteBuf.writerIndex();
|
|
||||||
if (available > 0) {
|
|
||||||
if (writerIndex + available > maxCapacity) {
|
|
||||||
byteBuf.capacity(maxCapacity);
|
|
||||||
} else {
|
|
||||||
byteBuf.ensureWritableBytes(available);
|
|
||||||
}
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (writerIndex != capacity) {
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
// FIXME: magic number
|
|
||||||
final int increment = 4096;
|
|
||||||
if (writerIndex + increment > maxCapacity) {
|
|
||||||
byteBuf.capacity(maxCapacity);
|
|
||||||
} else {
|
|
||||||
byteBuf.ensureWritableBytes(increment);
|
|
||||||
}
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
Loading…
Reference in New Issue
Block a user