Merge pull request #37 from netty/bytebuffer-impl

Add a ByteBuffer based implementation of Buffer

Author: chrisvest

.github/workflows/ci-workflow.yml

@@ -33,7 +33,7 @@ jobs:
       - uses: actions/checkout@v2
 
       # Enable caching of Docker layers
-      - uses: satackey/[email protected].8
+      - uses: satackey/[email protected].11
         continue-on-error: true
         with:
           key: docker-cache-${{ steps.cache-key.outputs.key }}-{hash}

Dockerfile

@@ -29,7 +29,7 @@ WORKDIR /home/build
 
 # Prepare our own build
 COPY pom.xml pom.xml
-RUN mvn dependency:go-offline surefire:test -ntp
+RUN mvn dependency:go-offline surefire:test checkstyle:check -ntp
 
 # Copy over the project code and run our build
 COPY . .

pom.xml

@@ -69,14 +69,11 @@
   <properties>
     <javaModuleName>io.netty.incubator.buffer</javaModuleName>
     <netty.version>5.0.0.Final-SNAPSHOT</netty.version>
-    <netty.build.version>28</netty.build.version>
+    <netty.build.version>29</netty.build.version>
     <java.version>16</java.version>
     <junit.version>5.7.0</junit.version>
     <surefire.version>3.0.0-M5</surefire.version>
     <skipTests>false</skipTests>
-    <argLine.java9.extras />
-    <!-- Export some stuff which is used during our tests -->
-    <argLine.java9>--illegal-access=deny ${argLine.java9.extras}</argLine.java9>
     <argLine.common>
       -server
       -dsa -da -ea:io.netty...
@@ -95,9 +92,21 @@
       </extension>
     </extensions>
     <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-dependency-plugin</artifactId>
+        <version>3.1.2</version>
+        <executions>
+          <execution>
+            <goals>
+              <goal>properties</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
       <plugin>
         <artifactId>maven-compiler-plugin</artifactId>
-        <version>3.8.0</version>
+        <version>3.8.1</version>
         <configuration>
           <compilerVersion>${java.version}</compilerVersion>
           <fork>true</fork>
@@ -114,15 +123,18 @@
           <compilerArgs>
             <arg>--add-modules</arg>
             <arg>jdk.incubator.foreign</arg>
+
+            <!--
+             These two are really only needed for test-compile, but the maven-compiler-plugin cannot express that
+             -->
+            <arg>--patch-module</arg>
+            <arg>io.netty.buffer=${io.netty:netty-buffer:test-jar:tests}</arg>
           </compilerArgs>
-          <excludes>
-            <exclude>**/package-info.java</exclude>
-          </excludes>
         </configuration>
       </plugin>
       <plugin>
         <artifactId>maven-checkstyle-plugin</artifactId>
-        <version>3.1.0</version>
+        <version>3.1.2</version>
         <executions>
           <execution>
             <id>check-style</id>
@@ -148,7 +160,7 @@
           <dependency>
             <groupId>com.puppycrawl.tools</groupId>
             <artifactId>checkstyle</artifactId>
-            <version>8.29</version>
+            <version>8.41</version>
           </dependency>
           <dependency>
             <groupId>io.netty</groupId>
@@ -164,8 +176,7 @@
           <includes>
             <include>**/*Test*.java</include>
           </includes>
-          <runOrder>random</runOrder>
-          <argLine>${argLine.common} ${argLine.printGC} ${argLine.java9} --add-modules jdk.incubator.foreign</argLine>
+          <argLine>${argLine.common} ${argLine.printGC} --patch-module io.netty.buffer=${io.netty:netty-buffer:test-jar:tests} --add-modules jdk.incubator.foreign</argLine>
           <!-- Ensure the whole stacktrace is preserved when an exception is thrown. See https://issues.apache.org/jira/browse/SUREFIRE-1457 -->
           <trimStackTrace>false</trimStackTrace>
         </configuration>
@@ -235,7 +246,7 @@
       <plugin>
         <groupId>org.apache.felix</groupId>
         <artifactId>maven-bundle-plugin</artifactId>
-        <version>2.5.4</version>
+        <version>5.1.1</version>
         <executions>
           <execution>
             <id>generate-manifest</id>
@@ -399,6 +410,7 @@
       <artifactId>netty-buffer</artifactId>
       <version>${netty.version}</version>
       <type>test-jar</type>
+      <classifier>tests</classifier>
       <scope>test</scope>
     </dependency>
     <dependency>

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

@@ -408,9 +408,7 @@ default Buffer reset() {
      *
      * @return A {@link ByteCursor} for iterating the readable bytes of this buffer.
      */
