Revert "Remove object recycling from these classes"

This reverts commit 3ba07d8

transport-raknet/src/main/java/org/cloudburstmc/netty/channel/raknet/packet/EncapsulatedPacket.java

@@ -20,12 +20,16 @@
 import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.CompositeByteBuf;
 import io.netty.util.AbstractReferenceCounted;
+import io.netty.util.internal.ObjectPool;
 import io.netty.util.ReferenceCountUtil;
 import org.cloudburstmc.netty.channel.raknet.RakConstants;
 import org.cloudburstmc.netty.channel.raknet.RakReliability;
 
 public class EncapsulatedPacket extends AbstractReferenceCounted {
 
+    private static final ObjectPool<EncapsulatedPacket> RECYCLER = ObjectPool.newPool(EncapsulatedPacket::new);
+
+    private final ObjectPool.Handle<EncapsulatedPacket> handle;
     private RakReliability reliability;
     private int reliabilityIndex;
     private int sequenceIndex;
@@ -38,6 +42,14 @@ public class EncapsulatedPacket extends AbstractReferenceCounted {
     private ByteBuf buffer;
     private boolean needsBAS;
 
+    public static EncapsulatedPacket newInstance() {
+        return RECYCLER.get();
+    }
+
+    private EncapsulatedPacket(ObjectPool.Handle<EncapsulatedPacket> handle) {
+        this.handle = handle;
+    }
+
     public void encode(CompositeByteBuf buffer) {
         RakReliability reliability = this.reliability;
         ByteBuf header = buffer.alloc().ioBuffer(3 + reliability.getSize() + (this.split ? 10 : 0));
@@ -111,7 +123,7 @@ public int getSize() {
     }
 
     public EncapsulatedPacket fromSplit(ByteBuf reassembled) {
-        EncapsulatedPacket packet = new EncapsulatedPacket();
+        EncapsulatedPacket packet = newInstance();
         packet.reliability = this.reliability;
         packet.reliabilityIndex = this.reliabilityIndex;
         packet.sequenceIndex = this.sequenceIndex;
@@ -135,6 +147,8 @@ protected void deallocate() {
         this.partId = 0;
         this.partIndex = 0;
         this.buffer = null;
+        setRefCnt(1);
+        this.handle.recycle(this);
     }
 
     @Override

transport-raknet/src/main/java/org/cloudburstmc/netty/channel/raknet/packet/RakDatagramPacket.java

@@ -17,6 +17,7 @@
 package org.cloudburstmc.netty.channel.raknet.packet;
 
 import io.netty.util.AbstractReferenceCounted;
+import io.netty.util.internal.ObjectPool;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -25,12 +26,23 @@
 
 public class RakDatagramPacket extends AbstractReferenceCounted {
 
+    private static final ObjectPool<RakDatagramPacket> RECYCLER = ObjectPool.newPool(RakDatagramPacket::new);
+
+    private final ObjectPool.Handle<RakDatagramPacket> handle;
     private final List<EncapsulatedPacket> packets = new ArrayList<>();
     private byte flags = FLAG_VALID | FLAG_NEEDS_B_AND_AS;
     private long sendTime;
     private long nextSend;
     private int sequenceIndex = -1;
 
+    public static RakDatagramPacket newInstance() {
+        return RECYCLER.get();
+    }
+
+    private RakDatagramPacket(ObjectPool.Handle<RakDatagramPacket> handle) {
+        this.handle = handle;
+    }
+
     @Override
     public RakDatagramPacket retain() {
         super.retain();
@@ -63,6 +75,11 @@ public boolean tryAddPacket(EncapsulatedPacket packet, int mtu) {
         return true;
     }
 
+    @Override
+    public boolean release() {
+        return super.release();
+    }
+
     @Override
     protected void deallocate() {
         for (EncapsulatedPacket packet : this.packets) {
@@ -73,6 +90,8 @@ protected void deallocate() {
         this.sendTime = 0;
         this.nextSend = 0;
         this.sequenceIndex = -1;
+        setRefCnt(1);
+        this.handle.recycle(this);
     }
 
     public int getSize() {
@@ -122,7 +141,8 @@ public void setSequenceIndex(int sequenceIndex) {
     @Override
     public String toString() {
         return "RakDatagramPacket{" +
-                "packets=" + packets +
+                "handle=" + handle +
+                ", packets=" + packets +
                 ", flags=" + flags +
                 ", sendTime=" + sendTime +
                 ", nextSend=" + nextSend +

transport-raknet/src/main/java/org/cloudburstmc/netty/handler/codec/raknet/common/RakDatagramCodec.java

@@ -68,12 +68,12 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> li
             return;
         }
 
-        RakDatagramPacket packet = new RakDatagramPacket();
+        RakDatagramPacket packet = RakDatagramPacket.newInstance();
         try {
             packet.setFlags(buffer.readByte());
             packet.setSequenceIndex(buffer.readUnsignedMediumLE());
             while (buffer.isReadable()) {
-                EncapsulatedPacket encapsulated = new EncapsulatedPacket();
+                EncapsulatedPacket encapsulated = EncapsulatedPacket.newInstance();
                 try {
                     encapsulated.decode(buffer);
                     packet.getPackets().add(encapsulated.retain());

transport-raknet/src/main/java/org/cloudburstmc/netty/handler/codec/raknet/common/RakSessionCodec.java

@@ -156,6 +156,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                 while ((packet = orderingHeap.poll()) != null) {
                     packet.release();
                 }
+                orderingHeap.release();
             }
         }
 
@@ -166,6 +167,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
             while ((packet = outgoingPackets.poll()) != null) {
                 packet.release();
             }
+            outgoingPackets.release();
         }
 
         if (log.isTraceEnabled()) {
@@ -175,7 +177,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 
     private void initHeapWeights() {
         for (int priorityLevel = 0; priorityLevel < 4; priorityLevel++) {
-            this.outgoingPacketNextWeights[priorityLevel] = (1L << priorityLevel) * priorityLevel + priorityLevel;
+            this.outgoingPacketNextWeights[priorityLevel] = (1 << priorityLevel) * priorityLevel + priorityLevel;
         }
     }
 
@@ -563,7 +565,7 @@ private void sendDatagrams(ChannelHandlerContext ctx, long curTime, int mtuSize)
         }
 
         int transmissionBandwidth = this.slidingWindow.getTransmissionBandwidth();
-        RakDatagramPacket datagram = new RakDatagramPacket();
+        RakDatagramPacket datagram = RakDatagramPacket.newInstance();
         datagram.setSendTime(curTime);
         EncapsulatedPacket packet;
 
@@ -580,7 +582,7 @@ private void sendDatagrams(ChannelHandlerContext ctx, long curTime, int mtuSize)
             if (!datagram.tryAddPacket(packet, mtuSize)) {
                 this.sendDatagram(ctx, datagram, curTime);
 
-                datagram = new RakDatagramPacket();
+                datagram = RakDatagramPacket.newInstance();
                 datagram.setSendTime(curTime);
                 if (!datagram.tryAddPacket(packet, mtuSize)) {
                     throw new IllegalArgumentException("Packet too large to fit in MTU (size: " + packet.getSize() + ", MTU: " + mtuSize + ")");
@@ -596,7 +598,7 @@ private void sendDatagrams(ChannelHandlerContext ctx, long curTime, int mtuSize)
     private void sendImmediate(ChannelHandlerContext ctx, EncapsulatedPacket[] packets) {
         long curTime = System.currentTimeMillis();
         for (EncapsulatedPacket packet : packets) {
-            RakDatagramPacket datagram = new RakDatagramPacket();
+            RakDatagramPacket datagram = RakDatagramPacket.newInstance();
             datagram.setSendTime(curTime);
             if (!datagram.tryAddPacket(packet, this.getMtu())) {
                 throw new IllegalArgumentException("Packet too large to fit in MTU (size: " + packet.getSize() + ", MTU: " + this.getMtu() + ")");
@@ -697,7 +699,7 @@ private EncapsulatedPacket[] createEncapsulated(RakMessage rakMessage) {
         // Now create the packets.
         EncapsulatedPacket[] packets = new EncapsulatedPacket[buffers.length];
         for (int i = 0, parts = buffers.length; i < parts; i++) {
-            EncapsulatedPacket packet = new EncapsulatedPacket();
+            EncapsulatedPacket packet = EncapsulatedPacket.newInstance();
             packet.setBuffer(buffers[i]);
             packet.setNeedsBAS(true);
             packet.setOrderingChannel((short) orderingChannel);

transport-raknet/src/main/java/org/cloudburstmc/netty/util/FastBinaryMinHeap.java

@@ -16,14 +16,18 @@
 
 package org.cloudburstmc.netty.util;
 
+import io.netty.util.AbstractReferenceCounted;
+import io.netty.util.internal.ObjectPool;
+
 import java.util.Arrays;
 import java.util.NoSuchElementException;
 import java.util.Objects;
 
-public class FastBinaryMinHeap<E> {
+public class FastBinaryMinHeap<E> extends AbstractReferenceCounted {
 
-    private static final Entry INFIMUM = new Entry(null, Long.MAX_VALUE);
-    private static final Entry SUPREMUM = new Entry(null, Long.MIN_VALUE);
+    private static final Entry INFIMUM = new Entry(Long.MAX_VALUE);
+    private static final Entry SUPREMUM = new Entry(Long.MIN_VALUE);
+    private static final ObjectPool<Entry> RECYCLER = ObjectPool.newPool(Entry::new);
     private int size;
 
     public FastBinaryMinHeap() {
@@ -38,6 +42,14 @@ public FastBinaryMinHeap(int initialCapacity) {
         this.heap[0] = SUPREMUM;
     }
 
+    private static Entry newEntry(Object element, long weight) {
+        Entry entry = RECYCLER.get();
+        entry.element = element;
+        entry.weight = weight;
+
+        return entry;
+    }
+
     private void resize(int capacity) {
         int adjustedSize = this.size + 1;
         int copyLength = Math.min(this.heap.length, adjustedSize);
@@ -80,7 +92,7 @@ private void insert0(long weight, E element) {
             predWeight = this.heap[pred].weight;
         }
 
-        this.heap[hole] = new Entry(element, weight);
+        this.heap[hole] = newEntry(element, weight);
     }
 
     public void insertSeries(long weight, E[] elements) {
@@ -106,7 +118,7 @@ public void insertSeries(long weight, E[] elements) {
             for (E element : elements) {
                 Objects.requireNonNull(element, "element");
 
-                this.heap[++this.size] = new Entry(element, weight);
+                this.heap[++this.size] = newEntry(element, weight);
             }
         } else {
             for (E element : elements) {
@@ -134,6 +146,7 @@ public void remove() {
         if (this.size == 0) {
             throw new NoSuchElementException("Heap is empty");
         }
+        this.heap[1].release();
         int hole = 1;
         int succ = 2;
         int sz = this.size;
@@ -178,13 +191,46 @@ public boolean isEmpty() {
         return this.size == 0;
     }
 
-    private static class Entry {
-        private final Object element;
-        private final long weight;
+    @Override
+    protected void deallocate() {
+        while (this.size > 0) {
+            Entry entry = this.heap[1];
+            this.remove();
+            entry.release();
+        }
+    }
+
+    @Override
+    public FastBinaryMinHeap<E> touch(Object hint) {
+        return this;
+    }
+
+    private static class Entry extends AbstractReferenceCounted {
+        private final ObjectPool.Handle<Entry> handle;
+        private Object element;
+        private long weight;
 
-        private Entry(Object element, long weight) {
-            this.element = element;
+        private Entry(long weight) {
             this.weight = weight;
+            this.handle = null;
+        }
+
+        private Entry(ObjectPool.Handle<Entry> handle) {
+            this.handle = handle;
+        }
+
+        @Override
+        protected void deallocate() {
+            setRefCnt(1);
+            if (handle == null) return;
+            this.element = null;
+            this.weight = 0;
+            this.handle.recycle(this);
+        }
+
+        @Override
+        public Entry touch(Object hint) {
+            return this;
         }
     }
 }