Merge pull request #36 from netty/rc-thread-safety

Allow slices to obtain ownership when parent is closed

Author: chrisvest

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

@@ -45,6 +45,7 @@ public void drop(Buffer buf) {
         GatedCleanable c = (GatedCleanable) CLEANABLE.getAndSet(this, null);
         if (c != null) {
             c.clean();
+            delegate.drop(buf);
         }
     }
 
@@ -57,24 +58,43 @@ public void attach(Buffer buf) {
             c.clean();
         }
 
-        var pool = this.pool;
         var mem = manager.unwrapRecoverableMemory(buf);
-        var delegate = this.delegate;
-        WeakReference<Buffer> ref = new WeakReference<>(buf);
+        WeakReference<CleanerPooledDrop> ref = new WeakReference<>(this);
         AtomicBoolean gate = new AtomicBoolean(true);
-        cleanable = new GatedCleanable(gate, CLEANER.register(this, () -> {
+        cleanable = new GatedCleanable(gate, CLEANER.register(this, new CleanAction(pool, mem, ref, gate)));
+    }
+
+    @Override
+    public String toString() {
+        return "CleanerPooledDrop(" + delegate + ')';
+    }
+
+    private static final class CleanAction implements Runnable {
+        private final SizeClassedMemoryPool pool;
+        private final Object mem;
+        private final WeakReference<CleanerPooledDrop> ref;
+        private final AtomicBoolean gate;
+
+        private CleanAction(SizeClassedMemoryPool pool, Object mem, WeakReference<CleanerPooledDrop> ref,
+                            AtomicBoolean gate) {
+            this.pool = pool;
+            this.mem = mem;
+            this.ref = ref;
+            this.gate = gate;
+        }
+
+        @Override
+        public void run() {
             if (gate.getAndSet(false)) {
-                Buffer b = ref.get();
-                if (b == null) {
+                var monitored = ref.get();
+                if (monitored == null) {
                     pool.recoverMemory(mem);
-                } else {
-                    delegate.drop(b);
                 }
             }
-        }));
+        }
     }
 