-    default ByteCursor openCursor() {
-        return openCursor(readerOffset(), readableBytes());
-    }
+    ByteCursor openCursor();
 
     /**
      * Open a cursor to iterate the given number bytes of this buffer, starting at the given offset.

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

@@ -15,6 +15,8 @@
  */
 package io.netty.buffer.api;
 
+import io.netty.buffer.api.internal.Statics;
+
 import java.nio.ByteOrder;
 
 /**
@@ -75,18 +77,18 @@ default void close() {
     }
 
     static BufferAllocator heap() {
-        return new ManagedBufferAllocator(MemoryManager.getHeapMemoryManager(), Statics.CLEANER);
+        return new ManagedBufferAllocator(MemoryManagers.getManagers().getHeapMemoryManager(), Statics.CLEANER);
     }
 
     static BufferAllocator direct() {
-        return new ManagedBufferAllocator(MemoryManager.getNativeMemoryManager(), Statics.CLEANER);
+        return new ManagedBufferAllocator(MemoryManagers.getManagers().getNativeMemoryManager(), Statics.CLEANER);
     }
 
     static BufferAllocator pooledHeap() {
-        return new SizeClassedMemoryPool(MemoryManager.getHeapMemoryManager());
+        return new SizeClassedMemoryPool(MemoryManagers.getManagers().getHeapMemoryManager());
     }
 
     static BufferAllocator pooledDirect() {
-        return new SizeClassedMemoryPool(MemoryManager.getNativeMemoryManager());
+        return new SizeClassedMemoryPool(MemoryManagers.getManagers().getNativeMemoryManager());
     }
 }

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

@@ -18,7 +18,7 @@
 import java.lang.invoke.VarHandle;
 import java.util.Objects;
 
-import static io.netty.buffer.api.Statics.findVarHandle;
+import static io.netty.buffer.api.internal.Statics.findVarHandle;
 import static java.lang.invoke.MethodHandles.lookup;
 
 /**

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

@@ -20,8 +20,8 @@
 import java.lang.ref.WeakReference;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import static io.netty.buffer.api.Statics.CLEANER;
-import static io.netty.buffer.api.Statics.findVarHandle;
+import static io.netty.buffer.api.internal.Statics.CLEANER;
+import static io.netty.buffer.api.internal.Statics.findVarHandle;
 import static java.lang.invoke.MethodHandles.lookup;
 
 class CleanerPooledDrop implements Drop<Buffer> {

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

@@ -396,6 +396,11 @@ public void copyInto(int srcPos, Buffer dest, int destPos, int length) {
         }
     }
 
+    @Override
+    public ByteCursor openCursor() {
+        return openCursor(readerOffset(), readableBytes());
+    }
+
     @Override
     public ByteCursor openCursor(int fromOffset, int length) {
         if (fromOffset < 0) {
@@ -1147,6 +1152,7 @@ protected Owned<CompositeBuffer> prepareSend() {
             }
             throw throwable;
         }
+        boolean readOnly = this.readOnly;
         makeInaccessible();
         return new Owned<CompositeBuffer>() {
             @Override
@@ -1167,6 +1173,7 @@ void makeInaccessible() {
         capacity = 0;
         roff = 0;
         woff = 0;
+        readOnly = false;
         closed = true;
     }
 

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

@@ -17,7 +17,7 @@
 
 import java.lang.ref.Cleaner;
 
-import static io.netty.buffer.api.Statics.NO_OP_DROP;
+import static io.netty.buffer.api.internal.Statics.NO_OP_DROP;
 
 class ManagedBufferAllocator implements BufferAllocator, AllocatorControl {
     private final MemoryManager manager;
@@ -44,6 +44,6 @@ public Object allocateUntethered(Buffer originator, int size) {
     @Override
     public void recoverMemory(Object memory) {
         // Free the recovered memory.
-        manager.recoverMemory(memory, manager.drop()).close();
+        manager.recoverMemory(this, memory, manager.drop()).close();
     }
 }

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

@@ -21,19 +21,11 @@
 import java.lang.ref.Cleaner;
 
 public interface MemoryManager {
-    static MemoryManager getHeapMemoryManager() {
-        return new HeapMemorySegmentManager();
-    }
-
-    static MemoryManager getNativeMemoryManager() {
-        return new NativeMemorySegmentManager();
-    }
-
     boolean isNative();
-    Buffer allocateConfined(AllocatorControl alloc, long size, Drop<Buffer> drop, Cleaner cleaner);
-    Buffer allocateShared(AllocatorControl allo, long size, Drop<Buffer> drop, Cleaner cleaner);
+    Buffer allocateConfined(AllocatorControl allocatorControl, long size, Drop<Buffer> drop, Cleaner cleaner);
+    Buffer allocateShared(AllocatorControl allocatorControl, long size, Drop<Buffer> drop, Cleaner cleaner);
     Drop<Buffer> drop();
     Object unwrapRecoverableMemory(Buffer buf);
     int capacityOfRecoverableMemory(Object memory);
-    Buffer recoverMemory(Object recoverableMemory, Drop<Buffer> drop);
+    Buffer recoverMemory(AllocatorControl allocatorControl, Object recoverableMemory, Drop<Buffer> drop);
 }

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

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+import java.util.ServiceLoader;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+
+/**
+ * The MemoryManagers interface is the handle through which {@link BufferAllocator buffer allocators} access the low
+ * level memory management APIs.
+ * <p>
+ * This is hidden behind this interface in order to make allocation and pool agnostic and reusable across buffer and
+ * memory implementations.
+ */
+public interface MemoryManagers {
+    /**
+     * Get the default, or currently configured, memory managers instance.
+     * @return A MemoryManagers instance.
+     */
+    static MemoryManagers getManagers() {
+        return MemoryManagersOverride.getManagers();
+    }
+
+    /**
+     * Temporarily override the default configured memory managers instance.
+     * <p>
+     * Calls to {@link #getManagers()} from within the given supplier will get the given managers instance.
+     *
+     * @param managers Override the default configured managers instance with this instance.
+     * @param supplier The supplier function to be called while the override is in place.
+     * @param <T> The result type from the supplier.
+     * @return The result from the supplier.
+     */
+    static <T> T using(MemoryManagers managers, Supplier<T> supplier) {
+        return MemoryManagersOverride.using(managers, supplier);
+    }
+
+    /**
+     * Get a lazy-loading stream of all available memory managers.
+     *
+     * @return A stream of providers of memory managers instances.
+     */
+    static Stream<ServiceLoader.Provider<MemoryManagers>> getAllManagers() {
+        var loader = ServiceLoader.load(MemoryManagers.class);
+        return loader.stream();
+    }
+
+    /**
+     * Get a {@link MemoryManager} instance that is suitable for allocating on-heap {@link Buffer} instances.
+     *
+     * @return An on-heap {@link MemoryManager}.
+     */
+    MemoryManager getHeapMemoryManager();
+
+    /**
+     * Get a {@link MemoryManager} instance that is suitable for allocating off-heap {@link Buffer} instances.
+     *
+     * @return An off-heap {@link MemoryManager}.
+     */
+    MemoryManager getNativeMemoryManager();
+}

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

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+import io.netty.buffer.api.memseg.SegmentMemoryManagers;
+
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+
+final class MemoryManagersOverride {
+    private static final MemoryManagers DEFAULT = new SegmentMemoryManagers();
+    private static final AtomicInteger OVERRIDES_AVAILABLE = new AtomicInteger();
+    private static final Map<Thread, MemoryManagers> OVERRIDES = Collections.synchronizedMap(new IdentityHashMap<>());
+
+    private MemoryManagersOverride() {
+    }
+
+    static MemoryManagers getManagers() {
+        if (OVERRIDES_AVAILABLE.get() > 0) {
+            return OVERRIDES.getOrDefault(Thread.currentThread(), DEFAULT);
+        }
+        return DEFAULT;
+    }
+
+    static <T> T using(MemoryManagers managers, Supplier<T> supplier) {
+        Thread thread = Thread.currentThread();
+        OVERRIDES.put(thread, managers);
+        OVERRIDES_AVAILABLE.incrementAndGet();
+        try {
+            return supplier.get();
+        } finally {
+            OVERRIDES_AVAILABLE.decrementAndGet();
+            OVERRIDES.remove(thread);
+        }
+    }
+}