Add support for read-only buffers

Motivation:
There are cases where you want a buffer to be "constant."
Buffers are inherently mutable, but it's possible to block off write access to the buffer contents.
This doesn't make it completely safe to share the buffer across multiple threads, but it does catch most races that could occur.

Modification:
Add a method to Buf for toggling read-only mode.
When a buffer is read-only, the write accessors throw exceptions when called.
In the MemSegBuf, this is implemented by having separate read and write references to the underlying memory segment.
In a read-only buffer, the write reference is redirected to point to a closed memory segment, thus preventing all writes to the memory backing the buffer.

Result:
It is now possible to make buffers read-only.
Note, however, that it is also possible to toggle a read-only buffer back to writable.
We need that in order for buffer pools to be able to fully reset the state of a buffer, regardless of the buffer implementation.

Author: chrisvest

src/main/java/io/netty/buffer/api/Allocator.java

@@ -70,7 +70,7 @@ default Buf allocate(int size, ByteOrder order) {
      * <pre>{@code
      *     try (Buf a = allocator.allocate(size);
      *          Buf b = allocator.allocate(size)) {
-     *         return Buf.compose(a, b); // Reference counts for 'a' and 'b' incremented here.
+     *         return allocator.compose(a, b); // Reference counts for 'a' and 'b' incremented here.
      *     } // Reference count for 'a' and 'b' decremented here; composite buffer now holds the last references.
      * }</pre>
      * <p>

src/main/java/io/netty/buffer/api/Buf.java

@@ -154,6 +154,7 @@ public interface Buf extends Rc<Buf>, BufAccessors {
      * @return This Buf.
      * @throws IndexOutOfBoundsException if the specified {@code offset} is less than the current
      * {@link #readerOffset()} or greater than {@link #capacity()}.
+     * @throws IllegalStateException if this buffer is {@linkplain #readOnly() read-only}.
      */
     Buf writerOffset(int offset);
 
@@ -178,6 +179,7 @@ default int writableBytes() {
      *
      * @param value The byte value to write at every position in the buffer.
      * @return This Buf.
+     * @throws IllegalStateException if this buffer is {@linkplain #readOnly() read-only}.
      */
     Buf fill(byte value);
 
@@ -187,6 +189,20 @@ default int writableBytes() {
      */
     long getNativeAddress();
 
+    /**
+     * Set the read-only state of this buffer.
+     *
+     * @return this buffer.
+     */
+    Buf readOnly(boolean readOnly);
+
+    /**
+     * Query if this buffer is read-only or not.
+     *
+     * @return {@code true} if this buffer is read-only, {@code false} otherwise.
+     */
+    boolean readOnly();
+
     /**
      * Returns a slice of this buffer's readable bytes.
      * Modifying the content of the returned buffer or this buffer affects each other's content,
@@ -371,8 +387,8 @@ default ByteCursor openReverseCursor() {
      * {@code false}.
      *
      * @param size The requested number of bytes of space that should be available for writing.
-     * @throws IllegalStateException if this buffer is not in an owned state.
-     * That is, if {@link #isOwned()} is {@code false}.
+     * @throws IllegalStateException if this buffer is not in an {@linkplain #isOwned() owned} state,
+     * or is {@linkplain #readOnly() read-only}.
      */
     default void ensureWritable(int size) {
         ensureWritable(size, true);
@@ -410,8 +426,8 @@ default void ensureWritable(int size) {
      * @param allowCompaction {@code true} if the method is allowed to modify the
      *                                   {@linkplain #readerOffset() reader offset} and
      *                                   {@linkplain #writerOffset() writer offset}, otherwise {@code false}.
-     * @throws IllegalStateException if this buffer is not in an owned state.
-     * That is, if {@link #isOwned()} is {@code false}.
+     * @throws IllegalStateException if this buffer is not in an {@linkplain #isOwned() owned} state,
+     *      * or is {@linkplain #readOnly() read-only}.
      */
     void ensureWritable(int size, boolean allowCompaction);
 
@@ -462,7 +478,8 @@ default void ensureWritable(int size) {
     /**
      * Discards the read bytes, and moves the buffer contents to the beginning of the buffer.
      *
-     * The buffer must be {@linkplain #isOwned() owned}, or an exception will be thrown.
+     * @throws IllegalStateException if this buffer is not in an {@linkplain #isOwned() owned} state,
+     * or is {@linkplain #readOnly() read-only}.
      */
     void compact();
 }

src/main/java/io/netty/buffer/api/CompositeBuf.java

@@ -20,9 +20,6 @@
 import java.util.Arrays;
 import java.util.Objects;
 
-import static jdk.incubator.foreign.MemoryAccess.setByteAtOffset;
-import static jdk.incubator.foreign.MemoryAccess.setLongAtOffset;
-
 final class CompositeBuf extends RcSupport<Buf, CompositeBuf> implements Buf {
     /**
      * The max array size is JVM implementation dependant, but most seem to settle on {@code Integer.MAX_VALUE - 8}.
@@ -48,6 +45,7 @@ final class CompositeBuf extends RcSupport<Buf, CompositeBuf> implements Buf {
     private int subOffset; // The next offset *within* a consituent buffer to read from or write to.
     private ByteOrder order;
     private boolean closed;
+    private boolean readOnly;
 
     CompositeBuf(Allocator allocator, Buf[] bufs) {
         this(allocator, true, bufs.clone(), COMPOSITE_DROP); // Clone prevents external modification of array.
@@ -68,6 +66,14 @@ private CompositeBuf(Allocator allocator, boolean isSendable, Buf[] bufs, Drop<C
                 }
             }
             order = bufs[0].order();
+
+            boolean targetReadOnly = bufs[0].readOnly();
+            for (Buf buf : bufs) {
+                if (buf.readOnly() != targetReadOnly) {
+                    throw new IllegalArgumentException("Constituent buffers have inconsistent read-only state.");
+                }
+            }
+            readOnly = targetReadOnly;
         } else {
             order = ByteOrder.nativeOrder();
         }
@@ -201,6 +207,20 @@ public long getNativeAddress() {
         return 0;
     }
 
+    @Override
+    public Buf readOnly(boolean readOnly) {
+        for (Buf buf : bufs) {
+            buf.readOnly(readOnly);
+        }
+        this.readOnly = readOnly;
+        return this;
+    }
+
+    @Override
+    public boolean readOnly() {
+        return readOnly;
+    }
+
     @Override
     public Buf slice(int offset, int length) {
         checkWriteBounds(offset, length);
@@ -236,7 +256,7 @@ public Buf slice(int offset, int length) {
                 slices = new Buf[] { choice.slice(subOffset, 0) };
             }
 
-            return new CompositeBuf(allocator, false, slices, drop).writerOffset(length);
+            return new CompositeBuf(allocator, false, slices, drop);
         } catch (Throwable throwable) {
             // We called acquire prior to the try-clause. We need to undo that if we're not creating a composite buffer:
             close();
@@ -267,10 +287,10 @@ private void copyInto(int srcPos, CopyInto dest, int destPos, int length) {
             throw new IndexOutOfBoundsException("Length cannot be negative: " + length + '.');
         }
         if (srcPos < 0) {
-            throw indexOutOfBounds(srcPos);
+            throw indexOutOfBounds(srcPos, false);
         }
         if (srcPos + length > capacity) {
-            throw indexOutOfBounds(srcPos + length);
+            throw indexOutOfBounds(srcPos + length, false);
         }
         while (length > 0) {
             var buf = (Buf) chooseBuffer(srcPos, 0);
@@ -293,10 +313,10 @@ public void copyInto(int srcPos, Buf dest, int destPos, int length) {
             throw new IndexOutOfBoundsException("Length cannot be negative: " + length + '.');
         }
         if (srcPos < 0) {
-            throw indexOutOfBounds(srcPos);
+            throw indexOutOfBounds(srcPos, false);
         }
         if (srcPos + length > capacity) {
-            throw indexOutOfBounds(srcPos + length);
+            throw indexOutOfBounds(srcPos + length, false);
         }
 
         // Iterate in reverse to account for src and dest buffer overlap.
@@ -530,6 +550,9 @@ public void ensureWritable(int size, boolean allowCompaction) {
         if (size < 0) {
             throw new IllegalArgumentException("Cannot ensure writable for a negative size: " + size + '.');
         }
+        if (readOnly) {
+            throw bufferIsReadOnly();
+        }
         if (writableBytes() >= size) {
             // We already have enough space.
             return;
@@ -586,15 +609,23 @@ void extendWith(Buf extension) {
                     "This buffer uses " + order() + " byte order, and cannot be extended with " +
                     "a buffer that uses " + extension.order() + " byte order.");
         }
-        if (bufs.length == 0) {
-            order = extension.order();
+        if (bufs.length > 0 && extension.readOnly() != readOnly()) {
+            throw new IllegalArgumentException(
+                    "This buffer is " + (readOnly? "read-only" : "writable") + ", " +
+                    "and cannot be extended with a buffer that is " +
+                    (extension.readOnly()? "read-only." : "writable."));
         }
+
         long newSize = capacity() + (long) extension.capacity();
         Allocator.checkSize(newSize);
 
         Buf[] restoreTemp = bufs; // We need this to restore our buffer array, in case offset computations fail.
         try {
             unsafeExtendWith(extension.acquire());
+            if (restoreTemp.length == 0) {
+                order = extension.order();
+                readOnly = extension.readOnly();
+            }
         } catch (Exception e) {
             bufs = restoreTemp;
             throw e;
@@ -642,6 +673,9 @@ public void compact() {
         if (!isOwned()) {
             throw new IllegalStateException("Buffer must be owned in order to compact.");
         }
+        if (readOnly()) {
+            throw new IllegalStateException("Buffer must be writable in order to compact, but was read-only.");
+        }
         int distance = roff;
         if (distance == 0) {
             return;
@@ -962,6 +996,7 @@ public CompositeBuf transferOwnership(Drop<CompositeBuf> drop) {
                     received[i] = sends[i].receive();
                 }
                 var composite = new CompositeBuf(allocator, true, received, drop);
+                composite.readOnly = readOnly;
                 drop.attach(composite);
                 return composite;
             }
@@ -1034,7 +1069,7 @@ private BufAccessors prepRead(int index, int size) {
 
     private void checkReadBounds(int index, int size) {
         if (index < 0 || woff < index + size) {
-            throw indexOutOfBounds(index);
+            throw indexOutOfBounds(index, false);
         }
     }
 
@@ -1051,19 +1086,30 @@ private BufAccessors prepWrite(int index, int size) {
 
     private void checkWriteBounds(int index, int size) {
         if (index < 0 || capacity < index + size) {
-            throw indexOutOfBounds(index);
+            throw indexOutOfBounds(index, true);
         }
     }
 
-    private RuntimeException indexOutOfBounds(int index) {
+    private RuntimeException indexOutOfBounds(int index, boolean write) {
         if (closed) {
-            return new IllegalStateException("This buffer is closed.");
+            return bufferIsClosed();
+        }
+        if (write && readOnly) {
+            return bufferIsReadOnly();
         }
         return new IndexOutOfBoundsException(
                 "Index " + index + " is out of bounds: [read 0 to " + woff + ", write 0 to " +
                 (capacity - 1) + "].");
     }
 
+    private static IllegalStateException bufferIsClosed() {
+        return new IllegalStateException("This buffer is closed.");
+    }
+
+    private static IllegalStateException bufferIsReadOnly() {
+        return new IllegalStateException("This buffer is read-only.");
+    }
+
     private BufAccessors chooseBuffer(int index, int size) {
         int i = searchOffsets(index);
         if (i == bufs.length) {

src/main/java/io/netty/buffer/api/SizeClassedMemoryPool.java

@@ -43,7 +43,11 @@ public Buf allocate(int size) {
         var sizeClassPool = getSizeClassPool(size);
         Send<Buf> send = sizeClassPool.poll();
         if (send != null) {
-            return send.receive().reset().fill((byte) 0).order(ByteOrder.nativeOrder());
+            return send.receive()
+                       .reset()
+                       .readOnly(false)
+                       .fill((byte) 0)
+                       .order(ByteOrder.nativeOrder());
         }
         return createBuf(size, getDrop());
     }