-    private static class GatedCleanable implements Cleanable {
+    private static final class GatedCleanable implements Cleanable {
         private final AtomicBoolean gate;
         private final Cleanable cleanable;
 

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

@@ -31,16 +31,23 @@ final class CompositeBuffer extends RcSupport<Buffer, CompositeBuffer> implement
      * non-composite copy of the buffer.
      */
     private static final int MAX_CAPACITY = Integer.MAX_VALUE - 8;
-    private static final Drop<CompositeBuffer> COMPOSITE_DROP = buf -> {
-        for (Buffer b : buf.bufs) {
-            b.close();
+    private static final Drop<CompositeBuffer> COMPOSITE_DROP = new Drop<CompositeBuffer>() {
+        @Override
+        public void drop(CompositeBuffer buf) {
+            for (Buffer b : buf.bufs) {
+                b.close();
+            }
+            buf.makeInaccessible();
+        }
+
+        @Override
+        public String toString() {
+            return "COMPOSITE_DROP";
         }
-        buf.makeInaccessible();
     };
 
     private final BufferAllocator allocator;
     private final TornBufferAccessors tornBufAccessors;
-    private final boolean isSendable;
     private Buffer[] bufs;
     private int[] offsets; // The offset, for the composite buffer, where each constituent buffer starts.
     private int capacity;
@@ -52,7 +59,7 @@ final class CompositeBuffer extends RcSupport<Buffer, CompositeBuffer> implement
     private boolean readOnly;
 
     CompositeBuffer(BufferAllocator allocator, Deref<Buffer>[] refs) {
-        this(allocator, true, filterExternalBufs(refs), COMPOSITE_DROP, false);
+        this(allocator, filterExternalBufs(refs), COMPOSITE_DROP, false);
     }
 
     private static Buffer[] filterExternalBufs(Deref<Buffer>[] refs) {
@@ -106,11 +113,10 @@ private static Stream<Buffer> flattenBuffer(Buffer buf) {
         return Stream.of(buf);
     }
 
-    private CompositeBuffer(BufferAllocator allocator, boolean isSendable, Buffer[] bufs, Drop<CompositeBuffer> drop,
+    private CompositeBuffer(BufferAllocator allocator, Buffer[] bufs, Drop<CompositeBuffer> drop,
                             boolean acquireBufs) {
         super(drop);
         this.allocator = allocator;
-        this.isSendable = isSendable;
         if (acquireBufs) {
             for (Buffer buf : bufs) {
                 buf.acquire();
@@ -297,46 +303,31 @@ public Buffer slice(int offset, int length) {
                     offset + ", and length was " + length + '.');
         }
         Buffer choice = (Buffer) chooseBuffer(offset, 0);
-        Buffer[] slices = null;
-        acquire(); // Increase reference count of the original composite buffer.
-        Drop<CompositeBuffer> drop = obj -> {
-            close(); // Decrement the reference count of the original composite buffer.
-            COMPOSITE_DROP.drop(obj);
-        };
-
-        try {
-            if (length > 0) {
-                slices = new Buffer[bufs.length];
-                int off = subOffset;
-                int cap = length;
-                int i;
-                for (i = searchOffsets(offset); cap > 0; i++) {
-                    var buf = bufs[i];
-                    int avail = buf.capacity() - off;
-                    slices[i] = buf.slice(off, Math.min(cap, avail));
-                    cap -= avail;
-                    off = 0;
-                }
-                slices = Arrays.copyOf(slices, i);
-            } else {
-                // Specialize for length == 0, since we must slice from at least one constituent buffer.
-                slices = new Buffer[] { choice.slice(subOffset, 0) };
-            }
-
-            return new CompositeBuffer(allocator, false, slices, drop, true);
-        } 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();
-            throw throwable;
-        } finally {
-            if (slices != null) {
-                for (Buffer slice : slices) {
-                    if (slice != null) {
-                        slice.close(); // Ownership now transfers to the composite buffer.
-                    }
-                }
-            }
+        Buffer[] slices;
+
+        if (length > 0) {
+            slices = new Buffer[bufs.length];
+            int off = subOffset;
+            int cap = length;
+            int i;
+            for (i = searchOffsets(offset); cap > 0; i++) {
+                var buf = bufs[i];
+                int avail = buf.capacity() - off;
+                slices[i] = buf.slice(off, Math.min(cap, avail));
+                cap -= avail;
+                off = 0;
+            }
+            slices = Arrays.copyOf(slices, i);
+        } else {
+            // Specialize for length == 0, since we must slice from at least one constituent buffer.
+            slices = new Buffer[] { choice.slice(subOffset, 0) };
         }
+
+        // Use the constructor that skips filtering out empty buffers, and skips acquiring on the buffers.
+        // This is important because 1) slice() already acquired the buffers, and 2) if this slice is empty
+        // then we need to keep holding on to it to prevent this originating composite buffer from getting
+        // ownership. If it did, its behaviour would be inconsistent with that of a non-composite buffer.
+        return new CompositeBuffer(allocator, slices, COMPOSITE_DROP, false);
     }
 
     @Override
@@ -749,7 +740,7 @@ public Buffer bifurcate() {
         }
         if (bufs.length == 0) {
             // Bifurcating a zero-length buffer is trivial.
-            return new CompositeBuffer(allocator, true, bufs, unsafeGetDrop(), true).order(order);
+            return new CompositeBuffer(allocator, bufs, unsafeGetDrop(), true).order(order);
         }
 
         int i = searchOffsets(woff);
@@ -761,7 +752,7 @@ public Buffer bifurcate() {
         }
         computeBufferOffsets();
         try {
-            var compositeBuf = new CompositeBuffer(allocator, true, bifs, unsafeGetDrop(), true);
+            var compositeBuf = new CompositeBuffer(allocator, bifs, unsafeGetDrop(), true);
             compositeBuf.order = order; // Preserve byte order even if bifs array is empty.
             return compositeBuf;
         } finally {
@@ -1164,7 +1155,7 @@ public CompositeBuffer transferOwnership(Drop<CompositeBuffer> drop) {
                 for (int i = 0; i < sends.length; i++) {
                     received[i] = sends[i].receive();
                 }
-                var composite = new CompositeBuffer(allocator, true, received, drop, true);
+                var composite = new CompositeBuffer(allocator, received, drop, true);
                 composite.readOnly = readOnly;
                 drop.attach(composite);
                 return composite;
@@ -1179,18 +1170,9 @@ void makeInaccessible() {
         closed = true;
     }
 
-    @Override
-    protected IllegalStateException notSendableException() {
-        if (!isSendable) {
-            return new IllegalStateException(
-                    "Cannot send() this buffer. This buffer might be a slice of another buffer.");
-        }
-        return super.notSendableException();
-    }
-
     @Override
     public boolean isOwned() {
-        return isSendable && super.isOwned() && allConstituentsAreOwned();
+        return super.isOwned() && allConstituentsAreOwned();
     }
 
     private boolean allConstituentsAreOwned() {

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

@@ -34,5 +34,6 @@ static MemoryManager getNativeMemoryManager() {
     Buffer allocateShared(AllocatorControl allo, 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);
 }

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

@@ -37,7 +37,7 @@ protected RcSupport(Drop<T> drop) {
     @Override
     public final I acquire() {
         if (acquires < 0) {
-            throw attachTrace(new IllegalStateException("Resource is closed."));
+            throw attachTrace(new IllegalStateException("This resource is closed: " + this + '.'));
         }
         if (acquires == Integer.MAX_VALUE) {
             throw new IllegalStateException("Cannot acquire more references; counter would overflow.");
@@ -98,7 +98,7 @@ protected <E extends Throwable> E attachTrace(E throwable) {
      */
     protected IllegalStateException notSendableException() {
         return new IllegalStateException(
-                "Cannot send() a reference counted object with " + acquires + " outstanding acquires: " + this + '.');
+                "Cannot send() a reference counted object with " + countBorrows() + " borrows: " + this + '.');
     }
 
     @Override

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

@@ -28,7 +28,7 @@ class SizeClassedMemoryPool implements BufferAllocator, AllocatorControl, Drop<B
     private static final VarHandle CLOSE = Statics.findVarHandle(
             lookup(), SizeClassedMemoryPool.class, "closed", boolean.class);
     private final MemoryManager manager;
-    private final ConcurrentHashMap<Integer, ConcurrentLinkedQueue<Send<Buffer>>> pool;
+    private final ConcurrentHashMap<Integer, ConcurrentLinkedQueue<Object>> pool;
     @SuppressWarnings("unused")
     private volatile boolean closed;
 
@@ -41,9 +41,9 @@ protected SizeClassedMemoryPool(MemoryManager manager) {
     public Buffer allocate(int size) {
         BufferAllocator.checkSize(size);
         var sizeClassPool = getSizeClassPool(size);
-        Send<Buffer> send = sizeClassPool.poll();
-        if (send != null) {
-            return send.receive()
+        Object memory = sizeClassPool.poll();
+        if (memory != null) {
+            return recoverMemoryIntoBuffer(memory)
                        .reset()
                        .readOnly(false)
                        .fill((byte) 0)
@@ -71,10 +71,10 @@ public void close() {
         if (CLOSE.compareAndSet(this, false, true)) {
             var capturedExceptions = new ArrayList<Exception>(4);
             pool.forEach((k, v) -> {
-                Send<Buffer> send;
-                while ((send = v.poll()) != null) {
+                Object memory;
+                while ((memory = v.poll()) != null) {
                     try {
-                        send.receive().close();
+                        dispose(recoverMemoryIntoBuffer(memory));
                     } catch (Exception e) {
                         capturedExceptions.add(e);
                     }
@@ -94,40 +94,47 @@ public void drop(Buffer buf) {
             dispose(buf);
             return;
         }
-        var sizeClassPool = getSizeClassPool(buf.capacity());
-        sizeClassPool.offer(buf.send());
+        Object mem = manager.unwrapRecoverableMemory(buf);
+        var sizeClassPool = getSizeClassPool(manager.capacityOfRecoverableMemory(mem));
+        sizeClassPool.offer(mem);
         if (closed) {
-            Send<Buffer> send;
-            while ((send = sizeClassPool.poll()) != null) {
-                send.receive().close();
+            Object memory;
+            while ((memory = sizeClassPool.poll()) != null) {
+                dispose(recoverMemoryIntoBuffer(memory));
             }
         }
     }
 
+    @Override
+    public String toString() {
+        return "SizeClassedMemoryPool";
+    }
+
     @Override
     public Object allocateUntethered(Buffer originator, int size) {
         var sizeClassPool = getSizeClassPool(size);
-        Send<Buffer> send = sizeClassPool.poll();
-        Buffer untetheredBuf;
-        if (send != null) {
-            var transfer = (TransferSend<Buffer, Buffer>) send;
-            var owned = transfer.unsafeUnwrapOwned();
-            untetheredBuf = owned.transferOwnership(NO_OP_DROP);
-        } else {
-            untetheredBuf = createBuf(size, NO_OP_DROP);
+        Object memory = sizeClassPool.poll();
+        if (memory == null) {
+            Buffer untetheredBuf = createBuf(size, NO_OP_DROP);
+            memory = manager.unwrapRecoverableMemory(untetheredBuf);
         }
-        return manager.unwrapRecoverableMemory(untetheredBuf);
+        return memory;
     }
 
     @Override
     public void recoverMemory(Object memory) {
+        Buffer buf = recoverMemoryIntoBuffer(memory);
+        buf.close();
+    }
+
+    private Buffer recoverMemoryIntoBuffer(Object memory) {
         var drop = getDrop();
         var buf = manager.recoverMemory(memory, drop);
         drop.attach(buf);
-        buf.close();
+        return buf;
     }
 
-    private ConcurrentLinkedQueue<Send<Buffer>> getSizeClassPool(int size) {
+    private ConcurrentLinkedQueue<Object> getSizeClassPool(int size) {
         return pool.computeIfAbsent(size, k -> new ConcurrentLinkedQueue<>());
     }
 

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

@@ -21,7 +21,15 @@
 
 interface Statics {
     Cleaner CLEANER = Cleaner.create();
-    Drop<Buffer> NO_OP_DROP = buf -> {
+    Drop<Buffer> NO_OP_DROP = new Drop<Buffer>() {
+        @Override
+        public void drop(Buffer obj) {
+        }
+
+        @Override
+        public String toString() {
+            return "NO_OP_DROP";
+        }
     };
 
     static VarHandle findVarHandle(Lookup lookup, Class<?> recv, String name, Class<?> type) {