andreacavalli/netty-incubator-buffer-api
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Address PR review comments

Browse Source
This commit is contained in:
Chris Vest 2021-01-18 11:57:35 +01:00
parent 1c1149395b
commit e22b57ddcd
4 changed files with 48 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
src/main/java/io/netty/buffer/api/Buf.java
View File

@ -15,8 +15,8 @@
*/
package io.netty.buffer.api;
import io.netty.buffer.api.ComponentProcessor.OfReadable;
import io.netty.buffer.api.ComponentProcessor.OfWritable;
import io.netty.buffer.api.ComponentProcessor.ReadableComponentProcessor;
import io.netty.buffer.api.ComponentProcessor.WritableComponentProcessor;
import io.netty.buffer.api.ComponentProcessor.ReadableComponent;
import io.netty.buffer.api.ComponentProcessor.WritableComponent;
@ -498,9 +498,9 @@ public interface Buf extends Rc<Buf>, BufAccessors {
/**
* Get the number of "components" in this buffer, that are readable. These are the components that would be
* processed by {@link #forEachReadable(int, OfReadable)}. For composite buffers, this is the number of
* transitive constituent buffers that are readable, while non-composite buffers only have at most one readable
* component.
* processed by {@link #forEachReadable(int, ReadableComponentProcessor)}. For composite buffers, this is the
* number of transitive constituent buffers that are readable, while non-composite buffers only have at most one
* readable component.
* <p>
* The number of readable components may be less than the {@link #countComponents() component count}, if not all of
* them have readable data.
@ -511,9 +511,9 @@ public interface Buf extends Rc<Buf>, BufAccessors {
/**
* Get the number of "components" in this buffer, that are writable. These are the components that would be
* processed by {@link #forEachWritable(int, OfWritable)}. For composite buffers, this is the number of
* transitive constituent buffers that are writable, while non-composite buffers only have at most one writable
* component.
* processed by {@link #forEachWritable(int, WritableComponentProcessor)}. For composite buffers, this is the
* number of transitive constituent buffers that are writable, while non-composite buffers only have at most one
* writable component.
* <p>
* The number of writable components may be less than the {@link #countComponents() component count}, if not all of
* them have space for writing.
@ -525,14 +525,15 @@ public interface Buf extends Rc<Buf>, BufAccessors {
/**
* Process all readable components of this buffer, and return the number of components processed.
* <p>
* The given {@linkplain OfReadable processor} is called for each readable component in this buffer,
* The given {@linkplain ReadableComponentProcessor processor} is called for each readable component in this buffer,
* and passed a component index, for the given component in the iteration, and a {@link ReadableComponent} object
* for accessing the data within the given component.
* <p>
* The component index is specific to the particular invokation of this method. The first call to the consumer will
* be passed the given initial index, and the next call will be passed the initial index plus one, and so on.
* <p>
* The {@linkplain OfReadable component processor} may stop the iteration at any time by returning {@code false}.
* The {@linkplain ReadableComponentProcessor component processor} may stop the iteration at any time by returning
* {@code false}.
* This will cause the number of components processed to be returned as a negative number (to signal early return),
* and the number of components processed may then be less than the
* {@linkplain #countReadableComponents() readable component count}.
@ -545,7 +546,7 @@ public interface Buf extends Rc<Buf>, BufAccessors {
* such changes, are any method that requires the buffer to be {@linkplain #isOwned() owned}.
* <p>
* The best way to ensure this doesn't cause any trouble, is to use the buffers directly as part of the iteration,
* or immediately after the iteration.
* or immediately after the iteration while we are still in the scope of the method that triggered the iteration.
* <p>
* <strong>Note</strong> that the arrays, memory addresses, and byte buffers exposed as components by this method,
* should not be used for changing the buffer contents. Doing so may cause undefined behaviour.
@ -554,26 +555,27 @@ public interface Buf extends Rc<Buf>, BufAccessors {
* this buffer instance.
*
* @param initialIndex The initial index of the iteration, and the index that will be passed to the first call to
* the {@linkplain OfReadable#process(int, ReadableComponent) processor}.
* the {@linkplain ReadableComponentProcessor#process(int, ReadableComponent) processor}.
* @param processor The processor that will be used to process the buffer components.
* @return The number of readable components processed, as a positive number of all readable components were
* processed, or as a negative number if the iteration was stopped because
* {@link OfReadable#process(int, ReadableComponent)} returned {@code false}.
* {@link ReadableComponentProcessor#process(int, ReadableComponent)} returned {@code false}.
* In any case, the number of components processed may be less than {@link #countComponents()}.
*/
int forEachReadable(int initialIndex, OfReadable processor);
int forEachReadable(int initialIndex, ReadableComponentProcessor processor);
/**
* Process all writable components of this buffer, and return the number of components processed.
* <p>
* The given {@linkplain OfWritable processor} is called for each writable component in this buffer,
* The given {@linkplain WritableComponentProcessor processor} is called for each writable component in this buffer,
* and passed a component index, for the given component in the iteration, and a {@link WritableComponent} object
* for accessing the data within the given component.
* <p>
* The component index is specific to the particular invokation of this method. The first call to the consumer will
* be passed the given initial index, and the next call will be passed the initial index plus one, and so on.
* <p>
* The {@link OfWritable component processor} may stop the iteration at any time by returning {@code false}.
* The {@link WritableComponentProcessor component processor} may stop the iteration at any time by returning
* {@code false}.
* This will cause the number of components processed to be returned as a negative number (to signal early return),
* and the number of components processed may then be less than the
* {@linkplain #countReadableComponents() readable component count}.
@ -586,18 +588,18 @@ public interface Buf extends Rc<Buf>, BufAccessors {
* such changes, are any method that requires the buffer to be {@linkplain #isOwned() owned}.
* <p>
* The best way to ensure this doesn't cause any trouble, is to use the buffers directly as part of the iteration,
* or immediately after the iteration.
* or immediately after the iteration while we are still in the scope of the method that triggered the iteration.
* <p>
* Changes to position and limit of the byte buffers exposed via the processed components, are not reflected back to
* this buffer instance.
*
* @param initialIndex The initial index of the iteration, and the index that will be passed to the first call to
* the {@linkplain OfWritable#process(int, WritableComponent) processor}.
* the {@linkplain WritableComponentProcessor#process(int, WritableComponent) processor}.
* @param processor The processor that will be used to process the buffer components.
* @return The number of writable components processed, as a positive number of all writable components were
* processed, or as a negative number if the iteration was stopped because
* {@link OfWritable#process(int, WritableComponent)} returned {@code false}.
* {@link WritableComponentProcessor#process(int, WritableComponent)} returned {@code false}.
* In any case, the number of components processed may be less than {@link #countComponents()}.
*/
int forEachWritable(int initialIndex, OfWritable processor);
int forEachWritable(int initialIndex, WritableComponentProcessor processor);
}

27
src/main/java/io/netty/buffer/api/ComponentProcessor.java
View File

@ -18,24 +18,25 @@ package io.netty.buffer.api;
import java.nio.ByteBuffer;
/**
* This interface contain a collection of APIs used in the {@link Buf#forEachReadable(int, OfReadable)} and
* {@link Buf#forEachWritable(int, OfWritable)} methods.
* This interface contain a collection of APIs used in the {@link Buf#forEachReadable(int, ReadableComponentProcessor)}
* and {@link Buf#forEachWritable(int, WritableComponentProcessor)} methods.
*/
public interface ComponentProcessor {
/**
* A processor of {@linkplain ReadableComponent readable components}.
*/
@FunctionalInterface
interface OfReadable extends ComponentProcessor {
interface ReadableComponentProcessor extends ComponentProcessor {
/**
* Process the given component at the given index in the {@link Buf#forEachReadable(int, OfReadable) iteration}.
* Process the given component at the given index in the
* {@link Buf#forEachReadable(int, ReadableComponentProcessor) iteration}.
* <p>
* The component object itself is only valid during this call, but the {@link ByteBuffer byte buffers}, arrays,
* and native address pointers obtained from it, will be valid until any
* {@link Buf#isOwned() ownership} requiring operation is performed on the buffer.
*
* @param index The current index of the given buffer component, based on the initial index passed to the
* {@link Buf#forEachReadable(int, OfReadable)} method.
* {@link Buf#forEachReadable(int, ReadableComponentProcessor)} method.
* @param component The current buffer component being processed.
* @return {@code true} if the iteration should continue and more components should be processed, otherwise
* {@code false} to stop the iteration early.
@ -47,17 +48,17 @@ public interface ComponentProcessor {
* A processor of {@linkplain WritableComponent writable components}.
*/
@FunctionalInterface
interface OfWritable extends ComponentProcessor {
interface WritableComponentProcessor extends ComponentProcessor {
/**
* Process the given component at the given index in the
* {@link Buf#forEachWritable(int, OfWritable)} iteration}.
* {@link Buf#forEachWritable(int, WritableComponentProcessor)} iteration}.
* <p>
* The component object itself is only valid during this call, but the {@link ByteBuffer byte buffers}, arrays,
* and native address pointers obtained from it, will be valid until any
* {@link Buf#isOwned() ownership} requiring operation is performed on the buffer.
*
* @param index The current index of the given buffer component, based on the initial index passed to the
* {@link Buf#forEachWritable(int, OfWritable)} method.
* {@link Buf#forEachWritable(int, WritableComponentProcessor)} method.
* @param component The current buffer component being processed.
* @return {@code true} if the iteration should continue and more components should be processed, otherwise
* {@code false} to stop the iteration early.
@ -67,7 +68,7 @@ public interface ComponentProcessor {
/**
* A view onto the buffer component being processed in a given iteration of
* {@link Buf#forEachReadable(int, OfReadable)}.
* {@link Buf#forEachReadable(int, ReadableComponentProcessor)}.
*/
interface ReadableComponent {
@ -115,16 +116,16 @@ public interface ComponentProcessor {
* Get a {@link ByteBuffer} instance for this memory component.
* <p>
* <strong>Note</strong> that the {@link ByteBuffer} is read-only, to prevent write accesses to the memory,
* when the buffer component is obtained through {@link Buf#forEachReadable(int, OfReadable)}.
* when the buffer component is obtained through {@link Buf#forEachReadable(int, ReadableComponentProcessor)}.
*
* @return A new {@link ByteBuffer} for this memory component.
* @return A new {@link ByteBuffer}, with its own position and limit, for this memory component.
*/
ByteBuffer readableBuffer();
}
/**
* A view onto the buffer component being processed in a given iteration of
* {@link Buf#forEachWritable(int, OfWritable)}.
* {@link Buf#forEachWritable(int, WritableComponentProcessor)}.
*/
interface WritableComponent {
@ -162,7 +163,7 @@ public interface ComponentProcessor {
* Get a {@link ByteBuffer} instance for this memory component, which can be used for modifying the buffer
* contents.
*
* @return A new {@link ByteBuffer} for this memory component.
* @return A new {@link ByteBuffer}, with its own position and limit, for this memory component.
*/
ByteBuffer writableBuffer();
}

7
src/main/java/io/netty/buffer/api/CompositeBuf.java
View File

@ -15,6 +15,9 @@
*/
package io.netty.buffer.api;
import io.netty.buffer.api.ComponentProcessor.ReadableComponentProcessor;
import io.netty.buffer.api.ComponentProcessor.WritableComponentProcessor;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.Arrays;
@ -750,7 +753,7 @@ final class CompositeBuf extends RcSupport<Buf, CompositeBuf> implements Buf {
}
@Override
public int forEachReadable(int initialIndex, ComponentProcessor.OfReadable processor) {
public int forEachReadable(int initialIndex, ReadableComponentProcessor processor) {
checkReadBounds(readerOffset(), Math.max(1, readableBytes()));
int visited = 0;
for (Buf buf : bufs) {
@ -768,7 +771,7 @@ final class CompositeBuf extends RcSupport<Buf, CompositeBuf> implements Buf {
}
@Override
public int forEachWritable(int initialIndex, ComponentProcessor.OfWritable processor) {
public int forEachWritable(int initialIndex, WritableComponentProcessor processor) {
checkWriteBounds(writerOffset(), Math.max(1, writableBytes()));
int visited = 0;
for (Buf buf : bufs) {

11
src/main/java/io/netty/buffer/api/memseg/MemSegBuf.java
View File

@ -19,9 +19,10 @@ import io.netty.buffer.api.Allocator;
import io.netty.buffer.api.AllocatorControl;
import io.netty.buffer.api.Buf;
import io.netty.buffer.api.ByteCursor;
import io.netty.buffer.api.ComponentProcessor;
import io.netty.buffer.api.ComponentProcessor.ReadableComponent;
import io.netty.buffer.api.ComponentProcessor.ReadableComponentProcessor;
import io.netty.buffer.api.ComponentProcessor.WritableComponent;
import io.netty.buffer.api.ComponentProcessor.WritableComponentProcessor;
import io.netty.buffer.api.Drop;
import io.netty.buffer.api.Owned;
import io.netty.buffer.api.RcSupport;
@ -61,7 +62,9 @@ class MemSegBuf extends RcSupport<Buf, MemSegBuf> implements Buf, ReadableCompon
private final AllocatorControl alloc;
private final boolean isSendable;
private final int baseOffset; // TODO remove this when JDK bug is fixed (slices of heap buffers)
// TODO remove baseOffset when JDK bug is fixed (slices of heap buffers)
// See https://mail.openjdk.java.net/pipermail/panama-dev/2021-January/011810.html
private final int baseOffset;
private MemorySegment seg;
private MemorySegment wseg;
private ByteOrder order;
@ -551,13 +554,13 @@ class MemSegBuf extends RcSupport<Buf, MemSegBuf> implements Buf, ReadableCompon
}
@Override
public int forEachReadable(int initialIndex, ComponentProcessor.OfReadable processor) {
public int forEachReadable(int initialIndex, ReadableComponentProcessor processor) {
checkRead(readerOffset(), Math.max(1, readableBytes()));
return processor.process(initialIndex, this)? 1 : -1;
}
@Override
public int forEachWritable(int initialIndex, ComponentProcessor.OfWritable processor) {
public int forEachWritable(int initialIndex, WritableComponentProcessor processor) {
checkWrite(writerOffset(), Math.max(1, writableBytes()));
return processor.process(initialIndex, this)? 1 : -1;
}